from __future__ import annotations import argparse import asyncio import copy import json import os import subprocess import sys from pathlib import Path from typing import Any import pytest REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "scripts")) import run_leo_clone_composition_checkpoint as composition # noqa: E402 RUN_MARKER = "composition-unit-20260711" CONVERSATION_MARKER = "composition-memory-unit-20260711" AGENT_ID = "11111111-1111-4111-8111-111111111111" def _packet(fixture: dict[str, Any]) -> dict[str, Any]: return { "claim_candidates": [ { "claim_key": "guarded_apply_required", "text": "Approval leaves canonical rows unchanged until guarded apply succeeds.", "type": "structural", "confidence": 0.98, "evidence": [{"source_key": "operating_document", "role": "grounds"}], "edges": [ { "edge_type": "contradicts", "target": "approval_is_canonical", } ], }, { "claim_key": "approval_is_canonical", "text": "Reviewer approval immediately makes a Leo proposal canonical.", "type": "empirical", "confidence": 0.4, "evidence": [{"source_key": "contradictory_post", "role": "grounds"}], }, ], "source_candidates": [ { "source_key": "operating_document", "source_type": "article", "storage_path": fixture["document"]["storage_path"], "url": None, "title": fixture["document"]["title"], "source_quality": "internal_operating_document", "key_excerpt": fixture["document"]["text"], "content_sha256": fixture["document"]["content_sha256"], }, { "source_key": "contradictory_post", "source_type": "post", "storage_path": None, "url": fixture["post"]["url"], "title": fixture["post"]["title"], "source_quality": "unverified_social_post", "key_excerpt": fixture["post"]["text"], "content_sha256": fixture["post"]["content_sha256"], }, ], "dedupe_conflict_assessment": { "database_search_query": fixture["project"], "potential_existing_claim_ids": [], "conflicts": [ { "from_claim_key": "guarded_apply_required", "to_claim_key": "approval_is_canonical", "relationship": "contradicts", } ], }, } def test_fixture_and_prompts_keep_source_bytes_and_memory_boundaries_explicit() -> None: fixture = composition.build_source_fixture(RUN_MARKER) prompts = composition.build_prompts(fixture, CONVERSATION_MARKER) assert fixture["project"] in fixture["document"]["text"] assert fixture["project"] in fixture["post"]["text"] assert fixture["document"]["content_sha256"] in prompts["extract"] assert fixture["post"]["content_sha256"] in prompts["extract"] assert CONVERSATION_MARKER in prompts["extract"] assert CONVERSATION_MARKER not in prompts["reason"] assert fixture["document"]["text"] not in prompts["reason"] assert fixture["post"]["text"] not in prompts["reason"] child = composition.normalized_stage.prepare_normalized_child( composition.build_parent_packet(_packet(fixture), fixture, RUN_MARKER, AGENT_ID) ) assert child["id"] not in prompts["reason"] assert all( row["id"] not in prompts["reason"] for collection in ("claims", "sources") for row in child["payload"]["apply_payload"][collection] ) def test_valid_packet_is_hash_bound_evidenced_and_conflict_aware() -> None: fixture = composition.build_source_fixture(RUN_MARKER) validation = composition.validate_composition_packet(_packet(fixture), fixture) assert validation["claim_count"] == 2 assert validation["source_count"] == 2 assert validation["evidence_source_keys"] == ["contradictory_post", "operating_document"] assert validation["conflict_pairs"] == [["guarded_apply_required", "approval_is_canonical"]] @pytest.mark.parametrize("mutation", ["hash", "excerpt", "evidence", "conflict"]) def test_packet_validation_fails_closed_on_provenance_or_reasoning_loss(mutation: str) -> None: fixture = composition.build_source_fixture(RUN_MARKER) packet = _packet(fixture) if mutation == "hash": packet["source_candidates"][0]["content_sha256"] = "0" * 64 elif mutation == "excerpt": packet["source_candidates"][0]["key_excerpt"] = "hallucinated excerpt" elif mutation == "evidence": packet["claim_candidates"][0]["evidence"] = [] else: packet["claim_candidates"][0]["edges"] = [] with pytest.raises(composition.bound.CheckpointError): composition.validate_composition_packet(packet, fixture) def test_packet_normalizes_to_deterministic_hash_bound_claim_graph() -> None: fixture = composition.build_source_fixture(RUN_MARKER) packet = _packet(fixture) parent = composition.build_parent_packet(packet, fixture, RUN_MARKER, AGENT_ID) child = composition.normalized_stage.prepare_normalized_child(parent) payload = child["payload"]["apply_payload"] assert parent["id"] == composition._stable_uuid(RUN_MARKER, "rich-source-packet") assert len(payload["claims"]) == 2 assert len(payload["sources"]) == 4 assert len(payload["evidence"]) == 4 assert len(payload["edges"]) == 1 assert payload["edges"][0]["edge_type"] == "contradicts" assert {fixture["document"]["content_sha256"], fixture["post"]["content_sha256"]} <= { row["hash"] for row in payload["sources"] } assert composition.expected_table_deltas(child) == { "public.claims": 2, "public.sources": 4, "public.claim_evidence": 4, "public.claim_edges": 1, "public.reasoning_tools": 0, "kb_stage.kb_proposals": 1, composition.lifecycle.APPROVAL_TABLE: 1, } def test_final_state_receipt_must_match_every_generated_row_id() -> None: fixture = composition.build_source_fixture(RUN_MARKER) child = composition.normalized_stage.prepare_normalized_child( composition.build_parent_packet(_packet(fixture), fixture, RUN_MARKER, AGENT_ID) ) payload = child["payload"]["apply_payload"] edge = payload["edges"][0] state = { "phase": "final", "conversation_marker": CONVERSATION_MARKER, "proposal_id": child["id"], "status": "applied", "claim_ids": [row["id"] for row in payload["claims"]], "source_ids": [row["id"] for row in payload["sources"]], "conflict_edge": { "from_claim": edge["from_claim"], "to_claim": edge["to_claim"], "edge_type": edge["edge_type"], }, } assert composition._state_matches(state, conversation_marker=CONVERSATION_MARKER, child=child) is True marker_alias = copy.deepcopy(state) marker_alias["marker"] = marker_alias.pop("conversation_marker") assert composition._state_matches(marker_alias, conversation_marker=CONVERSATION_MARKER, child=child) is True wrong = copy.deepcopy(state) wrong["source_ids"] = wrong["source_ids"][:-1] assert composition._state_matches(wrong, conversation_marker=CONVERSATION_MARKER, child=child) is False def test_unrelated_guard_excludes_every_planned_row_and_is_read_only( monkeypatch: pytest.MonkeyPatch, ) -> None: fixture = composition.build_source_fixture(RUN_MARKER) child = composition.normalized_stage.prepare_normalized_child( composition.build_parent_packet(_packet(fixture), fixture, RUN_MARKER, AGENT_ID) ) observed: dict[str, str] = {} def fake_psql(_container: str, _database: str, sql: str) -> dict[str, Any]: observed["sql"] = sql return {table: {"count": 1, "rowset_md5": "same"} for table in composition.TABLES} monkeypatch.setattr(composition.bound, "_psql_json", fake_psql) snapshot = composition.unrelated_guard_snapshot("clone-id", "teleo", child) lowered = observed["sql"].lower() assert set(snapshot) == set(composition.TABLES) assert "begin transaction read only" in lowered assert child["id"] in observed["sql"] for collection in ("claims", "sources"): for row in child["payload"]["apply_payload"][collection]: assert row["id"] in observed["sql"] assert "insert into" not in lowered assert "update " not in lowered assert "delete from" not in lowered def test_prefixed_receipts_are_exact_json_objects() -> None: payload = {"claim_candidates": [], "source_candidates": []} reply = "analysis\nCOMPOSITION_PACKET: " + json.dumps(payload, sort_keys=True) assert composition.parse_prefixed_json(reply, composition.PACKET_PREFIX) == payload with pytest.raises(composition.bound.CheckpointError, match="invalid"): composition.parse_prefixed_json("COMPOSITION_PACKET: nope", composition.PACKET_PREFIX) with pytest.raises(composition.bound.CheckpointError, match="did not contain"): composition.parse_prefixed_json("no receipt", composition.PACKET_PREFIX) def test_semantic_search_and_exact_read_helpers_require_database_commands_without_generated_ids() -> None: claim_id = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" proposal_id = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" project = "AURORA-TEST" turn = { "tool_invocations": [ {"argv": ["search", "approval apply canonical guarded operator review"]}, {"argv": ["search-proposals", project, "--status", "all", "--limit", "20"]}, {"argv": ["show-proposal", proposal_id]}, {"argv": ["show", claim_id]}, {"argv": ["evidence", claim_id, "--format", "json"]}, {"argv": ["edges", claim_id, "--format", "json"]}, ] } assert composition._searched_semantically_without_ids(turn, [claim_id, proposal_id]) is True assert composition._searched_composed_proposal(turn, project) is True assert composition._read_exact_proposal(turn, proposal_id) is True assert composition._read_claim(turn, claim_id) is True assert composition._read_claim_evidence(turn, claim_id) is True assert composition._read_claim_edges(turn, claim_id) is True evidence_only = {"tool_invocations": [{"argv": ["evidence", claim_id, "--format", "json"]}]} assert composition._read_claim_material(evidence_only, claim_id) is True turn["tool_invocations"][0]["argv"][1] = f"show {claim_id}" assert composition._searched_semantically_without_ids(turn, [claim_id, proposal_id]) is False def test_cumulative_transcript_merge_accepts_only_receipted_truncated_unknown_results() -> None: first = [ {"phase": "call", "tool_call_id": "call-1", "tool_name": "terminal"}, { "phase": "result", "tool_call_id": "call-1", "tool_name": "terminal", "content": '{"exit_code":0}', "content_truncated": False, }, ] second = [ *first, {"phase": "call", "tool_call_id": "call-2", "tool_name": "terminal"}, { "phase": "result", "tool_call_id": "call-2", "tool_name": "terminal", "content": '{"output":"large', "content_truncated": True, }, ] gateways = [ {"transcript_tool_trace": {"events": first}}, {"transcript_tool_trace": {"events": second}}, ] merged = composition.merge_transcript_events(gateways) consistency = composition.terminal_trace_consistency( merged, { "invocation_count": 2, "all_bound_to_supplied_target": True, "all_completed_successfully": True, }, ) assert len(merged) == 4 assert consistency["passes"] is True assert consistency["unknown_result_call_ids"] == ["call-2"] missing_receipt = [event for event in merged if event.get("tool_call_id") != "call-2" or event["phase"] != "result"] assert composition.terminal_trace_consistency( missing_receipt, { "invocation_count": 2, "all_bound_to_supplied_target": True, "all_completed_successfully": True, }, )["passes"] is False def test_source_contains_no_production_write_restart_or_delivery_adapter() -> None: source = (REPO_ROOT / "scripts" / "run_leo_clone_composition_checkpoint.py").read_text(encoding="utf-8").lower() assert "insert into public." not in source assert "update public." not in source assert "delete from public." not in source assert "systemctl restart" not in source assert "send_message(" not in source def test_source_manifest_covers_the_full_composition_and_gate_dependency_closure() -> None: paths = {item["repo_relative_path"] for item in composition.source_manifest()} assert paths == { "scripts/apply_proposal.py", "scripts/apply_worker.py", "scripts/approve_proposal.py", "scripts/bootstrap_clone_kb_gate.py", "scripts/kb_apply_prereqs.sql", "scripts/kb_proposal_normalize.py", "scripts/kb_proposal_review_packet.py", "scripts/kb_rich_proposal_creation_plan.py", "scripts/run_approve_claim_clone_canary.py", "scripts/run_leo_clone_bound_handler_checkpoint.py", "scripts/run_leo_clone_composition_checkpoint.py", "scripts/run_leo_clone_lifecycle_checkpoint.py", "scripts/stage_normalized_proposal.py", } def _args(tmp_path: Path) -> argparse.Namespace: return argparse.Namespace( container="composition-clone", db="teleo", output=tmp_path / "composition-report.json", run_marker=RUN_MARKER, conversation_marker=CONVERSATION_MARKER, chat_id=composition.bound.DEFAULT_CHAT_ID, user_id=composition.bound.DEFAULT_USER_ID, user_name="composition unit", turn_timeout=30, review_secrets_file=str(tmp_path / "kb-review.env"), apply_secrets_file=str(tmp_path / "kb-apply.env"), copy_model_auth=True, operator_review=True, guarded_apply=True, ) def _proposal(child: dict[str, Any], state: str) -> dict[str, Any]: reviewed = state in {"approved", "applied"} applied = state == "applied" return { "id": child["id"], "proposal_type": child["proposal_type"], "status": state, "channel": child["channel"], "source_ref": child["source_ref"], "rationale": child["rationale"], "payload": child["payload"], "reviewed_by_handle": composition.lifecycle.REVIEWER_HANDLE if reviewed else None, "reviewed_by_agent_id": AGENT_ID if reviewed else None, "reviewed_at": "2026-07-11T10:00:00+00:00" if reviewed else None, "review_note": composition.lifecycle.REVIEW_NOTE if reviewed else None, "applied_by_handle": composition.lifecycle.APPLIED_BY_HANDLE if applied else None, "applied_by_agent_id": AGENT_ID if applied else None, "applied_at": "2026-07-11T10:01:00+00:00" if applied else None, } def _approval_snapshot(proposal: dict[str, Any]) -> dict[str, Any]: return { "proposal_id": proposal["id"], "proposal_type": proposal["proposal_type"], "payload": proposal["payload"], "reviewed_by_handle": proposal["reviewed_by_handle"], "reviewed_by_agent_id": proposal["reviewed_by_agent_id"], "reviewed_at": proposal["reviewed_at"], "review_note": proposal["review_note"], } def test_mocked_full_composition_checkpoint_passes_with_exact_rows_and_cleanup( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: args = _args(tmp_path) Path(args.review_secrets_file).write_text("KB_REVIEW_PASSWORD=test\n", encoding="utf-8") Path(args.apply_secrets_file).write_text("KB_APPLY_PASSWORD=test\n", encoding="utf-8") fixture = composition.build_source_fixture(args.run_marker) packet = _packet(fixture) parent = composition.build_parent_packet(packet, fixture, args.run_marker, AGENT_ID) child = composition.normalized_stage.prepare_normalized_child(parent) expected_rows = composition.guarded._expected_bundle_rows(child["payload"]["apply_payload"]) empty_rows = {key: [] for key in expected_rows} base_counts = {table: 10 for table in composition.TABLES} final_counts = { table: base_counts[table] + composition.expected_table_deltas(child)[table] for table in composition.TABLES } base_snapshot = { "counts": base_counts, "rowset_md5": {table: f"base-{table}" for table in composition.TABLES}, } state: dict[str, Any] = {"phase": "initial", "invocations": [], "transcript_events": []} target_identity = { "id": "composition-clone-id", "name": args.container, "image": "postgres", "labels": { composition.bound.DISPOSABLE_LABEL_KEY: composition.bound.DISPOSABLE_LABEL_VALUE, }, "network_mode": "none", "ports": {}, "mount_sources": ["/composition-clone-data"], "started_at": "clone-start", } production_identity = { "id": "production-id", "name": composition.bound.PRODUCTION_CONTAINER, "image": "postgres", "labels": {}, "network_mode": "bridge", "ports": {}, "mount_sources": ["/production-data"], "started_at": "production-start", } monkeypatch.setattr( composition.lifecycle, "assert_disposable_target", lambda _container, _database: (dict(target_identity), dict(production_identity)), ) monkeypatch.setattr( composition.bound, "container_identity", lambda name: dict(production_identity if name == composition.bound.PRODUCTION_CONTAINER else target_identity), ) def guard_snapshot(container: str, _database: str, **_kwargs: Any) -> dict[str, Any]: snapshot = json.loads(json.dumps(base_snapshot)) production = container == composition.bound.PRODUCTION_CONTAINER snapshot["database_identity"] = { "current_database": args.db, "system_identifier": "production-system" if production else "clone-system", "transaction_read_only": "off", } if not production and state["phase"] == "applied": snapshot["counts"] = dict(final_counts) for table, delta in composition.expected_table_deltas(child).items(): if delta: snapshot["rowset_md5"][table] = "changed" return snapshot monkeypatch.setattr(composition.bound, "database_guard_snapshot", guard_snapshot) clone_gate = { "approval_table": composition.lifecycle.APPROVAL_TABLE, "review_principals_table": "kb_stage.kb_review_principals", "review_role_exists": True, "apply_role_exists": True, "approve_function": "kb_stage.approve_strict_proposal", "assert_function": "kb_stage.assert_approved_proposal", "finish_function": "kb_stage.finish_approved_proposal", } production_gate = { "approval_table": None, "review_principals_table": None, "review_role_exists": False, "apply_role_exists": True, "approve_function": None, "assert_function": None, "finish_function": None, } monkeypatch.setattr( composition, "gate_schema_manifest", lambda container, _database: ( dict(production_gate) if container == composition.bound.PRODUCTION_CONTAINER else dict(clone_gate) ), ) monkeypatch.setattr( composition.bound, "service_state", lambda: {"returncode": 0, "ActiveState": "active", "MainPID": "123", "NRestarts": "0"}, ) monkeypatch.setattr( composition.bound, "bridge_hashes", lambda: {"wrapper": "live-wrapper", "bridge_skill": "live-skill"}, ) runtime_root = tmp_path / "composition-runtime" runtime_profile = runtime_root / "profile" def copy_profile(**_kwargs: Any) -> tuple[Path, dict[str, Any]]: runtime_profile.mkdir(parents=True) composition.bound.register_temp_root(runtime_root) return runtime_profile, {"passes": True, "secret_contents_read_or_recorded": False} monkeypatch.setattr(composition.bound, "copy_profile", copy_profile) monkeypatch.setattr( composition.bound, "copy_ephemeral_model_auth", lambda _profile: {"mode": "ephemeral_file_copy", "source_contents_recorded": False}, ) monkeypatch.setattr( composition.bound, "load_ephemeral_provider_environment", lambda _profile: ({}, {"bound_credential_count": 0, "credential_contents_recorded": False}), ) tool_log = runtime_profile / "tool-log.jsonl" monkeypatch.setattr( composition, "patch_composition_bridge", lambda _profile, **_kwargs: { "tool_log_path": str(tool_log), "run_nonce": "composition-run-nonce", "wrapper_path": str(runtime_profile / "bin" / "teleo-kb"), "bridge_skill_path": str(runtime_profile / "skills" / "teleo-kb-bridge" / "SKILL.md"), }, ) monkeypatch.setattr(composition, "resolve_leo_agent_id", lambda _container, _database: AGENT_ID) monkeypatch.setattr( composition.normalized_stage, "stage_normalized_child", lambda _container, _database, _parent: ( state.update(phase="pending_review") or { "child": child, "bound_container_id": target_identity["id"], "database_receipt": { "created": True, "proposal_id": child["id"], "proposal_type": child["proposal_type"], "status": "pending_review", "source_ref": child["source_ref"], "payload_sha256": child["payload_sha256"], }, } ), ) monkeypatch.setattr( composition.guarded, "_exact_bundle_readback", lambda _container, _database, _payload: expected_rows if state["phase"] == "applied" else empty_rows, ) monkeypatch.setattr( composition.guarded, "_proposal_readback", lambda _container, _database, _proposal_id: ( None if state["phase"] == "initial" else _proposal(child, state["phase"]) ), ) monkeypatch.setattr( composition.guarded, "_approval_snapshot_readback", lambda _container, _database, _proposal_id: _approval_snapshot(_proposal(child, "approved")), ) unrelated = {table: {"count": 10, "rowset_md5": f"unrelated-{table}"} for table in composition.TABLES} monkeypatch.setattr( composition, "unrelated_guard_snapshot", lambda _container, _database, _child: json.loads(json.dumps(unrelated)), ) def review( _args: argparse.Namespace, _proposal_id: str ) -> tuple[subprocess.CompletedProcess[str], subprocess.CompletedProcess[str]]: state["phase"] = "approved" return ( subprocess.CompletedProcess([], 0, "select kb_stage.approve_strict_proposal(...);", ""), subprocess.CompletedProcess([], 0, "approved", ""), ) def apply( _args: argparse.Namespace, _proposal_id: str ) -> tuple[subprocess.CompletedProcess[str], subprocess.CompletedProcess[str]]: state["phase"] = "applied" return ( subprocess.CompletedProcess( [], 0, "select kb_stage.assert_approved_proposal(...); select kb_stage.finish_approved_proposal(...);", "", ), subprocess.CompletedProcess([], 0, "applied", ""), ) monkeypatch.setattr(composition.lifecycle, "run_operator_review", review) monkeypatch.setattr(composition.lifecycle, "run_guarded_apply", apply) async def run_turn( _args: argparse.Namespace, _profile: Path, *, phase: str, prompt: str, provider_environment: dict[str, str], tool_log: Path, ) -> dict[str, Any]: assert provider_environment == {} assert tool_log == runtime_profile / "tool-log.jsonl" if phase == "T1_extract_sources": reply = "Extracted two conflicting claims.\nCOMPOSITION_PACKET: " + json.dumps( packet, sort_keys=True, separators=(",", ":") ) argvs = [["context", packet["dedupe_conflict_assessment"]["database_search_query"]]] pid = 101 else: payload = child["payload"]["apply_payload"] edge = payload["edges"][0] receipt = { "phase": "final", "conversation_marker": args.conversation_marker, "proposal_id": child["id"], "status": "applied", "claim_ids": [row["id"] for row in payload["claims"]], "source_ids": [row["id"] for row in payload["sources"]], "conflict_edge": { "from_claim": edge["from_claim"], "to_claim": edge["to_claim"], "edge_type": edge["edge_type"], }, } reply = ( "Approval was not canonical; guarded apply made it applied and canonical. The sources conflict. " f"I recall {args.conversation_marker}.\nCOMPOSITION_STATE: " + json.dumps(receipt, sort_keys=True, separators=(",", ":")) ) argvs = [ ["search", "approval apply canonical guarded operator review"], ["search-proposals", fixture["project"], "--status", "all", "--limit", "20"], ["show-proposal", child["id"]], *[["show", row["id"]] for row in payload["claims"]], *[["evidence", row["id"], "--format", "json"] for row in payload["claims"]], ["edges", payload["claims"][0]["id"], "--format", "json"], ] pid = 202 invocations = [ { "invocation_id": f"{phase}-{index}", "container": target_identity["id"], "database": args.db, "argv": argv, "returncode": 0, } for index, argv in enumerate(argvs) ] state["invocations"].extend(invocations) current_events = [ event for invocation in invocations for event in ( { "phase": "call", "tool_name": "terminal", "tool_call_id": invocation["invocation_id"], }, { "phase": "result", "tool_name": "terminal", "tool_call_id": invocation["invocation_id"], "content": '{"exit_code":0}', "content_truncated": False, }, ) ] state["transcript_events"].extend(current_events) events = copy.deepcopy(state["transcript_events"]) return { "phase": phase, "prompt": prompt, "prompt_sha256": "prompt-hash", "tool_invocations": invocations, "gateway": { "authorized": True, "handler_invoked": True, "posted_to_telegram": False, "model_free_fallback_used": False, "reply": reply, "session_key": "telegram:composition", "transcript_tool_trace": { "session_id": "persisted-composition-session", "events": events, }, "tool_surface": { "allowed_tools": ["skills_list", "skill_view", "terminal"], "actual_registry_tools": ["skills_list", "skill_view", "terminal"], "gateway_adapters_verified_mapping": True, "gateway_adapter_count": 0, "send_message_tool_enabled": False, "terminal_restricted_to_clone_wrapper": True, "terminal_subprocess_inherits_provider_credentials": False, }, "child_process": { "pid": pid, "readiness_verified": True, "timed_out": False, "exitcode": 0, "result_transport": "private_temp_file", "termination": {"attempted": False}, "alive_after_readback": False, "process_group_alive_after_readback": False, }, }, } monkeypatch.setattr(composition, "run_turn", run_turn) monkeypatch.setattr( composition.bound, "read_tool_proof", lambda _path, _container, _database, **_kwargs: { "invocations": list(state["invocations"]), "invocation_count": len(state["invocations"]), "all_bound_to_supplied_target": bool(state["invocations"]), "all_completed_successfully": bool(state["invocations"]), }, ) report = asyncio.run(composition.run_checkpoint(args)) assert report["status"] == "pass", { "errors": report["errors"], "failed_checks": [name for name, passed in report["checks"].items() if not passed], } assert all(report["checks"].values()) assert report["final_canonical_rows"] == expected_rows assert report["checks"]["target_table_deltas_exact"] is True assert report["cleanup"]["active_gateway_registry_clear"] is True assert not os.path.lexists(runtime_root) retained = json.loads(args.output.read_text(encoding="utf-8")) assert retained["status"] == "pass" assert args.output.stat().st_mode & 0o777 == 0o600