"""Tests for scripts/assemble_telegram_visible_direct_claim_capture_receipt.py.""" from __future__ import annotations import json import sys from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "scripts")) import assemble_telegram_visible_direct_claim_capture_receipt as receipt # noqa: E402 STRUCTURED_FIXTURE_READBACKS = { "DC-01": "\nDB readback: Proposal: f004bbb2-ac9a-481f-b7b8-74319373ba6a; Status: applied; Applied at: 2026-07-05.", "DC-02": "\nDB readback: Proposal: a64df080-8502-42e2-98f4-9bbdecb8da73; Status: approved; Applied at: none.", "DC-03": "\nDB readback: Proposal: a64df080-8502-42e2-98f4-9bbdecb8da73; Status: approved; Applied at: none.", "DC-04": ( "\nDB readback: Proposal: e987b07c-9c23-45ba-b95c-d6d04c6536c7; Status: pending_review; " "Applied at: none; Source_ref: telegram-resolved-attachment:doc_996e2e86cd94." ), "DC-05": ( "\nDB readback: claims: 1837; sources: 4145; claim_edges: 4916; " "claim_evidence: 4670; kb_proposals: 26." ), "DC-06": "\nDB readback: Proposal: f004bbb2-ac9a-481f-b7b8-74319373ba6a; Status: applied; Applied at: 2026-07-05.", } KNOWN_GOOD_REPLIES = { "DC-01": ( "Mostly still proposals. Approved is not the same as applied. Canonical public.* changes require applied_at " "and row-level postflight proof; I cannot claim the DB changed without that readback. Next Cory-style " "follow-up: validate or rebuild a strict payload for the approved proposal before review and authorization." ), "DC-02": ( "No, not canonical. Helmer is approved and staged, but its public.sources, public.claims, evidence, edges, " "and reasoning tool rows are not applied. Explicit operator authorization comes only after strict payload " "validation. Next Cory-style follow-up: validate or rebuild the payload, review it, then authorize apply and " "postflight." ), "DC-03": ( "No. Fresh readback: the decision-matrix schema is absent; reviewer status is not a decision-matrix vote. " "matrix_voters, proposal_votes, and proposal_decisions do not exist, so canonical public.* state must be " "checked separately. Next Cory-style follow-up: validate the strict proposal payload before using the " "reviewer/admin path." ), "DC-04": ( "Not just pointer mismatch. Telegram file refs, document evaluations, proposal source_ref keys, raw files, " "and canonical public.sources rows are distinct layers; pending_review also reflects reviewer state. The " "pointer claim does not prove the whole cause. Next Cory-style follow-up: run a row-link audit and draft the " "guarded source/evidence apply contract." ), "DC-05": ( "Staging yes, canonical KB change not safe to demo from chat. Approved is not the same as applied. A staging " "write leaves a proposal UUID and row-level readback artifact; canonical public.claim_edges mutation requires " "explicit operator authorization and retained postflight proof. Next Cory-style follow-up: stage a separately " "reviewed strict add_edge canary with two existing canonical claim IDs." ), "DC-06": ( "No. SOUL.md is a runtime/rendered artifact, not canonical Postgres and not the source of truth. Canonical " "identity requires new or updated public.* row IDs plus renderer and postflight proof; direct edits do not " "change canonical identity. Next Cory-style follow-up: audit the identity rows and stage the missing proposal." ), } def load_packet() -> dict: return json.loads(receipt.DEFAULT_PACKET.read_text(encoding="utf-8")) def retained_visible_replies() -> dict[str, str]: data = json.loads( ( REPO_ROOT / "docs/reports/leo-working-state-20260709/telegram-visible-direct-claim-capture-results-current.json" ).read_text(encoding="utf-8") ) return {item["prompt_id"]: item["reply"] for item in data["results"]} def good_capture(packet: dict) -> dict: messages = [] for index, item in enumerate(packet["exact_messages"], start=1): messages.append( { "prompt_id": item["prompt_id"], "sent_text": item["message"], "reply": KNOWN_GOOD_REPLIES[item["prompt_id"]] + STRUCTURED_FIXTURE_READBACKS[item["prompt_id"]], "sent_at_utc": f"2026-07-10T03:0{index}:00Z", "reply_at_utc": f"2026-07-10T03:0{index}:30Z", "telegram_message_id": f"sent-{index}", "reply_message_id": f"reply-{index}", } ) return { "operator_authorization_text": packet["explicit_operator_authorization_text"], "capture_source": { "platform": "telegram", "chat_label": "Leo", "chat_id": "-5146042086", "captured_at_utc": "2026-07-10T03:08:00Z", "captured_by": "codex", "final_screenshot": "outputs/telegram-visible-direct-claim-suite-20260710/final.png", "final_accessibility": "outputs/telegram-visible-direct-claim-suite-20260710/final.txt", }, "db_before_after": { "before": {"public.claims": 1837}, "after": {"public.claims": 1837}, "db_counts_changed": False, }, "service_readback": { "before": {"ActiveState": "active", "SubState": "running"}, "after": {"ActiveState": "active", "SubState": "running"}, }, "messages": messages, } def test_receipt_without_capture_emits_awaiting_capture_template(): packet = load_packet() data = receipt.build_receipt(packet, None, source_capture_path=None) assert data["status"] == "awaiting_capture" assert data["receipt_ready"] is False assert data["capture_template"]["messages"][0]["prompt_id"] == "DC-01" assert data["checks"]["missing_prompt_ids"] == ["DC-01", "DC-02", "DC-03", "DC-04", "DC-05", "DC-06"] def test_good_capture_scores_and_receipt_passes(): packet = load_packet() data = receipt.build_receipt(packet, good_capture(packet), source_capture_path=Path("capture.json")) assert data["status"] == "pass" assert data["receipt_ready"] is True assert data["score"]["pass"] is True assert data["score"]["passes"] == data["score"]["expected_prompt_count"] == 6 assert len(data["results"]) == 6 assert data["results"][0]["evidence_tier"] == "telegram_visible_capture" def test_retained_visible_capture_exposes_dc05_apply_path_issue(): packet = load_packet() capture = good_capture(packet) replies = retained_visible_replies() capture["messages"][4]["reply"] = replies["DC-05"] + STRUCTURED_FIXTURE_READBACKS["DC-05"] data = receipt.build_receipt(packet, capture, source_capture_path=Path("capture.json")) assert data["status"] == "fail" assert data["score"]["failures"][0]["prompt_id"] == "DC-05" assert data["score"]["failures"][0]["response_issue_detected"] is True def test_capture_with_message_text_mismatch_fails(): packet = load_packet() capture = good_capture(packet) capture["messages"][0]["sent_text"] = "wrong" data = receipt.build_receipt(packet, capture, source_capture_path=Path("capture.json")) assert data["status"] == "fail" assert data["receipt_ready"] is False assert data["checks"]["sent_text_exact"] is False assert data["checks"]["text_mismatch_prompt_ids"] == ["DC-01"] def test_capture_with_empty_reply_fails(): packet = load_packet() capture = good_capture(packet) capture["messages"][1]["reply"] = "" data = receipt.build_receipt(packet, capture, source_capture_path=Path("capture.json")) assert data["status"] == "fail" assert data["receipt_ready"] is False assert data["checks"]["replies_nonempty"] is False assert data["checks"]["empty_reply_prompt_ids"] == ["DC-02"] def test_main_writes_awaiting_capture_artifacts(tmp_path: Path): out = tmp_path / "receipt.json" md = tmp_path / "receipt.md" results = tmp_path / "results.json" code = receipt.main(["--out", str(out), "--markdown-out", str(md), "--results-out", str(results)]) assert code == 0 assert json.loads(out.read_text(encoding="utf-8"))["status"] == "awaiting_capture" assert "Telegram-Visible Direct-Claim Capture Receipt" in md.read_text(encoding="utf-8") assert not results.exists() def test_receipt_records_cli_override_paths(tmp_path: Path): packet_path = tmp_path / "packet.json" results = tmp_path / "results.json" packet = load_packet() packet_path.write_text(json.dumps(packet), encoding="utf-8") data = receipt.build_receipt( packet, good_capture(packet), source_capture_path=tmp_path / "capture.json", packet_path=packet_path, results_out=results, ) assert data["packet_path"] == str(packet_path) assert data["results_out"] == str(results)