Add isolated Leo learning lifecycle
This commit is contained in:
twentyOne2x
parent
1472c8513a
commit
8bb10e5c94
2 changed files with 1033 additions and 0 deletions
900
scripts/run_leo_integrated_learning_lifecycle.py
Normal file
900
scripts/run_leo_integrated_learning_lifecycle.py
Normal file
|
|
@ -0,0 +1,900 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Run a real source -> review -> apply -> recompile lifecycle in disposable Postgres.
|
||||
|
||||
The harness composes the repository's existing source compiler, normalized
|
||||
proposal staging, review packet, guarded approval, guarded apply, and replay
|
||||
receipt surfaces. It starts its own networkless tmpfs Postgres container and
|
||||
never accepts an existing database target, so the explicit review/apply actions
|
||||
cannot reach VPS, GCP, or another persistent database.
|
||||
|
||||
The source packet must contain:
|
||||
|
||||
* ``prepared/extracted-text.txt``
|
||||
* ``prepared/source-manifest.json``
|
||||
* ``proposal-packet.json``
|
||||
|
||||
The extracted text is used as both the hash-bound source artifact and its strict
|
||||
UTF-8 extraction. ``proposal-packet.json`` is an independently retained
|
||||
compiler expectation: the harness refuses to continue unless a fresh compile
|
||||
matches it byte-for-byte under canonical JSON.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
REPO_ROOT = HERE.parent
|
||||
sys.path.insert(0, str(HERE))
|
||||
|
||||
import apply_proposal as ap # noqa: E402
|
||||
import compile_kb_source_packet as compiler # noqa: E402
|
||||
import kb_proposal_review_packet as review_packets # noqa: E402
|
||||
import proposal_apply_lifecycle as apply_lifecycle # noqa: E402
|
||||
import stage_normalized_proposal as stage # noqa: E402
|
||||
|
||||
RECEIPT_SCHEMA = "livingip.leoIntegratedLearningLifecycleReceipt.v1"
|
||||
CONTAINER_LABEL = "livingip.canary=leo-integrated-learning-lifecycle"
|
||||
DATABASE = "teleo"
|
||||
REVIEWER_HANDLE = "m3ta"
|
||||
REVIEWER_ID = "11111111-1111-4111-8111-111111111111"
|
||||
DEFAULT_REVIEW_NOTE = (
|
||||
"Approved the exact hash-bound candidate rows for this disposable integrated-learning lifecycle only."
|
||||
)
|
||||
POSTGRES_IMAGE = "postgres:16-alpine"
|
||||
|
||||
|
||||
class LifecycleError(RuntimeError):
|
||||
"""Raised when the isolated lifecycle cannot prove an exact transition."""
|
||||
|
||||
|
||||
def canonical_json_bytes(value: Any) -> bytes:
|
||||
return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8")
|
||||
|
||||
|
||||
def canonical_sha256(value: Any) -> str:
|
||||
return hashlib.sha256(canonical_json_bytes(value)).hexdigest()
|
||||
|
||||
|
||||
def sha256_text(value: str) -> str:
|
||||
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as handle:
|
||||
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def parse_last_json_line(output: str) -> Any:
|
||||
for line in reversed(output.splitlines()):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
return json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def _safe_error_text(value: str, *, limit: int = 1200) -> str:
|
||||
compact = " ".join(value.split())
|
||||
return compact[-limit:]
|
||||
|
||||
|
||||
def command_receipt(result: subprocess.CompletedProcess[str], *, expected_marker: str | None = None) -> dict[str, Any]:
|
||||
parsed = parse_last_json_line(result.stdout)
|
||||
if isinstance(parsed, dict):
|
||||
parsed = {key: value for key, value in parsed.items() if key not in {"receipt_path"}}
|
||||
return {
|
||||
"returncode": result.returncode,
|
||||
"stdout_sha256": sha256_text(result.stdout),
|
||||
"stderr_sha256": sha256_text(result.stderr),
|
||||
"stdout_json": parsed if isinstance(parsed, dict) else None,
|
||||
"expected_marker": expected_marker,
|
||||
"expected_marker_present": expected_marker in result.stdout if expected_marker else None,
|
||||
}
|
||||
|
||||
|
||||
def run_tool(arguments: list[str], *, action: str) -> subprocess.CompletedProcess[str]:
|
||||
result = subprocess.run(arguments, text=True, capture_output=True, check=False)
|
||||
if result.returncode != 0:
|
||||
detail = _safe_error_text(result.stderr or result.stdout)
|
||||
raise LifecycleError(f"{action} failed ({result.returncode}): {detail}")
|
||||
return result
|
||||
|
||||
|
||||
class DisposablePostgres:
|
||||
def __init__(self, name: str, admin_password: str) -> None:
|
||||
self.name = name
|
||||
self.admin_password = admin_password
|
||||
self.docker = shutil.which("docker")
|
||||
if not self.docker:
|
||||
raise LifecycleError("docker executable is unavailable")
|
||||
self.started = False
|
||||
self.environment: dict[str, Any] = {}
|
||||
|
||||
def _run(
|
||||
self,
|
||||
arguments: list[str],
|
||||
*,
|
||||
input_text: str | None = None,
|
||||
check: bool = True,
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
result = subprocess.run(
|
||||
[self.docker, *arguments],
|
||||
input=input_text,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
if check and result.returncode != 0:
|
||||
detail = _safe_error_text(result.stderr or result.stdout)
|
||||
raise LifecycleError(f"docker {arguments[0]} failed ({result.returncode}): {detail}")
|
||||
return result
|
||||
|
||||
def start(self) -> dict[str, Any]:
|
||||
result = self._run(
|
||||
[
|
||||
"run",
|
||||
"--rm",
|
||||
"-d",
|
||||
"--name",
|
||||
self.name,
|
||||
"--network",
|
||||
"none",
|
||||
"--label",
|
||||
CONTAINER_LABEL,
|
||||
"--label",
|
||||
f"livingip.canary.instance={self.name}",
|
||||
"--tmpfs",
|
||||
"/var/lib/postgresql/data:rw,nosuid,size=384m",
|
||||
"-e",
|
||||
f"POSTGRES_PASSWORD={self.admin_password}",
|
||||
"-e",
|
||||
f"POSTGRES_DB={DATABASE}",
|
||||
POSTGRES_IMAGE,
|
||||
]
|
||||
)
|
||||
self.started = True
|
||||
deadline = time.monotonic() + 45
|
||||
ready_streak = 0
|
||||
while time.monotonic() < deadline:
|
||||
ready = self._run(
|
||||
["exec", self.name, "pg_isready", "-U", "postgres", "-d", DATABASE],
|
||||
check=False,
|
||||
)
|
||||
if ready.returncode == 0:
|
||||
ready_streak += 1
|
||||
if ready_streak >= 2:
|
||||
break
|
||||
else:
|
||||
ready_streak = 0
|
||||
time.sleep(0.25)
|
||||
else:
|
||||
raise LifecycleError("disposable Postgres did not become stably ready")
|
||||
|
||||
inspect = json.loads(self._run(["inspect", self.name]).stdout)[0]
|
||||
mounts = [
|
||||
{
|
||||
"type": mount.get("Type"),
|
||||
"name": mount.get("Name"),
|
||||
"destination": mount.get("Destination"),
|
||||
}
|
||||
for mount in inspect.get("Mounts", [])
|
||||
]
|
||||
labels = inspect.get("Config", {}).get("Labels", {})
|
||||
self.environment = {
|
||||
"container_id": result.stdout.strip(),
|
||||
"container_name": self.name,
|
||||
"image": inspect.get("Config", {}).get("Image"),
|
||||
"network_mode": inspect.get("HostConfig", {}).get("NetworkMode"),
|
||||
"tmpfs_data_dir": "/var/lib/postgresql/data" in inspect.get("HostConfig", {}).get("Tmpfs", {}),
|
||||
"mounts": mounts,
|
||||
"canary_label": labels.get("livingip.canary"),
|
||||
"instance_label": labels.get("livingip.canary.instance"),
|
||||
}
|
||||
if self.environment["network_mode"] != "none":
|
||||
raise LifecycleError("disposable Postgres is not networkless")
|
||||
if not self.environment["tmpfs_data_dir"] or mounts:
|
||||
raise LifecycleError("disposable Postgres must use tmpfs without Docker volumes")
|
||||
if self.environment["canary_label"] != "leo-integrated-learning-lifecycle":
|
||||
raise LifecycleError("disposable Postgres canary label is missing")
|
||||
if self.environment["instance_label"] != self.name:
|
||||
raise LifecycleError("disposable Postgres instance label is missing")
|
||||
return self.environment
|
||||
|
||||
def psql(self, sql: str) -> str:
|
||||
result = self._run(
|
||||
[
|
||||
"exec",
|
||||
"-i",
|
||||
self.name,
|
||||
"psql",
|
||||
"-X",
|
||||
"-U",
|
||||
"postgres",
|
||||
"-d",
|
||||
DATABASE,
|
||||
"-v",
|
||||
"ON_ERROR_STOP=1",
|
||||
"-Atq",
|
||||
],
|
||||
input_text=sql,
|
||||
)
|
||||
return result.stdout.strip()
|
||||
|
||||
def psql_json(self, sql: str) -> Any:
|
||||
output = self.psql(sql)
|
||||
parsed = parse_last_json_line(output)
|
||||
if parsed is None:
|
||||
raise LifecycleError("Postgres readback did not contain JSON")
|
||||
return parsed
|
||||
|
||||
def cleanup(self) -> dict[str, Any]:
|
||||
remove = self._run(["rm", "-f", self.name], check=False) if self.started else None
|
||||
inspect = self._run(["inspect", self.name], check=False)
|
||||
names = self._run(
|
||||
["container", "ls", "-a", "--filter", f"name=^/{self.name}$", "--format", "{{.Names}}"],
|
||||
check=False,
|
||||
)
|
||||
labels = self._run(
|
||||
[
|
||||
"container",
|
||||
"ls",
|
||||
"-a",
|
||||
"--filter",
|
||||
"label=livingip.canary=leo-integrated-learning-lifecycle",
|
||||
"--filter",
|
||||
f"label=livingip.canary.instance={self.name}",
|
||||
"--format",
|
||||
"{{.Names}}",
|
||||
],
|
||||
check=False,
|
||||
)
|
||||
return {
|
||||
"remove_attempted": self.started,
|
||||
"remove_returncode": remove.returncode if remove else None,
|
||||
"container_absent": inspect.returncode != 0,
|
||||
"name_readback_clear": names.returncode == 0 and not names.stdout.strip(),
|
||||
"instance_label_readback_clear": labels.returncode == 0 and not labels.stdout.strip(),
|
||||
"network_mode_was_none": self.environment.get("network_mode") == "none",
|
||||
"tmpfs_data_dir_was_present": self.environment.get("tmpfs_data_dir") is True,
|
||||
"docker_volume_mounts_observed": self.environment.get("mounts", []),
|
||||
}
|
||||
|
||||
|
||||
def bootstrap_sql(review_password: str, apply_password: str) -> str:
|
||||
return f"""\\set ON_ERROR_STOP on
|
||||
create role kb_apply login noinherit password {ap.sql_literal(apply_password)};
|
||||
create role kb_review login noinherit password {ap.sql_literal(review_password)};
|
||||
create schema kb_stage;
|
||||
create type evidence_role as enum ('grounds', 'illustrates', 'contradicts');
|
||||
create type edge_type as enum (
|
||||
'supports', 'challenges', 'requires', 'relates', 'contradicts', 'supersedes',
|
||||
'derives_from', 'cites', 'causes', 'constrains', 'accelerates'
|
||||
);
|
||||
|
||||
create table public.agents (
|
||||
id uuid primary key,
|
||||
handle text not null unique,
|
||||
kind text not null
|
||||
);
|
||||
create table public.claims (
|
||||
id uuid primary key,
|
||||
type text not null,
|
||||
text text not null,
|
||||
status text not null,
|
||||
confidence numeric,
|
||||
tags text[] not null default '{{}}',
|
||||
created_by uuid references public.agents(id),
|
||||
superseded_by uuid references public.claims(id),
|
||||
created_at timestamptz not null default now(),
|
||||
updated_at timestamptz not null default now()
|
||||
);
|
||||
create table public.sources (
|
||||
id uuid primary key,
|
||||
source_type text not null,
|
||||
url text,
|
||||
storage_path text,
|
||||
excerpt text,
|
||||
hash text not null,
|
||||
created_by uuid references public.agents(id),
|
||||
captured_at timestamptz,
|
||||
created_at timestamptz not null default now()
|
||||
);
|
||||
create table public.claim_evidence (
|
||||
id uuid primary key default gen_random_uuid(),
|
||||
claim_id uuid not null references public.claims(id),
|
||||
source_id uuid not null references public.sources(id),
|
||||
role evidence_role not null,
|
||||
weight numeric,
|
||||
created_by uuid references public.agents(id),
|
||||
created_at timestamptz not null default now()
|
||||
);
|
||||
create table public.claim_edges (
|
||||
id uuid primary key,
|
||||
from_claim uuid not null references public.claims(id),
|
||||
to_claim uuid not null references public.claims(id),
|
||||
edge_type edge_type not null,
|
||||
weight numeric,
|
||||
created_by uuid references public.agents(id),
|
||||
created_at timestamptz not null default now()
|
||||
);
|
||||
create table public.reasoning_tools (
|
||||
id uuid primary key,
|
||||
agent_id uuid references public.agents(id),
|
||||
name text not null,
|
||||
description text not null,
|
||||
category text,
|
||||
created_at timestamptz not null default now(),
|
||||
updated_at timestamptz not null default now()
|
||||
);
|
||||
create table public.strategies (
|
||||
id uuid primary key default gen_random_uuid(),
|
||||
agent_id uuid not null references public.agents(id),
|
||||
diagnosis text,
|
||||
guiding_policy text,
|
||||
proximate_objectives jsonb,
|
||||
version integer not null default 1,
|
||||
active boolean not null default true,
|
||||
created_at timestamptz not null default now(),
|
||||
unique (agent_id, version)
|
||||
);
|
||||
create table public.strategy_nodes (
|
||||
id uuid primary key default gen_random_uuid(),
|
||||
agent_id uuid references public.agents(id),
|
||||
node_type text,
|
||||
title text,
|
||||
body text,
|
||||
rank integer,
|
||||
status text not null default 'active',
|
||||
horizon text,
|
||||
measure text,
|
||||
source_ref text,
|
||||
metadata jsonb not null default '{{}}',
|
||||
created_at timestamptz not null default now(),
|
||||
updated_at timestamptz not null default now()
|
||||
);
|
||||
create table kb_stage.kb_proposals (
|
||||
id uuid primary key,
|
||||
proposal_type text not null,
|
||||
status text not null,
|
||||
proposed_by_handle text,
|
||||
proposed_by_agent_id uuid references public.agents(id),
|
||||
channel text,
|
||||
source_ref text,
|
||||
rationale text,
|
||||
payload jsonb not null,
|
||||
reviewed_by_handle text,
|
||||
reviewed_by_agent_id uuid references public.agents(id),
|
||||
reviewed_at timestamptz,
|
||||
review_note text,
|
||||
applied_by_handle text,
|
||||
applied_by_agent_id uuid references public.agents(id),
|
||||
applied_at timestamptz,
|
||||
created_at timestamptz not null default now(),
|
||||
updated_at timestamptz not null default now()
|
||||
);
|
||||
insert into public.agents (id, handle, kind)
|
||||
values ({ap.sql_literal(REVIEWER_ID)}::uuid, {ap.sql_literal(REVIEWER_HANDLE)}, 'human');
|
||||
"""
|
||||
|
||||
|
||||
def compile_source_packet(source_packet: Path) -> tuple[dict[str, Any], dict[str, Any], dict[str, Path]]:
|
||||
paths = {
|
||||
"artifact": source_packet / "prepared" / "extracted-text.txt",
|
||||
"text": source_packet / "prepared" / "extracted-text.txt",
|
||||
"manifest": source_packet / "prepared" / "source-manifest.json",
|
||||
"expected": source_packet / "proposal-packet.json",
|
||||
}
|
||||
missing = [str(path) for path in paths.values() if not path.is_file()]
|
||||
if missing:
|
||||
raise LifecycleError(f"source packet is missing required files: {missing}")
|
||||
expected = json.loads(paths["expected"].read_text(encoding="utf-8"))
|
||||
if not isinstance(expected, dict):
|
||||
raise LifecycleError("proposal-packet.json must contain one JSON object")
|
||||
compiled = compiler.compile_source_packet(paths["artifact"], paths["text"], paths["manifest"])
|
||||
if canonical_json_bytes(compiled) != canonical_json_bytes(expected):
|
||||
raise LifecycleError("fresh compiler output does not match retained proposal-packet.json")
|
||||
return compiled, expected, paths
|
||||
|
||||
|
||||
def proposal_readback(postgres: DisposablePostgres, proposal_id: str) -> dict[str, Any] | None:
|
||||
sql = f"""select jsonb_build_object(
|
||||
'id', id::text,
|
||||
'proposal_type', proposal_type,
|
||||
'status', status,
|
||||
'proposed_by_handle', proposed_by_handle,
|
||||
'proposed_by_agent_id', proposed_by_agent_id::text,
|
||||
'channel', channel,
|
||||
'source_ref', source_ref,
|
||||
'rationale', rationale,
|
||||
'payload', payload,
|
||||
'reviewed_by_handle', reviewed_by_handle,
|
||||
'reviewed_by_agent_id', reviewed_by_agent_id::text,
|
||||
'reviewed_at', reviewed_at::text,
|
||||
'review_note', review_note,
|
||||
'applied_by_handle', applied_by_handle,
|
||||
'applied_by_agent_id', applied_by_agent_id::text,
|
||||
'applied_at', applied_at::text,
|
||||
'created_at', created_at::text,
|
||||
'updated_at', updated_at::text
|
||||
)::text
|
||||
from kb_stage.kb_proposals
|
||||
where id = {ap.sql_literal(proposal_id)}::uuid;"""
|
||||
parsed = parse_last_json_line(postgres.psql(sql))
|
||||
if parsed is not None and not isinstance(parsed, dict):
|
||||
raise LifecycleError("proposal readback was not an object")
|
||||
return parsed
|
||||
|
||||
|
||||
def approval_readback(postgres: DisposablePostgres, proposal_id: str) -> dict[str, Any] | None:
|
||||
sql = f"""select jsonb_build_object(
|
||||
'proposal_id', proposal_id::text,
|
||||
'proposal_type', proposal_type,
|
||||
'payload', payload,
|
||||
'reviewed_by_handle', reviewed_by_handle,
|
||||
'reviewed_by_agent_id', reviewed_by_agent_id::text,
|
||||
'reviewed_by_db_role', reviewed_by_db_role::text,
|
||||
'reviewed_at', reviewed_at::text,
|
||||
'review_note', review_note
|
||||
)::text
|
||||
from kb_stage.kb_proposal_approvals
|
||||
where proposal_id = {ap.sql_literal(proposal_id)}::uuid;"""
|
||||
parsed = parse_last_json_line(postgres.psql(sql))
|
||||
if parsed is not None and not isinstance(parsed, dict):
|
||||
raise LifecycleError("approval readback was not an object")
|
||||
return parsed
|
||||
|
||||
|
||||
def probe_applied_proposal(child: dict[str, Any]) -> dict[str, Any]:
|
||||
return {
|
||||
**child,
|
||||
"status": "applied",
|
||||
"applied_by_handle": ap.SERVICE_AGENT_HANDLE,
|
||||
"applied_at": "2000-01-01T00:00:00+00:00",
|
||||
}
|
||||
|
||||
|
||||
def target_rows_readback(postgres: DisposablePostgres, child: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
|
||||
sql = ap.replay_receipt.build_postflight_sql(probe_applied_proposal(child))
|
||||
parsed = postgres.psql_json(sql)
|
||||
if not isinstance(parsed, dict) or any(not isinstance(rows, list) for rows in parsed.values()):
|
||||
raise LifecycleError("canonical target-row readback was malformed")
|
||||
return parsed
|
||||
|
||||
|
||||
def agents_readback(postgres: DisposablePostgres) -> list[dict[str, Any]]:
|
||||
parsed = postgres.psql_json(
|
||||
"select coalesce(jsonb_agg(to_jsonb(a) order by a.id::text), '[]'::jsonb)::text from public.agents a;"
|
||||
)
|
||||
if not isinstance(parsed, list):
|
||||
raise LifecycleError("agent readback was malformed")
|
||||
return parsed
|
||||
|
||||
|
||||
def database_state(postgres: DisposablePostgres, child: dict[str, Any]) -> dict[str, Any]:
|
||||
proposal_id = str(child["id"])
|
||||
canonical_rows = target_rows_readback(postgres, child)
|
||||
state = {
|
||||
"agents": agents_readback(postgres),
|
||||
"canonical_rows": canonical_rows,
|
||||
"proposal": proposal_readback(postgres, proposal_id),
|
||||
"immutable_approval": approval_readback(postgres, proposal_id),
|
||||
}
|
||||
return {
|
||||
"database_fingerprint": canonical_sha256(state),
|
||||
"canonical_fingerprint": canonical_sha256(canonical_rows),
|
||||
"state": state,
|
||||
}
|
||||
|
||||
|
||||
def exact_identities(
|
||||
child: dict[str, Any], canonical_rows: dict[str, list[dict[str, Any]]] | None = None
|
||||
) -> dict[str, Any]:
|
||||
apply_payload = child["payload"]["apply_payload"]
|
||||
evidence_rows = canonical_rows.get("claim_evidence", []) if canonical_rows else []
|
||||
return {
|
||||
"proposal_id": child["id"],
|
||||
"claim_ids": [row["id"] for row in apply_payload.get("claims", [])],
|
||||
"source_ids": [row["id"] for row in apply_payload.get("sources", [])],
|
||||
"planned_evidence_keys": [
|
||||
{
|
||||
"claim_id": row["claim_id"],
|
||||
"source_id": row["source_id"],
|
||||
"role": row.get("role") or "grounds",
|
||||
}
|
||||
for row in apply_payload.get("evidence", [])
|
||||
],
|
||||
"evidence_ids": [row["id"] for row in evidence_rows if row.get("id")],
|
||||
"edge_ids": [row["id"] for row in canonical_rows.get("claim_edges", [])] if canonical_rows else [],
|
||||
}
|
||||
|
||||
|
||||
def review_packet(proposal: dict[str, Any]) -> dict[str, Any]:
|
||||
packet = review_packets.classify_proposal(proposal)
|
||||
apply_payload = proposal["payload"]["apply_payload"]
|
||||
packet["candidate_rows"] = {
|
||||
family: apply_payload.get(family, [])
|
||||
for family in ("claims", "sources", "evidence", "edges", "reasoning_tools")
|
||||
}
|
||||
packet["packet_sha256"] = canonical_sha256(packet)
|
||||
return packet
|
||||
|
||||
|
||||
def write_secret_file(path: Path, key: str, value: str) -> None:
|
||||
path.write_text(f"{key}={value}\n", encoding="utf-8")
|
||||
path.chmod(0o600)
|
||||
|
||||
|
||||
def _tool_base(script: Path, proposal_id: str, secrets_file: Path, role: str, container: str) -> list[str]:
|
||||
return [
|
||||
sys.executable,
|
||||
str(script),
|
||||
proposal_id,
|
||||
"--secrets-file",
|
||||
str(secrets_file),
|
||||
"--container",
|
||||
container,
|
||||
"--db",
|
||||
DATABASE,
|
||||
"--host",
|
||||
"127.0.0.1",
|
||||
"--role",
|
||||
role,
|
||||
]
|
||||
|
||||
|
||||
def _all_target_rows_empty(rows: dict[str, list[dict[str, Any]]]) -> bool:
|
||||
return set(rows) == set(apply_lifecycle.ROW_TABLES) and all(not table_rows for table_rows in rows.values())
|
||||
|
||||
|
||||
def _row_counts(rows: dict[str, list[dict[str, Any]]]) -> dict[str, int]:
|
||||
return {family: len(table_rows) for family, table_rows in rows.items()}
|
||||
|
||||
|
||||
def normalized_canonical_rows(rows: dict[str, list[dict[str, Any]]]) -> dict[str, list[dict[str, Any]]]:
|
||||
return {family: sorted(table_rows, key=canonical_json_bytes) for family, table_rows in sorted(rows.items())}
|
||||
|
||||
|
||||
def run_lifecycle(
|
||||
source_packet: Path,
|
||||
output: Path,
|
||||
*,
|
||||
review_action: str,
|
||||
reviewed_by: str = REVIEWER_HANDLE,
|
||||
review_note: str = DEFAULT_REVIEW_NOTE,
|
||||
) -> dict[str, Any]:
|
||||
if review_action != "approve":
|
||||
raise LifecycleError("the only supported explicit review action is 'approve'")
|
||||
if reviewed_by != REVIEWER_HANDLE:
|
||||
raise LifecycleError(f"disposable reviewer must be {REVIEWER_HANDLE!r}")
|
||||
if not review_note.strip():
|
||||
raise LifecycleError("review note must be non-empty")
|
||||
|
||||
source_packet = source_packet.resolve()
|
||||
output = output.resolve()
|
||||
temp_root = Path(tempfile.mkdtemp(prefix="leo-integrated-learning-"))
|
||||
private_apply_receipt_path = temp_root / "apply-receipt.json"
|
||||
container_name = f"leo-integrated-learning-{os.getpid()}-{uuid.uuid4().hex[:10]}"
|
||||
postgres = DisposablePostgres(container_name, secrets.token_urlsafe(32))
|
||||
receipt: dict[str, Any] = {
|
||||
"schema": RECEIPT_SCHEMA,
|
||||
"status": "fail",
|
||||
"required_tier": "realistic_isolated_local_container_end_to_end",
|
||||
"current_tier": "not_proven",
|
||||
"source_packet": str(source_packet),
|
||||
"review_action": review_action,
|
||||
"safety_boundary": {
|
||||
"self_created_container_only": True,
|
||||
"existing_database_target_argument_supported": False,
|
||||
"vps_or_gcp_target_supported": False,
|
||||
"telegram_send_performed": False,
|
||||
},
|
||||
"checks": {},
|
||||
}
|
||||
|
||||
try:
|
||||
compiled, expected, paths = compile_source_packet(source_packet)
|
||||
child = compiled["strict_child_proposal"]
|
||||
parent = compiled["parent_proposal"]
|
||||
proposal_id = str(child["id"])
|
||||
apply_payload = child["payload"]["apply_payload"]
|
||||
manifest = json.loads(paths["manifest"].read_text(encoding="utf-8"))
|
||||
receipt["source"] = {
|
||||
"source_identity": manifest["source"]["identity"],
|
||||
"source_locator": manifest["source"]["locator"],
|
||||
"artifact_sha256": sha256_file(paths["artifact"]),
|
||||
"manifest_canonical_sha256": canonical_sha256(manifest),
|
||||
"retained_proposal_packet_canonical_sha256": canonical_sha256(expected),
|
||||
}
|
||||
receipt["compiled"] = {
|
||||
"status": compiled["status"],
|
||||
"bundle_sha256": canonical_sha256(compiled),
|
||||
"parent_proposal_id": parent["id"],
|
||||
"strict_proposal_id": proposal_id,
|
||||
"candidate_counts": {
|
||||
family: len(apply_payload.get(family, []))
|
||||
for family in ("claims", "sources", "evidence", "edges", "reasoning_tools")
|
||||
},
|
||||
}
|
||||
compiled_identities = exact_identities(child)
|
||||
receipt["identities"] = compiled_identities
|
||||
|
||||
environment = postgres.start()
|
||||
receipt["environment"] = environment
|
||||
review_password = secrets.token_urlsafe(32)
|
||||
apply_password = secrets.token_urlsafe(32)
|
||||
review_secrets = temp_root / "kb-review.env"
|
||||
apply_secrets = temp_root / "kb-apply.env"
|
||||
preflight_path = temp_root / "fresh-preflight.json"
|
||||
write_secret_file(review_secrets, "KB_REVIEW_PASSWORD", review_password)
|
||||
write_secret_file(apply_secrets, "KB_APPLY_PASSWORD", apply_password)
|
||||
|
||||
postgres.psql(bootstrap_sql(review_password, apply_password))
|
||||
postgres.psql((HERE / "kb_apply_prereqs.sql").read_text(encoding="utf-8"))
|
||||
|
||||
initial = database_state(postgres, child)
|
||||
stage_result = postgres.psql_json(stage.build_stage_sql(child))
|
||||
pending = proposal_readback(postgres, proposal_id)
|
||||
if pending is None:
|
||||
raise LifecycleError("staging did not persist the exact proposal")
|
||||
staged = database_state(postgres, child)
|
||||
packet = review_packet(pending)
|
||||
receipt["phases"] = {
|
||||
"initial": initial,
|
||||
"staged_pending_review": {
|
||||
"database_fingerprint": staged["database_fingerprint"],
|
||||
"canonical_fingerprint": staged["canonical_fingerprint"],
|
||||
"proposal": pending,
|
||||
"stage_result": stage_result,
|
||||
"review_packet": packet,
|
||||
},
|
||||
}
|
||||
|
||||
approve_base = [
|
||||
*_tool_base(
|
||||
HERE / "approve_proposal.py",
|
||||
proposal_id,
|
||||
review_secrets,
|
||||
"kb_review",
|
||||
container_name,
|
||||
),
|
||||
"--reviewed-by",
|
||||
reviewed_by,
|
||||
"--review-note",
|
||||
review_note,
|
||||
]
|
||||
approve_dry_run = run_tool([*approve_base, "--dry-run"], action="approval dry-run")
|
||||
if "kb_stage.approve_strict_proposal" not in approve_dry_run.stdout:
|
||||
raise LifecycleError("approval dry-run did not bind the guarded approval function")
|
||||
approve_execute = run_tool(approve_base, action="explicit review approval")
|
||||
approved = proposal_readback(postgres, proposal_id)
|
||||
immutable_approval = approval_readback(postgres, proposal_id)
|
||||
if approved is None or immutable_approval is None:
|
||||
raise LifecycleError("explicit review did not create both approved state and immutable approval")
|
||||
approved_state = database_state(postgres, child)
|
||||
receipt["phases"]["approved_unapplied"] = {
|
||||
"database_fingerprint": approved_state["database_fingerprint"],
|
||||
"canonical_fingerprint": approved_state["canonical_fingerprint"],
|
||||
"proposal": approved,
|
||||
"immutable_approval": immutable_approval,
|
||||
"review_action": {
|
||||
"reviewed_by": reviewed_by,
|
||||
"review_note": review_note,
|
||||
"dry_run": command_receipt(
|
||||
approve_dry_run,
|
||||
expected_marker="kb_stage.approve_strict_proposal",
|
||||
),
|
||||
"execute": command_receipt(approve_execute),
|
||||
},
|
||||
}
|
||||
|
||||
target_rows_before = target_rows_readback(postgres, child)
|
||||
preflight = apply_lifecycle.build_fresh_preflight(approved, target_rows_before)
|
||||
preflight_path.write_text(json.dumps(preflight, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
preflight_path.chmod(0o600)
|
||||
apply_base = [
|
||||
*_tool_base(
|
||||
HERE / "apply_proposal.py",
|
||||
proposal_id,
|
||||
apply_secrets,
|
||||
"kb_apply",
|
||||
container_name,
|
||||
),
|
||||
"--fresh-preflight",
|
||||
str(preflight_path),
|
||||
"--receipt-out",
|
||||
str(private_apply_receipt_path),
|
||||
]
|
||||
apply_dry_run = run_tool([*apply_base, "--dry-run"], action="guarded apply dry-run")
|
||||
apply_markers = (
|
||||
"kb_stage.assert_approved_proposal",
|
||||
"kb_stage.finish_approved_proposal",
|
||||
)
|
||||
if any(marker not in apply_dry_run.stdout for marker in apply_markers):
|
||||
raise LifecycleError("apply dry-run did not bind both approval guards")
|
||||
apply_execute = run_tool(apply_base, action="guarded canonical apply")
|
||||
if not private_apply_receipt_path.is_file():
|
||||
raise LifecycleError("guarded apply did not retain its private replay receipt")
|
||||
apply_receipt = json.loads(private_apply_receipt_path.read_text(encoding="utf-8"))
|
||||
applied = proposal_readback(postgres, proposal_id)
|
||||
if applied is None:
|
||||
raise LifecycleError("applied proposal readback is missing")
|
||||
final_rows = target_rows_readback(postgres, child)
|
||||
applied_state = database_state(postgres, child)
|
||||
|
||||
recompiled, _expected_again, _paths_again = compile_source_packet(source_packet)
|
||||
recompiled_child = recompiled["strict_child_proposal"]
|
||||
final_identities = exact_identities(recompiled_child, final_rows)
|
||||
receipt["identities"] = final_identities
|
||||
receipt["phases"]["applied"] = {
|
||||
"database_fingerprint": applied_state["database_fingerprint"],
|
||||
"canonical_fingerprint": applied_state["canonical_fingerprint"],
|
||||
"proposal": applied,
|
||||
"canonical_rows": final_rows,
|
||||
"apply_action": {
|
||||
"dry_run": {
|
||||
**command_receipt(apply_dry_run),
|
||||
"approval_guards_present": all(marker in apply_dry_run.stdout for marker in apply_markers),
|
||||
},
|
||||
"execute": command_receipt(apply_execute),
|
||||
},
|
||||
"apply_receipt": apply_receipt,
|
||||
}
|
||||
receipt["phases"]["recompiled_readback"] = {
|
||||
"bundle_sha256": canonical_sha256(recompiled),
|
||||
"matches_initial_compile": canonical_json_bytes(recompiled) == canonical_json_bytes(compiled),
|
||||
"proposal_id_matches": recompiled_child["id"] == proposal_id,
|
||||
"claim_ids_match": final_identities["claim_ids"] == compiled_identities["claim_ids"],
|
||||
"source_ids_match": final_identities["source_ids"] == compiled_identities["source_ids"],
|
||||
"canonical_rows_sha256": canonical_sha256(normalized_canonical_rows(final_rows)),
|
||||
"apply_receipt_rows_sha256": canonical_sha256(
|
||||
normalized_canonical_rows(apply_receipt.get("canonical_rows"))
|
||||
),
|
||||
}
|
||||
|
||||
expected_counts = ap.replay_receipt.expected_row_counts(applied)
|
||||
actual_counts = _row_counts(final_rows)
|
||||
checks = {
|
||||
"fresh_compile_matches_retained_packet": canonical_json_bytes(compiled) == canonical_json_bytes(expected),
|
||||
"compiler_emits_pending_review": compiled.get("status") == "pending_review",
|
||||
"initial_target_rows_absent": _all_target_rows_empty(initial["state"]["canonical_rows"]),
|
||||
"staged_proposal_exact": pending.get("id") == proposal_id and pending.get("payload") == child["payload"],
|
||||
"staged_status_pending_review": pending.get("status") == "pending_review",
|
||||
"review_packet_requires_human_review": packet.get("review_state") == "needs_human_review",
|
||||
"stage_did_not_change_canonical_rows": staged["canonical_fingerprint"] == initial["canonical_fingerprint"],
|
||||
"explicit_review_status_approved": approved.get("status") == "approved",
|
||||
"approved_is_not_applied": approved.get("applied_at") is None,
|
||||
"immutable_approval_matches_payload": immutable_approval.get("payload") == child["payload"],
|
||||
"immutable_approval_matches_reviewer": immutable_approval.get("reviewed_by_handle") == reviewed_by,
|
||||
"review_did_not_change_canonical_rows": approved_state["canonical_fingerprint"]
|
||||
== initial["canonical_fingerprint"],
|
||||
"apply_status_applied": applied.get("status") == "applied" and bool(applied.get("applied_at")),
|
||||
"apply_changed_canonical_fingerprint": applied_state["canonical_fingerprint"]
|
||||
!= initial["canonical_fingerprint"],
|
||||
"apply_row_counts_exact": actual_counts == expected_counts,
|
||||
"apply_receipt_replay_ready": apply_receipt.get("replay_ready") is True,
|
||||
"apply_receipt_rows_exact": normalized_canonical_rows(apply_receipt.get("canonical_rows"))
|
||||
== normalized_canonical_rows(final_rows),
|
||||
"recompile_matches_initial": canonical_json_bytes(recompiled) == canonical_json_bytes(compiled),
|
||||
"recompile_proposal_id_exact": recompiled_child["id"] == proposal_id,
|
||||
"database_fingerprints_distinguish_all_transitions": len(
|
||||
{
|
||||
initial["database_fingerprint"],
|
||||
staged["database_fingerprint"],
|
||||
approved_state["database_fingerprint"],
|
||||
applied_state["database_fingerprint"],
|
||||
}
|
||||
)
|
||||
== 4,
|
||||
}
|
||||
receipt["checks"] = checks
|
||||
if not all(checks.values()):
|
||||
failed = [name for name, passed in checks.items() if not passed]
|
||||
raise LifecycleError(f"lifecycle checks failed: {failed}")
|
||||
except (LifecycleError, OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc:
|
||||
receipt["error"] = {"type": type(exc).__name__, "message": str(exc)}
|
||||
finally:
|
||||
cleanup = postgres.cleanup()
|
||||
shutil.rmtree(temp_root, ignore_errors=True)
|
||||
cleanup["temporary_secret_root_absent"] = not temp_root.exists()
|
||||
cleanup["private_apply_receipt_removed_with_secret_root"] = not private_apply_receipt_path.exists()
|
||||
cleanup["all"] = all(
|
||||
(
|
||||
cleanup.get("container_absent") is True,
|
||||
cleanup.get("name_readback_clear") is True,
|
||||
cleanup.get("instance_label_readback_clear") is True,
|
||||
cleanup.get("network_mode_was_none") is True,
|
||||
cleanup.get("tmpfs_data_dir_was_present") is True,
|
||||
cleanup.get("docker_volume_mounts_observed") == [],
|
||||
cleanup.get("temporary_secret_root_absent") is True,
|
||||
cleanup.get("private_apply_receipt_removed_with_secret_root") is True,
|
||||
)
|
||||
)
|
||||
receipt["cleanup"] = cleanup
|
||||
receipt["checks"]["container_and_private_temp_cleanup_proven"] = cleanup["all"]
|
||||
receipt["status"] = (
|
||||
"pass" if "error" not in receipt and receipt["checks"] and all(receipt["checks"].values()) else "fail"
|
||||
)
|
||||
receipt["current_tier"] = (
|
||||
"realistic_isolated_local_container_end_to_end"
|
||||
if receipt["status"] == "pass"
|
||||
else "runtime_verification_failed"
|
||||
)
|
||||
receipt["strongest_claim"] = (
|
||||
"A retained real source packet was freshly compiled, staged, explicitly reviewed, guarded-applied "
|
||||
"into networkless disposable Postgres, deterministically recompiled, and read back with exact rows; "
|
||||
"the container and private temporary files were then independently absent."
|
||||
if receipt["status"] == "pass"
|
||||
else "The integrated-learning lifecycle did not pass all fail-closed checks."
|
||||
)
|
||||
receipt["not_proven"] = [
|
||||
"VPS or GCP canonical apply",
|
||||
"live Telegram send or reply",
|
||||
"human acceptance of these candidate claims for production",
|
||||
]
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary_output = output.with_name(f".{output.name}.{uuid.uuid4().hex}.tmp")
|
||||
temporary_output.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
os.replace(temporary_output, output)
|
||||
return receipt
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--source-packet", required=True, type=Path)
|
||||
parser.add_argument("--output", required=True, type=Path)
|
||||
parser.add_argument(
|
||||
"--review-action",
|
||||
required=True,
|
||||
choices=("approve",),
|
||||
help="explicit disposable-only review action; no default is accepted",
|
||||
)
|
||||
parser.add_argument("--reviewed-by", default=REVIEWER_HANDLE)
|
||||
parser.add_argument("--review-note", default=DEFAULT_REVIEW_NOTE)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parse_args(argv)
|
||||
receipt = run_lifecycle(
|
||||
args.source_packet,
|
||||
args.output,
|
||||
review_action=args.review_action,
|
||||
reviewed_by=args.reviewed_by,
|
||||
review_note=args.review_note,
|
||||
)
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"status": receipt["status"],
|
||||
"receipt": str(args.output.resolve()),
|
||||
"proposal_id": receipt.get("identities", {}).get("proposal_id"),
|
||||
"current_tier": receipt["current_tier"],
|
||||
"cleanup": receipt["cleanup"].get("all"),
|
||||
},
|
||||
sort_keys=True,
|
||||
)
|
||||
)
|
||||
return 0 if receipt["status"] == "pass" else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
133
tests/test_run_leo_integrated_learning_lifecycle.py
Normal file
133
tests/test_run_leo_integrated_learning_lifecycle.py
Normal file
|
|
@ -0,0 +1,133 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
||||
|
||||
import compile_kb_source_packet as compiler # noqa: E402
|
||||
import run_leo_integrated_learning_lifecycle as lifecycle # noqa: E402
|
||||
|
||||
|
||||
def _write_source_packet(tmp_path: Path) -> Path:
|
||||
packet = tmp_path / "source-packet"
|
||||
prepared = packet / "prepared"
|
||||
prepared.mkdir(parents=True)
|
||||
artifact = REPO_ROOT / "fixtures" / "working-leo" / "document-ingestion-v1.json"
|
||||
manifest = REPO_ROOT / "fixtures" / "working-leo" / "source-compiler-manifest-v1.json"
|
||||
extracted = prepared / "extracted-text.txt"
|
||||
manifest_target = prepared / "source-manifest.json"
|
||||
shutil.copyfile(artifact, extracted)
|
||||
shutil.copyfile(manifest, manifest_target)
|
||||
expected = compiler.compile_source_packet(extracted, extracted, manifest_target)
|
||||
(packet / "proposal-packet.json").write_text(
|
||||
json.dumps(expected, indent=2, sort_keys=True) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return packet
|
||||
|
||||
|
||||
def _docker_available() -> bool:
|
||||
if not shutil.which("docker"):
|
||||
return False
|
||||
result = subprocess.run(
|
||||
["docker", "info", "--format", "{{.ServerVersion}}"],
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
return result.returncode == 0
|
||||
|
||||
|
||||
def test_source_packet_recompiles_to_retained_expectation(tmp_path: Path) -> None:
|
||||
packet = _write_source_packet(tmp_path)
|
||||
|
||||
compiled, expected, paths = lifecycle.compile_source_packet(packet)
|
||||
|
||||
assert compiled == expected
|
||||
assert lifecycle.sha256_file(paths["artifact"]) == expected["validated_manifest"]["artifact_sha256"]
|
||||
assert compiled["status"] == "pending_review"
|
||||
assert compiled["strict_child_proposal"]["status"] == "pending_review"
|
||||
|
||||
|
||||
def test_source_packet_refuses_retained_packet_drift(tmp_path: Path) -> None:
|
||||
packet = _write_source_packet(tmp_path)
|
||||
expected_path = packet / "proposal-packet.json"
|
||||
expected = json.loads(expected_path.read_text(encoding="utf-8"))
|
||||
expected["strict_child_proposal"]["status"] = "approved"
|
||||
expected_path.write_text(json.dumps(expected), encoding="utf-8")
|
||||
|
||||
with pytest.raises(lifecycle.LifecycleError, match="does not match retained"):
|
||||
lifecycle.compile_source_packet(packet)
|
||||
|
||||
|
||||
def test_exact_identities_adds_observed_evidence_ids(tmp_path: Path) -> None:
|
||||
packet = _write_source_packet(tmp_path)
|
||||
compiled, _expected, _paths = lifecycle.compile_source_packet(packet)
|
||||
child = compiled["strict_child_proposal"]
|
||||
apply_payload = child["payload"]["apply_payload"]
|
||||
planned = apply_payload["evidence"][0]
|
||||
evidence_id = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"
|
||||
|
||||
identities = lifecycle.exact_identities(
|
||||
child,
|
||||
{
|
||||
"claim_evidence": [{"id": evidence_id, **planned}],
|
||||
"claim_edges": [],
|
||||
},
|
||||
)
|
||||
|
||||
assert identities["proposal_id"] == child["id"]
|
||||
assert identities["claim_ids"] == [row["id"] for row in apply_payload["claims"]]
|
||||
assert identities["source_ids"] == [row["id"] for row in apply_payload["sources"]]
|
||||
assert identities["planned_evidence_keys"] == [
|
||||
{
|
||||
"claim_id": row["claim_id"],
|
||||
"source_id": row["source_id"],
|
||||
"role": row["role"],
|
||||
}
|
||||
for row in apply_payload["evidence"]
|
||||
]
|
||||
assert identities["evidence_ids"] == [evidence_id]
|
||||
|
||||
|
||||
def test_lifecycle_requires_explicit_approve_action(tmp_path: Path) -> None:
|
||||
with pytest.raises(lifecycle.LifecycleError, match="only supported explicit review action"):
|
||||
lifecycle.run_lifecycle(
|
||||
tmp_path / "unused",
|
||||
tmp_path / "receipt.json",
|
||||
review_action="inspect",
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.skipif(not _docker_available(), reason="Docker daemon is required for the isolated lifecycle")
|
||||
def test_disposable_postgres_lifecycle_end_to_end(tmp_path: Path) -> None:
|
||||
packet = _write_source_packet(tmp_path)
|
||||
output = tmp_path / "integrated-learning-receipt.json"
|
||||
|
||||
receipt = lifecycle.run_lifecycle(
|
||||
packet,
|
||||
output,
|
||||
review_action="approve",
|
||||
review_note="Approved exact fixture payload for isolated lifecycle regression coverage.",
|
||||
)
|
||||
|
||||
assert receipt["status"] == "pass", receipt.get("error")
|
||||
assert receipt["current_tier"] == "realistic_isolated_local_container_end_to_end"
|
||||
assert receipt["phases"]["staged_pending_review"]["proposal"]["status"] == "pending_review"
|
||||
assert receipt["phases"]["approved_unapplied"]["proposal"]["status"] == "approved"
|
||||
assert receipt["phases"]["approved_unapplied"]["proposal"]["applied_at"] is None
|
||||
assert receipt["phases"]["applied"]["proposal"]["status"] == "applied"
|
||||
assert receipt["phases"]["applied"]["apply_receipt"]["replay_ready"] is True
|
||||
assert receipt["phases"]["recompiled_readback"]["matches_initial_compile"] is True
|
||||
assert receipt["identities"]["evidence_ids"]
|
||||
assert len(receipt["checks"]) >= 20
|
||||
assert all(receipt["checks"].values())
|
||||
assert receipt["cleanup"]["all"] is True
|
||||
assert json.loads(output.read_text(encoding="utf-8")) == receipt
|
||||
Loading…
Reference in a new issue