#!/usr/bin/env python3 """Verify retained authenticated-Chrome proof for Leo's unseen Telegram chain.""" from __future__ import annotations import argparse import hashlib import json import re from datetime import datetime, timezone from pathlib import Path from typing import Any try: import build_telegram_visible_unseen_chain_packet as packet_builder import verify_leo_unseen_reasoning_chain as unseen except ImportError: # pragma: no cover - imported as scripts.* in tests from scripts import build_telegram_visible_unseen_chain_packet as packet_builder from scripts import verify_leo_unseen_reasoning_chain as unseen SCHEMA = "livingip.telegramVisibleUnseenChainReceipt.v1" REPORT_DIR = Path("docs/reports/leo-working-state-20260709") DEFAULT_PACKET = packet_builder.DEFAULT_OUT DEFAULT_PREFLIGHT = REPORT_DIR / "telegram-visible-unseen-chain-preflight-current.json" DEFAULT_POSTFLIGHT = REPORT_DIR / "telegram-visible-unseen-chain-postflight-current.json" DEFAULT_OUT = REPORT_DIR / "telegram-visible-unseen-chain-capture-receipt-current.json" DEFAULT_MARKDOWN_OUT = REPORT_DIR / "telegram-visible-unseen-chain-capture-receipt-current.md" def _load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def _sha256_file(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() 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 blank_capture_template(packet: dict[str, Any], preflight: dict[str, Any]) -> dict[str, Any]: target = packet.get("target") or {} return { "protocol_sha256": packet.get("protocol_sha256"), "preflight_state_token_sha256": preflight.get("state_token_sha256"), "operator_authorization_text": packet.get("explicit_operator_authorization_text"), "operator_authorization_captured_at_utc": "", "extra_messages_sent": 0, "capture_source": { "platform": "telegram", "transport": "authenticated_chrome_telegram_ui", "chat_label": target.get("chat_label"), "chat_id": target.get("chat_id"), "captured_at_utc": "", "captured_by": "", "final_screenshot": "", "final_accessibility": "", }, "messages": [ { "sequence": item["sequence"], "prompt_id": item["prompt_id"], "sent_text": item["message"], "reply": "", "sent_at_utc": "", "reply_at_utc": "", "telegram_message_id": "", "reply_message_id": "", "turn_screenshot": "", "turn_accessibility": "", } for item in packet.get("exact_messages", []) ], } def _resolve_evidence(path_value: str, evidence_root: Path) -> dict[str, Any]: if not path_value.strip(): return {"path": path_value, "exists": False, "bytes": 0, "sha256": ""} path = Path(path_value) if not path.is_absolute(): path = evidence_root / path resolved = path.resolve() exists = resolved.is_file() return { "path": str(resolved), "exists": exists, "bytes": resolved.stat().st_size if exists else 0, "sha256": _sha256_file(resolved) if exists else "", } 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 _normalized_text(value: str) -> str: return re.sub(r"\s+", " ", value).strip().casefold() def _accessibility_binds_message(evidence: dict[str, Any], *, sent_text: str, reply: str) -> bool: if not evidence.get("exists") or not evidence.get("path"): return False text = Path(str(evidence["path"])).read_text(encoding="utf-8", errors="replace") normalized = _normalized_text(text) return _normalized_text(sent_text) in normalized and _normalized_text(reply) in normalized def _subject_match(postflight: dict[str, Any]) -> dict[str, Any]: subject = (postflight.get("remote") or {}).get("subject_readback") or {} matches = subject.get("matches") if isinstance(subject, dict) else [] return matches[0] if isinstance(matches, list) and len(matches) == 1 else {} def _semantic_checks(replies: list[str], *, subject_ref: str, subject_match: dict[str, Any]) -> dict[str, bool]: if len(replies) != 3: return { "visible_db_object_selected_without_operator_id": False, "visible_weak_support_explained": False, "same_subject_survived_challenge": False, "assumption_support_and_falsifier_separated": False, "multiple_review_only_candidates_proposed": False, "objection_revised_candidate": False, "resolving_source_named": False, "review_approval_apply_boundary_explicit": False, "reply_claimed_no_mutation": False, "independent_db_object_matches_visible_subject": False, } first, second, third = replies first_refs = unseen._short_refs(first) second_refs = unseen._short_refs(second) shared_terms = unseen._shared_subject_terms(first, second) selected = subject_ref[:8].casefold() if subject_ref else "" visible_subject = bool(selected and (selected in first.casefold() or selected in second.casefold())) same_subject = bool( visible_subject and ( (selected in first.casefold() and selected in second.casefold()) or first_refs & second_refs or len(shared_terms) >= 2 ) ) row = subject_match.get("row") if isinstance(subject_match.get("row"), dict) else {} row_id = str(row.get("id") or "").casefold() proposals = unseen._contains(second, "proposal a", "proposal b") or ( unseen._contains_any(second, "two proposals", "two replacement claims") and "1" in second and "2" in second ) return { "visible_db_object_selected_without_operator_id": unseen._contains_any(first, "claim", "belief", "axiom") and unseen._contains_any(first, "database", "db", "public.") and unseen._contains_any(first, "confidence", "support", "evidence"), "visible_weak_support_explained": unseen._contains_any( first, "nothing supports", "zero evidence", "no evidence", "no claim-evidence", "no_source_pointer", "no external primary source", ), "same_subject_survived_challenge": same_subject, "assumption_support_and_falsifier_separated": unseen._contains_any(second, "assumption") and unseen._contains_any(second, "observed support", "evidence") and unseen._contains_any(second, "falsifier", "counterexample", "would falsify"), "multiple_review_only_candidates_proposed": proposals and unseen._contains_any(second, "review", "staged", "proposal") and unseen._contains_any( second, "nothing applied", "not live knowledge", "staged only", "not an apply", ), "objection_revised_candidate": unseen._contains_any(third, "revised proposal", "revision") and unseen._contains(third, "mechanism", "outcome"), "resolving_source_named": unseen._contains_any( third, "what would resolve", "source would resolve", "evidence would resolve" ) and unseen._contains_any( third, "study", "dataset", "source", "randomized", "quasi-experimental", "replication", "ostrom", ), "review_approval_apply_boundary_explicit": unseen._contains_any( third, "review", "pending_review", "human review" ) and unseen._contains_any(third, "approve", "approved", "approval") and unseen._contains_any(third, "apply", "applied_at", "canonical"), "reply_claimed_no_mutation": unseen._contains_any( third, "not an apply", "nothing staged or changed", "canonical unchanged", "staged intent, not knowledge", "proposal (not an apply)", ), "independent_db_object_matches_visible_subject": bool( visible_subject and row_id and row_id.startswith(selected) ), } def verify_capture( packet: dict[str, Any], preflight: dict[str, Any], capture: dict[str, Any] | None, postflight: dict[str, Any] | None, *, packet_path: Path, evidence_root: Path, ) -> dict[str, Any]: if capture is None: return { "schema": SCHEMA, "generated_at_utc": datetime.now(timezone.utc).isoformat(), "mode": "telegram_visible_unseen_chain_capture_verifier", "status": "awaiting_action_time_authorization", "receipt_ready": False, "current_tier": "T0_packet_plus_T3_preflight", "required_tier": "T3_live_readonly", "telegram_visible_messages_sent": False, "mutates_db": False, "packet_path": str(packet_path), "preflight_path": str(DEFAULT_PREFLIGHT), "postflight_path": str(DEFAULT_POSTFLIGHT), "capture_template": blank_capture_template(packet, preflight), "checks": { "packet_ready": packet.get("ready_to_request_action_time_authorization") is True, "preflight_passed": preflight.get("status") == "pass", "action_time_authorization_present": False, "capture_present": False, "postflight_present": False, }, "claim_ceiling": { "proven": "The exact packet and live zero-mutation preflight are ready.", "not_proven": [ "Telegram-visible message delivery", "Telegram-visible Leo replies", "post-send fingerprint equality", ], }, } messages = _capture_messages(capture) expected = packet.get("exact_messages") or [] capture_source = capture.get("capture_source") if isinstance(capture.get("capture_source"), dict) else {} replies = [str(item.get("reply") or "") for item in messages] ids = [str(item.get("telegram_message_id") or "") for item in messages] reply_ids = [str(item.get("reply_message_id") or "") for item in messages] screenshot = _resolve_evidence(str(capture_source.get("final_screenshot") or ""), evidence_root) accessibility = _resolve_evidence(str(capture_source.get("final_accessibility") or ""), evidence_root) turn_evidence = [ { "prompt_id": item.get("prompt_id"), "screenshot": _resolve_evidence(str(item.get("turn_screenshot") or ""), evidence_root), "accessibility": _resolve_evidence(str(item.get("turn_accessibility") or ""), evidence_root), } for item in messages ] postflight = postflight or {} subject_ref = str(postflight.get("subject_ref") or "") subject_match = _subject_match(postflight) semantics = _semantic_checks(replies, subject_ref=subject_ref, subject_match=subject_match) pre_fingerprint = (preflight.get("remote") or {}).get("fingerprint_after") or {} post_fingerprint = (postflight.get("remote") or {}).get("fingerprint_after") or {} pre_service = (preflight.get("remote") or {}).get("service_after") or {} post_service = (postflight.get("remote") or {}).get("service_after") or {} service_keys = ("MainPID", "NRestarts", "ExecMainStartTimestamp") handler_binding = packet.get("handler_receipt_binding") or {} handler_path = Path(str(handler_binding.get("path") or "")) if not handler_path.is_absolute(): handler_path = evidence_root / handler_path expected_ids = [str(item.get("prompt_id") or "") for item in expected] actual_ids = [str(item.get("prompt_id") or "") for item in messages] checks = { "packet_ready": packet.get("ready_to_request_action_time_authorization") is True, "packet_file_hash_matches_preflight": _sha256_file(packet_path) == preflight.get("packet_file_sha256"), "handler_receipt_binding_unchanged": handler_path.is_file() and _sha256_file(handler_path) == handler_binding.get("sha256"), "preflight_passed": preflight.get("status") == "pass" and preflight.get("phase") == "preflight", "postflight_passed": postflight.get("status") == "pass" and postflight.get("phase") == "postflight", "protocol_bound_across_all_artifacts": capture.get("protocol_sha256") == preflight.get("protocol_sha256") == postflight.get("protocol_sha256") == packet.get("protocol_sha256"), "preflight_state_token_bound_to_capture": capture.get("preflight_state_token_sha256") == preflight.get("state_token_sha256"), "exact_action_time_authorization_captured": capture.get("operator_authorization_text") == packet.get("explicit_operator_authorization_text") and _present(capture.get("operator_authorization_captured_at_utc")), "authenticated_chrome_transport_proven": capture_source.get("platform") == "telegram" and capture_source.get("transport") == "authenticated_chrome_telegram_ui", "bot_api_not_substituted": "bot_api" not in json.dumps(capture_source).casefold(), "exact_destination_matches": capture_source.get("chat_label") == (packet.get("target") or {}).get("chat_label") and capture_source.get("chat_id") == (packet.get("target") or {}).get("chat_id"), "exact_three_messages_and_no_extras": len(messages) == len(expected) == 3 and capture.get("extra_messages_sent") == 0, "prompt_ids_and_sequence_exact": actual_ids == expected_ids and [item.get("sequence") for item in messages] == [1, 2, 3], "sent_text_exact": [item.get("sent_text") for item in messages] == [item.get("message") for item in expected], "visible_replies_nonempty": len(replies) == 3 and all(reply.strip() for reply in replies), "message_and_reply_ids_present_unique": all(ids) and all(reply_ids) and len(set(ids)) == 3 and len(set(reply_ids)) == 3, "message_and_reply_timestamps_present": all( _present(item.get("sent_at_utc")) and _present(item.get("reply_at_utc")) for item in messages ), "capture_metadata_present": all( _present(capture_source.get(key)) for key in ("captured_at_utc", "captured_by") ), "visual_evidence_files_present": screenshot["exists"] and screenshot["bytes"] > 0 and accessibility["exists"] and accessibility["bytes"] > 0 and len(turn_evidence) == 3 and all( item["screenshot"]["exists"] and item["screenshot"]["bytes"] > 0 and item["accessibility"]["exists"] and item["accessibility"]["bytes"] > 0 for item in turn_evidence ), "turn_accessibility_binds_exact_messages_and_replies": len(turn_evidence) == len(messages) == 3 and all( _accessibility_binds_message( evidence["accessibility"], sent_text=str(message.get("sent_text") or ""), reply=str(message.get("reply") or ""), ) for evidence, message in zip(turn_evidence, messages, strict=True) ), "canonical_fingerprint_unchanged_from_preflight": bool( pre_fingerprint.get("fingerprint_sha256") and pre_fingerprint.get("fingerprint_sha256") == post_fingerprint.get("fingerprint_sha256") ), "gateway_process_unchanged_from_preflight": all( pre_service.get(key) == post_service.get(key) for key in service_keys ), **semantics, } outcomes = { "telegram_visible_three_turn_chain": all( checks[key] for key in ( "authenticated_chrome_transport_proven", "exact_destination_matches", "exact_three_messages_and_no_extras", "sent_text_exact", "visible_replies_nonempty", "visual_evidence_files_present", "turn_accessibility_binds_exact_messages_and_replies", ) ), "visible_db_grounded_challenge": all( checks[key] for key in ( "visible_db_object_selected_without_operator_id", "visible_weak_support_explained", "same_subject_survived_challenge", "assumption_support_and_falsifier_separated", "independent_db_object_matches_visible_subject", ) ), "visible_review_only_revision": all( checks[key] for key in ( "multiple_review_only_candidates_proposed", "objection_revised_candidate", "resolving_source_named", "review_approval_apply_boundary_explicit", "reply_claimed_no_mutation", ) ), "canonical_state_unchanged": checks["canonical_fingerprint_unchanged_from_preflight"], } passed = all(checks.values()) and all(outcomes.values()) return { "schema": SCHEMA, "generated_at_utc": datetime.now(timezone.utc).isoformat(), "mode": "telegram_visible_unseen_chain_capture_verifier", "status": "pass" if passed else "fail", "receipt_ready": passed, "current_tier": "T3_live_readonly" if passed else "T2_or_lower_incomplete_visible_proof", "required_tier": "T3_live_readonly", "telegram_visible_messages_sent": True, "production_apply_executed": False, "mutates_db": False, "packet_path": str(packet_path), "checks": checks, "outcomes": outcomes, "selected_subject": { "ref": subject_ref, "table": subject_match.get("table"), "row_id": (subject_match.get("row") or {}).get("id"), }, "evidence": { "final_screenshot": screenshot, "final_accessibility": accessibility, "turn_evidence": turn_evidence, "telegram_message_ids": ids, "telegram_reply_ids": reply_ids, "preflight_state_token_sha256": preflight.get("state_token_sha256"), "postflight_state_token_sha256": postflight.get("state_token_sha256"), "canonical_fingerprint_sha256": post_fingerprint.get("fingerprint_sha256"), }, "claim_ceiling": { "proven": ( "One exact authenticated-Chrome Telegram chain visibly selected and challenged a DB object, " "proposed and revised review-only knowledge, preserved the approval/apply boundary, and left the " "canonical fingerprint unchanged." if passed else "No T3 claim; at least one visible, semantic, binding, or state check failed." ), "not_proven": [ "canonical proposal staging or apply", "agent-to-agent review", "GCP runtime parity for the Telegram-visible chain", ], }, } 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: lines = [ "# Telegram-Visible Unseen Chain Capture Receipt", "", f"Generated UTC: `{data['generated_at_utc']}`", f"Status: `{data['status']}`", f"Receipt ready: `{data['receipt_ready']}`", f"Current tier: `{data['current_tier']}`", f"Required tier: `{data['required_tier']}`", f"Telegram-visible messages sent: `{data['telegram_visible_messages_sent']}`", f"Mutates DB: `{data['mutates_db']}`", "", "## Checks", "", ] lines.extend(f"- `{key}`: `{value}`" for key, value in sorted(data.get("checks", {}).items())) if data.get("capture_template"): lines.extend( [ "", "## Awaiting Authorized Capture", "", "The packet and preflight are ready. No Telegram message has been sent by this verifier.", ] ) lines.extend( [ "", "## Claim Ceiling", "", data["claim_ceiling"]["proven"], "", ] ) 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("--preflight", type=Path, default=DEFAULT_PREFLIGHT) parser.add_argument("--capture-json", type=Path) parser.add_argument("--postflight", type=Path, default=DEFAULT_POSTFLIGHT) parser.add_argument("--evidence-root", type=Path, default=Path(".")) parser.add_argument("--out", type=Path, default=DEFAULT_OUT) parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MARKDOWN_OUT) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) packet = _load_json(args.packet) preflight = _load_json(args.preflight) capture = _load_json(args.capture_json) if args.capture_json else None postflight = _load_json(args.postflight) if capture is not None and args.postflight.exists() else None receipt = verify_capture( packet, preflight, capture, postflight, packet_path=args.packet, evidence_root=args.evidence_root, ) write_json(args.out, receipt) write_markdown(args.markdown_out, receipt) print( json.dumps( { "out": str(args.out), "markdown_out": str(args.markdown_out), "status": receipt["status"], "receipt_ready": receipt["receipt_ready"], }, indent=2, sort_keys=True, ) ) return 0 if receipt["status"] in {"pass", "awaiting_action_time_authorization"} else 2 if __name__ == "__main__": raise SystemExit(main())