diff --git a/docs/kb-rebuild-and-recompile.md b/docs/kb-rebuild-and-recompile.md index 2baf932..00996dc 100644 --- a/docs/kb-rebuild-and-recompile.md +++ b/docs/kb-rebuild-and-recompile.md @@ -148,7 +148,7 @@ 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 +## Genesis Plus Strict Ledger: Working Deterministic Slice `ops/run_local_genesis_ledger_rebuild.py` now executes the first exact genesis-plus-ledger slice in one command: @@ -206,27 +206,65 @@ Each referenced material object has exact top-level keys 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 +The command verifies every hash before starting Docker, requires a +SHA-256-pinned Postgres image (and defaults to a pinned PostgreSQL 16 Alpine +multi-platform digest), restores it on tmpfs with `--network none`, reapplies +the current guarded prerequisites, and proves genesis parity. Insert-only +entries seed the receipt's exact canonical row IDs and timestamps before the +existing `kb_apply` payload-bound guard executes. For `revise_strategy`, the +runner seeds only the proposal and approval, captures the target agent's +prestate, executes the real guarded transition, validates the generated delta, +and then replaces only those generated rows with the receipt-pinned IDs and +timestamps. Every path checks exact proposal and canonical row readbacks before +verifying 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 legacy `seed_exact` summary is the insert-only aggregate of +`proposal_seed_exact` and `canonical_seed_exact`; it is intentionally false for +successful mutating entries, which instead report +`mutating_delta_validated` and `mutating_poststate_normalized`. +Fresh guard bootstrap rows use a fixed baseline timestamp rather than wall +clock time, so repeated clean restores have identical row hashes. -The exact v1 claim ceiling is intentionally narrow: +The exact v1 claim ceiling is intentionally bounded: -- `add_edge`, `attach_evidence`, and `approve_claim` strict receipts execute; +- `add_edge`, `attach_evidence`, `approve_claim`, and `revise_strategy` 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; +- `revise_strategy` is accepted only when the receipt pins the exact SQL that + matches the current guarded apply engine. Immediately before each entry, the + runner captures the target agent's strategy/node IDs, active strategy, + non-retired nodes, and maximum version. It then validates the generated + post-minus-pre delta, requires `version = previous maximum + 1`, and replaces + only the generated strategy/node rows with the exact receipt rows; +- the original transaction timestamp is derived from the fresh strategy + `created_at` and must equal every fresh node's `created_at` and `updated_at`. + Only node IDs observed as non-retired before apply receive that timestamp; + already-retired, unrelated-agent, and shared NULL-agent rows stay untouched; +- generated nodes must have no anchors before normalization, preventing a + delete-and-reinsert step from silently cascading future trigger-created rows; - 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 transition contract for `revise_strategy` is final-state deterministic, not +a claim that the v1 receipt independently contains a historical before-image: + +```text +exact genesis/pre-entry state + + exact current/original guarded SQL + + receipt-pinned fresh strategy and nodes + -> prior active strategy inactive + -> exactly the prior non-retired nodes retired at the original transaction time + -> one receipt-exact active strategy and receipt-exact node set +``` + +The genesis and final manifests remain mandatory oracles. Any incorrect +prestate, unrelated-row mutation, missing/extra generated row, semantic drift, +version drift, hash drift, or final rowset difference fails the reconstruction. + The source compiler now turns one raw artifact, its strict UTF-8 extraction, and a reviewed extraction manifest into a deterministic, hash-bound `pending_review` proposal bundle: @@ -272,10 +310,9 @@ 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 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 +The remaining reconstruction work is to backfill or explicitly reject legacy +freeform applies and extend beyond genesis recovery to a blank-schema source +compiler. The strict 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 index ef871a3..a3361e1 100644 --- a/ops/run_local_genesis_ledger_rebuild.py +++ b/ops/run_local_genesis_ledger_rebuild.py @@ -1,12 +1,14 @@ #!/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. +The v1 replay boundary supports strict ``add_edge``, ``attach_evidence``, +``approve_claim``, and ``revise_strategy`` receipts. Insert-only actions seed +their receipt-pinned rows before the guarded transition. ``revise_strategy`` +instead captures the exact prestate identities, executes the exact hash-matched +guarded SQL, validates the generated delta, and normalizes only that delta to the +receipt-pinned transaction timestamp, IDs, and rows. Every entry must retain the +full approved and applied proposal rows plus its immutable approval snapshot. +Legacy and freeform entries still fail before Docker starts. """ from __future__ import annotations @@ -46,15 +48,21 @@ 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"}) +DEFAULT_REBUILD_IMAGE = ( + "docker.io/library/postgres:16-alpine@" + "sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777" +) +SUPPORTED_PROPOSAL_TYPES = frozenset( + {"add_edge", "attach_evidence", "approve_claim", "revise_strategy"} +) 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" + "exact genesis plus ordered strict proposal replay; supported types are add_edge, " + "attach_evidence, approve_claim, and revise_strategy; mutating strategy replay " + "captures prestate identities and normalizes the exact receipt-pinned poststate; " + "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", ) @@ -77,6 +85,7 @@ FIXED_ENGINE_PATHS = frozenset( ) SHA256_RE = re.compile(r"[0-9a-f]{64}\Z") +IMAGE_DIGEST_RE = re.compile(r"\S+@sha256:[0-9a-f]{64}\Z") PROPOSAL_TRANSITION_FIELDS = frozenset( {"status", "applied_by_handle", "applied_by_agent_id", "applied_at", "updated_at"} ) @@ -121,6 +130,35 @@ CANONICAL_TABLE_ORDER = ( "claim_evidence", "claim_edges", ) +STRATEGY_REQUIRED_FIELDS = frozenset( + { + "id", + "agent_id", + "diagnosis", + "guiding_policy", + "proximate_objectives", + "version", + "active", + "created_at", + } +) +STRATEGY_NODE_REQUIRED_FIELDS = frozenset( + { + "id", + "agent_id", + "node_type", + "title", + "body", + "rank", + "status", + "horizon", + "measure", + "source_ref", + "metadata", + "created_at", + "updated_at", + } +) class ReconstructionError(RuntimeError): @@ -308,6 +346,119 @@ def _validate_proposal_rows( ) +def _parse_aware_timestamp( + value: Any, field: str, *, code: str = "invalid_mutating_receipt" +) -> datetime: + if not isinstance(value, str) or not value: + raise ReconstructionError(code, f"{field} must be an ISO-8601 timestamp") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ReconstructionError(code, f"{field} must be an ISO-8601 timestamp") from exc + if parsed.tzinfo is None: + raise ReconstructionError(code, f"{field} must include a timezone") + return parsed + + +def _validated_uuid_list( + value: Any, field: str, *, code: str = "invalid_mutating_state" +) -> list[str]: + if not isinstance(value, list): + raise ReconstructionError(code, f"{field} must be a UUID list") + normalized: list[str] = [] + for item in value: + try: + normalized.append(str(uuid.UUID(str(item)))) + except (TypeError, ValueError, AttributeError) as exc: + raise ReconstructionError(code, f"{field} must contain only UUIDs") from exc + if len(normalized) != len(set(normalized)): + raise ReconstructionError(code, f"{field} contains duplicate UUIDs") + return normalized + + +def _validate_revise_strategy_receipt_rows( + proposal: dict[str, Any], canonical_rows: dict[str, Any] +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + payload = proposal.get("payload") + apply_payload = payload.get("apply_payload") if isinstance(payload, dict) else None + if not isinstance(apply_payload, dict): + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy receipt requires a strict apply payload" + ) + try: + agent_id = str(uuid.UUID(str(apply_payload.get("agent_id")))) + except (TypeError, ValueError, AttributeError) as exc: + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy receipt requires one agent UUID" + ) from exc + + strategies = canonical_rows.get("strategies") + nodes = canonical_rows.get("strategy_nodes") + if not isinstance(strategies, list) or len(strategies) != 1 or not isinstance(strategies[0], dict): + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy receipt must pin one fresh strategy row" + ) + expected_nodes = apply_payload.get("strategy_nodes") + if ( + not isinstance(expected_nodes, list) + or not isinstance(nodes, list) + or len(nodes) != len(expected_nodes) + or any(not isinstance(row, dict) for row in nodes) + ): + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy receipt must pin every fresh strategy node row" + ) + + strategy = strategies[0] + if frozenset(strategy) != STRATEGY_REQUIRED_FIELDS: + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy receipt strategy row fields are not canonical" + ) + if any(frozenset(row) != STRATEGY_NODE_REQUIRED_FIELDS for row in nodes): + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy receipt strategy node row fields are not canonical" + ) + strategy_id = _validated_uuid_list( + [strategy.get("id")], "receipt strategy id", code="invalid_mutating_receipt" + )[0] + node_ids = _validated_uuid_list( + [row.get("id") for row in nodes], + "receipt strategy node ids", + code="invalid_mutating_receipt", + ) + if strategy.get("agent_id") != agent_id or any(row.get("agent_id") != agent_id for row in nodes): + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy receipt rows do not belong to the payload agent" + ) + if strategy.get("active") is not True or any(row.get("status") != "active" for row in nodes): + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy receipt must pin the fresh active poststate" + ) + version = strategy.get("version") + if isinstance(version, bool) or not isinstance(version, int) or version < 1: + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy receipt strategy version must be positive" + ) + + transaction_timestamp = strategy.get("created_at") + transaction_time = _parse_aware_timestamp(transaction_timestamp, "receipt strategy created_at") + for row in nodes: + if row.get("created_at") != transaction_timestamp or row.get("updated_at") != transaction_timestamp: + raise ReconstructionError( + "invalid_mutating_receipt", + "revise_strategy fresh strategy and node timestamps must share one transaction timestamp", + ) + applied_time = _parse_aware_timestamp(proposal.get("applied_at"), "receipt proposal applied_at") + if transaction_time > applied_time: + raise ReconstructionError( + "invalid_mutating_receipt", "revise_strategy transaction timestamp is after proposal apply time" + ) + + return {**strategy, "id": strategy_id}, [ + {**row, "id": normalized_id} for row, normalized_id in zip(nodes, node_ids, strict=True) + ] + + def _validate_entry_material( material: dict[str, Any], *, @@ -340,11 +491,6 @@ def _validate_entry_material( 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" @@ -358,6 +504,8 @@ def _validate_entry_material( 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") + if proposal_type == "revise_strategy": + _validate_revise_strategy_receipt_rows(proposal, canonical_rows) try: rebuilt = replay_receipt.build_replay_receipt( proposal, @@ -382,6 +530,7 @@ def _validate_entry_material( raise ReconstructionError( "replay_material_hash_mismatch", "ledger replay-material pin does not match the private receipt" ) + receipt = {**receipt, "canonical_rows": rebuilt["canonical_rows"]} approved = material["approved_proposal"] approval = material["approval_snapshot"] @@ -394,6 +543,15 @@ def _validate_entry_material( "current_apply_engine_rejected_material", "current guarded apply engine rejected the strict replay material", ) from exc + current_apply_sql_sha256 = _sha256_text(current_apply_sql) + if proposal_type == "revise_strategy" and ( + apply_metadata.get("source") != "exact_executed_sql" + or apply_metadata.get("apply_sql_sha256") != current_apply_sql_sha256 + ): + raise ReconstructionError( + "mutating_apply_engine_mismatch", + "revise_strategy replay requires the exact executed SQL to match the current guarded apply engine", + ) return LedgerEntry( sequence=expected_sequence, @@ -405,7 +563,7 @@ def _validate_entry_material( applied_proposal=applied, receipt=receipt, current_apply_sql=current_apply_sql, - current_apply_sql_sha256=_sha256_text(current_apply_sql), + current_apply_sql_sha256=current_apply_sql_sha256, ) @@ -550,6 +708,280 @@ def build_seed_sql(entry: LedgerEntry) -> str: return "\n".join(statements) + "\n" +def _uuid_membership_sql(column: str, values: list[str], *, negate: bool = False) -> str: + if not values: + return "true" if negate else "false" + rendered = ", ".join(f"{apply_engine.sql_literal(value)}::uuid" for value in values) + operator = "not in" if negate else "in" + return f"{column} {operator} ({rendered})" + + +def _uuid_array_sql(values: list[str]) -> str: + if not values: + return "array[]::uuid[]" + rendered = ", ".join(f"{apply_engine.sql_literal(value)}::uuid" for value in values) + return f"array[{rendered}]::uuid[]" + + +def build_revise_strategy_prestate_sql(entry: LedgerEntry) -> str: + apply_payload = entry.receipt["proposal"]["payload"]["apply_payload"] + agent = apply_engine.sql_literal(apply_payload["agent_id"]) + return f"""with strategy_state as ( + select coalesce(jsonb_agg(s.id::text order by s.id::text), '[]'::jsonb) as strategy_ids, + coalesce( + jsonb_agg(s.id::text order by s.id::text) filter (where s.active), + '[]'::jsonb + ) as active_strategy_ids, + coalesce(max(s.version), 0) as max_strategy_version + from public.strategies s + where s.agent_id = {agent}::uuid +), node_state as ( + select coalesce(jsonb_agg(n.id::text order by n.id::text), '[]'::jsonb) as strategy_node_ids, + coalesce( + jsonb_agg(n.id::text order by n.id::text) filter (where n.status <> 'retired'), + '[]'::jsonb + ) as non_retired_strategy_node_ids + from public.strategy_nodes n + where n.agent_id = {agent}::uuid +) +select jsonb_build_object( + 'strategy_ids', s.strategy_ids, + 'active_strategy_ids', s.active_strategy_ids, + 'max_strategy_version', s.max_strategy_version, + 'strategy_node_ids', n.strategy_node_ids, + 'non_retired_strategy_node_ids', n.non_retired_strategy_node_ids +)::text +from strategy_state s cross join node_state n; +""" + + +def _validate_revise_strategy_prestate(value: Any) -> dict[str, Any]: + expected = frozenset( + { + "strategy_ids", + "active_strategy_ids", + "max_strategy_version", + "strategy_node_ids", + "non_retired_strategy_node_ids", + } + ) + state = _require_exact_keys(value, expected, "revise_strategy prestate") + strategy_ids = _validated_uuid_list(state["strategy_ids"], "prestate strategy ids") + active_strategy_ids = _validated_uuid_list( + state["active_strategy_ids"], "prestate active strategy ids" + ) + strategy_node_ids = _validated_uuid_list(state["strategy_node_ids"], "prestate strategy node ids") + non_retired_node_ids = _validated_uuid_list( + state["non_retired_strategy_node_ids"], "prestate non-retired strategy node ids" + ) + max_version = state["max_strategy_version"] + if isinstance(max_version, bool) or not isinstance(max_version, int) or max_version < 0: + raise ReconstructionError( + "invalid_mutating_state", "prestate maximum strategy version must be a non-negative integer" + ) + if len(active_strategy_ids) > 1 or not set(active_strategy_ids).issubset(strategy_ids): + raise ReconstructionError( + "invalid_mutating_state", "prestate active strategies violate the one-active invariant" + ) + if not set(non_retired_node_ids).issubset(strategy_node_ids): + raise ReconstructionError( + "invalid_mutating_state", "prestate non-retired strategy nodes are not a subset of all nodes" + ) + return { + "strategy_ids": strategy_ids, + "active_strategy_ids": active_strategy_ids, + "max_strategy_version": max_version, + "strategy_node_ids": strategy_node_ids, + "non_retired_strategy_node_ids": non_retired_node_ids, + } + + +def build_revise_strategy_generated_delta_sql(entry: LedgerEntry, prestate: dict[str, Any]) -> str: + state = _validate_revise_strategy_prestate(prestate) + apply_payload = entry.receipt["proposal"]["payload"]["apply_payload"] + agent = apply_engine.sql_literal(apply_payload["agent_id"]) + strategy_filter = _uuid_membership_sql("s.id", state["strategy_ids"], negate=True) + node_filter = _uuid_membership_sql("n.id", state["strategy_node_ids"], negate=True) + return f"""select jsonb_build_object( + 'strategies', coalesce(( + select jsonb_agg(to_jsonb(s) order by s.id::text) + from public.strategies s + where s.agent_id = {agent}::uuid and {strategy_filter} + ), '[]'::jsonb), + 'strategy_nodes', coalesce(( + select jsonb_agg(to_jsonb(n) order by n.id::text) + from public.strategy_nodes n + where n.agent_id = {agent}::uuid and {node_filter} + ), '[]'::jsonb) +)::text; +""" + + +def _validate_revise_strategy_generated_delta( + entry: LedgerEntry, + prestate: dict[str, Any], + generated_rows: Any, +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + state = _validate_revise_strategy_prestate(prestate) + if not isinstance(generated_rows, dict): + raise ReconstructionError( + "invalid_mutating_delta", "revise_strategy generated delta must be a row object" + ) + try: + replay_receipt.build_replay_receipt( + entry.receipt["proposal"], + generated_rows, + apply_sql_sha256=entry.current_apply_sql_sha256, + apply_sql_source="exact_executed_sql", + exported_at_utc=entry.receipt.get("exported_at_utc"), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ReconstructionError( + "invalid_mutating_delta", + "revise_strategy generated rows do not match the strict payload", + ) from exc + + strategies = generated_rows.get("strategies") + nodes = generated_rows.get("strategy_nodes") + if not isinstance(strategies, list) or len(strategies) != 1 or not isinstance(strategies[0], dict): + raise ReconstructionError( + "invalid_mutating_delta", "revise_strategy apply did not generate exactly one strategy" + ) + if not isinstance(nodes, list) or any(not isinstance(row, dict) for row in nodes): + raise ReconstructionError( + "invalid_mutating_delta", "revise_strategy apply generated invalid strategy nodes" + ) + strategy = strategies[0] + if strategy.get("version") != state["max_strategy_version"] + 1: + raise ReconstructionError( + "invalid_mutating_delta", "revise_strategy generated an unexpected strategy version" + ) + transaction_timestamp = strategy.get("created_at") + _parse_aware_timestamp( + transaction_timestamp, "generated strategy created_at", code="invalid_mutating_delta" + ) + if any( + row.get("created_at") != transaction_timestamp or row.get("updated_at") != transaction_timestamp + for row in nodes + ): + raise ReconstructionError( + "invalid_mutating_delta", + "revise_strategy generated rows do not share one transaction timestamp", + ) + _validated_uuid_list( + [strategy.get("id")], "generated strategy id", code="invalid_mutating_delta" + ) + _validated_uuid_list( + [row.get("id") for row in nodes], + "generated strategy node ids", + code="invalid_mutating_delta", + ) + return strategy, nodes + + +def build_revise_strategy_normalization_sql( + entry: LedgerEntry, + prestate: dict[str, Any], + generated_rows: dict[str, Any], +) -> str: + state = _validate_revise_strategy_prestate(prestate) + generated_strategy, generated_nodes = _validate_revise_strategy_generated_delta( + entry, state, generated_rows + ) + receipt_strategy, receipt_nodes = _validate_revise_strategy_receipt_rows( + entry.receipt["proposal"], entry.receipt["canonical_rows"] + ) + if receipt_strategy["version"] != state["max_strategy_version"] + 1: + raise ReconstructionError( + "mutating_receipt_version_mismatch", + "revise_strategy receipt version does not follow the observed prestate", + ) + if receipt_strategy["id"] in state["strategy_ids"] or set( + row["id"] for row in receipt_nodes + ).intersection(state["strategy_node_ids"]): + raise ReconstructionError( + "mutating_receipt_id_collision", "revise_strategy receipt IDs collide with the observed prestate" + ) + + apply_payload = entry.receipt["proposal"]["payload"]["apply_payload"] + agent = apply_engine.sql_literal(apply_payload["agent_id"]) + generated_strategy_id = str(uuid.UUID(str(generated_strategy["id"]))) + generated_node_ids = [str(uuid.UUID(str(row["id"]))) for row in generated_nodes] + previous_active = state["active_strategy_ids"] + changed_node_ids = state["non_retired_strategy_node_ids"] + transaction_timestamp = apply_engine.sql_literal(receipt_strategy["created_at"]) + generated_node_array = _uuid_array_sql(generated_node_ids) + previous_active_array = _uuid_array_sql(previous_active) + changed_node_array = _uuid_array_sql(changed_node_ids) + + prior_strategy_assertion = "" + if previous_active: + prior_strategy_assertion = f""" + if (select count(*) from public.strategies + where agent_id = {agent}::uuid and id = any(previous_active_ids) and not active) <> {len(previous_active)} then + raise exception 'revise_strategy prior active strategy transition mismatch'; + end if;""" + prior_node_normalization = "" + if changed_node_ids: + prior_node_normalization = f""" + if (select count(*) from public.strategy_nodes + where agent_id = {agent}::uuid and id = any(changed_node_ids) and status = 'retired') <> {len(changed_node_ids)} then + raise exception 'revise_strategy prior node transition mismatch'; + end if; + update public.strategy_nodes + set status = 'retired', updated_at = {transaction_timestamp}::timestamptz + where agent_id = {agent}::uuid and id = any(changed_node_ids); + get diagnostics changed_rows = row_count; + if changed_rows <> {len(changed_node_ids)} then + raise exception 'revise_strategy prior node normalization mismatch'; + end if;""" + + return f"""begin; +set local standard_conforming_strings = on; +do $reconstruction$ +declare + changed_rows integer; + generated_strategy_id constant uuid := {apply_engine.sql_literal(generated_strategy_id)}::uuid; + generated_node_ids constant uuid[] := {generated_node_array}; + previous_active_ids constant uuid[] := {previous_active_array}; + changed_node_ids constant uuid[] := {changed_node_array}; +begin{prior_strategy_assertion} + if exists ( + select 1 from public.strategy_node_anchors + where from_node_id = any(generated_node_ids) or to_node_id = any(generated_node_ids) + ) then + raise exception 'revise_strategy generated nodes unexpectedly have anchors'; + end if; + delete from public.strategy_nodes + where agent_id = {agent}::uuid and id = any(generated_node_ids); + get diagnostics changed_rows = row_count; + if changed_rows <> {len(generated_node_ids)} then + raise exception 'revise_strategy generated node delta mismatch'; + end if; + delete from public.strategies + where agent_id = {agent}::uuid and id = generated_strategy_id; + get diagnostics changed_rows = row_count; + if changed_rows <> 1 then + raise exception 'revise_strategy generated strategy delta mismatch'; + end if;{prior_node_normalization} + insert into public.strategies + select seeded.* from jsonb_populate_record( + null::public.strategies, {_jsonb_literal(receipt_strategy)} + ) seeded; + insert into public.strategy_nodes + select seeded.* from jsonb_populate_recordset( + null::public.strategy_nodes, {_jsonb_literal(receipt_nodes)} + ) seeded; + if (select count(*) from public.strategies + where agent_id = {agent}::uuid and active) <> 1 then + raise exception 'revise_strategy normalized one-active invariant mismatch'; + end if; +end +$reconstruction$; +commit; +""" + + def build_proposal_readback_sql(proposal_id: str) -> str: literal = apply_engine.sql_literal(proposal_id) return f"""select jsonb_build_object( @@ -694,7 +1126,12 @@ def _entry_summary(entry: LedgerEntry) -> dict[str, Any]: "current_apply_sql_sha256": entry.current_apply_sql_sha256, "expected_row_counts": entry.receipt["expected_row_counts"], "seed_exact": False, + "proposal_seed_exact": False, + "canonical_seed_exact": False, "guarded_apply_executed": False, + "mutating_prestate_captured": False, + "mutating_delta_validated": False, + "mutating_poststate_normalized": False, "proposal_exact": False, "canonical_rows_exact": False, "status": "pending", @@ -707,6 +1144,7 @@ def apply_entry( entry: LedgerEntry, summary: dict[str, Any], ) -> None: + mutating_prestate: dict[str, Any] | None = None try: _psql( args, @@ -733,20 +1171,34 @@ def apply_entry( 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 + summary["proposal_seed_exact"] = True + if entry.applied_proposal["proposal_type"] == "revise_strategy": + mutating_prestate = _validate_revise_strategy_prestate( + _read_json( + args, + container, + build_revise_strategy_prestate_sql(entry), + code="mutating_prestate_readback_failed", + action=f"ledger entry {entry.sequence} revise_strategy prestate readback", + ) + ) + summary["mutating_prestate_captured"] = True + else: + 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["canonical_seed_exact"] = True + summary["seed_exact"] = summary["proposal_seed_exact"] and summary["canonical_seed_exact"] _psql( args, @@ -759,6 +1211,29 @@ def apply_entry( ) summary["guarded_apply_executed"] = True + if mutating_prestate is not None: + generated_rows = _read_json( + args, + container, + build_revise_strategy_generated_delta_sql(entry, mutating_prestate), + code="mutating_delta_readback_failed", + action=f"ledger entry {entry.sequence} revise_strategy generated delta readback", + ) + normalization_sql = build_revise_strategy_normalization_sql( + entry, mutating_prestate, generated_rows + ) + summary["mutating_delta_validated"] = True + _psql( + args, + container, + normalization_sql, + role="postgres", + timeout=args.apply_timeout, + code="mutating_poststate_normalization_failed", + action=f"ledger entry {entry.sequence} revise_strategy exact poststate normalization", + ) + summary["mutating_poststate_normalized"] = True + _psql( args, container, @@ -1199,7 +1674,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: 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("--image", default=DEFAULT_REBUILD_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) @@ -1218,8 +1693,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: 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 IMAGE_DIGEST_RE.fullmatch(args.image): + parser.error("--image must be pinned by a lowercase sha256 digest") 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: diff --git a/scripts/kb_apply_prereqs.sql b/scripts/kb_apply_prereqs.sql index 9779c5a..839f6b5 100644 --- a/scripts/kb_apply_prereqs.sql +++ b/scripts/kb_apply_prereqs.sql @@ -315,8 +315,9 @@ alter table kb_stage.kb_review_principals owner to kb_gate_owner; alter table kb_stage.kb_proposal_approvals owner to kb_gate_owner; insert into kb_stage.kb_review_principals - (db_role, reviewed_by_handle, reviewed_by_agent_id, active) -select 'kb_review'::name, a.handle, a.id, true + (db_role, reviewed_by_handle, reviewed_by_agent_id, active, created_at) +select 'kb_review'::name, a.handle, a.id, true, + '1970-01-01 00:00:00+00'::timestamptz from public.agents a where a.handle = 'm3ta' on conflict (db_role) do nothing; diff --git a/tests/test_kb_apply_prereqs.py b/tests/test_kb_apply_prereqs.py index c4dbd2e..e66395c 100644 --- a/tests/test_kb_apply_prereqs.py +++ b/tests/test_kb_apply_prereqs.py @@ -203,6 +203,15 @@ def _lines(completed: subprocess.CompletedProcess[str]) -> set[str]: return {line for line in completed.stdout.splitlines() if line} +def test_reviewer_principal_seed_timestamp_is_deterministic(migrated_postgres: str) -> None: + created_at = _psql( + migrated_postgres, + "select created_at::text from kb_stage.kb_review_principals where db_role = 'kb_review';", + ).stdout.strip() + + assert created_at == "1970-01-01 00:00:00+00" + + def _catalog_snapshot(container: str) -> str: return _psql( container, diff --git a/tests/test_run_local_genesis_ledger_rebuild.py b/tests/test_run_local_genesis_ledger_rebuild.py index 4919b1c..721772c 100644 --- a/tests/test_run_local_genesis_ledger_rebuild.py +++ b/tests/test_run_local_genesis_ledger_rebuild.py @@ -23,6 +23,18 @@ 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" +OTHER_AGENT_ID = "22222222-2222-4222-8222-222222222222" +REVISE_PROPOSAL_ID = "88888888-8888-4888-8888-888888888888" +PRIOR_STRATEGY_ID = "33333333-3333-4333-8333-333333333333" +PRIOR_ACTIVE_NODE_ID = "55555555-5555-4555-8555-555555555555" +PRIOR_RETIRED_NODE_ID = "66666666-6666-4666-8666-666666666666" +OTHER_STRATEGY_ID = "77777777-7777-4777-8777-777777777777" +OTHER_ACTIVE_NODE_ID = "12121212-1212-4212-8212-121212121212" +NULL_AGENT_NODE_ID = "13131313-1313-4313-8313-131313131313" +PRIOR_NODE_ANCHOR_ID = "19191919-1919-4919-8919-191919191919" +RECEIPT_STRATEGY_ID = "14141414-1414-4414-8414-141414141414" +RECEIPT_NODE_A_ID = "15151515-1515-4515-8515-151515151515" +RECEIPT_NODE_B_ID = "16161616-1616-4616-8616-161616161616" PRIVATE_MARKER = "PRIVATE-REPLAY-MATERIAL-MUST-NOT-LEAK" @@ -155,6 +167,58 @@ def proposal_rows() -> tuple[dict, dict, dict]: return approved, applied, approval +def revise_strategy_proposal_rows() -> tuple[dict, dict, dict]: + approved, _, approval = proposal_rows() + payload = { + "apply_payload": { + "agent_id": REVIEWER_ID, + "strategy": { + "diagnosis": "Deterministic reconstruction needs exact mutation deltas.", + "guiding_policy": "Replay every guarded transition and fail closed on drift.", + "proximate_objectives": ["exact parity", "zero orphan containers"], + }, + "strategy_nodes": [ + { + "node_type": "policy", + "title": "Replay exactly", + "body": "Bind the guarded transition to its canonical receipt.", + "rank": 1, + }, + { + "node_type": "policy", + "title": "Replay exactly", + "body": "Bind the guarded transition to its canonical receipt.", + "rank": 1, + }, + ], + }, + "private_title": PRIVATE_MARKER, + } + approved = { + **approved, + "id": REVISE_PROPOSAL_ID, + "proposal_type": "revise_strategy", + "payload": payload, + "created_at": "2026-07-14T00:58: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 = { + **approval, + "proposal_id": REVISE_PROPOSAL_ID, + "proposal_type": "revise_strategy", + "payload": payload, + } + return approved, applied, approval + + def applied_envelope(applied: dict) -> dict: fields = ( "id", @@ -202,6 +266,62 @@ def replay_material() -> dict: } +def revise_strategy_replay_material( + *, apply_sql_source: str = "exact_executed_sql", version: int = 2 +) -> dict: + approved, applied, approval = revise_strategy_proposal_rows() + proposal = applied_envelope(applied) + transaction_timestamp = "2026-07-14T01:00:30+00:00" + canonical_rows = { + "strategies": [ + { + "id": RECEIPT_STRATEGY_ID, + "agent_id": REVIEWER_ID, + **proposal["payload"]["apply_payload"]["strategy"], + "version": version, + "active": True, + "created_at": transaction_timestamp, + } + ], + "strategy_nodes": [ + { + "id": node_id, + "agent_id": REVIEWER_ID, + **node, + "status": "active", + "horizon": None, + "measure": None, + "source_ref": None, + "metadata": {}, + "created_at": transaction_timestamp, + "updated_at": transaction_timestamp, + } + for node_id, node in zip( + (RECEIPT_NODE_A_ID, RECEIPT_NODE_B_ID), + proposal["payload"]["apply_payload"]["strategy_nodes"], + strict=True, + ) + ], + } + apply_sql = rebuild.apply_engine.build_apply_sql(proposal, applied["applied_by_handle"]) + receipt = rebuild.replay_receipt.build_replay_receipt( + proposal, + canonical_rows, + apply_sql_sha256=rebuild.replay_receipt.sha256_text(apply_sql), + apply_sql_source=apply_sql_source, + 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") @@ -238,7 +358,7 @@ def write_bundle(tmp_path: Path, *, material: dict | None = None) -> argparse.Na ledger=ledger_path, ledger_sha256=sha256_file(ledger_path), database="teleo", - image=rebuild.canonical_rebuild.DEFAULT_IMAGE, + image=rebuild.DEFAULT_REBUILD_IMAGE, container_prefix="teleo-genesis-ledger-test", tmpfs_mb=256, startup_timeout=30.0, @@ -302,18 +422,115 @@ def test_dump_hash_mismatch_fails_before_container_start(tmp_path: Path, monkeyp 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" +def test_revise_strategy_material_validates_exact_engine_and_canonicalizes_row_order( + tmp_path: Path, +) -> None: + material = revise_strategy_replay_material() + material["replay_receipt"]["canonical_rows"]["strategy_nodes"].reverse() args = write_bundle(tmp_path, material=material) - with pytest.raises(rebuild.ReconstructionError, match="prior strategy") as exc_info: + entry = rebuild.load_bundle(args).entries[0] + + assert entry.applied_proposal["proposal_type"] == "revise_strategy" + assert entry.receipt["apply_engine"]["source"] == "exact_executed_sql" + assert entry.receipt["apply_engine"]["apply_sql_sha256"] == entry.current_apply_sql_sha256 + assert entry.receipt["canonical_rows"]["strategy_nodes"] == sorted( + entry.receipt["canonical_rows"]["strategy_nodes"], key=rebuild._canonical_json + ) + + +def test_revise_strategy_material_rejects_reconstructed_apply_engine(tmp_path: Path) -> None: + args = write_bundle( + tmp_path, + material=revise_strategy_replay_material(apply_sql_source="reconstructed_current_engine"), + ) + + with pytest.raises(rebuild.ReconstructionError) as exc_info: rebuild.load_bundle(args) - assert exc_info.value.code == "unsupported_mutating_receipt" + assert exc_info.value.code == "mutating_apply_engine_mismatch" + + +@pytest.mark.parametrize( + "invalid_case", + ("extra_strategy_field", "duplicate_node_id", "node_timestamp_drift", "transaction_after_apply"), +) +def test_revise_strategy_receipt_rejects_impossible_poststate( + tmp_path: Path, invalid_case: str +) -> None: + material = revise_strategy_replay_material() + rows = material["replay_receipt"]["canonical_rows"] + if invalid_case == "extra_strategy_field": + rows["strategies"][0]["unexpected"] = "ignored by jsonb_populate_record" + elif invalid_case == "duplicate_node_id": + rows["strategy_nodes"][1]["id"] = rows["strategy_nodes"][0]["id"] + elif invalid_case == "node_timestamp_drift": + rows["strategy_nodes"][0]["updated_at"] = "2026-07-14T01:00:31+00:00" + else: + rows["strategies"][0]["created_at"] = "2026-07-14T01:01:01+00:00" + for row in rows["strategy_nodes"]: + row["created_at"] = "2026-07-14T01:01:01+00:00" + row["updated_at"] = "2026-07-14T01:01:01+00:00" + 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_mutating_receipt" + + +def test_revise_strategy_transition_builders_preserve_historical_prestate(tmp_path: Path) -> None: + entry = rebuild.load_bundle( + write_bundle(tmp_path, material=revise_strategy_replay_material()) + ).entries[0] + prestate = { + "strategy_ids": [PRIOR_STRATEGY_ID], + "active_strategy_ids": [PRIOR_STRATEGY_ID], + "max_strategy_version": 1, + "strategy_node_ids": [PRIOR_ACTIVE_NODE_ID, PRIOR_RETIRED_NODE_ID], + "non_retired_strategy_node_ids": [PRIOR_ACTIVE_NODE_ID], + } + generated_rows = copy.deepcopy(entry.receipt["canonical_rows"]) + generated_rows["strategies"][0]["id"] = "17171717-1717-4717-8717-171717171717" + generated_rows["strategies"][0]["created_at"] = "2026-07-15T01:00:30+00:00" + for index, row in enumerate(generated_rows["strategy_nodes"], start=1): + row["id"] = f"18181818-1818-4818-8818-18181818181{index}" + row["created_at"] = "2026-07-15T01:00:30+00:00" + row["updated_at"] = "2026-07-15T01:00:30+00:00" + + sql = rebuild.build_revise_strategy_normalization_sql(entry, prestate, generated_rows) + + assert PRIOR_ACTIVE_NODE_ID in sql + assert PRIOR_RETIRED_NODE_ID not in sql + assert PRIOR_STRATEGY_ID in sql + assert RECEIPT_STRATEGY_ID in sql + assert "updated_at = E'2026-07-14T01:00:30+00:00'::timestamptz" in sql + + +def test_revise_strategy_transition_builders_allow_empty_prestate(tmp_path: Path) -> None: + entry = rebuild.load_bundle( + write_bundle(tmp_path, material=revise_strategy_replay_material(version=1)) + ).entries[0] + prestate = { + "strategy_ids": [], + "active_strategy_ids": [], + "max_strategy_version": 0, + "strategy_node_ids": [], + "non_retired_strategy_node_ids": [], + } + generated_rows = copy.deepcopy(entry.receipt["canonical_rows"]) + generated_rows["strategies"][0]["id"] = "17171717-1717-4717-8717-171717171717" + generated_rows["strategies"][0]["created_at"] = "2026-07-15T01:00:30+00:00" + for index, row in enumerate(generated_rows["strategy_nodes"], start=1): + row["id"] = f"18181818-1818-4818-8818-18181818181{index}" + row["created_at"] = "2026-07-15T01:00:30+00:00" + row["updated_at"] = "2026-07-15T01:00:30+00:00" + + sql = rebuild.build_revise_strategy_normalization_sql(entry, prestate, generated_rows) + + assert "prior active strategy transition mismatch" not in sql + assert "prior node normalization mismatch" not in sql + assert RECEIPT_STRATEGY_ID in sql def test_legacy_freeform_receipt_fails_closed(tmp_path: Path) -> None: @@ -331,6 +548,17 @@ def test_legacy_freeform_receipt_fails_closed(tmp_path: Path) -> None: assert PRIVATE_MARKER not in exc_info.value.safe_message +def test_unknown_proposal_operation_fails_closed_before_replay(tmp_path: Path) -> None: + material = replay_material() + material["replay_receipt"]["proposal"]["proposal_type"] = "delete_claim" + args = write_bundle(tmp_path, material=material) + + with pytest.raises(rebuild.ReconstructionError) as exc_info: + rebuild.load_bundle(args) + + assert exc_info.value.code == "unsupported_proposal_type" + + def test_proposal_metadata_outside_apply_transition_is_rejected(tmp_path: Path) -> None: material = replay_material() material["applied_proposal"]["rationale"] = "silently changed after review" @@ -354,6 +582,28 @@ def test_output_cannot_overwrite_a_fixed_guard_or_engine_file(tmp_path: Path) -> assert exc_info.value.code == "unsafe_output_path" +def test_cli_defaults_to_pinned_image_and_rejects_moving_tag(tmp_path: Path) -> None: + bundle = write_bundle(tmp_path) + argv = [ + "--genesis-dump", + str(bundle.genesis_dump), + "--genesis-manifest", + str(bundle.genesis_manifest), + "--ledger", + str(bundle.ledger), + "--ledger-sha256", + bundle.ledger_sha256, + "--output", + str(tmp_path / "receipt.json"), + ] + + assert rebuild.parse_args(argv).image == rebuild.DEFAULT_REBUILD_IMAGE + with pytest.raises(SystemExit) as exc_info: + rebuild.parse_args([*argv, "--image", "postgres:16-alpine"]) + + assert exc_info.value.code == 2 + + def test_unknown_private_field_name_is_not_echoed_in_public_error(tmp_path: Path) -> None: material = replay_material() material[PRIVATE_MARKER] = "private value" @@ -484,17 +734,18 @@ create table public.reasoning_tools ( created_at timestamptz not null default now() ); create table public.strategies ( - id uuid primary key, + id uuid primary key default gen_random_uuid(), agent_id uuid not null references public.agents(id), diagnosis text, guiding_policy text, proximate_objectives jsonb, version integer not null default 1, active boolean not null default true, - created_at timestamptz not null default now() + created_at timestamptz not null default now(), + unique (agent_id, version) ); create table public.strategy_nodes ( - id uuid primary key, + id uuid primary key default gen_random_uuid(), agent_id uuid references public.agents(id), node_type text, title text, @@ -508,6 +759,20 @@ create table public.strategy_nodes ( created_at timestamptz not null default now(), updated_at timestamptz not null default now() ); +create table public.strategy_node_anchors ( + id uuid primary key default gen_random_uuid(), + from_node_id uuid not null references public.strategy_nodes(id) on delete cascade, + anchor_role text not null, + to_node_id uuid references public.strategy_nodes(id) on delete cascade, + to_shared_root_id uuid, + belief_id uuid, + claim_id uuid, + source_id uuid, + weight numeric, + note text, + created_at timestamptz not null default now(), + check (num_nonnulls(to_node_id, to_shared_root_id, belief_id, claim_id, source_id) = 1) +); create table kb_stage.kb_proposals ( id uuid primary key, proposal_type text not null, @@ -530,10 +795,34 @@ create table kb_stage.kb_proposals ( ); insert into public.agents (id, handle, kind) values - ('{REVIEWER_ID}', 'm3ta', 'human'); + ('{REVIEWER_ID}', 'm3ta', 'human'), + ('{OTHER_AGENT_ID}', 'other-agent', 'system'); 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}'); +insert into public.strategies + (id, agent_id, diagnosis, guiding_policy, proximate_objectives, version, active, created_at) +values + ('{PRIOR_STRATEGY_ID}', '{REVIEWER_ID}', 'Prior diagnosis', 'Prior policy', '["prior"]', 1, true, + '2026-07-01T00:00:00+00:00'), + ('{OTHER_STRATEGY_ID}', '{OTHER_AGENT_ID}', 'Other diagnosis', 'Other policy', '["other"]', 1, true, + '2026-07-01T00:00:00+00:00'); +insert into public.strategy_nodes + (id, agent_id, node_type, title, body, rank, status, created_at, updated_at) +values + ('{PRIOR_ACTIVE_NODE_ID}', '{REVIEWER_ID}', 'policy', 'Prior active', 'Retire exactly once', 1, + 'active', '2026-07-01T00:00:00+00:00', '2026-07-01T00:00:00+00:00'), + ('{PRIOR_RETIRED_NODE_ID}', '{REVIEWER_ID}', 'policy', 'Prior retired', 'Leave timestamp unchanged', 2, + 'retired', '2026-06-01T00:00:00+00:00', '2026-06-02T00:00:00+00:00'), + ('{OTHER_ACTIVE_NODE_ID}', '{OTHER_AGENT_ID}', 'policy', 'Other active', 'Leave other agent unchanged', 1, + 'active', '2026-07-01T00:00:00+00:00', '2026-07-01T00:00:00+00:00'), + ('{NULL_AGENT_NODE_ID}', null, 'policy', 'Shared active', 'Leave shared node unchanged', 1, + 'active', '2026-07-01T00:00:00+00:00', '2026-07-01T00:00:00+00:00'); +insert into public.strategy_node_anchors + (id, from_node_id, anchor_role, to_node_id, weight, note, created_at) +values + ('{PRIOR_NODE_ANCHOR_ID}', '{PRIOR_ACTIVE_NODE_ID}', 'serves', '{PRIOR_RETIRED_NODE_ID}', 1.0, + 'Existing anchor must survive strategy revision replay', '2026-07-01T00:00:00+00:00'); """ @@ -631,13 +920,17 @@ def source_postgres() -> Iterator[tuple[str, str]]: "run", "--rm", "--detach", + "--network", + "none", "--name", container, + "--label", + f"{rebuild.canonical_rebuild.CANARY_LABEL}={container}", "--env", f"POSTGRES_DB={database}", "--env", "POSTGRES_HOST_AUTH_METHOD=trust", - rebuild.canonical_rebuild.DEFAULT_IMAGE, + rebuild.DEFAULT_REBUILD_IMAGE, ], text=True, capture_output=True, @@ -692,14 +985,24 @@ def source_postgres() -> Iterator[tuple[str, str]]: capture_output=True, check=False, ) + inspected = subprocess.run( + ["docker", "container", "inspect", container], + text=True, + capture_output=True, + check=False, + ) + if inspected.returncode == 0: + pytest.fail(f"fixture PostgreSQL container was not removed: {container}") -def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup( +def capture_source_snapshot( + source_container: str, + database: str, tmp_path: Path, - source_postgres: tuple[str, str], -) -> None: - source_container, database = source_postgres - genesis_dump = tmp_path / "genesis.dump" + *, + stem: str, +) -> tuple[Path, Path]: + dump = tmp_path / f"{stem}.dump" dump_result = subprocess.run( [ "docker", @@ -711,7 +1014,7 @@ def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup( "-d", database, "--format=custom", - "--file=/tmp/genesis.dump", + f"--file=/tmp/{stem}.dump", ], text=True, capture_output=True, @@ -719,15 +1022,130 @@ def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup( ) assert dump_result.returncode == 0, dump_result.stderr copy_result = subprocess.run( - ["docker", "cp", f"{source_container}:/tmp/genesis.dump", str(genesis_dump)], + ["docker", "cp", f"{source_container}:/tmp/{stem}.dump", str(dump)], text=True, capture_output=True, check=False, ) assert copy_result.returncode == 0, copy_result.stderr + manifest = tmp_path / f"{stem}-manifest.jsonl" + capture_manifest(source_container, database, manifest) + return dump, manifest - genesis_manifest = tmp_path / "genesis-manifest.jsonl" - capture_manifest(source_container, database, genesis_manifest) + +def seed_proposal_and_approval( + source_container: str, + database: str, + approved: dict, + approval: dict, +) -> dict: + sql = f"""begin; +set local standard_conforming_strings = on; +insert into kb_stage.kb_proposals +select seeded.* from jsonb_populate_record( + null::kb_stage.kb_proposals, {rebuild._jsonb_literal(approved)} +) seeded; +insert into kb_stage.kb_proposal_approvals +select seeded.* from jsonb_populate_record( + null::kb_stage.kb_proposal_approvals, {rebuild._jsonb_literal(approval)} +) seeded; +commit; +""" + docker_psql(source_container, database, sql) + raw = docker_psql( + source_container, + database, + rebuild.build_proposal_readback_sql(approved["id"]), + ).stdout.strip() + return json.loads(raw) + + +def run_genesis_ledger_command( + *, + tmp_path: Path, + database: str, + genesis_dump: Path, + genesis_manifest: Path, + material_path: Path, + final_manifest: Path, + artifact_stem: str, + container_prefix: str, + run_number: int, +) -> tuple[subprocess.CompletedProcess[str], dict, Path]: + 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": json.loads(material_path.read_text(encoding="utf-8"))[ + "replay_receipt" + ]["hashes"]["replay_material_sha256"], + } + ], + "final_parity": { + "manifest": final_manifest.name, + "sha256": sha256_file(final_manifest), + }, + } + ledger_path = write_json(tmp_path / f"{artifact_stem}-ledger-run-{run_number}.json", ledger) + output = tmp_path / f"{artifact_stem}-reconstruction-receipt-run-{run_number}.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", + container_prefix, + "--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 + assert output.is_file(), "rebuild command succeeded without writing its receipt" + receipt = json.loads(output.read_text(encoding="utf-8")) + return completed, receipt, output + + +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, genesis_manifest = capture_source_snapshot( + source_container, + database, + tmp_path, + stem="insert-only-genesis", + ) material = replay_material() material_path = write_json(tmp_path / "entry-0001.json", material) @@ -753,64 +1171,18 @@ def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup( 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, + completed, receipt, output = run_genesis_ledger_command( + tmp_path=tmp_path, + database=database, + genesis_dump=genesis_dump, + genesis_manifest=genesis_manifest, + material_path=material_path, + final_manifest=final_manifest, + artifact_stem="insert-only", + container_prefix="teleo-genesis-ledger-live-test", + run_number=1, ) - 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" @@ -831,3 +1203,132 @@ def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup( check=False, ) assert inspect.returncode != 0 + + +def test_live_revise_strategy_rebuild_is_exact_repeatable_and_leaves_no_container( + tmp_path: Path, + source_postgres: tuple[str, str], +) -> None: + source_container, database = source_postgres + genesis_dump, genesis_manifest = capture_source_snapshot( + source_container, + database, + tmp_path, + stem="revise-strategy-genesis", + ) + + approved, _, approval = revise_strategy_proposal_rows() + pre_apply = seed_proposal_and_approval( + source_container, + database, + approved, + approval, + ) + apply_sql = rebuild.apply_engine.build_apply_sql(pre_apply["proposal"], "kb-apply") + docker_psql(source_container, database, apply_sql, role="kb_apply") + post_apply = json.loads( + docker_psql( + source_container, + database, + rebuild.build_proposal_readback_sql(REVISE_PROPOSAL_ID), + ).stdout.strip() + ) + applied_proposal = post_apply["proposal"] + proposal_envelope = applied_envelope(applied_proposal) + source_receipt_path, receipt = rebuild.apply_engine.capture_replay_receipt( + argparse.Namespace( + container=source_container, + role="kb_apply", + host="127.0.0.1", + db=database, + receipt_out=str(tmp_path / "source-revise-strategy-private-receipt.json"), + receipt_dir=None, + ), + proposal_envelope, + "", + apply_sql=apply_sql, + ) + assert source_receipt_path.is_file() + assert stat.S_IMODE(source_receipt_path.stat().st_mode) == 0o600 + material = { + "artifact": rebuild.MATERIAL_ARTIFACT, + "contract_version": rebuild.MATERIAL_CONTRACT_VERSION, + "sequence": 1, + "approved_proposal": pre_apply["proposal"], + "approval_snapshot": pre_apply["approval_snapshot"], + "applied_proposal": applied_proposal, + "replay_receipt": receipt, + } + material_path = write_json(tmp_path / "revise-strategy-entry-0001.json", material) + final_manifest = tmp_path / "revise-strategy-final-manifest.jsonl" + capture_manifest(source_container, database, final_manifest) + + runs = [ + run_genesis_ledger_command( + tmp_path=tmp_path, + database=database, + genesis_dump=genesis_dump, + genesis_manifest=genesis_manifest, + material_path=material_path, + final_manifest=final_manifest, + artifact_stem="revise-strategy", + container_prefix="teleo-genesis-ledger-revise", + run_number=run_number, + ) + for run_number in (1, 2) + ] + + for completed, reconstruction, output in runs: + assert reconstruction["status"] == "pass" + assert reconstruction["genesis_parity"]["status"] == "pass" + assert reconstruction["final_parity"]["status"] == "pass" + entry_summary = reconstruction["ledger"]["entries"][0] + assert entry_summary["proposal_type"] == "revise_strategy" + assert entry_summary["proposal_seed_exact"] is True + assert entry_summary["canonical_seed_exact"] is False + assert entry_summary["seed_exact"] is False + assert entry_summary["guarded_apply_executed"] is True + assert entry_summary["mutating_prestate_captured"] is True + assert entry_summary["mutating_delta_validated"] is True + assert entry_summary["mutating_poststate_normalized"] is True + assert entry_summary["proposal_exact"] is True + assert entry_summary["canonical_rows_exact"] is True + assert reconstruction["cleanup"]["container_absent"] is True + assert reconstruction["safety"]["network_access_available_to_container"] is False + assert reconstruction["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 + + container_name = reconstruction["container"]["name"] + inspect = subprocess.run( + ["docker", "container", "inspect", container_name], + text=True, + capture_output=True, + check=False, + ) + assert inspect.returncode != 0 + labeled = subprocess.run( + [ + "docker", + "ps", + "-aq", + "--filter", + f"label={rebuild.canonical_rebuild.CANARY_LABEL}={container_name}", + ], + text=True, + capture_output=True, + check=False, + ) + assert labeled.returncode == 0 + assert labeled.stdout.strip() == "" + + first = runs[0][1] + second = runs[1][1] + assert first["inputs"] == second["inputs"] + assert first["ledger"]["entries"][0]["material_sha256"] == second["ledger"]["entries"][0][ + "material_sha256" + ] + assert first["ledger"]["entries"][0]["replay_material_sha256"] == second["ledger"]["entries"][0][ + "replay_material_sha256" + ]