#!/usr/bin/env python3 """Independent live verifier for Task26 post-claim adjudication v2.""" from __future__ import annotations import argparse import hashlib import json import os import subprocess from datetime import datetime from pathlib import Path from typing import Any HERE = Path(__file__).resolve().parent ROOT = HERE.parents[3] V1 = ROOT / ".omo/evidence/task26/task26-post-claim-adjudicator-st_01a007d9" V3 = ROOT / ".omo/evidence/task26/task26-invite-harness-st_01a0056a-v3" EXPECTED = { "post_claim_adjudicator_v2.py": "1ee766c54ee881c4b64dc55e5484bed4871024e7d513eec1c47bc7f631ed8889", "test_post_claim_adjudicator_v2.py": "0a36713430ecbb7cf0ce3548bdd3ec36548273ee89d717eac6385a55dc3d9488", "schema-v2.json": "b1cf370fac1a5551a4168561bc23bae9486566b36fd1f4f2fd6c5dd50c3658f7", } CURRENT_DIGEST = "a9b73fcd6f46884857c14fdb94db0618f306225e9c17ed36b9a5b3a343bcdd48" SESSION = "cb_PCczfFXoI4GjvCxLBs1oRA" ACTOR = "8527916639" CUSTOMER_KEY = "task26_live_2e_r2_20260815_8527916639" ABSENT = ( "data/customers", "data/activation-completion-notices.jsonl", "data/customer-activation-audit.jsonl", "data/customer-activation-journal.json", "data/customer-activation-receipt.json", "data/nutrition-onboarding-projection-journal.jsonl", "data/scheduled-deliveries.jsonl", "data/scheduled-deliveries-fence.json", "data/customer-schedule-claims", "data/recovery-audits", "data/activation-readiness", ) def canonical(value: Any) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def write_once(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, 0o600) try: os.write(fd, canonical(value) + b"\n") os.fsync(fd) finally: os.close(fd) def verify_v1_immutable() -> None: if sha(V1 / "hashes.json") != "3c6d7111ac1c900fbeaa3f7f7bbd93139c60b1ec107516a6c386ce55b4e8d613": raise RuntimeError("v1 inventory changed") inventory = json.loads((V1 / "hashes.json").read_text()) artifacts = inventory.get("artifacts") if not isinstance(artifacts, dict): raise RuntimeError("v1 inventory is invalid") for name, proof in artifacts.items(): if not isinstance(proof, dict) or sha(V1 / name) != proof.get("sha256"): raise RuntimeError(f"v1 artifact changed: {name}") def verify(profile: Path, live_receipt: Path, evidence_root: Path) -> dict[str, Any]: for name, wanted in EXPECTED.items(): if sha(HERE / name) != wanted: raise RuntimeError(f"sealed v2 artifact mismatch: {name}") verify_v1_immutable() dry = json.loads((V3 / "task26-live-2e-r2-dry-run.redacted.json").read_text()) preparation = json.loads((V3 / "task26-live-2e-r2-preparation.redacted.json").read_text()) if ( sha(V3 / "task26-live-2e-r2-dry-run.redacted.json") != "cda348a3a9500a541d9d8851ec8ebad51be4738dd6a6e8d5c629ec1e65e64a4c" or dry.get("registry_customer_count") != 0 or dry.get("accepted_claim_count") != 0 or preparation.get("ledger_sha256") != "c7c53db41858d53a6860baa98a1bc03f33e19a82048501c2a4b7c25dfddd3855" ): raise RuntimeError("zero-row preparation provenance mismatch") ledger_path = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" ledger = json.loads(ledger_path.read_text()) payload = {"schema": ledger["schema"], "sessions": ledger["sessions"]} if ledger.get("digest") != hashlib.sha256(canonical(payload)).hexdigest() or ledger.get("digest") != CURRENT_DIGEST: raise RuntimeError("current ledger digest mismatch") sessions = ledger.get("sessions") if not isinstance(sessions, list) or len(sessions) != 2: raise RuntimeError("session cardinality mismatch") old, current = sessions claim = {"role": "customer", "user_id": ACTOR, "chat_id": ACTOR, "topic_id": "0", "message_id": "158"} if not ( old.get("state") == "EXPIRED" and old.get("session_id") == "cb_v4olwxbpSQatMtVLR4QLmw" and current.get("session_id") == SESSION and current.get("sid_hash") == "53b4f95b5b4db9a611128976a7b951bbe22d9619e05b2ff07fadf5b02f9a5380" and current.get("state") == "AWAITING_CONSENT" and current.get("generation") == 4 and current.get("recovery_attempt_generation") == 1 and current.get("role_claims") == [claim] and current.get("consent_publication_attempt") == 1 and current.get("consent_card_message_id") == "159" ): raise RuntimeError("claim or consent authority mismatch") registry_path = profile / "customers/registry.json" registry = json.loads(registry_path.read_text()) rows = registry.get("customers") if not isinstance(rows, list) or len(rows) != 1 or not isinstance(rows[0], dict): raise RuntimeError("canonical registry cardinality mismatch") row = rows[0] if not ( row.get("customer_key") == CUSTOMER_KEY and row.get("enabled") is False and row.get("telegram") == {"user_id": ACTOR, "chat_id": ACTOR, "topic_id": "0"} and row.get("ai_processing_consent") == {"granted": False, "recorded_on": None, "notice_version": None} and isinstance(row.get("plan"), dict) and len(row["plan"].get("weeks", [])) == 12 ): raise RuntimeError("disabled registry row mismatch") created = datetime.fromisoformat(current["created_at"]).timestamp() updated = datetime.fromisoformat(current["updated_at"]).timestamp() if not created < registry_path.stat().st_mtime < updated: raise RuntimeError("registry provenance timestamp mismatch") present = [name for name in ABSENT if (profile / name).exists()] if present: raise RuntimeError("downstream authority exists: " + ",".join(present)) service = subprocess.run( ["systemctl", "--user", "show", "hermes-gateway-dualcoachtest.service", "-p", "ActiveState", "-p", "SubState", "-p", "MainPID"], text=True, capture_output=True, check=False, ) values = dict(line.split("=", 1) for line in service.stdout.splitlines() if "=" in line) if service.returncode or values.get("ActiveState") != "active" or values.get("SubState") != "running" or values.get("MainPID") in {None, "0"}: raise RuntimeError("service is not active candidate runtime") site = Path("/home/cube/projects/richard/hermes-agent/.venv/lib/python3.12/site-packages/gateway/platforms") source_pins = { "telegram_customer_bootstrap.py": "145515d5e110dcebb94fcaa554bcee29544b3a75dcfe058fae944ea3b70042b2", "telegram_customer_bootstrap_registration.py": "cc0e4b697633150f3626daa8cf5bb4825bc96243f1a3dfadb6c9f540cad9906c", "telegram.py": "b41060dea28eb3bbb83217068f5218d5df2c879dba00e73c9ca4e577a6049dad", } if any(sha(site / name) != wanted for name, wanted in source_pins.items()): raise RuntimeError("active candidate source bytes mismatch") handoff = json.loads((V3 / "task26-live-2e-r2-invite-handoff.private.json").read_text()) token = handoff.get("start_token") if not isinstance(token, str): raise RuntimeError("private token binding UNKNOWN") hits = [] for path in evidence_root.rglob("*"): if path.is_file() and not path.is_symlink(): try: if token.encode() in path.read_bytes(): hits.append(path.resolve()) except (OSError, PermissionError): pass if hits != [(V3 / "task26-live-2e-r2-invite-handoff.private.json").resolve()]: raise RuntimeError("raw token leakage mismatch") receipt = json.loads(live_receipt.read_text()) registration = receipt.get("disabled_registration_proof") if ( receipt.get("status") != "PASS_SUCCESSFUL_CLAIM_AND_DISABLED_REGISTRATION" or not isinstance(registration, dict) or registration.get("customer_registry_rows") != 1 or registration.get("status") != "disabled/not_activated" or registration.get("checkin_generation_delivery_authorities") != 0 ): raise RuntimeError("v2 live receipt mismatch") return { "schema": "task26-post-claim-independent-verification-v2", "status": "PASS_INDEPENDENTLY_VERIFIED", "claim_and_sid_binding": True, "disabled_registration_exact": True, "zero_row_to_one_disabled_row_candidate_path": True, "consent_card_159_exactly_once": True, "downstream_authority_count": 0, "service_candidate_bytes_exact": True, "raw_token_leakage": False, "v1_artifacts_unchanged": True, "check_count": 27, } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--profile", type=Path, required=True) parser.add_argument("--live-receipt", type=Path, required=True) parser.add_argument("--evidence-root", type=Path, required=True) parser.add_argument("--receipt", type=Path, required=True) args = parser.parse_args() try: result = verify(args.profile, args.live_receipt, args.evidence_root) code = 0 except (OSError, ValueError, KeyError, TypeError, RuntimeError, json.JSONDecodeError) as exc: result = {"schema": "task26-post-claim-independent-verification-v2", "status": "FAIL", "blocker": str(exc), "unknown_is_failure": True} code = 2 write_once(args.receipt, result) print(json.dumps({"status": result["status"], "receipt": str(args.receipt)}, sort_keys=True)) return code if __name__ == "__main__": raise SystemExit(main())