#!/usr/bin/env python3 """Append-only, read-only adjudication of the completed Task26 replacement claim.""" from __future__ import annotations import argparse import hashlib import json import os import stat import subprocess import sys import zipfile from datetime import datetime from pathlib import Path from typing import Any CANDIDATE = "2e0894eac92bc396cc4723bf1f18ebc653b95018dd41574df435941c235da925" WHEEL_SHA256 = "af4a9d0a1ffffb6eb7551c1d6dc2b32853ca6d024332a4f8f5702bbf992f141b" PLAN_SHA256 = "d816dbbd6a27d5826b251d279a9831de096ef7cd79b5fd131ceff7253b7369be" PREPARATION_SHA256 = "ec6acee7743c4cd50c84fb28d2e2c2fafd0d62165cacbbaadc2a72e0307e206d" PREPARATION_LEDGER_SHA256 = "c7c53db41858d53a6860baa98a1bc03f33e19a82048501c2a4b7c25dfddd3855" CURRENT_LEDGER_DIGEST = "a9b73fcd6f46884857c14fdb94db0618f306225e9c17ed36b9a5b3a343bcdd48" DRAFT_SHA256 = "ee05174e7287a87f9c5e40dae1034cf8b3228edc6b9516df8e6349667baffc7c" HANDOFF_SHA256 = "ebb97e82ac464abb900bc3a690a5c6c321c1cfd6401d3fdf40c283ae2faf8795" START_TOKEN_SHA256 = "4415e362f61e7500f98f99c5b468828076f61f30a34eebe8b2863fc9faabeaad" PRIOR_SESSION_ID = "cb_v4olwxbpSQatMtVLR4QLmw" PRIOR_SID_HASH = "63bdc993abc7b4f0025e69b10246faf8b5f4da3b7c040f02987a77bef3fa7c59" SESSION_ID = "cb_PCczfFXoI4GjvCxLBs1oRA" SID_HASH = "53b4f95b5b4db9a611128976a7b951bbe22d9619e05b2ff07fadf5b02f9a5380" CUSTOMER_KEY = "task26_live_2e_r2_20260815_8527916639" ACTOR = "8527916639" CLAIM_MESSAGE = "158" CONSENT_CARD = "159" BOOTSTRAP_MODULE_SHA256 = "145515d5e110dcebb94fcaa554bcee29544b3a75dcfe058fae944ea3b70042b2" REGISTRATION_MODULE_SHA256 = "cc0e4b697633150f3626daa8cf5bb4825bc96243f1a3dfadb6c9f540cad9906c" TELEGRAM_MODULE_SHA256 = "b41060dea28eb3bbb83217068f5218d5df2c879dba00e73c9ca4e577a6049dad" SERVICE = "hermes-gateway-dualcoachtest.service" O_FLAGS = os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0) HERE = Path(__file__).resolve().parent ROOT = HERE.parents[3] V3 = ROOT / ".omo/evidence/task26/task26-invite-harness-st_01a0056a-v3" WHEEL = ROOT / ".omo/evidence/task26" / ("task26-repaired-archive-successor-" + CANDIDATE) / "artifacts/hermes_agent-0.17.0-py3-none-any.whl" PLAN = ROOT / ".omo/plans/dualcoach-production-readiness.md" PREPARATION = V3 / "task26-live-2e-r2-preparation.redacted.json" DRAFT = V3 / "task26-live-2e-r2-customer-draft.private.json" HANDOFF = V3 / "task26-live-2e-r2-invite-handoff.private.json" ABSENT_AUTHORITIES = ( "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", ) class AdjudicationError(RuntimeError): pass def sha(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def canonical(value: Any) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") def read_regular(path: Path, *, private: bool = True) -> bytes: fd = os.open(path, os.O_RDONLY | O_FLAGS) try: before = os.fstat(fd) if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1 or before.st_uid != os.getuid(): raise AdjudicationError(f"unsafe evidence file: {path}") if private and stat.S_IMODE(before.st_mode) not in {0o400, 0o600}: raise AdjudicationError(f"non-private evidence file: {path}") chunks: list[bytes] = [] while block := os.read(fd, 1024 * 1024): chunks.append(block) after = os.fstat(fd) fields = ("st_dev", "st_ino", "st_mode", "st_nlink", "st_uid", "st_size", "st_mtime_ns", "st_ctime_ns") if tuple(getattr(before, key) for key in fields) != tuple(getattr(after, key) for key in fields): raise AdjudicationError(f"evidence changed during read: {path}") return b"".join(chunks) finally: os.close(fd) def read_json(path: Path, *, private: bool = True) -> dict[str, Any]: try: value = json.loads(read_regular(path, private=private)) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise AdjudicationError(f"invalid JSON: {path}") from exc if not isinstance(value, dict): raise AdjudicationError(f"JSON root is not an object: {path}") return value def exclusive_receipt(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) if path.parent.is_symlink(): raise AdjudicationError("receipt directory is a symlink") fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_FLAGS, 0o600) try: raw = canonical(value) + b"\n" os.fchmod(fd, 0o600) offset = 0 while offset < len(raw): offset += os.write(fd, raw[offset:]) os.fsync(fd) finally: os.close(fd) def validate_pins() -> dict[str, Any]: pins = ((WHEEL, WHEEL_SHA256), (PLAN, PLAN_SHA256), (PREPARATION, PREPARATION_SHA256), (DRAFT, DRAFT_SHA256), (HANDOFF, HANDOFF_SHA256)) for path, wanted in pins: if sha(read_regular(path, private=path not in {WHEEL, PLAN})) != wanted: raise AdjudicationError(f"sealed input hash mismatch: {path.name}") preparation = read_json(PREPARATION) expected = { "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": PLAN_SHA256, "ledger_sha256": PREPARATION_LEDGER_SHA256, "session_id": SESSION_ID, "sid_hash": SID_HASH, "generation": 1, "state": "PREPARED", "handoff_sha256": HANDOFF_SHA256, "start_token_sha256": START_TOKEN_SHA256, } if any(preparation.get(key) != value for key, value in expected.items()): raise AdjudicationError("v3 preparation binding mismatch") return preparation def validate_ledger(profile: Path) -> tuple[dict[str, Any], os.stat_result]: path = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" raw = read_regular(path) info = path.stat() value = json.loads(raw) if not isinstance(value, dict) or set(value) != {"schema", "sessions", "digest"}: raise AdjudicationError("bootstrap ledger shape mismatch") payload = {"schema": value["schema"], "sessions": value["sessions"]} if value["schema"] != "telegram-customer-bootstrap-v1" or value["digest"] != sha(canonical(payload)): raise AdjudicationError("bootstrap ledger digest mismatch") if value["digest"] != CURRENT_LEDGER_DIGEST: raise AdjudicationError("current bootstrap ledger is not sealed state") if not isinstance(value["sessions"], list) or len(value["sessions"]) != 2: raise AdjudicationError("bootstrap session cardinality mismatch") return value, info def exact_sessions(ledger: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: prior, replacement = ledger["sessions"] prior_expected = { "session_id": PRIOR_SESSION_ID, "sid_hash": PRIOR_SID_HASH, "state": "EXPIRED", "generation": 2, "role_claims": [], "recovery_attempts": [], "consent_publication_attempt": 0, "consent_card_message_id": None, "recovery_attempt_generation": 0, } if any(prior.get(key) != value for key, value in prior_expected.items()): raise AdjudicationError("old EXPIRED session changed") prior_draft = prior.get("customer_draft") if not isinstance(prior_draft, dict) or prior_draft.get("customer_user_id") is not None: raise AdjudicationError("old EXPIRED customer binding changed") if replacement.get("session_id") != SESSION_ID or replacement.get("sid_hash") != SID_HASH: raise AdjudicationError("replacement identity binding mismatch") if SESSION_ID == PRIOR_SESSION_ID or SID_HASH == PRIOR_SID_HASH: raise AdjudicationError("replacement identity is not distinct") expected_final = { "state": "AWAITING_CONSENT", "generation": 4, "recovery_attempt_generation": 1, "recovery_attempts": [], "consent_publication_attempt": 1, "consent_card_message_id": CONSENT_CARD, "failure_code": None, } if any(replacement.get(key) != value for key, value in expected_final.items()): raise AdjudicationError("replacement final state mismatch") claims = replacement.get("role_claims") wanted_claim = {"role": "customer", "user_id": ACTOR, "chat_id": ACTOR, "topic_id": "0", "message_id": CLAIM_MESSAGE} if claims != [wanted_claim]: raise AdjudicationError("accepted role claim is not exactly one sealed private-DM claim") draft_before = read_json(DRAFT) draft_after = replacement.get("customer_draft") if not isinstance(draft_after, dict): raise AdjudicationError("replacement customer draft is invalid") changed = {key for key in set(draft_before) | set(draft_after) if draft_before.get(key) != draft_after.get(key)} if changed != {"customer_user_id"} or draft_before.get("customer_user_id") is not None or draft_after.get("customer_user_id") != ACTOR: raise AdjudicationError("customer draft changed beyond customer_user_id claim") if replacement.get("customer_draft_digest") != sha(canonical(draft_after)): raise AdjudicationError("claimed customer draft digest mismatch") return prior, replacement, draft_before def digest_relationship(ledger: dict[str, Any], prior: dict[str, Any], replacement: dict[str, Any], draft: dict[str, Any]) -> dict[str, Any]: prepared = dict(replacement) prepared.update({ "customer_draft": draft, "customer_draft_digest": sha(canonical(draft)), "state": "PREPARED", "generation": 1, "updated_at": replacement["created_at"], "role_claims": [], "consent_publication_attempt": 0, "consent_card_message_id": None, "recovery_attempt_generation": 0, "recovery_attempts": [], }) payload = {"schema": ledger["schema"], "sessions": [prior, prepared]} prepared_document = {**payload, "digest": sha(canonical(payload))} reconstructed = sha(canonical(prepared_document)) if reconstructed != PREPARATION_LEDGER_SHA256: raise AdjudicationError("preparation-to-current exact digest relationship mismatch") return { "preparation_ledger_sha256": reconstructed, "prepared_payload_digest": prepared_document["digest"], "current_payload_digest": ledger["digest"], "claimed_draft_digest": replacement["customer_draft_digest"], } def token_binding(evidence_root: Path) -> dict[str, Any]: handoff = read_json(HANDOFF) token = handoff.get("start_token") if not isinstance(token, str) or not token.startswith("rc1_"): raise AdjudicationError("private handoff token is invalid") if handoff.get("session_id") != SESSION_ID or sha(token.encode()) != START_TOKEN_SHA256: raise AdjudicationError("private handoff token/session mismatch") if sha(token[4:].encode("ascii")) != SID_HASH: raise AdjudicationError("private handoff SID hash mismatch") hits: list[Path] = [] for path in evidence_root.rglob("*"): if not path.is_file() or path.is_symlink(): continue try: if token.encode() in read_regular(path, private=False): hits.append(path.resolve()) except (AdjudicationError, OSError, PermissionError): continue if hits != [HANDOFF.resolve()]: raise AdjudicationError("raw invite token leaked outside sealed private handoff") return {"token_sha256": START_TOKEN_SHA256, "sid_hash": SID_HASH, "raw_token_hit_count": 1, "raw_token_only_private_handoff": True} def authority_absence(profile: Path) -> dict[str, Any]: registry = read_json(profile / "customers/registry.json") customers = registry.get("customers") if not isinstance(customers, list): raise AdjudicationError("customer registry cardinality is UNKNOWN") if customers: raise AdjudicationError(f"required no customer registry row, observed {len(customers)} row(s)") present = [relative for relative in ABSENT_AUTHORITIES if (profile / relative).exists()] if present: raise AdjudicationError("activation/checkin/generation/delivery authority exists: " + ",".join(present)) return {"customer_registry_rows": 0, "activation_authorities": 0, "checkin_generation_delivery_authorities": 0} def source_semantics_and_service() -> dict[str, Any]: result = subprocess.run(["systemctl", "--user", "show", SERVICE, "-p", "ActiveState", "-p", "SubState", "-p", "MainPID"], text=True, capture_output=True, check=False) values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line) if result.returncode or values.get("ActiveState") != "active" or values.get("SubState") != "running": raise AdjudicationError("sealed service is not active/running") try: pid = int(values.get("MainPID", "0")) cmdline = Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0") environment = dict(item.split(b"=", 1) for item in Path(f"/proc/{pid}/environ").read_bytes().split(b"\0") if b"=" in item) except (OSError, ValueError) as exc: raise AdjudicationError("service process identity is UNKNOWN") from exc wanted_cmd = [b"/home/cube/projects/richard/hermes-agent/.venv/bin/python", b"-m", b"hermes_cli.main", b"--profile", b"dualcoachtest", b"gateway", b"run"] if cmdline[:-1] != wanted_cmd or environment.get(b"VIRTUAL_ENV") != b"/home/cube/projects/richard/hermes-agent/.venv": raise AdjudicationError("service process binding mismatch") site = Path("/home/cube/projects/richard/hermes-agent/.venv/lib/python3.12/site-packages/gateway/platforms") modules = { "telegram_customer_bootstrap.py": BOOTSTRAP_MODULE_SHA256, "telegram_customer_bootstrap_registration.py": REGISTRATION_MODULE_SHA256, "telegram.py": TELEGRAM_MODULE_SHA256, } sources: dict[str, str] = {} for name, wanted in modules.items(): raw = read_regular(site / name, private=False) if sha(raw) != wanted: raise AdjudicationError(f"active candidate module mismatch: {name}") sources[name] = wanted with zipfile.ZipFile(WHEEL) as archive: for name, wanted in modules.items(): member = "gateway/platforms/" + name if sha(archive.read(member)) != wanted: raise AdjudicationError(f"candidate wheel module mismatch: {name}") bootstrap = (site / "telegram_customer_bootstrap.py").read_text() registration = (site / "telegram_customer_bootstrap_registration.py").read_text() telegram = (site / "telegram.py").read_text() fragments = ( (bootstrap, "customer_draft=draft", "state=BootstrapState.REGISTERING", "generation=current.generation + 1"), (registration, "admin.register_customer(", "target=BootstrapState.AWAITING_CONSENT"), (telegram, "reserve_consent_publication(", "reserve_recovery_attempt(", "bind_recovery_receipt("), ) if any(any(fragment not in source for fragment in wanted) for source, *wanted in fragments): raise AdjudicationError("candidate atomic claim/publication semantics mismatch") return {"service": "active/running", "main_pid": pid, "candidate_modules": sources, "normal_path_transitions": [ "PREPARED/g1 -> REGISTERING/g2 + one role claim + customer_user_id", "register one disabled customer row", "REGISTERING/g2 -> AWAITING_CONSENT/g3", "bind owner no-op durable rewrite at g3", "reserve consent publication -> attempt1/g4", "reserve recovery -> recovery_attempt_generation1/g4", "bind provider receipt card159 -> AWAITING_CONSENT/g4", ]} def timestamp_evidence(replacement: dict[str, Any], ledger_info: os.stat_result) -> dict[str, Any]: created = datetime.fromisoformat(replacement["created_at"]) updated = datetime.fromisoformat(replacement["updated_at"]) mtime = datetime.fromtimestamp(ledger_info.st_mtime, tz=updated.tzinfo) if not created < updated or abs((mtime - updated).total_seconds()) > 0.01: raise AdjudicationError("ledger event timestamp relationship is UNKNOWN") return {"prepared_at": created.isoformat(), "one_claim_event_finalized_at": updated.isoformat(), "ledger_mtime_matches_final_transition": True, "one_event_semantics": "one durable RoleClaim(message_id=158) accepted by one start handler invocation"} def adjudicate(profile: Path, evidence_root: Path) -> dict[str, Any]: validate_pins() ledger, info = validate_ledger(profile) prior, replacement, draft = exact_sessions(ledger) relationship = digest_relationship(ledger, prior, replacement, draft) token = token_binding(evidence_root) timestamps = timestamp_evidence(replacement, info) service = source_semantics_and_service() absence = authority_absence(profile) return { "schema": "task26-post-claim-adjudication-v1", "status": "PASS_SUCCESSFUL_CLAIM", "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": PLAN_SHA256, "session_id": SESSION_ID, "actor_id": ACTOR, "claim_message_id": CLAIM_MESSAGE, "consent_card_message_id": CONSENT_CARD, "state": "AWAITING_CONSENT", "generation": 4, "recovery_attempt_generation": 1, "old_expired_session_unchanged": True, "replacement_identity_distinct": True, "accepted_role_claim_count": 1, "consent_publication_count": 1, "customer_draft_only_customer_user_id_changed": True, "profile_mutations": 0, "network_calls": 0, "replay_count": 0, "privacy": "redacted; hashes and identifiers only; no raw token", **relationship, "token_binding": token, "timestamp_evidence": timestamps, "service_proof": service, "absence_proof": absence, } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--profile", 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 = adjudicate(args.profile, args.evidence_root) code = 0 except (AdjudicationError, OSError, ValueError, KeyError, TypeError, zipfile.BadZipFile) as exc: result = {"schema": "task26-post-claim-adjudication-v1", "status": "FAIL", "blocker": str(exc), "unknown_is_failure": True, "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": PLAN_SHA256, "profile_mutations": 0, "network_calls": 0, "replay_count": 0, "privacy": "redacted; no raw token"} code = 2 exclusive_receipt(args.receipt, result) sys.stdout.buffer.write(canonical({"status": result["status"], "receipt": str(args.receipt)}) + b"\n") return code if __name__ == "__main__": raise SystemExit(main())