#!/usr/bin/env python3 """Assemble and score a Telegram-visible direct-claim capture receipt. This script does not send Telegram messages and does not mutate the database. It validates a retained capture from the authorized DC-01..DC-06 Telegram group run against the authorization packet, emits scorer-ready results, and records a receipt/status artifact. With no capture JSON it emits an awaiting-capture template so the required evidence shape is explicit. """ from __future__ import annotations import argparse import json from datetime import datetime, timezone from pathlib import Path from typing import Any import working_leo_open_ended_benchmark as benchmark REPORT_DIR = Path("docs/reports/leo-working-state-20260709") DEFAULT_PACKET = REPORT_DIR / "telegram-visible-direct-claim-authorization-packet-current.json" DEFAULT_OUT = REPORT_DIR / "telegram-visible-direct-claim-capture-receipt-current.json" DEFAULT_MARKDOWN_OUT = REPORT_DIR / "telegram-visible-direct-claim-capture-receipt-current.md" DEFAULT_RESULTS_OUT = REPORT_DIR / "telegram-visible-direct-claim-capture-results-current.json" def _load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def expected_messages(packet: dict[str, Any]) -> list[dict[str, Any]]: return [ { "prompt_id": item["prompt_id"], "dimension": item["dimension"], "sent_text": item["message"], "required_signals": item.get("required_signals", []), } for item in packet.get("exact_messages", []) ] def blank_capture_template(packet: dict[str, Any]) -> dict[str, Any]: target = packet.get("target", {}) return { "operator_authorization_text": packet.get("explicit_operator_authorization_text", ""), "capture_source": { "platform": target.get("platform", "telegram"), "chat_label": target.get("chat_label", ""), "chat_id": target.get("chat_id", ""), "captured_at_utc": "", "captured_by": "", "final_screenshot": "", "final_accessibility": "", }, "db_before_after": { "before": {}, "after": {}, "db_counts_changed": None, }, "service_readback": { "before": {}, "after": {}, }, "messages": [ { "prompt_id": item["prompt_id"], "sent_text": item["sent_text"], "reply": "", "sent_at_utc": "", "reply_at_utc": "", "telegram_message_id": "", "reply_message_id": "", } for item in expected_messages(packet) ], } def _present(value: Any) -> bool: if value is None: return False if isinstance(value, str): return bool(value.strip()) if isinstance(value, dict | list): return bool(value) return True def _capture_messages(capture: dict[str, Any]) -> list[dict[str, Any]]: messages = capture.get("messages") return [item for item in messages if isinstance(item, dict)] if isinstance(messages, list) else [] def validate_capture(packet: dict[str, Any], capture: dict[str, Any] | None) -> dict[str, Any]: expected = expected_messages(packet) expected_by_id = {item["prompt_id"]: item for item in expected} expected_ids = [item["prompt_id"] for item in expected] if capture is None: return { "capture_present": False, "authorization_text_matches": False, "target_matches_packet": False, "message_ids_exact": False, "sent_text_exact": False, "replies_nonempty": False, "db_before_after_present": False, "service_readback_present": False, "final_visual_proof_present": False, "capture_complete_for_scoring": False, "missing_prompt_ids": expected_ids, "unexpected_prompt_ids": [], "text_mismatch_prompt_ids": [], "empty_reply_prompt_ids": expected_ids, } target = packet.get("target", {}) source = capture.get("capture_source") if isinstance(capture.get("capture_source"), dict) else {} messages = _capture_messages(capture) by_id = {str(item.get("prompt_id")): item for item in messages if item.get("prompt_id")} found_ids = [str(item.get("prompt_id")) for item in messages if item.get("prompt_id")] missing = [prompt_id for prompt_id in expected_ids if prompt_id not in by_id] unexpected = [prompt_id for prompt_id in found_ids if prompt_id not in expected_by_id] text_mismatch = [ prompt_id for prompt_id in expected_ids if prompt_id in by_id and str(by_id[prompt_id].get("sent_text") or "") != expected_by_id[prompt_id]["sent_text"] ] empty_replies = [ prompt_id for prompt_id in expected_ids if prompt_id in by_id and not str(by_id[prompt_id].get("reply") or "").strip() ] authorization_text = str(capture.get("operator_authorization_text") or "") target_matches = ( source.get("platform") == target.get("platform") and source.get("chat_label") == target.get("chat_label") and source.get("chat_id") == target.get("chat_id") ) db_before_after = capture.get("db_before_after") service_readback = capture.get("service_readback") final_visual = _present(source.get("final_screenshot")) and _present(source.get("final_accessibility")) message_ids_exact = found_ids == expected_ids sent_text_exact = not text_mismatch and message_ids_exact replies_nonempty = not empty_replies and message_ids_exact return { "capture_present": True, "authorization_text_matches": authorization_text == packet.get("explicit_operator_authorization_text"), "target_matches_packet": target_matches, "message_ids_exact": message_ids_exact, "sent_text_exact": sent_text_exact, "replies_nonempty": replies_nonempty, "db_before_after_present": isinstance(db_before_after, dict) and _present(db_before_after.get("before")) and _present(db_before_after.get("after")), "service_readback_present": _present(service_readback), "final_visual_proof_present": final_visual, "capture_complete_for_scoring": message_ids_exact and sent_text_exact and replies_nonempty, "missing_prompt_ids": missing, "unexpected_prompt_ids": unexpected, "text_mismatch_prompt_ids": text_mismatch, "empty_reply_prompt_ids": empty_replies, } def scorer_results(packet: dict[str, Any], capture: dict[str, Any]) -> list[dict[str, Any]]: by_id = {str(item.get("prompt_id")): item for item in _capture_messages(capture) if item.get("prompt_id")} results = [] for item in expected_messages(packet): captured = by_id[item["prompt_id"]] results.append( { "prompt_id": item["prompt_id"], "dimension": item["dimension"], "prompt": item["sent_text"], "reply": str(captured.get("reply") or ""), "ok": True, "mutates_kb": False, "evidence_tier": "telegram_visible_capture", "sent_at_utc": captured.get("sent_at_utc", ""), "reply_at_utc": captured.get("reply_at_utc", ""), "telegram_message_id": captured.get("telegram_message_id", ""), "reply_message_id": captured.get("reply_message_id", ""), } ) return results def score_capture(packet: dict[str, Any], results: list[dict[str, Any]]) -> dict[str, Any]: catalog = benchmark.prompt_catalog(include_direct_claim_followups=True) return benchmark.score_result_subset( results, catalog=catalog, expected_prompt_ids=list(packet.get("target", {}).get("expected_prompt_ids", [])), ) def build_receipt( packet: dict[str, Any], capture: dict[str, Any] | None, *, source_capture_path: Path | None, packet_path: Path = DEFAULT_PACKET, results_out: Path = DEFAULT_RESULTS_OUT, ) -> dict[str, Any]: checks = validate_capture(packet, capture) results: list[dict[str, Any]] = [] score: dict[str, Any] | None = None if capture is not None and checks["capture_complete_for_scoring"]: results = scorer_results(packet, capture) score = score_capture(packet, results) score_passes = bool(score and score.get("pass") is True) checks["score_passes"] = score_passes receipt_ready = ( checks["capture_present"] and checks["authorization_text_matches"] and checks["target_matches_packet"] and checks["capture_complete_for_scoring"] and checks["db_before_after_present"] and checks["service_readback_present"] and checks["final_visual_proof_present"] and score_passes ) if not checks["capture_present"]: status = "awaiting_capture" else: status = "pass" if receipt_ready else "fail" return { "generated_at_utc": datetime.now(timezone.utc).isoformat(), "mode": "telegram_visible_direct_claim_capture_receipt", "status": status, "receipt_ready": receipt_ready, "capture_present": checks["capture_present"], "telegram_visible_messages_sent": bool(checks["capture_present"]), "production_apply_executed": False, "mutates_db": False, "packet_path": str(packet_path), "source_capture_path": str(source_capture_path) if source_capture_path else "", "results_out": str(results_out), "checks": checks, "results": results, "score": score, "capture_template": blank_capture_template(packet) if capture is None else None, "claim_ceiling": ( "This receipt validates retained Telegram-visible capture evidence only. It does not send Telegram " "messages, does not run production apply, and does not prove canonical DB mutation. A pass requires exact " "DC-01..DC-06 messages, matching operator authorization, nonempty visible replies, DB/service readbacks, " "visual proof, and a 6/6 direct-claim score." ), } def write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") def write_markdown(path: Path, data: dict[str, Any]) -> None: checks = data["checks"] lines = [ "# Telegram-Visible Direct-Claim Capture Receipt", "", f"Generated UTC: `{data['generated_at_utc']}`", f"Mode: `{data['mode']}`", f"Status: `{data['status']}`", f"Receipt ready: `{data['receipt_ready']}`", f"Capture present: `{data['capture_present']}`", f"Telegram-visible messages sent: `{data['telegram_visible_messages_sent']}`", f"Production apply executed: `{data['production_apply_executed']}`", f"Mutates DB: `{data['mutates_db']}`", "", "## Checks", "", ] for key, value in sorted(checks.items()): lines.append(f"- `{key}`: `{value}`") if data.get("score"): score = data["score"] lines.extend( [ "", "## Score", "", f"- Pass: `{score['pass']}`", f"- Prompts scored: `{score['passes']}/{score['expected_prompt_count']}`", f"- Missing prompt IDs: `{', '.join(score['missing_prompt_ids'])}`", f"- Unexpected prompt IDs: `{', '.join(score['unexpected_prompt_ids'])}`", ] ) if data.get("capture_template"): lines.extend( [ "", "## Capture Template", "", "Populate `messages[].reply`, timestamp/id fields, `db_before_after`, `service_readback`, and " "`capture_source.final_screenshot` / `capture_source.final_accessibility`, then rerun this script " "with `--capture-json`.", ] ) lines.extend(["", "## Claim Ceiling", "", data["claim_ceiling"], ""]) path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(lines), encoding="utf-8") def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--packet", type=Path, default=DEFAULT_PACKET) parser.add_argument("--capture-json", type=Path) parser.add_argument("--out", type=Path, default=DEFAULT_OUT) parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MARKDOWN_OUT) parser.add_argument("--results-out", type=Path, default=DEFAULT_RESULTS_OUT) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) packet = _load_json(args.packet) capture = _load_json(args.capture_json) if args.capture_json else None receipt = build_receipt( packet, capture, source_capture_path=args.capture_json, packet_path=args.packet, results_out=args.results_out, ) if receipt["results"]: write_json(args.results_out, {"results": receipt["results"]}) receipt["results_out"] = str(args.results_out) write_json(args.out, receipt) write_markdown(args.markdown_out, receipt) print( json.dumps( { "out": str(args.out), "markdown_out": str(args.markdown_out), "results_out": str(args.results_out) if receipt["results"] else "", "status": receipt["status"], "receipt_ready": receipt["receipt_ready"], }, indent=2, sort_keys=True, ) ) return 0 if receipt["status"] in {"pass", "awaiting_capture"} else 2 if __name__ == "__main__": raise SystemExit(main())