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)