retain replay receipts for guarded KB applies
This commit is contained in:
twentyOne2x
parent
83141bc97d
commit
7ea07cc4fc
7 changed files with 1147 additions and 12 deletions
|
|
@ -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
|
||||
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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
482
scripts/kb_apply_replay_receipt.py
Normal file
482
scripts/kb_apply_replay_receipt.py
Normal file
|
|
@ -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
|
||||
401
tests/test_kb_apply_replay_receipt.py
Normal file
401
tests/test_kb_apply_replay_receipt.py
Normal file
|
|
@ -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)
|
||||
Loading…
Reference in a new issue