From 1a6b7d51a4401dc7c5472880512b8377f9766a01 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Wed, 15 Jul 2026 00:33:29 +0200 Subject: [PATCH] Add deterministic genesis plus ledger rebuild --- docs/kb-rebuild-and-recompile.md | 95 +- ops/run_local_genesis_ledger_rebuild.py | 1284 +++++++++++++++++ .../test_run_local_genesis_ledger_rebuild.py | 833 +++++++++++ 3 files changed, 2205 insertions(+), 7 deletions(-) create mode 100644 ops/run_local_genesis_ledger_rebuild.py create mode 100644 tests/test_run_local_genesis_ledger_rebuild.py diff --git a/docs/kb-rebuild-and-recompile.md b/docs/kb-rebuild-and-recompile.md index 841e222..2442702 100644 --- a/docs/kb-rebuild-and-recompile.md +++ b/docs/kb-rebuild-and-recompile.md @@ -141,9 +141,89 @@ Normal applies mark the SQL hash as `exact_executed_sql`. A later pretends the current engine hash is historical proof of the originally executed program. -This closes replay-receipt loss for new strict applies. It does not reconstruct -legacy freeform applies, and it does not yet execute the complete -genesis-plus-ledger blank-database replay. +This closes replay-receipt loss for new strict applies. The receipt alone does +not retain every column of the proposal ledger, so exact reconstruction also +needs the full approved proposal row, immutable approval snapshot, and final +applied proposal row. + +## Genesis Plus Strict Ledger: Working Insert-Only Slice + +`ops/run_local_genesis_ledger_rebuild.py` now executes the first exact +genesis-plus-ledger slice in one command: + +```bash +.venv/bin/python ops/run_local_genesis_ledger_rebuild.py \ + --genesis-dump /private/path/genesis.dump \ + --genesis-manifest /private/path/genesis-manifest.jsonl \ + --ledger /private/path/ledger.json \ + --ledger-sha256 "$LEDGER_SHA256" \ + --output /tmp/genesis-ledger-rebuild-receipt.json +``` + +The v1 ledger pins the genesis dump and manifest, final parity manifest, +reconstruction/restore/guard/apply/replay/parity engines, and every ordered +private material file. Each material file contains one existing `kb_apply_replay_receipt`, the +exact full proposal row immediately before apply, its immutable +`kb_proposal_approvals` row, and the exact full proposal row after apply. These +files can contain claim text or source excerpts and must remain private. + +The ledger shape is: + +```json +{ + "artifact": "teleo_genesis_plus_ledger", + "contract_version": 1, + "engine": { + "reconstruction_command_sha256": "", + "base_rebuild_engine_sha256": "", + "apply_engine_sha256": "", + "replay_receipt_engine_sha256": "", + "guard_prerequisites_sha256": "", + "parity_sql_sha256": "" + }, + "genesis": { + "dump_sha256": "", + "parity_manifest_sha256": "" + }, + "entries": [{ + "sequence": 1, + "material": "private/0001.json", + "sha256": "", + "replay_material_sha256": "" + }], + "final_parity": { + "manifest": "final-manifest.jsonl", + "sha256": "" + } +} +``` + +Each referenced material object has exact top-level keys +`artifact`, `contract_version`, `sequence`, `approved_proposal`, +`approval_snapshot`, `applied_proposal`, and `replay_receipt`. Proposal objects +must contain every current `kb_stage.kb_proposals` column; partial envelopes +are rejected. + +The command verifies every hash before starting Docker, restores a tmpfs-backed +Postgres with `--network none`, reapplies the current guarded prerequisites, +and proves genesis parity. For each entry it seeds the receipt's exact row IDs +and timestamps in the disposable clone, executes the existing `kb_apply` +payload-bound guard and ledger transition, checks exact proposal and canonical +row readbacks, then verifies the complete final parity manifest. Its public +mode-`0600` receipt contains hashes, IDs, types, counts, parity summaries, and +cleanup proof, but no payloads, rows, SQL, source paths, or command errors. + +The exact v1 claim ceiling is intentionally narrow: + +- `add_edge`, `attach_evidence`, and `approve_claim` strict receipts execute; +- sequence gaps, hash drift, engine drift, duplicate proposals, and legacy or + freeform payloads fail before container start; +- `revise_strategy` fails closed because its current receipt omits the prior + strategies and nodes that the apply engine deactivates or retires; +- full proposal before/after rows are mandatory because the current receipt + envelope does not retain proposal origin fields or exact `updated_at`; +- this proves only an isolated local reconstruction. It does not touch or prove + VPS, GCP, Telegram, a live database, or blank-schema source recompilation. The source compiler now turns one raw artifact, its strict UTF-8 extraction, and a reviewed extraction manifest into a deterministic, hash-bound @@ -190,10 +270,11 @@ neither command applies canonical rows. The existing full-data clone canary separately proves that a reviewed packet can create source, claim, evidence, and edge rows and affect later reasoning. -The remaining recompilation capability is to join genesis, accepted packets, -and their row-level receipts into a complete corpus runner. One-document -preparation, staging, and receipt capture do not yet prove a clean database can -be rebuilt semantically from every source. +The remaining reconstruction work is to retain complete deltas for mutating +contracts such as `revise_strategy`, backfill or explicitly reject legacy +freeform applies, and extend beyond genesis recovery to a blank-schema source +compiler. The insert-only ledger runner does not prove that every historical +canonical row can be rebuilt semantically from retained sources. ## Definition Of Working diff --git a/ops/run_local_genesis_ledger_rebuild.py b/ops/run_local_genesis_ledger_rebuild.py new file mode 100644 index 0000000..ef871a3 --- /dev/null +++ b/ops/run_local_genesis_ledger_rebuild.py @@ -0,0 +1,1284 @@ +#!/usr/bin/env python3 +"""Deterministically rebuild a disposable Postgres from genesis plus a strict ledger. + +The v1 replay boundary is intentionally narrow. It supports strict, insert-only +``add_edge``, ``attach_evidence``, and ``approve_claim`` receipts. Each ledger +entry must also retain the exact approved and applied proposal rows plus the +immutable approval snapshot. Legacy/freeform entries and ``revise_strategy`` +fail before Docker starts because their retained material is not a complete +database delta. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import signal +import subprocess +import sys +import time +import uuid +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +try: + from . import run_local_canonical_postgres_rebuild as canonical_rebuild + from .verify_postgres_parity_manifest import load_manifest +except ImportError: # pragma: no cover - direct script execution + import run_local_canonical_postgres_rebuild as canonical_rebuild + from verify_postgres_parity_manifest import load_manifest + + +REPO_ROOT = Path(__file__).resolve().parents[1] +SCRIPTS_DIR = REPO_ROOT / "scripts" +if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) + +import apply_proposal as apply_engine # noqa: E402, I001 +import kb_apply_replay_receipt as replay_receipt # noqa: E402 + + +LEDGER_ARTIFACT = "teleo_genesis_plus_ledger" +MATERIAL_ARTIFACT = "teleo_strict_proposal_replay_material" +LEDGER_CONTRACT_VERSION = 1 +MATERIAL_CONTRACT_VERSION = 1 +SUPPORTED_PROPOSAL_TYPES = frozenset({"add_edge", "attach_evidence", "approve_claim"}) +CLAIM_CEILING = ( + "exact genesis plus ordered strict insert-only proposal replay; supported types are " + "add_edge, attach_evidence, and approve_claim; full approved/applied proposal rows " + "and immutable approval snapshots are required" +) +NOT_PROVEN = ( + "legacy or freeform proposal replay", + "revise_strategy replay because prior-row mutations are absent from v1 receipts", + "source-corpus recompilation from a blank schema", + "VPS, GCP, Telegram, or any live database mutation", +) + +APPLY_ENGINE_PATH = SCRIPTS_DIR / "apply_proposal.py" +REPLAY_RECEIPT_ENGINE_PATH = SCRIPTS_DIR / "kb_apply_replay_receipt.py" +GUARD_PREREQUISITES_PATH = SCRIPTS_DIR / "kb_apply_prereqs.sql" +PARITY_SQL_PATH = REPO_ROOT / "ops" / "postgres_parity_manifest.sql" +BASE_REBUILD_PATH = REPO_ROOT / "ops" / "run_local_canonical_postgres_rebuild.py" +GUARD_PREREQUISITES_DESTINATION = "/tmp/kb-apply-prereqs.sql" +FIXED_ENGINE_PATHS = frozenset( + { + APPLY_ENGINE_PATH.resolve(), + REPLAY_RECEIPT_ENGINE_PATH.resolve(), + GUARD_PREREQUISITES_PATH.resolve(), + PARITY_SQL_PATH.resolve(), + BASE_REBUILD_PATH.resolve(), + Path(__file__).resolve(), + } +) + +SHA256_RE = re.compile(r"[0-9a-f]{64}\Z") +PROPOSAL_TRANSITION_FIELDS = frozenset( + {"status", "applied_by_handle", "applied_by_agent_id", "applied_at", "updated_at"} +) +PROPOSAL_FIELDS = frozenset( + { + "id", + "proposal_type", + "status", + "proposed_by_handle", + "proposed_by_agent_id", + "channel", + "source_ref", + "rationale", + "payload", + "reviewed_by_handle", + "reviewed_by_agent_id", + "reviewed_at", + "review_note", + "applied_by_handle", + "applied_by_agent_id", + "applied_at", + "created_at", + "updated_at", + } +) +APPROVAL_FIELDS = frozenset( + { + "proposal_id", + "proposal_type", + "payload", + "reviewed_by_handle", + "reviewed_by_agent_id", + "reviewed_by_db_role", + "reviewed_at", + "review_note", + } +) +CANONICAL_TABLE_ORDER = ( + "claims", + "sources", + "reasoning_tools", + "claim_evidence", + "claim_edges", +) + + +class ReconstructionError(RuntimeError): + """A fail-closed error whose message is safe for the public receipt.""" + + def __init__(self, code: str, safe_message: str) -> None: + super().__init__(safe_message) + self.code = code + self.safe_message = safe_message + + +@dataclass(frozen=True) +class LedgerEntry: + sequence: int + material_path: Path + material_sha256: str + replay_material_sha256: str + approved_proposal: dict[str, Any] + approval_snapshot: dict[str, Any] + applied_proposal: dict[str, Any] + receipt: dict[str, Any] + current_apply_sql: str + current_apply_sql_sha256: str + + +@dataclass(frozen=True) +class ReconstructionBundle: + ledger_sha256: str + ledger_bytes: int + genesis_dump_sha256: str + genesis_dump_bytes: int + genesis_manifest: dict[str, Any] + genesis_manifest_sha256: str + genesis_manifest_bytes: int + final_manifest_path: Path + final_manifest: dict[str, Any] + final_manifest_sha256: str + final_manifest_bytes: int + engine_hashes: dict[str, str] + entries: tuple[LedgerEntry, ...] + protected_paths: frozenset[Path] + + +def _canonical_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _sha256_text(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _validate_sha256(value: Any, field: str) -> str: + if not isinstance(value, str) or not SHA256_RE.fullmatch(value): + raise ReconstructionError("invalid_hash_pin", f"{field} must be a lowercase SHA-256 digest") + return value + + +def _require_exact_keys(value: Any, expected: frozenset[str], field: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise ReconstructionError("invalid_contract", f"{field} must be an object") + observed = frozenset(value) + if observed != expected: + raise ReconstructionError( + "invalid_contract", + f"{field} fields do not match v1 contract; " + f"missing_count={len(expected - observed)}, extra_count={len(observed - expected)}", + ) + return value + + +def _load_json_object(path: Path, field: str) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise ReconstructionError("invalid_json_artifact", f"{field} is not valid UTF-8 JSON") from exc + if not isinstance(value, dict): + raise ReconstructionError("invalid_json_artifact", f"{field} must contain one JSON object") + return value + + +def _resolve_artifact(ledger_path: Path, declared: Any, field: str) -> Path: + if not isinstance(declared, str) or not declared.strip(): + raise ReconstructionError("invalid_contract", f"{field} must be a non-empty path string") + candidate = Path(declared).expanduser() + if not candidate.is_absolute(): + candidate = ledger_path.parent / candidate + candidate = candidate.resolve() + if not candidate.is_file(): + raise ReconstructionError("missing_artifact", f"{field} is not an existing regular file") + return candidate + + +def _assert_file_hash(path: Path, expected: Any, field: str) -> str: + pin = _validate_sha256(expected, field) + actual = canonical_rebuild.sha256_file(path) + if actual != pin: + raise ReconstructionError("hash_pin_mismatch", f"{field} does not match the supplied artifact") + return actual + + +def _engine_hashes() -> dict[str, str]: + paths = { + "reconstruction_command_sha256": Path(__file__), + "base_rebuild_engine_sha256": BASE_REBUILD_PATH, + "apply_engine_sha256": APPLY_ENGINE_PATH, + "replay_receipt_engine_sha256": REPLAY_RECEIPT_ENGINE_PATH, + "guard_prerequisites_sha256": GUARD_PREREQUISITES_PATH, + "parity_sql_sha256": PARITY_SQL_PATH, + } + return {name: canonical_rebuild.sha256_file(path) for name, path in paths.items()} + + +def _validate_proposal_rows( + approved: dict[str, Any], + applied: dict[str, Any], + approval: dict[str, Any], + receipt_proposal: dict[str, Any], +) -> None: + _require_exact_keys(approved, PROPOSAL_FIELDS, "material.approved_proposal") + _require_exact_keys(applied, PROPOSAL_FIELDS, "material.applied_proposal") + _require_exact_keys(approval, APPROVAL_FIELDS, "material.approval_snapshot") + + try: + proposal_id = str(uuid.UUID(str(applied["id"]))) + except (TypeError, ValueError, AttributeError) as exc: + raise ReconstructionError("invalid_material", "material proposal id must be a UUID") from exc + + if approved["id"] != proposal_id or approval["proposal_id"] != proposal_id: + raise ReconstructionError("material_mismatch", "proposal and approval ids do not match") + if approved["status"] != "approved" or applied["status"] != "applied": + raise ReconstructionError( + "unsupported_proposal_state", + "material requires one exact approved row and one exact applied row", + ) + if any(approved[field] is not None for field in ("applied_by_handle", "applied_by_agent_id", "applied_at")): + raise ReconstructionError( + "unsupported_proposal_state", "approved proposal material already contains apply metadata" + ) + if not applied["applied_by_handle"] or not applied["applied_by_agent_id"] or not applied["applied_at"]: + raise ReconstructionError("invalid_material", "applied proposal material lacks exact apply metadata") + if not approved["reviewed_by_agent_id"] or not approval["reviewed_by_db_role"]: + raise ReconstructionError( + "invalid_material", "strict replay requires a reviewer agent id and reviewer database role" + ) + + immutable_fields = PROPOSAL_FIELDS - PROPOSAL_TRANSITION_FIELDS + changed_immutable = sorted(field for field in immutable_fields if approved[field] != applied[field]) + if changed_immutable: + raise ReconstructionError( + "material_mismatch", + f"proposal changed outside the guarded apply transition: {changed_immutable}", + ) + + approval_projection = { + "proposal_id": approved["id"], + "proposal_type": approved["proposal_type"], + "payload": approved["payload"], + "reviewed_by_handle": approved["reviewed_by_handle"], + "reviewed_by_agent_id": approved["reviewed_by_agent_id"], + "reviewed_at": approved["reviewed_at"], + "review_note": approved["review_note"], + } + for field, expected in approval_projection.items(): + if approval[field] != expected: + raise ReconstructionError( + "material_mismatch", f"approval snapshot does not match approved proposal field {field}" + ) + + envelope_fields = ( + "id", + "proposal_type", + "status", + "payload", + "reviewed_by_handle", + "reviewed_by_agent_id", + "reviewed_at", + "review_note", + "applied_by_handle", + "applied_by_agent_id", + "applied_at", + ) + if any(receipt_proposal.get(field) != applied[field] for field in envelope_fields): + raise ReconstructionError( + "material_mismatch", "replay receipt proposal envelope does not match applied proposal row" + ) + + +def _validate_entry_material( + material: dict[str, Any], + *, + expected_sequence: int, + expected_replay_hash: str, + material_path: Path, + material_sha256: str, +) -> LedgerEntry: + expected_keys = frozenset( + { + "artifact", + "contract_version", + "sequence", + "approved_proposal", + "approval_snapshot", + "applied_proposal", + "replay_receipt", + } + ) + _require_exact_keys(material, expected_keys, "ledger entry material") + if material["artifact"] != MATERIAL_ARTIFACT or material["contract_version"] != MATERIAL_CONTRACT_VERSION: + raise ReconstructionError("unsupported_material_contract", "ledger entry material is not v1") + if material["sequence"] != expected_sequence: + raise ReconstructionError("ledger_order_mismatch", "material sequence does not match ledger order") + + receipt = material["replay_receipt"] + if not isinstance(receipt, dict): + raise ReconstructionError("invalid_material", "material.replay_receipt must be an object") + proposal = receipt.get("proposal") + if not isinstance(proposal, dict): + raise ReconstructionError("invalid_material", "replay receipt has no proposal object") + proposal_type = proposal.get("proposal_type") + if proposal_type == "revise_strategy": + raise ReconstructionError( + "unsupported_mutating_receipt", + "revise_strategy receipts omit prior strategy and node mutations and cannot replay exactly", + ) + if proposal_type not in SUPPORTED_PROPOSAL_TYPES: + raise ReconstructionError( + "unsupported_proposal_type", "ledger entry proposal type is not supported by v1 reconstruction" + ) + if receipt.get("artifact") != "kb_apply_replay_receipt" or receipt.get("replay_ready") is not True: + raise ReconstructionError("invalid_replay_receipt", "ledger entry is not a replay-ready apply receipt") + if receipt.get("receipt_contract_version") != replay_receipt.RECEIPT_CONTRACT_VERSION: + raise ReconstructionError("unsupported_receipt_contract", "replay receipt contract version is unsupported") + + apply_metadata = receipt.get("apply_engine") + canonical_rows = receipt.get("canonical_rows") + if not isinstance(apply_metadata, dict) or not isinstance(canonical_rows, dict): + raise ReconstructionError("invalid_replay_receipt", "replay receipt is missing apply or row material") + try: + rebuilt = replay_receipt.build_replay_receipt( + proposal, + canonical_rows, + apply_sql_sha256=apply_metadata.get("apply_sql_sha256"), + apply_sql_source=apply_metadata.get("source"), + exported_at_utc=receipt.get("exported_at_utc"), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ReconstructionError( + "invalid_replay_receipt", "replay receipt failed strict payload and exact-row validation" + ) from exc + + if receipt.get("hashes") != rebuilt["hashes"]: + raise ReconstructionError("replay_receipt_hash_mismatch", "replay receipt hashes are not recoverable") + if receipt.get("expected_row_counts") != rebuilt["expected_row_counts"]: + raise ReconstructionError("replay_receipt_count_mismatch", "replay receipt expected counts are invalid") + if receipt.get("actual_row_counts") != rebuilt["actual_row_counts"]: + raise ReconstructionError("replay_receipt_count_mismatch", "replay receipt actual counts are invalid") + actual_replay_hash = rebuilt["hashes"]["replay_material_sha256"] + if actual_replay_hash != expected_replay_hash: + raise ReconstructionError( + "replay_material_hash_mismatch", "ledger replay-material pin does not match the private receipt" + ) + + approved = material["approved_proposal"] + approval = material["approval_snapshot"] + applied = material["applied_proposal"] + _validate_proposal_rows(approved, applied, approval, proposal) + try: + current_apply_sql = apply_engine.build_apply_sql(proposal, applied["applied_by_handle"]) + except (KeyError, TypeError, ValueError) as exc: + raise ReconstructionError( + "current_apply_engine_rejected_material", + "current guarded apply engine rejected the strict replay material", + ) from exc + + return LedgerEntry( + sequence=expected_sequence, + material_path=material_path, + material_sha256=material_sha256, + replay_material_sha256=actual_replay_hash, + approved_proposal=approved, + approval_snapshot=approval, + applied_proposal=applied, + receipt=receipt, + current_apply_sql=current_apply_sql, + current_apply_sql_sha256=_sha256_text(current_apply_sql), + ) + + +def load_bundle(args: argparse.Namespace) -> ReconstructionBundle: + canonical_rebuild.validate_custom_dump(args.genesis_dump) + ledger_sha256 = _assert_file_hash(args.ledger, args.ledger_sha256, "--ledger-sha256") + ledger = _load_json_object(args.ledger, "ledger") + top_level_keys = frozenset({"artifact", "contract_version", "engine", "genesis", "entries", "final_parity"}) + _require_exact_keys(ledger, top_level_keys, "ledger") + if ledger["artifact"] != LEDGER_ARTIFACT or ledger["contract_version"] != LEDGER_CONTRACT_VERSION: + raise ReconstructionError("unsupported_ledger_contract", "ledger is not the v1 genesis-plus-ledger contract") + + actual_engine_hashes = _engine_hashes() + expected_engine = _require_exact_keys(ledger["engine"], frozenset(actual_engine_hashes), "ledger.engine") + for field, actual_hash in actual_engine_hashes.items(): + if _validate_sha256(expected_engine[field], f"ledger.engine.{field}") != actual_hash: + raise ReconstructionError("engine_hash_mismatch", f"ledger engine pin does not match current {field}") + + genesis = _require_exact_keys( + ledger["genesis"], frozenset({"dump_sha256", "parity_manifest_sha256"}), "ledger.genesis" + ) + genesis_dump_sha256 = _assert_file_hash(args.genesis_dump, genesis["dump_sha256"], "ledger.genesis.dump_sha256") + genesis_manifest_sha256 = _assert_file_hash( + args.genesis_manifest, + genesis["parity_manifest_sha256"], + "ledger.genesis.parity_manifest_sha256", + ) + try: + genesis_manifest = load_manifest(args.genesis_manifest) + except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc: + raise ReconstructionError("invalid_genesis_manifest", "genesis parity manifest is invalid") from exc + + final_spec = _require_exact_keys(ledger["final_parity"], frozenset({"manifest", "sha256"}), "ledger.final_parity") + final_manifest_path = _resolve_artifact(args.ledger, final_spec["manifest"], "final parity manifest") + final_manifest_sha256 = _assert_file_hash(final_manifest_path, final_spec["sha256"], "ledger.final_parity.sha256") + try: + final_manifest = load_manifest(final_manifest_path) + except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc: + raise ReconstructionError("invalid_final_manifest", "final parity manifest is invalid") from exc + + raw_entries = ledger["entries"] + if not isinstance(raw_entries, list): + raise ReconstructionError("invalid_contract", "ledger.entries must be an ordered list") + entries: list[LedgerEntry] = [] + protected_paths = { + args.genesis_dump.resolve(), + args.genesis_manifest.resolve(), + args.ledger.resolve(), + final_manifest_path, + *FIXED_ENGINE_PATHS, + } + seen_proposal_ids: set[str] = set() + for expected_sequence, raw_entry in enumerate(raw_entries, 1): + entry_spec = _require_exact_keys( + raw_entry, + frozenset({"sequence", "material", "sha256", "replay_material_sha256"}), + f"ledger.entries[{expected_sequence - 1}]", + ) + if entry_spec["sequence"] != expected_sequence: + raise ReconstructionError( + "ledger_order_mismatch", "ledger entry sequences must be contiguous and start at 1" + ) + material_path = _resolve_artifact( + args.ledger, entry_spec["material"], f"ledger entry {expected_sequence} material" + ) + material_sha256 = _assert_file_hash( + material_path, entry_spec["sha256"], f"ledger entry {expected_sequence} sha256" + ) + expected_replay_hash = _validate_sha256( + entry_spec["replay_material_sha256"], + f"ledger entry {expected_sequence} replay_material_sha256", + ) + entry = _validate_entry_material( + _load_json_object(material_path, f"ledger entry {expected_sequence} material"), + expected_sequence=expected_sequence, + expected_replay_hash=expected_replay_hash, + material_path=material_path, + material_sha256=material_sha256, + ) + proposal_id = entry.applied_proposal["id"] + if proposal_id in seen_proposal_ids: + raise ReconstructionError("duplicate_proposal", "ledger contains the same proposal more than once") + seen_proposal_ids.add(proposal_id) + entries.append(entry) + protected_paths.add(material_path) + + if args.output.resolve() in protected_paths: + raise ReconstructionError("unsafe_output_path", "--output must not overwrite any reconstruction input") + + return ReconstructionBundle( + ledger_sha256=ledger_sha256, + ledger_bytes=args.ledger.stat().st_size, + genesis_dump_sha256=genesis_dump_sha256, + genesis_dump_bytes=args.genesis_dump.stat().st_size, + genesis_manifest=genesis_manifest, + genesis_manifest_sha256=genesis_manifest_sha256, + genesis_manifest_bytes=args.genesis_manifest.stat().st_size, + final_manifest_path=final_manifest_path, + final_manifest=final_manifest, + final_manifest_sha256=final_manifest_sha256, + final_manifest_bytes=final_manifest_path.stat().st_size, + engine_hashes=actual_engine_hashes, + entries=tuple(entries), + protected_paths=frozenset(protected_paths), + ) + + +def _jsonb_literal(value: Any) -> str: + return apply_engine.sql_literal(_canonical_json(value)) + "::jsonb" + + +def build_seed_sql(entry: LedgerEntry) -> str: + statements = [ + "begin;", + "set local standard_conforming_strings = on;", + "insert into kb_stage.kb_proposals", + "select seeded.* from jsonb_populate_record(", + f" null::kb_stage.kb_proposals, {_jsonb_literal(entry.approved_proposal)}", + ") seeded", + "on conflict (id) do nothing;", + "insert into kb_stage.kb_proposal_approvals", + "select seeded.* from jsonb_populate_record(", + f" null::kb_stage.kb_proposal_approvals, {_jsonb_literal(entry.approval_snapshot)}", + ") seeded", + "on conflict (proposal_id) do nothing;", + ] + rows = entry.receipt["canonical_rows"] + for table in CANONICAL_TABLE_ORDER: + table_rows = rows.get(table) + if not table_rows: + continue + statements.extend( + [ + f"insert into public.{table}", + "select seeded.* from jsonb_populate_recordset(", + f" null::public.{table}, {_jsonb_literal(table_rows)}", + ") seeded", + "on conflict do nothing;", + ] + ) + statements.append("commit;") + return "\n".join(statements) + "\n" + + +def build_proposal_readback_sql(proposal_id: str) -> str: + literal = apply_engine.sql_literal(proposal_id) + return f"""select jsonb_build_object( + 'proposal', (select to_jsonb(p) from kb_stage.kb_proposals p + where p.id = {literal}::uuid), + 'approval_snapshot', (select to_jsonb(a) from kb_stage.kb_proposal_approvals a + where a.proposal_id = {literal}::uuid) +)::text; +""" + + +def build_applied_timestamp_normalization_sql(entry: LedgerEntry) -> str: + applied = entry.applied_proposal + return f"""do $reconstruction$ +declare + changed_rows integer; +begin + update kb_stage.kb_proposals + set applied_at = {apply_engine.sql_literal(applied["applied_at"])}::timestamptz, + updated_at = {apply_engine.sql_literal(applied["updated_at"])}::timestamptz + where id = {apply_engine.sql_literal(applied["id"])}::uuid + and status = 'applied' + and applied_by_handle = {apply_engine.sql_literal(applied["applied_by_handle"])} + and applied_by_agent_id is not distinct from + {apply_engine.sql_literal(applied["applied_by_agent_id"])}::uuid; + get diagnostics changed_rows = row_count; + if changed_rows <> 1 then + raise exception 'deterministic proposal timestamp normalization failed'; + end if; +end +$reconstruction$; +""" + + +def _run_stdin( + command: list[str], + *, + stdin: str, + timeout: float, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + input=stdin, + text=True, + capture_output=True, + check=False, + timeout=timeout, + ) + + +def _diagnostic_fingerprint(completed: subprocess.CompletedProcess[str]) -> str: + output = (completed.stderr or "") + "\n" + (completed.stdout or "") + return _sha256_text(output) + + +def _require_command_success( + completed: subprocess.CompletedProcess[str], + *, + code: str, + action: str, +) -> str: + if completed.returncode != 0: + fingerprint = _diagnostic_fingerprint(completed) + raise ReconstructionError( + code, + f"{action} failed; private command output withheld; diagnostic_sha256={fingerprint}", + ) + return completed.stdout + + +def _psql( + args: argparse.Namespace, + container: str, + sql: str, + *, + role: str, + timeout: float, + code: str, + action: str, +) -> str: + command = [ + args.docker_bin, + "exec", + "-i", + container, + "psql", + "-X", + "-U", + role, + "-h", + "127.0.0.1", + "-d", + args.database, + "-Atq", + "-v", + "ON_ERROR_STOP=1", + ] + try: + completed = _run_stdin(command, stdin=sql, timeout=timeout) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ReconstructionError(code, f"{action} did not complete; private details withheld") from exc + return _require_command_success(completed, code=code, action=action) + + +def _read_json( + args: argparse.Namespace, + container: str, + sql: str, + *, + code: str, + action: str, +) -> Any: + raw = _psql( + args, + container, + sql, + role="postgres", + timeout=args.command_timeout, + code=code, + action=action, + ).strip() + try: + return json.loads(raw) + except json.JSONDecodeError as exc: + raise ReconstructionError(code, f"{action} returned an invalid private JSON readback") from exc + + +def _assert_exact_json(actual: Any, expected: Any, *, code: str, action: str) -> None: + if _canonical_json(actual) != _canonical_json(expected): + raise ReconstructionError(code, f"{action} did not match the hash-pinned material") + + +def _entry_summary(entry: LedgerEntry) -> dict[str, Any]: + proposal = entry.applied_proposal + return { + "sequence": entry.sequence, + "proposal_id": proposal["id"], + "proposal_type": proposal["proposal_type"], + "material_sha256": entry.material_sha256, + "replay_material_sha256": entry.replay_material_sha256, + "original_apply_sql_sha256": entry.receipt["apply_engine"]["apply_sql_sha256"], + "current_apply_sql_sha256": entry.current_apply_sql_sha256, + "expected_row_counts": entry.receipt["expected_row_counts"], + "seed_exact": False, + "guarded_apply_executed": False, + "proposal_exact": False, + "canonical_rows_exact": False, + "status": "pending", + } + + +def apply_entry( + args: argparse.Namespace, + container: str, + entry: LedgerEntry, + summary: dict[str, Any], +) -> None: + try: + _psql( + args, + container, + build_seed_sql(entry), + role="postgres", + timeout=args.apply_timeout, + code="entry_seed_failed", + action=f"ledger entry {entry.sequence} exact seed", + ) + proposal_readback = _read_json( + args, + container, + build_proposal_readback_sql(entry.approved_proposal["id"]), + code="entry_seed_readback_failed", + action=f"ledger entry {entry.sequence} proposal seed readback", + ) + _assert_exact_json( + proposal_readback, + { + "proposal": entry.approved_proposal, + "approval_snapshot": entry.approval_snapshot, + }, + code="entry_seed_mismatch", + action=f"ledger entry {entry.sequence} proposal seed", + ) + seeded_rows = _read_json( + args, + container, + replay_receipt.build_postflight_sql(entry.receipt["proposal"]), + code="entry_seed_readback_failed", + action=f"ledger entry {entry.sequence} canonical seed readback", + ) + _assert_exact_json( + seeded_rows, + entry.receipt["canonical_rows"], + code="entry_seed_mismatch", + action=f"ledger entry {entry.sequence} canonical seed", + ) + summary["seed_exact"] = True + + _psql( + args, + container, + entry.current_apply_sql, + role="kb_apply", + timeout=args.apply_timeout, + code="guarded_apply_failed", + action=f"ledger entry {entry.sequence} guarded apply", + ) + summary["guarded_apply_executed"] = True + + _psql( + args, + container, + build_applied_timestamp_normalization_sql(entry), + role="postgres", + timeout=args.command_timeout, + code="proposal_timestamp_normalization_failed", + action=f"ledger entry {entry.sequence} deterministic ledger timestamp normalization", + ) + proposal_readback = _read_json( + args, + container, + build_proposal_readback_sql(entry.applied_proposal["id"]), + code="entry_postflight_failed", + action=f"ledger entry {entry.sequence} applied proposal readback", + ) + _assert_exact_json( + proposal_readback, + { + "proposal": entry.applied_proposal, + "approval_snapshot": entry.approval_snapshot, + }, + code="entry_postflight_mismatch", + action=f"ledger entry {entry.sequence} applied proposal", + ) + summary["proposal_exact"] = True + + final_rows = _read_json( + args, + container, + replay_receipt.build_postflight_sql(entry.receipt["proposal"]), + code="entry_postflight_failed", + action=f"ledger entry {entry.sequence} exact row readback", + ) + _assert_exact_json( + final_rows, + entry.receipt["canonical_rows"], + code="entry_postflight_mismatch", + action=f"ledger entry {entry.sequence} canonical rows", + ) + summary["canonical_rows_exact"] = True + summary["status"] = "pass" + except ReconstructionError: + summary["status"] = "fail" + raise + + +def _sanitize_parity(result: dict[str, Any]) -> dict[str, Any]: + source = dict(result["source_manifest"]) + source.pop("path", None) + return { + "status": result["status"], + "verifier": result["verifier"], + "source_manifest": source, + "target_manifest": result["target_manifest"], + "problems": result["problems"], + "details": result["details"], + } + + +def _run_parity( + args: argparse.Namespace, + container: str, + manifest_path: Path, + manifest: dict[str, Any], + *, + code: str, + action: str, +) -> dict[str, Any]: + try: + result = canonical_rebuild.run_parity_check( + args.docker_bin, + container, + args.database, + source_manifest_path=manifest_path, + source_manifest=manifest, + manifest_sql=PARITY_SQL_PATH, + command_timeout=args.manifest_timeout, + max_target_query_ms=args.max_target_query_ms, + max_target_source_ratio=args.max_target_source_ratio, + ) + except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired, json.JSONDecodeError) as exc: + raise ReconstructionError(code, f"{action} could not complete; private details withheld") from exc + sanitized = _sanitize_parity(result) + if sanitized["status"] != "pass": + raise ReconstructionError(code, f"{action} found a manifest mismatch") + return sanitized + + +def _initial_receipt(args: argparse.Namespace, container: str) -> dict[str, Any]: + return { + "artifact": "local_genesis_plus_ledger_reconstruction", + "contract_version": 1, + "generated_at_utc": datetime.now(UTC).isoformat(), + "status": "running", + "claim_ceiling": CLAIM_CEILING, + "not_proven": list(NOT_PROVEN), + "database": args.database, + "inputs": { + "genesis_dump": {"bytes": None, "sha256": None, "pin_verified": False}, + "genesis_manifest": {"bytes": None, "sha256": None, "pin_verified": False}, + "ledger": {"bytes": None, "sha256": None, "pin_verified": False}, + "final_manifest": {"bytes": None, "sha256": None, "pin_verified": False}, + "engine_hashes": {}, + }, + "container": { + "name": container, + "image": args.image, + "network_mode_required": "none", + "postgres_data_tmpfs_required": (f"rw,nosuid,nodev,noexec,size={args.tmpfs_mb}m"), + "preexisting_absence_proven": False, + "started": False, + "isolation": None, + }, + "guard_prerequisites": {"status": "pending", "sha256": None}, + "genesis_parity": {"status": "pending"}, + "genesis_key_counts": {}, + "ledger": {"entry_count": None, "entries": []}, + "final_parity": {"status": "pending"}, + "final_key_counts": {}, + "cleanup": { + "attempted": False, + "container": container, + "container_absent": None, + "status": "pending", + }, + "timings_seconds": {}, + "safety": { + "production_database_touched": False, + "network_access_available_to_container": False, + "persistent_postgres_storage_used": False, + "private_replay_material_emitted": False, + "vps_gcp_telegram_or_live_db_used": False, + }, + "error": None, + } + + +def run_reconstruction(args: argparse.Namespace) -> dict[str, Any]: + started_at = time.monotonic() + container = canonical_rebuild.new_container_name(args.container_prefix) + receipt = _initial_receipt(args, container) + phase = "input_preflight" + primary_error: dict[str, Any] | None = None + + try: + bundle = load_bundle(args) + receipt["inputs"] = { + "genesis_dump": { + "bytes": bundle.genesis_dump_bytes, + "sha256": bundle.genesis_dump_sha256, + "pin_verified": True, + }, + "genesis_manifest": { + "bytes": bundle.genesis_manifest_bytes, + "sha256": bundle.genesis_manifest_sha256, + "pin_verified": True, + }, + "ledger": { + "bytes": bundle.ledger_bytes, + "sha256": bundle.ledger_sha256, + "pin_verified": True, + }, + "final_manifest": { + "bytes": bundle.final_manifest_bytes, + "sha256": bundle.final_manifest_sha256, + "pin_verified": True, + }, + "engine_hashes": bundle.engine_hashes, + } + receipt["ledger"]["entry_count"] = len(bundle.entries) + receipt["ledger"]["entries"] = [_entry_summary(entry) for entry in bundle.entries] + + phase = "container_name_preflight" + absent, absence_error = canonical_rebuild.inspect_container_absence( + args.docker_bin, container, timeout=args.command_timeout + ) + if absent is not True: + raise ReconstructionError( + "container_name_not_proven_unused", + "generated disposable container name could not be proven unused", + ) from RuntimeError(absence_error or "container exists") + receipt["container"]["preexisting_absence_proven"] = True + + phase = "container_start" + tmpfs_options = f"{canonical_rebuild.DATA_DIR}:rw,nosuid,nodev,noexec,size={args.tmpfs_mb}m" + completed = canonical_rebuild._run( + [ + args.docker_bin, + "run", + "--detach", + "--rm", + "--network", + "none", + "--name", + container, + "--label", + f"{canonical_rebuild.CANARY_LABEL}={container}", + "--tmpfs", + tmpfs_options, + "--env", + f"POSTGRES_DB={args.database}", + "--env", + "POSTGRES_HOST_AUTH_METHOD=trust", + args.image, + ], + timeout=args.command_timeout, + ) + container_id = _require_command_success( + completed, + code="container_start_failed", + action="networkless disposable PostgreSQL start", + ).strip() + if not container_id: + raise ReconstructionError("container_start_failed", "container start returned no id") + receipt["container"]["started"] = True + receipt["container"]["id_prefix"] = container_id[:12] + receipt["container"]["isolation"] = canonical_rebuild.inspect_container_isolation( + args.docker_bin, + container, + expected_image=args.image, + expected_label=container, + timeout=args.command_timeout, + ) + + phase = "database_readiness" + receipt["container"]["named_database_psql_attempts"] = canonical_rebuild.wait_for_named_database( + args.docker_bin, + container, + args.database, + startup_timeout=args.startup_timeout, + command_timeout=args.command_timeout, + poll_interval=args.poll_interval, + ) + receipt["container"]["named_database_psql_ready"] = True + + phase = "application_role_bootstrap" + receipt["application_roles_bootstrapped"] = canonical_rebuild.execute_application_role_bootstrap( + args.docker_bin, + container, + args.database, + bundle.genesis_manifest, + timeout=args.command_timeout, + ) + + phase = "genesis_restore" + completed = canonical_rebuild._run( + [ + args.docker_bin, + "cp", + str(args.genesis_dump.resolve()), + f"{container}:{canonical_rebuild.DUMP_DESTINATION}", + ], + timeout=args.command_timeout, + ) + _require_command_success(completed, code="genesis_copy_failed", action="genesis dump copy") + completed = canonical_rebuild._run( + [ + args.docker_bin, + "exec", + container, + "pg_restore", + "-U", + "postgres", + "-d", + args.database, + "--no-owner", + "--no-privileges", + "--exit-on-error", + canonical_rebuild.DUMP_DESTINATION, + ], + timeout=args.restore_timeout, + ) + _require_command_success(completed, code="genesis_restore_failed", action="genesis restore") + + phase = "guard_prerequisites" + completed = canonical_rebuild._run( + [ + args.docker_bin, + "cp", + str(GUARD_PREREQUISITES_PATH), + f"{container}:{GUARD_PREREQUISITES_DESTINATION}", + ], + timeout=args.command_timeout, + ) + _require_command_success( + completed, + code="guard_prerequisites_copy_failed", + action="guarded apply prerequisites copy", + ) + completed = canonical_rebuild._run( + [ + args.docker_bin, + "exec", + container, + "psql", + "-X", + "-U", + "postgres", + "-d", + args.database, + "-Atq", + "-v", + "ON_ERROR_STOP=1", + "-f", + GUARD_PREREQUISITES_DESTINATION, + ], + timeout=args.apply_timeout, + ) + _require_command_success( + completed, + code="guard_prerequisites_failed", + action="guarded apply prerequisites", + ) + receipt["guard_prerequisites"] = { + "status": "pass", + "sha256": bundle.engine_hashes["guard_prerequisites_sha256"], + } + + phase = "genesis_parity" + receipt["genesis_parity"] = _run_parity( + args, + container, + args.genesis_manifest, + bundle.genesis_manifest, + code="genesis_parity_failed", + action="genesis parity verification", + ) + receipt["genesis_key_counts"], _ = canonical_rebuild.collect_key_counts( + args.docker_bin, + container, + args.database, + timeout=args.command_timeout, + ) + + phase = "ordered_ledger_replay" + for entry, summary in zip(bundle.entries, receipt["ledger"]["entries"], strict=True): + apply_entry(args, container, entry, summary) + + phase = "final_parity" + receipt["final_parity"] = _run_parity( + args, + container, + bundle.final_manifest_path, + bundle.final_manifest, + code="final_parity_failed", + action="final parity verification", + ) + receipt["final_key_counts"], _ = canonical_rebuild.collect_key_counts( + args.docker_bin, + container, + args.database, + timeout=args.command_timeout, + ) + receipt["status"] = "pass" + except ReconstructionError as exc: + primary_error = { + "phase": phase, + "code": exc.code, + "type": type(exc).__name__, + "message": exc.safe_message, + } + receipt["status"] = "fail" + receipt["error"] = primary_error + except ( + OSError, + ValueError, + RuntimeError, + subprocess.TimeoutExpired, + json.JSONDecodeError, + KeyboardInterrupt, + canonical_rebuild.TerminationRequested, + ) as exc: + primary_error = { + "phase": phase, + "code": f"{phase}_failed", + "type": type(exc).__name__, + "message": f"{phase} failed; private details withheld", + } + receipt["status"] = "fail" + receipt["error"] = primary_error + finally: + cleanup_started = time.monotonic() + cleanup = canonical_rebuild.cleanup_container( + args.docker_bin, + container, + command_timeout=args.command_timeout, + poll_interval=args.poll_interval, + ) + receipt["cleanup"] = cleanup + receipt["timings_seconds"]["cleanup"] = round(time.monotonic() - cleanup_started, 6) + receipt["timings_seconds"]["total"] = round(time.monotonic() - started_at, 6) + if cleanup.get("container_absent") is not True: + receipt["status"] = "fail" + cleanup_error = { + "phase": "cleanup", + "code": "cleanup_absence_unproven", + "type": "CleanupProofError", + "message": "disposable container absence was not independently proven", + } + if primary_error is None: + receipt["error"] = cleanup_error + else: + receipt["cleanup_error"] = cleanup_error + if cleanup.get("interrupted"): + receipt["status"] = "fail" + interruption_error = { + "phase": "cleanup", + "code": "cleanup_interrupted", + "type": "TerminationRequested", + "message": "cleanup was interrupted; private details withheld", + } + if primary_error is None: + receipt["error"] = interruption_error + else: + receipt["cleanup_interruption"] = interruption_error + return receipt + + +def _declared_artifact_paths(ledger_path: Path) -> set[Path]: + ledger = _load_json_object(ledger_path, "ledger") + paths = {ledger_path.resolve()} + final = ledger.get("final_parity") + if isinstance(final, dict) and isinstance(final.get("manifest"), str): + paths.add(_resolve_artifact(ledger_path, final["manifest"], "final parity manifest")) + entries = ledger.get("entries") + if isinstance(entries, list): + for index, entry in enumerate(entries): + if isinstance(entry, dict) and isinstance(entry.get("material"), str): + paths.add(_resolve_artifact(ledger_path, entry["material"], f"entry {index + 1} material")) + return paths + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--genesis-dump", required=True, type=Path) + parser.add_argument("--genesis-manifest", required=True, type=Path) + parser.add_argument("--ledger", required=True, type=Path) + parser.add_argument("--ledger-sha256", required=True) + parser.add_argument("--database", default="teleo") + parser.add_argument("--image", default=canonical_rebuild.DEFAULT_IMAGE) + parser.add_argument("--container-prefix", default="teleo-genesis-ledger-rebuild") + parser.add_argument("--tmpfs-mb", default=1024, type=int) + parser.add_argument("--startup-timeout", default=60.0, type=float) + parser.add_argument("--restore-timeout", default=300.0, type=float) + parser.add_argument("--apply-timeout", default=180.0, type=float) + parser.add_argument("--manifest-timeout", default=180.0, type=float) + parser.add_argument("--command-timeout", default=30.0, type=float) + parser.add_argument("--poll-interval", default=0.25, type=float) + parser.add_argument("--max-target-query-ms", default=2000.0, type=float) + parser.add_argument("--max-target-source-ratio", default=20.0, type=float) + parser.add_argument("--docker-bin", default="docker") + parser.add_argument("--output", required=True, type=Path) + args = parser.parse_args(argv) + + if not canonical_rebuild.DATABASE_NAME_RE.fullmatch(args.database): + parser.error("--database must be a safe PostgreSQL identifier up to 63 characters") + if not canonical_rebuild.CONTAINER_PREFIX_RE.fullmatch(args.container_prefix): + parser.error("--container-prefix must be 1-40 Docker-safe characters") + if not args.image or any(character.isspace() for character in args.image): + parser.error("--image must be a non-empty Docker image reference without whitespace") + if not args.docker_bin or any(character.isspace() for character in args.docker_bin): + parser.error("--docker-bin must be one executable name or path without whitespace") + if not 64 <= args.tmpfs_mb <= 65536: + parser.error("--tmpfs-mb must be between 64 and 65536") + for value, flag in ( + (args.startup_timeout, "--startup-timeout"), + (args.restore_timeout, "--restore-timeout"), + (args.apply_timeout, "--apply-timeout"), + (args.manifest_timeout, "--manifest-timeout"), + (args.command_timeout, "--command-timeout"), + (args.poll_interval, "--poll-interval"), + (args.max_target_query_ms, "--max-target-query-ms"), + (args.max_target_source_ratio, "--max-target-source-ratio"), + ): + if value <= 0: + parser.error(f"{flag} must be positive") + for path, flag in ( + (args.genesis_dump, "--genesis-dump"), + (args.genesis_manifest, "--genesis-manifest"), + (args.ledger, "--ledger"), + ): + if not path.is_file(): + parser.error(f"{flag} must be an existing regular file") + if not SHA256_RE.fullmatch(args.ledger_sha256): + parser.error("--ledger-sha256 must be a lowercase SHA-256 digest") + try: + protected = { + args.genesis_dump.resolve(), + args.genesis_manifest.resolve(), + *_declared_artifact_paths(args.ledger), + *FIXED_ENGINE_PATHS, + } + except ReconstructionError as exc: + parser.error(exc.safe_message) + if args.output.resolve() in protected: + parser.error("--output must not overwrite a reconstruction input") + return args + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + previous_handlers: dict[signal.Signals, Any] = {} + for watched_signal in (signal.SIGINT, signal.SIGTERM): + previous_handlers[watched_signal] = signal.getsignal(watched_signal) + signal.signal(watched_signal, canonical_rebuild._termination_handler) + try: + payload = run_reconstruction(args) + finally: + for watched_signal, previous in previous_handlers.items(): + signal.signal(watched_signal, previous) + + try: + canonical_rebuild.write_receipt(args.output, payload) + except OSError: + payload["status"] = "fail" + payload["receipt_write_error"] = "receipt write failed; private details withheld" + print(json.dumps(payload, indent=2, sort_keys=True)) + return 0 if payload["status"] == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_run_local_genesis_ledger_rebuild.py b/tests/test_run_local_genesis_ledger_rebuild.py new file mode 100644 index 0000000..4919b1c --- /dev/null +++ b/tests/test_run_local_genesis_ledger_rebuild.py @@ -0,0 +1,833 @@ +from __future__ import annotations + +import argparse +import copy +import hashlib +import json +import shutil +import stat +import subprocess +import sys +import time +import uuid +from collections.abc import Iterator +from pathlib import Path + +import pytest + +from ops import run_local_genesis_ledger_rebuild as rebuild + +CLAIM_A = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" +CLAIM_B = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" +REVIEWER_ID = "11111111-1111-4111-8111-111111111111" +SERVICE_ID = "44444444-4444-4444-4444-444444444444" +PROPOSAL_ID = "99999999-9999-4999-8999-999999999999" +EDGE_ID = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee" +PRIVATE_MARKER = "PRIVATE-REPLAY-MATERIAL-MUST-NOT-LEAK" + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + digest.update(path.read_bytes()) + return digest.hexdigest() + + +def write_json(path: Path, value: object) -> Path: + path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return path + + +def manifest_rows(database: str = "teleo_fixture") -> list[dict]: + role = { + "name": "kb_apply", + "can_login": True, + "superuser": False, + "create_db": False, + "create_role": False, + "inherit": False, + "replication": False, + "bypass_rls": False, + } + rows = [ + { + "kind": "identity", + "database": database, + "server_version_num": 160010, + "transaction_read_only": "on", + "server_address": None, + "server_port": None, + "ssl": False, + "ssl_version": None, + "database_collation": "C", + "database_ctype": "C", + }, + {"kind": "schemas", "items": ["kb_stage", "public"]}, + {"kind": "extensions", "items": []}, + {"kind": "application_roles", "items": [role]}, + ] + rows.extend( + {"kind": kind, "items": []} + for kind in ( + "columns", + "constraints", + "indexes", + "sequences", + "views", + "functions", + "triggers", + "types", + "policies", + ) + ) + rows.extend( + [ + { + "kind": "table", + "schema": "public", + "table": "claims", + "row_count": 2, + "rowset_md5": "fixture", + }, + { + "kind": "performance", + "query": "count_claims", + "elapsed_ms": 1.0, + "observed_rows": 2, + }, + ] + ) + return rows + + +def write_manifest(path: Path) -> Path: + path.write_text("".join(json.dumps(row) + "\n" for row in manifest_rows()), encoding="utf-8") + return path + + +def proposal_rows() -> tuple[dict, dict, dict]: + payload = { + "apply_payload": { + "from_claim": CLAIM_A, + "to_claim": CLAIM_B, + "edge_type": "supports", + "weight": 0.8, + }, + "private_title": PRIVATE_MARKER, + } + approved = { + "id": PROPOSAL_ID, + "proposal_type": "add_edge", + "status": "approved", + "proposed_by_handle": "fixture-agent", + "proposed_by_agent_id": REVIEWER_ID, + "channel": "reconstruction_fixture", + "source_ref": "private://fixture/source", + "rationale": PRIVATE_MARKER, + "payload": payload, + "reviewed_by_handle": "m3ta", + "reviewed_by_agent_id": REVIEWER_ID, + "reviewed_at": "2026-07-14T01:00:00+00:00", + "review_note": "Reviewed exact strict fixture payload.", + "applied_by_handle": None, + "applied_by_agent_id": None, + "applied_at": None, + "created_at": "2026-07-14T00:59:00+00:00", + "updated_at": "2026-07-14T01:00:00+00:00", + } + applied = { + **approved, + "status": "applied", + "applied_by_handle": "kb-apply", + "applied_by_agent_id": SERVICE_ID, + "applied_at": "2026-07-14T01:01:00+00:00", + "updated_at": "2026-07-14T01:01:00.000001+00:00", + } + approval = { + "proposal_id": PROPOSAL_ID, + "proposal_type": "add_edge", + "payload": payload, + "reviewed_by_handle": "m3ta", + "reviewed_by_agent_id": REVIEWER_ID, + "reviewed_by_db_role": "kb_review", + "reviewed_at": "2026-07-14T01:00:00+00:00", + "review_note": "Reviewed exact strict fixture payload.", + } + return approved, applied, approval + + +def applied_envelope(applied: dict) -> dict: + fields = ( + "id", + "proposal_type", + "status", + "payload", + "reviewed_by_handle", + "reviewed_by_agent_id", + "reviewed_at", + "review_note", + "applied_by_handle", + "applied_by_agent_id", + "applied_at", + ) + return {field: applied[field] for field in fields} + + +def replay_material() -> dict: + approved, applied, approval = proposal_rows() + proposal = applied_envelope(applied) + edge = { + "id": EDGE_ID, + "from_claim": CLAIM_A, + "to_claim": CLAIM_B, + "edge_type": "supports", + "weight": 0.8, + "created_by": None, + "created_at": "2026-07-14T01:00:30+00:00", + } + receipt = rebuild.replay_receipt.build_replay_receipt( + proposal, + {"claim_edges": [edge]}, + apply_sql_sha256=rebuild.replay_receipt.sha256_text("fixture original guarded apply SQL"), + apply_sql_source="exact_executed_sql", + exported_at_utc="2026-07-14T01:02:00+00:00", + ) + return { + "artifact": rebuild.MATERIAL_ARTIFACT, + "contract_version": rebuild.MATERIAL_CONTRACT_VERSION, + "sequence": 1, + "approved_proposal": approved, + "approval_snapshot": approval, + "applied_proposal": applied, + "replay_receipt": receipt, + } + + +def write_bundle(tmp_path: Path, *, material: dict | None = None) -> argparse.Namespace: + dump = tmp_path / "genesis.dump" + dump.write_bytes(b"PGDMPfixture") + genesis_manifest = write_manifest(tmp_path / "genesis-manifest.jsonl") + final_manifest = write_manifest(tmp_path / "final-manifest.jsonl") + material = copy.deepcopy(material or replay_material()) + material_path = write_json(tmp_path / "entry-0001.json", material) + receipt_hash = material["replay_receipt"]["hashes"]["replay_material_sha256"] + ledger = { + "artifact": rebuild.LEDGER_ARTIFACT, + "contract_version": rebuild.LEDGER_CONTRACT_VERSION, + "engine": rebuild._engine_hashes(), + "genesis": { + "dump_sha256": sha256_file(dump), + "parity_manifest_sha256": sha256_file(genesis_manifest), + }, + "entries": [ + { + "sequence": 1, + "material": material_path.name, + "sha256": sha256_file(material_path), + "replay_material_sha256": receipt_hash, + } + ], + "final_parity": { + "manifest": final_manifest.name, + "sha256": sha256_file(final_manifest), + }, + } + ledger_path = write_json(tmp_path / "ledger.json", ledger) + return argparse.Namespace( + genesis_dump=dump, + genesis_manifest=genesis_manifest, + ledger=ledger_path, + ledger_sha256=sha256_file(ledger_path), + database="teleo", + image=rebuild.canonical_rebuild.DEFAULT_IMAGE, + container_prefix="teleo-genesis-ledger-test", + tmpfs_mb=256, + startup_timeout=30.0, + restore_timeout=60.0, + apply_timeout=60.0, + manifest_timeout=60.0, + command_timeout=20.0, + poll_interval=0.05, + max_target_query_ms=5000.0, + max_target_source_ratio=100.0, + docker_bin="docker", + output=tmp_path / "receipt.json", + ) + + +def rewrite_material_and_ledger(args: argparse.Namespace, material: dict) -> None: + material_path = args.ledger.parent / "entry-0001.json" + write_json(material_path, material) + ledger = json.loads(args.ledger.read_text(encoding="utf-8")) + ledger["entries"][0]["sha256"] = sha256_file(material_path) + write_json(args.ledger, ledger) + args.ledger_sha256 = sha256_file(args.ledger) + + +def test_load_bundle_validates_every_pin_and_current_apply_engine(tmp_path: Path) -> None: + args = write_bundle(tmp_path) + + bundle = rebuild.load_bundle(args) + + assert bundle.genesis_dump_sha256 == sha256_file(args.genesis_dump) + assert bundle.genesis_manifest_sha256 == sha256_file(args.genesis_manifest) + assert bundle.final_manifest_sha256 == sha256_file(tmp_path / "final-manifest.jsonl") + assert bundle.engine_hashes == rebuild._engine_hashes() + assert len(bundle.entries) == 1 + entry = bundle.entries[0] + assert entry.applied_proposal["proposal_type"] == "add_edge" + assert entry.current_apply_sql_sha256 == hashlib.sha256(entry.current_apply_sql.encode("utf-8")).hexdigest() + assert "kb_stage.assert_approved_proposal" in entry.current_apply_sql + + +def test_dump_hash_mismatch_fails_before_container_start(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + args = write_bundle(tmp_path) + args.genesis_dump.write_bytes(b"PGDMPtampered") + commands: list[list[str]] = [] + + def absent_docker(command: list[str], *, timeout: float): + commands.append(command) + if command[1:3] == ["rm", "--force"]: + return subprocess.CompletedProcess(command, 1, "", "No such container") + if command[1:3] == ["container", "inspect"]: + return subprocess.CompletedProcess(command, 1, "", "No such object") + raise AssertionError(f"unexpected command after failed preflight: {command}") + + monkeypatch.setattr(rebuild.canonical_rebuild, "_run", absent_docker) + + receipt = rebuild.run_reconstruction(args) + + assert receipt["status"] == "fail" + assert receipt["error"]["code"] == "hash_pin_mismatch" + assert receipt["cleanup"]["container_absent"] is True + assert not any(len(command) > 1 and command[1] == "run" for command in commands) + + +def test_revise_strategy_receipt_fails_closed_before_receipt_rows_are_used(tmp_path: Path) -> None: + material = replay_material() + material["replay_receipt"]["proposal"]["proposal_type"] = "revise_strategy" + material["applied_proposal"]["proposal_type"] = "revise_strategy" + material["approved_proposal"]["proposal_type"] = "revise_strategy" + material["approval_snapshot"]["proposal_type"] = "revise_strategy" + args = write_bundle(tmp_path, material=material) + + with pytest.raises(rebuild.ReconstructionError, match="prior strategy") as exc_info: + rebuild.load_bundle(args) + + assert exc_info.value.code == "unsupported_mutating_receipt" + + +def test_legacy_freeform_receipt_fails_closed(tmp_path: Path) -> None: + material = replay_material() + material["replay_receipt"]["proposal"]["payload"] = {"legacy": PRIVATE_MARKER} + material["approved_proposal"]["payload"] = {"legacy": PRIVATE_MARKER} + material["applied_proposal"]["payload"] = {"legacy": PRIVATE_MARKER} + material["approval_snapshot"]["payload"] = {"legacy": PRIVATE_MARKER} + args = write_bundle(tmp_path, material=material) + + with pytest.raises(rebuild.ReconstructionError) as exc_info: + rebuild.load_bundle(args) + + assert exc_info.value.code == "invalid_replay_receipt" + assert PRIVATE_MARKER not in exc_info.value.safe_message + + +def test_proposal_metadata_outside_apply_transition_is_rejected(tmp_path: Path) -> None: + material = replay_material() + material["applied_proposal"]["rationale"] = "silently changed after review" + args = write_bundle(tmp_path) + rewrite_material_and_ledger(args, material) + + with pytest.raises(rebuild.ReconstructionError) as exc_info: + rebuild.load_bundle(args) + + assert exc_info.value.code == "material_mismatch" + assert "rationale" in exc_info.value.safe_message + + +def test_output_cannot_overwrite_a_fixed_guard_or_engine_file(tmp_path: Path) -> None: + args = write_bundle(tmp_path) + args.output = rebuild.GUARD_PREREQUISITES_PATH + + with pytest.raises(rebuild.ReconstructionError) as exc_info: + rebuild.load_bundle(args) + + assert exc_info.value.code == "unsafe_output_path" + + +def test_unknown_private_field_name_is_not_echoed_in_public_error(tmp_path: Path) -> None: + material = replay_material() + material[PRIVATE_MARKER] = "private value" + args = write_bundle(tmp_path, material=material) + + with pytest.raises(rebuild.ReconstructionError) as exc_info: + rebuild.load_bundle(args) + + assert exc_info.value.code == "invalid_contract" + assert PRIVATE_MARKER not in exc_info.value.safe_message + + +def test_seed_sql_preserves_exact_rows_and_uses_only_fixed_tables(tmp_path: Path) -> None: + entry = rebuild.load_bundle(write_bundle(tmp_path)).entries[0] + + sql = rebuild.build_seed_sql(entry) + + assert "jsonb_populate_record" in sql + assert "kb_stage.kb_proposals" in sql + assert "kb_stage.kb_proposal_approvals" in sql + assert "public.claim_edges" in sql + assert "on conflict do nothing" in sql + assert PRIVATE_MARKER in sql + assert "public.strategies" not in sql + + +def test_private_psql_failure_output_is_replaced_with_a_fingerprint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, object] = {} + + def fail(command: list[str], *, stdin: str, timeout: float): + captured.update(command=command, stdin=stdin, timeout=timeout) + return subprocess.CompletedProcess(command, 1, "", f"ERROR: {PRIVATE_MARKER}") + + monkeypatch.setattr(rebuild, "_run_stdin", fail) + args = argparse.Namespace(docker_bin="docker", database="teleo") + + with pytest.raises(rebuild.ReconstructionError) as exc_info: + rebuild._psql( + args, + "fixture-container", + f"select '{PRIVATE_MARKER}';", + role="postgres", + timeout=3.0, + code="fixture_failure", + action="private fixture SQL", + ) + + assert PRIVATE_MARKER not in exc_info.value.safe_message + assert "diagnostic_sha256=" in exc_info.value.safe_message + assert PRIVATE_MARKER not in " ".join(captured["command"]) + assert PRIVATE_MARKER in captured["stdin"] + + +def test_entry_summary_is_secret_safe(tmp_path: Path) -> None: + entry = rebuild.load_bundle(write_bundle(tmp_path)).entries[0] + + encoded = json.dumps(rebuild._entry_summary(entry), sort_keys=True) + + assert PRIVATE_MARKER not in encoded + assert entry.applied_proposal["id"] in encoded + assert entry.replay_material_sha256 in encoded + + +BOOTSTRAP_SQL = f""" +create role kb_apply login noinherit; +create role kb_review login noinherit; +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, + 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() +); +create table public.strategies ( + id uuid primary key, + 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() +); +create table public.strategy_nodes ( + id uuid primary key, + 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 + ('{REVIEWER_ID}', 'm3ta', 'human'); +insert into public.claims (id, type, text, status, confidence, tags, created_by) values + ('{CLAIM_A}', 'structural', 'Fixture claim A', 'open', 0.8, array['fixture'], '{REVIEWER_ID}'), + ('{CLAIM_B}', 'structural', 'Fixture claim B', 'open', 0.8, array['fixture'], '{REVIEWER_ID}'); +""" + + +def docker_psql( + container: str, + database: str, + sql: str, + *, + role: str = "postgres", + check: bool = True, +) -> subprocess.CompletedProcess[str]: + completed = subprocess.run( + [ + "docker", + "exec", + "-i", + container, + "psql", + "-X", + "-U", + role, + "-h", + "127.0.0.1", + "-d", + database, + "-Atq", + "-v", + "ON_ERROR_STOP=1", + ], + input=sql, + text=True, + capture_output=True, + check=False, + ) + if check and completed.returncode != 0: + pytest.fail( + f"fixture psql failed ({completed.returncode})\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}" + ) + return completed + + +def wait_for_postgres(container: str, database: str) -> None: + for _ in range(120): + completed = docker_psql(container, database, "select 1;", check=False) + if completed.returncode == 0: + return + time.sleep(0.25) + pytest.fail("fixture PostgreSQL did not become ready") + + +def capture_manifest(container: str, database: str, destination: Path) -> None: + copy_result = subprocess.run( + ["docker", "cp", str(rebuild.PARITY_SQL_PATH), f"{container}:/tmp/parity.sql"], + text=True, + capture_output=True, + check=False, + ) + if copy_result.returncode != 0: + pytest.fail(f"could not copy parity SQL: {copy_result.stderr}") + completed = subprocess.run( + [ + "docker", + "exec", + container, + "psql", + "-X", + "-U", + "postgres", + "-d", + database, + "-Atq", + "-v", + "ON_ERROR_STOP=1", + "-f", + "/tmp/parity.sql", + ], + text=True, + capture_output=True, + check=False, + ) + if completed.returncode != 0: + pytest.fail(f"could not capture parity manifest: {completed.stderr}") + destination.write_text(completed.stdout, encoding="utf-8") + + +@pytest.fixture(scope="module") +def source_postgres() -> Iterator[tuple[str, str]]: + if shutil.which("docker") is None: + pytest.skip("Docker is required for the genesis-plus-ledger lifecycle test") + container = f"genesis-ledger-source-{uuid.uuid4().hex[:12]}" + database = "teleo_fixture" + started = subprocess.run( + [ + "docker", + "run", + "--rm", + "--detach", + "--name", + container, + "--env", + f"POSTGRES_DB={database}", + "--env", + "POSTGRES_HOST_AUTH_METHOD=trust", + rebuild.canonical_rebuild.DEFAULT_IMAGE, + ], + text=True, + capture_output=True, + check=False, + ) + if started.returncode != 0: + pytest.fail(f"could not start fixture PostgreSQL: {started.stderr}") + try: + wait_for_postgres(container, database) + docker_psql(container, database, BOOTSTRAP_SQL) + copy_result = subprocess.run( + [ + "docker", + "cp", + str(rebuild.GUARD_PREREQUISITES_PATH), + f"{container}:/tmp/kb-apply-prereqs.sql", + ], + text=True, + capture_output=True, + check=False, + ) + if copy_result.returncode != 0: + pytest.fail(f"could not copy apply prerequisites: {copy_result.stderr}") + migration = subprocess.run( + [ + "docker", + "exec", + container, + "psql", + "-X", + "-U", + "postgres", + "-d", + database, + "-Atq", + "-v", + "ON_ERROR_STOP=1", + "-f", + "/tmp/kb-apply-prereqs.sql", + ], + text=True, + capture_output=True, + check=False, + ) + if migration.returncode != 0: + pytest.fail(f"could not apply guarded prerequisites: {migration.stderr}") + yield container, database + finally: + subprocess.run( + ["docker", "rm", "--force", container], + text=True, + capture_output=True, + check=False, + ) + + +def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup( + tmp_path: Path, + source_postgres: tuple[str, str], +) -> None: + source_container, database = source_postgres + genesis_dump = tmp_path / "genesis.dump" + dump_result = subprocess.run( + [ + "docker", + "exec", + source_container, + "pg_dump", + "-U", + "postgres", + "-d", + database, + "--format=custom", + "--file=/tmp/genesis.dump", + ], + text=True, + capture_output=True, + check=False, + ) + assert dump_result.returncode == 0, dump_result.stderr + copy_result = subprocess.run( + ["docker", "cp", f"{source_container}:/tmp/genesis.dump", str(genesis_dump)], + text=True, + capture_output=True, + check=False, + ) + assert copy_result.returncode == 0, copy_result.stderr + + genesis_manifest = tmp_path / "genesis-manifest.jsonl" + capture_manifest(source_container, database, genesis_manifest) + + material = replay_material() + material_path = write_json(tmp_path / "entry-0001.json", material) + entry = rebuild._validate_entry_material( + material, + expected_sequence=1, + expected_replay_hash=material["replay_receipt"]["hashes"]["replay_material_sha256"], + material_path=material_path, + material_sha256=sha256_file(material_path), + ) + docker_psql(source_container, database, rebuild.build_seed_sql(entry)) + docker_psql( + source_container, + database, + entry.current_apply_sql, + role="kb_apply", + ) + docker_psql( + source_container, + database, + rebuild.build_applied_timestamp_normalization_sql(entry), + ) + + final_manifest = tmp_path / "final-manifest.jsonl" + capture_manifest(source_container, database, final_manifest) + ledger = { + "artifact": rebuild.LEDGER_ARTIFACT, + "contract_version": rebuild.LEDGER_CONTRACT_VERSION, + "engine": rebuild._engine_hashes(), + "genesis": { + "dump_sha256": sha256_file(genesis_dump), + "parity_manifest_sha256": sha256_file(genesis_manifest), + }, + "entries": [ + { + "sequence": 1, + "material": material_path.name, + "sha256": sha256_file(material_path), + "replay_material_sha256": entry.replay_material_sha256, + } + ], + "final_parity": { + "manifest": final_manifest.name, + "sha256": sha256_file(final_manifest), + }, + } + ledger_path = write_json(tmp_path / "ledger.json", ledger) + output = tmp_path / "reconstruction-receipt.json" + command = [ + sys.executable, + str(Path(rebuild.__file__).resolve()), + "--genesis-dump", + str(genesis_dump), + "--genesis-manifest", + str(genesis_manifest), + "--ledger", + str(ledger_path), + "--ledger-sha256", + sha256_file(ledger_path), + "--database", + database, + "--container-prefix", + "teleo-genesis-ledger-live-test", + "--tmpfs-mb", + "256", + "--max-target-query-ms", + "5000", + "--max-target-source-ratio", + "100", + "--output", + str(output), + ] + completed = subprocess.run( + command, + cwd=rebuild.REPO_ROOT, + text=True, + capture_output=True, + check=False, + timeout=180, + ) + + assert completed.returncode == 0, completed.stderr + completed.stdout + receipt = json.loads(output.read_text(encoding="utf-8")) + assert receipt["status"] == "pass" + assert receipt["genesis_parity"]["status"] == "pass" + assert receipt["ledger"]["entries"][0]["status"] == "pass" + assert receipt["ledger"]["entries"][0]["guarded_apply_executed"] is True + assert receipt["ledger"]["entries"][0]["canonical_rows_exact"] is True + assert receipt["final_parity"]["status"] == "pass" + assert receipt["cleanup"]["container_absent"] is True + assert receipt["safety"]["network_access_available_to_container"] is False + assert receipt["safety"]["private_replay_material_emitted"] is False + assert PRIVATE_MARKER not in completed.stdout + assert PRIVATE_MARKER not in output.read_text(encoding="utf-8") + assert stat.S_IMODE(output.stat().st_mode) == 0o600 + + inspect = subprocess.run( + ["docker", "container", "inspect", receipt["container"]["name"]], + text=True, + capture_output=True, + check=False, + ) + assert inspect.returncode != 0