From 7ea07cc4fccf0f43d4a20ad9184bace6a6d3fae8 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Tue, 14 Jul 2026 03:26:42 +0200 Subject: [PATCH] retain replay receipts for guarded KB applies --- docs/kb-rebuild-and-recompile.md | 36 +- .../current-truth-index.md | 5 +- .../kb-apply-replay-receipt-current.json | 78 +++ ...e-day-outcome-and-forward-path-20260714.md | 24 +- scripts/apply_proposal.py | 133 ++++- scripts/kb_apply_replay_receipt.py | 482 ++++++++++++++++++ tests/test_kb_apply_replay_receipt.py | 401 +++++++++++++++ 7 files changed, 1147 insertions(+), 12 deletions(-) create mode 100644 docs/reports/leo-working-state-20260709/kb-apply-replay-receipt-current.json create mode 100644 scripts/kb_apply_replay_receipt.py create mode 100644 tests/test_kb_apply_replay_receipt.py diff --git a/docs/kb-rebuild-and-recompile.md b/docs/kb-rebuild-and-recompile.md index 789e759..841e222 100644 --- a/docs/kb-rebuild-and-recompile.md +++ b/docs/kb-rebuild-and-recompile.md @@ -116,6 +116,35 @@ receipt. Every accepted change after that epoch must carry a replayable strict apply payload and row-level postflight receipt. This prevents the historical gap from growing while the old import rules are reconstructed. +The guarded apply CLI now enforces the receipt half of that policy. After a +successful apply it reloads the immutable applied proposal, selects the exact +canonical rows described by the strict payload, binds generated row IDs and +timestamps, hashes the exact executed apply SQL, payload, and rows, and +atomically writes a private mode `0600` receipt. It also supports read-only +recovery for a committed apply whose receipt file was lost, provided every +payload-controlled row still matches the immutable reviewed payload: + +```bash +python3 scripts/apply_proposal.py --receipt-only +``` + +The live VPS recovery canary used strict applied proposal +`00957f6c-9883-4015-95a4-6b09367efb0e`. It recovered exactly one canonical edge, +kept all database counts and the proposal payload hash unchanged, left the Leo +gateway on the same PID with zero restarts, and removed the temporary private +receipt. The full receipt is deliberately not committed because it can contain +claim bodies or source excerpts; the sanitized proof is retained as +`reports/leo-working-state-20260709/kb-apply-replay-receipt-current.json`. + +Normal applies mark the SQL hash as `exact_executed_sql`. A later +`--receipt-only` recovery marks it as `reconstructed_current_engine`; it never +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. + 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: @@ -161,9 +190,10 @@ 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 accepted packets into a -complete corpus runner and replay ledger. One-document preparation and staging -do not yet prove a clean database can be rebuilt semantically from every source. +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. ## Definition Of Working diff --git a/docs/reports/leo-working-state-20260709/current-truth-index.md b/docs/reports/leo-working-state-20260709/current-truth-index.md index a0d7b61..9dca26d 100644 --- a/docs/reports/leo-working-state-20260709/current-truth-index.md +++ b/docs/reports/leo-working-state-20260709/current-truth-index.md @@ -29,7 +29,9 @@ supersedes the July 13 verdict below. `29`-proposal VPS dump restored locally with zero mismatches and complete cleanup; it has not yet been restored to GCP. - Exact snapshot recovery works. Full source-to-blank-database recompilation is - partial; see `../../kb-rebuild-and-recompile.md`. + partial. The guarded apply path now emits or recovers private row-level replay + receipts for strict applied proposals; the live recovery canary changed no DB + or service state. See `../../kb-rebuild-and-recompile.md`. Newest artifacts: @@ -39,6 +41,7 @@ Newest artifacts: - `leo-v3-document-source-lifecycle-current.md` - `gcp-refresh-post-v3-snapshot-current.json` - `gcp-refresh-post-v3-local-rebuild-current.json` +- `kb-apply-replay-receipt-current.json` ## Live Addendum - 2026-07-13 diff --git a/docs/reports/leo-working-state-20260709/kb-apply-replay-receipt-current.json b/docs/reports/leo-working-state-20260709/kb-apply-replay-receipt-current.json new file mode 100644 index 0000000..c980c8a --- /dev/null +++ b/docs/reports/leo-working-state-20260709/kb-apply-replay-receipt-current.json @@ -0,0 +1,78 @@ +{ + "artifact": "kb_apply_replay_receipt_live_recovery_canary", + "claim_ceiling": "The guarded apply path now has a private row-level receipt and receipt-only recovery implementation. The live VPS canary proved recovery for the existing strict add_edge proposal without a database write. Other supported apply types are covered by static tests but still require disposable-clone lifecycle proof. This does not yet provide a complete genesis-plus-ledger blank-database runner.", + "current_canary": "Recover the exact canonical rows and immutable reviewed payload for an already-applied strict proposal, write a mode-0600 private receipt, and leave the production database and Leo gateway unchanged.", + "generated_at_utc": "2026-07-14T01:08:23Z", + "live_vps_read_only": { + "actual_row_counts": { + "claim_edges": 1 + }, + "apply_engine": { + "apply_sql_sha256": "8c54e2092e29f1137b3c907d670ea9181b35a9c9bd3007e9025a786d05745779", + "source": "reconstructed_current_engine" + }, + "checks": { + "applied_at_present": true, + "proposal_applied": true, + "row_level_postflight_exact": true, + "semantic_rows_match_strict_payload": true, + "strict_apply_payload_present": true + }, + "expected_row_counts": { + "claim_edges": 1 + }, + "proposal_id": "00957f6c-9883-4015-95a4-6b09367efb0e", + "proposal_payload_sha256": "004f51e2b035b1217a8b9fb62deefafe228f46f230f066c8c7ea1f9410178e15", + "proposal_type": "add_edge", + "receipt_file_mode": "0600", + "replay_material_sha256": "0d959d3e7c5ec29aff47e75a83c5a7082bbbeaa95c380603c974a1fc6f37596d", + "replay_ready": true + }, + "production_state": { + "after": { + "claims": 1837, + "edges": 4916, + "evidence": 4670, + "proposals": 29, + "sources": 4145 + }, + "before": { + "claims": 1837, + "edges": 4916, + "evidence": 4670, + "proposals": 29, + "sources": 4145 + }, + "canonical_database_mutated": false, + "gateway": { + "active_state": "active", + "main_pid_after": 366950, + "main_pid_before": 366950, + "nrestarts_after": 0, + "nrestarts_before": 0, + "sub_state": "running", + "unchanged": true + }, + "proposal_payload_unchanged": true, + "proposal_status_after": "applied", + "proposal_status_before": "applied", + "telegram_send_attempted": false + }, + "privacy_and_cleanup": { + "full_receipt_committed_to_git": false, + "full_receipt_may_contain_claim_or_source_bodies": true, + "temporary_remote_directory_absent": true + }, + "status": "pass", + "verification": { + "focused_tests_passed": 66, + "full_tests_passed": 1170, + "ruff_passed": true, + "supported_apply_types_in_static_suite": [ + "add_edge", + "attach_evidence", + "approve_claim", + "revise_strategy" + ] + } +} diff --git a/docs/reports/leo-working-state-20260709/working-leo-three-day-outcome-and-forward-path-20260714.md b/docs/reports/leo-working-state-20260709/working-leo-three-day-outcome-and-forward-path-20260714.md index 6c0212b..56c7969 100644 --- a/docs/reports/leo-working-state-20260709/working-leo-three-day-outcome-and-forward-path-20260714.md +++ b/docs/reports/leo-working-state-20260709/working-leo-three-day-outcome-and-forward-path-20260714.md @@ -1,6 +1,6 @@ # Working Leo: Three-Day Outcome And Forward Path -Current as of `2026-07-14 00:28 UTC`. +Current as of `2026-07-14 01:08 UTC`. This is the current high-level handoff for what the July 10-14 delivery wave achieved, what is working on the VPS, what still has to move to GCP, and how the @@ -27,9 +27,10 @@ replay is still pending. **Database rebuild answer: exact recovery works; semantic recompilation is partial.** A verified snapshot can recreate the current Postgres database in -seconds. Rebuilding a blank database from every retained document, extraction, -review decision, and apply receipt is not yet complete because historical rows -predate a complete replay ledger. +seconds. New strict applies now emit a recoverable private row-level replay +receipt so the provenance gap stops growing. Rebuilding a blank database from +every retained document, extraction, review decision, and apply receipt is not +yet complete because historical rows predate a complete replay ledger. ## What Working Leo Means @@ -323,6 +324,19 @@ The practical policy is therefore: This is how Leo improves by composing its database rather than by repeatedly training its chat style. +The first implementation step is now live-proven in read-only recovery mode. +The guarded apply CLI captures the immutable strict payload plus exact canonical +rows, generated UUIDs, timestamps, and the exact executed apply-SQL hash into a +private mode-`0600` receipt. `--receipt-only` can regenerate that receipt after +a committed apply, while the payload-controlled rows still match, without +reapplying or mutating the database; recovered engine hashes are explicitly +labeled as reconstructed rather than original execution proof. The VPS canary recovered strict +applied proposal `00957f6c-9883-4015-95a4-6b09367efb0e`, found exactly one +canonical edge, left all counts and the gateway unchanged, and removed the +private temporary file. This prevents future strict applies from becoming +unreplayable; it does not fill the pre-genesis historical gaps or run the final +blank-database compiler. + ## Current Outcome Matrix | Outcome | Current tier | Verdict | @@ -338,6 +352,7 @@ training its chat style. | Prior GCP copy and six-question replay | private Cloud SQL and GCP runtime | Proven historically | | Current GCP parity with VPS `29` proposals | fresh staging replay | Not yet run | | Exact snapshot rebuild | disposable local Postgres | Working | +| New strict-apply replay receipt | live VPS read-only recovery plus tests | Working for receipt capture; full replay runner pending | | Full source-to-blank-DB recompilation | complete corpus plus ledger | Partial | ## Immediate Execution Order @@ -365,6 +380,7 @@ training its chat style. - `gcp-refresh-local-rebuild-current.json` - `gcp-refresh-post-v3-snapshot-current.json` - `gcp-refresh-post-v3-local-rebuild-current.json` +- `kb-apply-replay-receipt-current.json` - `../../kb-rebuild-and-recompile.md` ## Final Claim Ceiling diff --git a/scripts/apply_proposal.py b/scripts/apply_proposal.py index d1b08f8..7fd4bd3 100644 --- a/scripts/apply_proposal.py +++ b/scripts/apply_proposal.py @@ -92,6 +92,8 @@ import uuid from pathlib import Path from typing import Any +import kb_apply_replay_receipt as replay_receipt + DEFAULT_SECRETS_FILE = "/home/teleo/.hermes/profiles/leoclean/secrets/kb-apply.env" DEFAULT_CONTAINER = "teleo-pg" DEFAULT_DB = "teleo" @@ -840,6 +842,15 @@ def assert_applyable(proposal: dict[str, Any]) -> None: ) +def assert_receiptable(proposal: dict[str, Any]) -> None: + if proposal.get("status") != "applied": + raise SystemExit( + f"proposal {proposal['id']} has status {proposal.get('status')!r}; " + "--receipt-only requires an applied proposal" + ) + replay_receipt.expected_row_counts(proposal) + + # --------------------------------------------------------------------------- # # Execution (docker exec as the kb_apply role) # # --------------------------------------------------------------------------- # @@ -857,7 +868,13 @@ def load_password(secrets_file: str) -> str: raise SystemExit(f"no KB_APPLY_PASSWORD/KB_APPLY_DB_PASSWORD/PGPASSWORD entry in {secrets_file}") -def run_psql(args: argparse.Namespace, sql: str, password: str) -> str: +def run_psql( + args: argparse.Namespace, + sql: str, + password: str, + *, + redact_output_on_error: bool = False, +) -> str: docker_binary = shutil.which("docker") or "docker" command = [ docker_binary, "exec", "-e", "PGPASSWORD", "-i", args.container, @@ -870,6 +887,10 @@ def run_psql(args: argparse.Namespace, sql: str, password: str) -> str: check=False, ) if result.returncode != 0: + if redact_output_on_error: + raise SystemExit( + f"psql failed ({result.returncode}); private postflight output redacted" + ) raise SystemExit( f"psql failed ({result.returncode})\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" ) @@ -885,7 +906,10 @@ def load_proposal(args: argparse.Namespace, password: str) -> dict[str, Any]: 'reviewed_by_handle', reviewed_by_handle, 'reviewed_by_agent_id', reviewed_by_agent_id::text, 'reviewed_at', reviewed_at::text, - 'review_note', review_note)::text + 'review_note', review_note, + 'applied_by_handle', applied_by_handle, + 'applied_by_agent_id', applied_by_agent_id::text, + 'applied_at', applied_at::text)::text from kb_stage.kb_proposals where id = {sql_literal(args.proposal_id)}::uuid;""" out = run_psql(args, sql, password).strip() @@ -894,6 +918,45 @@ def load_proposal(args: argparse.Namespace, password: str) -> dict[str, Any]: return json.loads(out) +def capture_replay_receipt( + args: argparse.Namespace, + proposal: dict[str, Any], + password: str, + *, + apply_sql: str | None = None, +) -> tuple[Path, dict[str, Any]]: + """Read exact postflight rows and atomically write a private replay receipt.""" + postflight_sql = replay_receipt.build_postflight_sql(proposal) + raw = run_psql( + args, + postflight_sql, + password, + redact_output_on_error=True, + ).strip() + if not raw: + raise RuntimeError("postflight query returned no row") + canonical_rows = json.loads(raw) + if not isinstance(canonical_rows, dict): + raise RuntimeError("postflight query did not return a JSON object") + apply_sql_source = "exact_executed_sql" + if apply_sql is None: + apply_sql = build_apply_sql(proposal, proposal.get("applied_by_handle")) + apply_sql_source = "reconstructed_current_engine" + receipt = replay_receipt.build_replay_receipt( + proposal, + canonical_rows, + apply_sql_sha256=replay_receipt.sha256_text(apply_sql), + apply_sql_source=apply_sql_source, + ) + path = replay_receipt.resolve_receipt_path( + str(proposal["id"]), + receipt_out=args.receipt_out, + receipt_dir=args.receipt_dir, + ) + replay_receipt.write_private_receipt(path, receipt) + return path, receipt + + def parse_args(argv: list[str]) -> argparse.Namespace: p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("proposal_id", help="UUID of the approved proposal to apply") @@ -909,12 +972,49 @@ def parse_args(argv: list[str]) -> argparse.Namespace: p.add_argument("--db", default=DEFAULT_DB) p.add_argument("--host", default=DEFAULT_HOST) p.add_argument("--role", default=DEFAULT_ROLE) - return p.parse_args(argv) + p.add_argument( + "--receipt-only", + action="store_true", + help="recover the private row-level replay receipt for an already applied proposal; no writes", + ) + p.add_argument( + "--receipt-dir", + default=None, + help="private receipt directory (default: KB_APPLY_RECEIPT_DIR or " + f"{replay_receipt.DEFAULT_RECEIPT_DIR})", + ) + p.add_argument( + "--receipt-out", + default=None, + help="exact private receipt path; overrides --receipt-dir", + ) + args = p.parse_args(argv) + if args.receipt_only and args.dry_run: + p.error("--receipt-only and --dry-run are mutually exclusive") + return args def main(argv: list[str] | None = None) -> int: args = parse_args(sys.argv[1:] if argv is None else argv) + if args.receipt_only: + password = load_password(args.secrets_file) + proposal = load_proposal(args, password) + assert_receiptable(proposal) + path, receipt = capture_replay_receipt(args, proposal, password) + print( + json.dumps( + { + "proposal_id": proposal["id"], + "receipt_path": str(path), + "replay_material_sha256": receipt["hashes"]["replay_material_sha256"], + "replay_ready": receipt["replay_ready"], + }, + sort_keys=True, + ) + ) + return 0 + if args.dry_run: # Dry-run still needs the proposal to build SQL; read it as kb_apply. password = load_password(args.secrets_file) @@ -928,7 +1028,32 @@ def main(argv: list[str] | None = None) -> int: assert_applyable(proposal) sql = build_apply_sql(proposal, args.applied_by) run_psql(args, sql, password) - print(f"applied proposal {proposal['id']} ({proposal['proposal_type']})") + applied_proposal = load_proposal(args, password) + try: + path, receipt = capture_replay_receipt( + args, + applied_proposal, + password, + apply_sql=sql, + ) + except Exception as exc: + raise SystemExit( + f"proposal {proposal['id']} committed, but private replay receipt capture failed: {exc}. " + f"Recover without reapplying via: {Path(__file__).name} {proposal['id']} --receipt-only" + ) from exc + print( + json.dumps( + { + "applied": True, + "proposal_id": proposal["id"], + "proposal_type": proposal["proposal_type"], + "receipt_path": str(path), + "replay_material_sha256": receipt["hashes"]["replay_material_sha256"], + "replay_ready": receipt["replay_ready"], + }, + sort_keys=True, + ) + ) return 0 diff --git a/scripts/kb_apply_replay_receipt.py b/scripts/kb_apply_replay_receipt.py new file mode 100644 index 0000000..83678a9 --- /dev/null +++ b/scripts/kb_apply_replay_receipt.py @@ -0,0 +1,482 @@ +#!/usr/bin/env python3 +"""Build a private, row-level replay receipt for one applied KB proposal. + +The guarded apply transaction keeps the immutable reviewed payload in +``kb_stage.kb_proposals``. This module binds that payload to the exact canonical +rows observed after commit, including generated UUIDs and timestamps. The +result is sufficient input for a future genesis-plus-ledger replayer without +putting claim bodies or source excerpts in logs or stdout. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +RECEIPT_CONTRACT_VERSION = 1 +DEFAULT_RECEIPT_DIR = "/opt/teleo-eval/logs/kb-apply-receipts" + + +def _canonical_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _sha256_json(value: Any) -> str: + return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest() + + +def sha256_text(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _sql_literal(value: Any) -> str: + if value is None: + return "null" + escaped = str(value).replace("\\", "\\\\").replace("'", "''") + return "E'" + escaped + "'" + + +def _jsonb(value: Any) -> str: + return _sql_literal(_canonical_json(value)) + "::jsonb" + + +def _require_applied_strict_proposal(proposal: dict[str, Any]) -> dict[str, Any]: + if proposal.get("status") != "applied": + raise ValueError("replay receipt requires proposal status='applied'") + if not proposal.get("applied_at"): + raise ValueError("replay receipt requires a non-null applied_at") + if not proposal.get("applied_by_handle"): + raise ValueError("replay receipt requires applied_by_handle") + payload = proposal.get("payload") + apply_payload = payload.get("apply_payload") if isinstance(payload, dict) else None + if not isinstance(apply_payload, dict): + raise ValueError("replay receipt requires a strict payload.apply_payload object") + return apply_payload + + +def _coalesced_json_agg(table: str, alias: str, where: str, order_by: str) -> str: + return ( + "coalesce((select jsonb_agg(to_jsonb(" + alias + ") order by " + order_by + ") " + "from " + table + " " + alias + " where " + where + "), '[]'::jsonb)" + ) + + +def _id_filter(alias: str, values: list[Any]) -> str: + if not values: + return "false" + rendered = ", ".join(f"{_sql_literal(value)}::uuid" for value in values) + return f"{alias}.id in ({rendered})" + + +def _or_filter(clauses: list[str]) -> str: + if not clauses: + return "false" + return "(" + ") or (".join(clauses) + ")" + + +def _build_add_edge_postflight_sql(apply_payload: dict[str, Any]) -> str: + where = ( + f"e.from_claim = {_sql_literal(apply_payload.get('from_claim'))}::uuid " + f"and e.to_claim = {_sql_literal(apply_payload.get('to_claim'))}::uuid " + f"and e.edge_type = {_sql_literal(apply_payload.get('edge_type'))}::edge_type" + ) + rows = _coalesced_json_agg("public.claim_edges", "e", where, "e.id::text") + return f"select jsonb_build_object('claim_edges', {rows})::text;" + + +def _build_attach_evidence_postflight_sql(apply_payload: dict[str, Any]) -> str: + clauses = [] + for row in apply_payload.get("evidence") or []: + role = row.get("role") or "grounds" + clauses.append( + f"ce.claim_id = {_sql_literal(row.get('claim_id'))}::uuid " + f"and ce.source_id = {_sql_literal(row.get('source_id'))}::uuid " + f"and ce.role = {_sql_literal(role)}::evidence_role" + ) + rows = _coalesced_json_agg( + "public.claim_evidence", + "ce", + _or_filter(clauses), + "ce.claim_id::text, ce.source_id::text, ce.role::text", + ) + return f"select jsonb_build_object('claim_evidence', {rows})::text;" + + +def _build_approve_claim_postflight_sql(apply_payload: dict[str, Any]) -> str: + claims = apply_payload.get("claims") or [] + sources = apply_payload.get("sources") or [] + evidence = apply_payload.get("evidence") or [] + edges = apply_payload.get("edges") or [] + tools = apply_payload.get("reasoning_tools") or [] + + evidence_clauses = [] + for row in evidence: + role = row.get("role") or "grounds" + evidence_clauses.append( + f"ce.claim_id = {_sql_literal(row.get('claim_id'))}::uuid " + f"and ce.source_id = {_sql_literal(row.get('source_id'))}::uuid " + f"and ce.role = {_sql_literal(role)}::evidence_role" + ) + edge_clauses = [ + f"e.from_claim = {_sql_literal(row.get('from_claim'))}::uuid " + f"and e.to_claim = {_sql_literal(row.get('to_claim'))}::uuid " + f"and e.edge_type = {_sql_literal(row.get('edge_type'))}::edge_type" + for row in edges + ] + + fields = { + "claims": _coalesced_json_agg( + "public.claims", "c", _id_filter("c", [row.get("id") for row in claims]), "c.id::text" + ), + "sources": _coalesced_json_agg( + "public.sources", "s", _id_filter("s", [row.get("id") for row in sources]), "s.id::text" + ), + "claim_evidence": _coalesced_json_agg( + "public.claim_evidence", + "ce", + _or_filter(evidence_clauses), + "ce.claim_id::text, ce.source_id::text, ce.role::text", + ), + "claim_edges": _coalesced_json_agg("public.claim_edges", "e", _or_filter(edge_clauses), "e.id::text"), + "reasoning_tools": _coalesced_json_agg( + "public.reasoning_tools", + "rt", + _id_filter("rt", [row.get("id") for row in tools]), + "rt.id::text", + ), + } + arguments = ", ".join(f"{_sql_literal(name)}, {expression}" for name, expression in fields.items()) + return f"select jsonb_build_object({arguments})::text;" + + +def _build_revise_strategy_postflight_sql(proposal: dict[str, Any], apply_payload: dict[str, Any]) -> str: + strategy = apply_payload.get("strategy") or {} + agent_id = apply_payload.get("agent_id") + node_clauses = [] + for row in apply_payload.get("strategy_nodes") or []: + node_clauses.append( + f"n.node_type = {_sql_literal(row.get('node_type'))} " + f"and n.title = {_sql_literal(row.get('title'))} " + f"and n.body = {_sql_literal(row.get('body'))} " + f"and n.rank = {int(row.get('rank', 1))}" + ) + node_filter = _or_filter(node_clauses) + return f"""with matched_strategy as ( + select s.* + from public.strategies s + where s.agent_id = {_sql_literal(agent_id)}::uuid + and s.diagnosis = {_sql_literal(strategy.get("diagnosis"))} + and s.guiding_policy = {_sql_literal(strategy.get("guiding_policy"))} + and s.proximate_objectives = {_jsonb(strategy.get("proximate_objectives"))} + and s.created_at <= {_sql_literal(proposal.get("applied_at"))}::timestamptz + order by s.version desc, s.id::text desc + limit 1 +), matched_nodes as ( + select n.* + from public.strategy_nodes n + join matched_strategy s + on n.agent_id = s.agent_id and n.created_at = s.created_at + where {node_filter} +) +select jsonb_build_object( + 'strategies', coalesce((select jsonb_agg(to_jsonb(s) order by s.id::text) from matched_strategy s), '[]'::jsonb), + 'strategy_nodes', coalesce((select jsonb_agg(to_jsonb(n) order by n.rank, n.id::text) from matched_nodes n), '[]'::jsonb) +)::text;""" + + +def build_postflight_sql(proposal: dict[str, Any]) -> str: + apply_payload = _require_applied_strict_proposal(proposal) + proposal_type = proposal.get("proposal_type") + if proposal_type == "add_edge": + return _build_add_edge_postflight_sql(apply_payload) + if proposal_type == "attach_evidence": + return _build_attach_evidence_postflight_sql(apply_payload) + if proposal_type == "approve_claim": + return _build_approve_claim_postflight_sql(apply_payload) + if proposal_type == "revise_strategy": + return _build_revise_strategy_postflight_sql(proposal, apply_payload) + raise ValueError(f"replay receipt does not support proposal_type {proposal_type!r}") + + +def expected_row_counts(proposal: dict[str, Any]) -> dict[str, int]: + apply_payload = _require_applied_strict_proposal(proposal) + proposal_type = proposal.get("proposal_type") + if proposal_type == "add_edge": + return {"claim_edges": 1} + if proposal_type == "attach_evidence": + return {"claim_evidence": len(apply_payload.get("evidence") or [])} + if proposal_type == "approve_claim": + return { + "claims": len(apply_payload.get("claims") or []), + "sources": len(apply_payload.get("sources") or []), + "claim_evidence": len(apply_payload.get("evidence") or []), + "claim_edges": len(apply_payload.get("edges") or []), + "reasoning_tools": len(apply_payload.get("reasoning_tools") or []), + } + if proposal_type == "revise_strategy": + return { + "strategies": 1, + "strategy_nodes": len(apply_payload.get("strategy_nodes") or []), + } + raise ValueError(f"replay receipt does not support proposal_type {proposal_type!r}") + + +def _normalize_rows(canonical_rows: dict[str, Any], expected_counts: dict[str, int]) -> dict[str, list[dict[str, Any]]]: + normalized: dict[str, list[dict[str, Any]]] = {} + for table in expected_counts: + rows = canonical_rows.get(table) + if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows): + raise ValueError(f"postflight {table} must be a list of row objects") + normalized[table] = sorted(rows, key=_canonical_json) + return normalized + + +def _semantic_projection(row: dict[str, Any], fields: tuple[str, ...]) -> dict[str, Any]: + return {field: row.get(field) for field in fields} + + +def _expected_semantic_rows(proposal: dict[str, Any]) -> dict[str, list[dict[str, Any]]]: + apply_payload = _require_applied_strict_proposal(proposal) + proposal_type = proposal.get("proposal_type") + if proposal_type == "add_edge": + return { + "claim_edges": [ + { + "from_claim": apply_payload.get("from_claim"), + "to_claim": apply_payload.get("to_claim"), + "edge_type": apply_payload.get("edge_type"), + "weight": apply_payload.get("weight"), + "created_by": None, + } + ] + } + if proposal_type == "attach_evidence": + return { + "claim_evidence": [ + { + "claim_id": row.get("claim_id"), + "source_id": row.get("source_id"), + "role": row.get("role") or "grounds", + "weight": row.get("weight"), + "created_by": None, + } + for row in apply_payload.get("evidence") or [] + ] + } + if proposal_type == "approve_claim": + agent_id = apply_payload.get("agent_id") + return { + "claims": [ + { + "id": row.get("id"), + "type": row.get("type"), + "text": row.get("text"), + "status": row.get("status", "open"), + "confidence": row.get("confidence"), + "tags": row.get("tags") or [], + "created_by": row.get("created_by", agent_id), + "superseded_by": row.get("superseded_by"), + } + for row in apply_payload.get("claims") or [] + ], + "sources": [ + { + "id": row.get("id"), + "source_type": row.get("source_type"), + "url": row.get("url"), + "storage_path": row.get("storage_path"), + "excerpt": row.get("excerpt"), + "hash": row.get("hash"), + "created_by": row.get("created_by", agent_id), + } + for row in apply_payload.get("sources") or [] + ], + "claim_evidence": [ + { + "claim_id": row.get("claim_id"), + "source_id": row.get("source_id"), + "role": row.get("role") or "grounds", + "weight": row.get("weight"), + "created_by": row.get("created_by", agent_id), + } + for row in apply_payload.get("evidence") or [] + ], + "claim_edges": [ + { + "from_claim": row.get("from_claim"), + "to_claim": row.get("to_claim"), + "edge_type": row.get("edge_type"), + "weight": row.get("weight"), + "created_by": row.get("created_by", agent_id), + } + for row in apply_payload.get("edges") or [] + ], + "reasoning_tools": [ + { + "id": row.get("id"), + "agent_id": row.get("agent_id", agent_id), + "name": row.get("name"), + "description": row.get("description"), + "category": row.get("category"), + } + for row in apply_payload.get("reasoning_tools") or [] + ], + } + if proposal_type == "revise_strategy": + strategy = apply_payload.get("strategy") or {} + agent_id = apply_payload.get("agent_id") + return { + "strategies": [ + { + "agent_id": agent_id, + "diagnosis": strategy.get("diagnosis"), + "guiding_policy": strategy.get("guiding_policy"), + "proximate_objectives": strategy.get("proximate_objectives"), + "active": True, + } + ], + "strategy_nodes": [ + { + "agent_id": agent_id, + "node_type": row.get("node_type"), + "title": row.get("title"), + "body": row.get("body"), + "rank": row.get("rank", 1), + "status": "active", + "horizon": None, + "measure": None, + "source_ref": None, + "metadata": {}, + } + for row in apply_payload.get("strategy_nodes") or [] + ], + } + raise ValueError(f"replay receipt does not support proposal_type {proposal_type!r}") + + +def _assert_semantic_rows_match(proposal: dict[str, Any], rows: dict[str, list[dict[str, Any]]]) -> None: + expected = _expected_semantic_rows(proposal) + for table, expected_rows in expected.items(): + fields = tuple(expected_rows[0]) if expected_rows else () + actual_projected = [_semantic_projection(row, fields) for row in rows.get(table, [])] + if sorted(expected_rows, key=_canonical_json) != sorted(actual_projected, key=_canonical_json): + raise ValueError( + f"row-level postflight semantic mismatch for {table}; " + "canonical values do not match the immutable strict payload" + ) + + +def build_replay_receipt( + proposal: dict[str, Any], + canonical_rows: dict[str, Any], + *, + apply_sql_sha256: str, + apply_sql_source: str, + exported_at_utc: str | None = None, +) -> dict[str, Any]: + if len(apply_sql_sha256) != 64 or any(character not in "0123456789abcdef" for character in apply_sql_sha256): + raise ValueError("apply_sql_sha256 must be a lowercase SHA-256 hex digest") + if apply_sql_source not in {"exact_executed_sql", "reconstructed_current_engine"}: + raise ValueError("invalid apply_sql_source") + apply_payload = _require_applied_strict_proposal(proposal) + expected = expected_row_counts(proposal) + rows = _normalize_rows(canonical_rows, expected) + actual = {table: len(table_rows) for table, table_rows in rows.items()} + mismatches = { + table: {"expected": expected[table], "actual": actual.get(table, 0)} + for table in expected + if actual.get(table, 0) != expected[table] + } + if mismatches: + raise ValueError(f"row-level postflight count mismatch: {mismatches}") + _assert_semantic_rows_match(proposal, rows) + + proposal_envelope = { + "id": str(proposal.get("id")), + "proposal_type": proposal.get("proposal_type"), + "status": proposal.get("status"), + "payload": proposal.get("payload"), + "reviewed_by_handle": proposal.get("reviewed_by_handle"), + "reviewed_by_agent_id": proposal.get("reviewed_by_agent_id"), + "reviewed_at": proposal.get("reviewed_at"), + "review_note": proposal.get("review_note"), + "applied_by_handle": proposal.get("applied_by_handle"), + "applied_by_agent_id": proposal.get("applied_by_agent_id"), + "applied_at": proposal.get("applied_at"), + } + row_hashes = {table: [_sha256_json(row) for row in table_rows] for table, table_rows in rows.items()} + table_hashes = {table: _sha256_json(table_rows) for table, table_rows in rows.items()} + replay_material = { + "receipt_contract_version": RECEIPT_CONTRACT_VERSION, + "apply_engine": { + "apply_sql_sha256": apply_sql_sha256, + "source": apply_sql_source, + }, + "proposal": proposal_envelope, + "canonical_rows": rows, + } + return { + **replay_material, + "artifact": "kb_apply_replay_receipt", + "exported_at_utc": exported_at_utc or datetime.now(timezone.utc).isoformat(), + "expected_row_counts": expected, + "actual_row_counts": actual, + "checks": { + "proposal_applied": True, + "strict_apply_payload_present": True, + "applied_at_present": True, + "row_level_postflight_exact": True, + "semantic_rows_match_strict_payload": True, + }, + "hashes": { + "apply_payload_sha256": _sha256_json(apply_payload), + "proposal_payload_sha256": _sha256_json(proposal.get("payload")), + "canonical_rows_sha256": _sha256_json(rows), + "row_sha256": row_hashes, + "table_sha256": table_hashes, + "replay_material_sha256": _sha256_json(replay_material), + }, + "replay_ready": True, + "privacy": "private receipt; may contain claim bodies and source excerpts; do not commit", + } + + +def resolve_receipt_path( + proposal_id: str, + *, + receipt_out: str | None = None, + receipt_dir: str | None = None, +) -> Path: + if receipt_out: + return Path(receipt_out) + root = Path(receipt_dir or os.environ.get("KB_APPLY_RECEIPT_DIR") or DEFAULT_RECEIPT_DIR) + return root / f"{proposal_id}.json" + + +def write_private_receipt(path: Path, receipt: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + payload = json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True) + "\n" + fd, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent)) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_name, path) + os.chmod(path, 0o600) + except Exception: + try: + os.close(fd) + except OSError: + pass + try: + os.unlink(temporary_name) + except FileNotFoundError: + pass + raise diff --git a/tests/test_kb_apply_replay_receipt.py b/tests/test_kb_apply_replay_receipt.py new file mode 100644 index 0000000..0bc302e --- /dev/null +++ b/tests/test_kb_apply_replay_receipt.py @@ -0,0 +1,401 @@ +import json +import stat +import sys +from argparse import Namespace +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "scripts")) + +import apply_proposal as ap # noqa: E402 +import kb_apply_replay_receipt as receipt # noqa: E402 + +CLAIM_A = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" +CLAIM_B = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" +SOURCE_A = "cccccccc-cccc-4ccc-8ccc-cccccccccccc" +TOOL_A = "dddddddd-dddd-4ddd-8ddd-dddddddddddd" +PROPOSAL_ID = "99999999-9999-4999-8999-999999999999" +APPLY_SQL_SHA256 = receipt.sha256_text("begin; select guarded_apply(); commit;") + + +def _build_receipt(proposal: dict, rows: dict, **kwargs: str) -> dict: + return receipt.build_replay_receipt( + proposal, + rows, + apply_sql_sha256=APPLY_SQL_SHA256, + apply_sql_source="exact_executed_sql", + **kwargs, + ) + + +def _applied(proposal_type: str, apply_payload: dict) -> dict: + return { + "id": PROPOSAL_ID, + "proposal_type": proposal_type, + "status": "applied", + "payload": {"apply_payload": apply_payload, "title": "Private reviewed packet"}, + "reviewed_by_handle": "m3taversal", + "reviewed_by_agent_id": None, + "reviewed_at": "2026-07-14T01:00:00+00:00", + "review_note": "Reviewed exact strict payload.", + "applied_by_handle": "kb-apply", + "applied_by_agent_id": "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee", + "applied_at": "2026-07-14T01:01:00+00:00", + } + + +def _edge_proposal() -> dict: + return _applied( + "add_edge", + { + "from_claim": CLAIM_A, + "to_claim": CLAIM_B, + "edge_type": "supports", + "weight": 0.8, + }, + ) + + +def _edge_row() -> dict: + return { + "id": "11111111-1111-4111-8111-111111111111", + "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", + } + + +def test_add_edge_postflight_query_selects_exact_semantic_key() -> None: + sql = receipt.build_postflight_sql(_edge_proposal()) + assert "public.claim_edges" in sql + assert CLAIM_A in sql + assert CLAIM_B in sql + assert "supports" in sql + assert "to_jsonb(e)" in sql + + +def test_attach_evidence_and_approve_claim_queries_cover_all_row_types() -> None: + evidence_proposal = _applied( + "attach_evidence", + {"evidence": [{"claim_id": CLAIM_A, "source_id": SOURCE_A, "role": "grounds"}]}, + ) + evidence_sql = receipt.build_postflight_sql(evidence_proposal) + assert "public.claim_evidence" in evidence_sql + assert CLAIM_A in evidence_sql + assert SOURCE_A in evidence_sql + + bundle = { + "contract_version": 2, + "claims": [{"id": CLAIM_A}], + "sources": [{"id": SOURCE_A}], + "evidence": [{"claim_id": CLAIM_A, "source_id": SOURCE_A, "role": "grounds"}], + "edges": [{"from_claim": CLAIM_A, "to_claim": CLAIM_B, "edge_type": "supports"}], + "reasoning_tools": [{"id": TOOL_A}], + } + sql = receipt.build_postflight_sql(_applied("approve_claim", bundle)) + for table in ( + "public.claims", + "public.sources", + "public.claim_evidence", + "public.claim_edges", + "public.reasoning_tools", + ): + assert table in sql + + +def test_revise_strategy_query_pins_generated_rows_to_apply_transaction_time() -> None: + proposal = _applied( + "revise_strategy", + { + "agent_id": "11111111-1111-4111-8111-111111111111", + "strategy": { + "diagnosis": "Current diagnosis", + "guiding_policy": "Current policy", + "proximate_objectives": ["one"], + }, + "strategy_nodes": [{"node_type": "policy", "title": "Policy", "body": "Body", "rank": 1}], + }, + ) + sql = receipt.build_postflight_sql(proposal) + assert "matched_strategy" in sql + assert "n.created_at = s.created_at" in sql + assert proposal["applied_at"] in sql + assert "public.strategy_nodes" in sql + + +def test_receipt_hash_is_recoverable_and_independent_of_export_time() -> None: + proposal = _edge_proposal() + rows = {"claim_edges": [_edge_row()]} + first = _build_receipt(proposal, rows, exported_at_utc="2026-07-14T02:00:00+00:00") + second = _build_receipt(proposal, rows, exported_at_utc="2026-07-14T03:00:00+00:00") + assert first["replay_ready"] is True + assert first["apply_engine"] == { + "apply_sql_sha256": APPLY_SQL_SHA256, + "source": "exact_executed_sql", + } + assert first["expected_row_counts"] == {"claim_edges": 1} + assert first["actual_row_counts"] == {"claim_edges": 1} + assert first["hashes"]["replay_material_sha256"] == second["hashes"]["replay_material_sha256"] + assert first["exported_at_utc"] != second["exported_at_utc"] + + +def test_attach_evidence_receipt_matches_every_payload_controlled_field() -> None: + proposal = _applied( + "attach_evidence", + { + "evidence": [ + { + "claim_id": CLAIM_A, + "source_id": SOURCE_A, + "role": "grounds", + "weight": 0.7, + } + ] + }, + ) + row = { + "claim_id": CLAIM_A, + "source_id": SOURCE_A, + "role": "grounds", + "weight": 0.7, + "created_by": None, + "created_at": "2026-07-14T01:00:30+00:00", + } + data = _build_receipt(proposal, {"claim_evidence": [row]}) + assert data["checks"]["semantic_rows_match_strict_payload"] is True + + +def test_approve_claim_receipt_matches_all_supported_canonical_row_types() -> None: + agent_id = "11111111-1111-4111-8111-111111111111" + payload = { + "contract_version": 2, + "agent_id": agent_id, + "claims": [ + { + "id": CLAIM_A, + "type": "structural", + "text": "Atomic canonical claim", + "status": "open", + "confidence": 0.8, + "tags": ["replay"], + "created_by": None, + } + ], + "sources": [ + { + "id": SOURCE_A, + "source_type": "article", + "url": "https://example.test/source", + "storage_path": None, + "excerpt": "Exact source excerpt", + "hash": "sha256:test", + "created_by": None, + } + ], + "evidence": [ + { + "claim_id": CLAIM_A, + "source_id": SOURCE_A, + "role": "grounds", + "weight": 0.9, + "created_by": None, + } + ], + "edges": [ + { + "from_claim": CLAIM_A, + "to_claim": CLAIM_B, + "edge_type": "supports", + "weight": 0.6, + "created_by": None, + } + ], + "reasoning_tools": [ + { + "id": TOOL_A, + "agent_id": None, + "name": "Replay tool", + "description": "A replayable reasoning tool.", + "category": "test", + } + ], + } + rows = { + "claims": [ + { + **payload["claims"][0], + "superseded_by": None, + "created_at": "2026-07-14T01:00:30+00:00", + "updated_at": "2026-07-14T01:00:30+00:00", + } + ], + "sources": [ + { + **payload["sources"][0], + "captured_at": None, + "created_at": "2026-07-14T01:00:30+00:00", + } + ], + "claim_evidence": [ + { + **payload["evidence"][0], + "created_at": "2026-07-14T01:00:30+00:00", + } + ], + "claim_edges": [ + { + "id": "22222222-2222-4222-8222-222222222222", + **payload["edges"][0], + "created_at": "2026-07-14T01:00:30+00:00", + } + ], + "reasoning_tools": [ + { + **payload["reasoning_tools"][0], + "created_at": "2026-07-14T01:00:30+00:00", + } + ], + } + data = _build_receipt(_applied("approve_claim", payload), rows) + assert data["actual_row_counts"] == { + "claims": 1, + "sources": 1, + "claim_evidence": 1, + "claim_edges": 1, + "reasoning_tools": 1, + } + + +def test_revise_strategy_receipt_requires_the_fresh_active_rows() -> None: + agent_id = "11111111-1111-4111-8111-111111111111" + payload = { + "agent_id": agent_id, + "strategy": { + "diagnosis": "Current diagnosis", + "guiding_policy": "Current policy", + "proximate_objectives": ["one"], + }, + "strategy_nodes": [{"node_type": "policy", "title": "Policy", "body": "Body", "rank": 1}], + } + rows = { + "strategies": [ + { + "id": "33333333-3333-4333-8333-333333333333", + "agent_id": agent_id, + **payload["strategy"], + "version": 2, + "active": True, + "created_at": "2026-07-14T01:00:30+00:00", + } + ], + "strategy_nodes": [ + { + "id": "44444444-4444-4444-8444-444444444444", + "agent_id": agent_id, + **payload["strategy_nodes"][0], + "status": "active", + "horizon": None, + "measure": None, + "source_ref": None, + "metadata": {}, + "created_at": "2026-07-14T01:00:30+00:00", + "updated_at": "2026-07-14T01:00:30+00:00", + } + ], + } + proposal = _applied("revise_strategy", payload) + assert _build_receipt(proposal, rows)["replay_ready"] is True + rows["strategies"][0]["active"] = False + with pytest.raises(ValueError, match="semantic mismatch"): + _build_receipt(proposal, rows) + + +def test_receipt_refuses_missing_or_ambiguous_postflight_rows() -> None: + proposal = _edge_proposal() + with pytest.raises(ValueError, match="count mismatch"): + _build_receipt(proposal, {"claim_edges": []}) + with pytest.raises(ValueError, match="count mismatch"): + _build_receipt(proposal, {"claim_edges": [_edge_row(), _edge_row()]}) + + +def test_receipt_refuses_same_key_with_payload_value_mismatch() -> None: + row = _edge_row() + row["weight"] = 0.1 + with pytest.raises(ValueError, match="semantic mismatch"): + _build_receipt(_edge_proposal(), {"claim_edges": [row]}) + + +def test_receipt_refuses_legacy_applied_proposal_without_strict_payload() -> None: + proposal = _edge_proposal() + proposal["payload"] = {"rationale": "legacy freeform packet"} + with pytest.raises(ValueError, match="strict payload"): + receipt.build_postflight_sql(proposal) + + +def test_private_receipt_is_atomically_written_mode_0600(tmp_path: Path) -> None: + data = _build_receipt(_edge_proposal(), {"claim_edges": [_edge_row()]}) + output = tmp_path / "private" / "receipt.json" + receipt.write_private_receipt(output, data) + assert json.loads(output.read_text(encoding="utf-8"))["replay_ready"] is True + assert stat.S_IMODE(output.stat().st_mode) == 0o600 + assert not list(output.parent.glob(f".{output.name}.*")) + + +def test_apply_capture_writes_receipt_without_printing_canonical_rows( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + proposal = _edge_proposal() + output = tmp_path / "receipt.json" + args = Namespace( + container="teleo-pg", + db="teleo", + host="127.0.0.1", + role="kb_apply", + receipt_out=str(output), + receipt_dir=None, + ) + seen: list[str] = [] + + def fake_run_psql( + _args: Namespace, + sql: str, + _password: str, + *, + redact_output_on_error: bool = False, + ) -> str: + seen.append(sql) + assert redact_output_on_error is True + return json.dumps({"claim_edges": [_edge_row()]}) + + monkeypatch.setattr(ap, "run_psql", fake_run_psql) + path, data = ap.capture_replay_receipt(args, proposal, "not-logged") + assert path == output + assert output.exists() + assert data["replay_ready"] is True + assert data["apply_engine"]["source"] == "reconstructed_current_engine" + assert len(seen) == 1 + assert "public.claim_edges" in seen[0] + + args.receipt_out = str(tmp_path / "exact-executed.json") + _path, exact = ap.capture_replay_receipt( + args, + proposal, + "not-logged", + apply_sql="begin; select exact_executed_apply(); commit;", + ) + assert exact["apply_engine"] == { + "apply_sql_sha256": receipt.sha256_text("begin; select exact_executed_apply(); commit;"), + "source": "exact_executed_sql", + } + + +def test_receipt_only_guard_requires_applied_status() -> None: + proposal = _edge_proposal() + proposal["status"] = "approved" + with pytest.raises(SystemExit, match="requires an applied proposal"): + ap.assert_receiptable(proposal)