Merge pull request #136 from living-ip/codex/kb-apply-replay-receipts-20260714

Retain replay receipts for guarded KB applies

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 <proposal-uuid> --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
+The remaining recompilation capability is to join genesis, accepted packets,
-complete corpus runner and replay ledger. One-document preparation and staging
+and their row-level receipts into a complete corpus runner. One-document
-do not yet prove a clean database can be rebuilt semantically from every source.
+preparation, staging, and receipt capture do not yet prove a clean database can
+be rebuilt semantically from every source.
 
 ## Definition Of Working
 

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
 

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"
+    ]
+  }
+}

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,
+seconds. New strict applies now emit a recoverable private row-level replay
-review decision, and apply receipt is not yet complete because historical rows
+receipt so the provenance gap stops growing. Rebuilding a blank database from
-predate a complete replay ledger.
+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

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
 
 

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

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)