teleo-infrastructure/tests/test_kb_apply_replay_receipt.py

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)