#!/usr/bin/env python3 """Verify a source Golden Path from native production artifacts only.""" from __future__ import annotations import argparse import hashlib import importlib import importlib.util import json import os import stat import sys from collections.abc import Mapping from datetime import date from pathlib import Path from typing import Any, cast from gateway.platforms.task26_runtime_authority import validate_snapshot _PROVENANCE_SPEC = importlib.util.spec_from_file_location( "golden_verifier_installed_wheel_provenance", Path(__file__).with_name("installed_wheel_provenance.py"), ) if _PROVENANCE_SPEC is None or _PROVENANCE_SPEC.loader is None: raise RuntimeError("installed provenance helper is unavailable") _PROVENANCE_MODULE = importlib.util.module_from_spec(_PROVENANCE_SPEC) _PROVENANCE_SPEC.loader.exec_module(_PROVENANCE_MODULE) collect_installed_runtime = _PROVENANCE_MODULE.collect_installed_runtime loaded_module_receipts = _PROVENANCE_MODULE.loaded_module_receipts read_private_json = _PROVENANCE_MODULE.read_private_json portable_installed_runtime = _PROVENANCE_MODULE.portable_installed_runtime nonportable_installed_record_audit = ( _PROVENANCE_MODULE.nonportable_installed_record_audit ) validate_nonportable_installed_record_audit = ( _PROVENANCE_MODULE.validate_nonportable_installed_record_audit ) DEFAULT_SOURCE = Path("/home/cube/.cache/task26-strict-successor-1786976146/src-p") TASK26_CLAUSES = ( "onboarding_22_clear", "deterministic_ambiguity_revision", "preview_isolation", "approved_card_projection_gate", "unknown_delivery_no_retry", "successful_lifecycle", "cleanup_resume_terminal", ) TASK26_ANSWERS = ( "1990년 1월 15일입니다.", "남성입니다.", "키는 180cm입니다.", "현재 체중은 80kg입니다.", "보통 수준입니다.", "주 3회 60분 근력 운동을 하고 평일에는 하루 8천 보 정도 걷습니다.", "현재 체중을 유지하고 싶습니다.", "유지가 목표라 목표 체중은 정하지 않겠습니다.", "유지가 목표라 목표 날짜도 정하지 않겠습니다.", "음식 알레르기는 없습니다.", "음식 불내증은 없습니다.", "종교적 또는 윤리적으로 제외하는 음식은 없습니다.", "싫어해서 피하는 음식은 없습니다.", "특별한 식단 선호는 없습니다.", "진단받은 질환은 없습니다.", "복용 중인 약이나 보충제는 없습니다.", "임신 또는 수유에 해당하지 않습니다.", "섭식장애 위험이나 과거력은 없습니다.", "가스레인지, 전자레인지, 냉장고를 사용할 수 있고 기본 조리가 가능합니다.", "식비 예산은 보통입니다.", "하루 세 끼를 먹습니다.", "평일 점심은 12시, 저녁은 운동 후 8시쯤이며 그 외 제약은 없습니다.", ) PHASES = ( "OBSERVER_SUBSCRIBED", "INVITE_RECEIPTED", "CONSENT_COMMITTED", "ONBOARDING_22_COMMITTED", "RECONCILIATION_COMMITTED", "CUSTOMER_ATTESTED", "OWNER_REVIEW_PUBLISHED", "READY_DISABLED", "ACTIVATED", "CHECKIN_FINALIZED", "GENERATION_DRAFTED", "OWNER_APPROVED", "APPROVED_CARD_PROJECTED", "CAPABILITY_ISSUED", "CAPABILITY_CONSUMED", "SENT_AUDITED", "SURFACE_CAPTURED", "QUIESCED", "ARCHIVED", "CLEANUP_COMPLETE", "SOURCE_GOLDEN_PATH_PASS", ) ZERO = "0" * 64 TERMINAL_CATEGORY_FIELDS = frozenset( { "registry", "bootstrap", "onboarding", "outbox", "key", "owner_action", "service_state", "event", "customer_data", "activation", "sent_event", "candidate", } ) TERMINAL_INVENTORY_FIELDS = frozenset( { "customer_key", "disabled", "nonconsenting", "categories", "active_count", "pending_count", "unknown_count", "orphan_count", "archive_verified", "journal_committed", "terminal", } ) def canonical(value: object, *, ascii_only: bool = False) -> bytes: return json.dumps(value, ensure_ascii=ascii_only, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def obj(value: object, label: str) -> dict[str, Any]: if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): raise ValueError(f"{label} is not an object") return cast(dict[str, Any], value) def read_json(path: Path, label: str) -> dict[str, Any]: return obj(json.loads(path.read_text(encoding="utf-8")), label) def serialize_terminal_inventory(terminal: object) -> dict[str, object]: fields = getattr(type(terminal), "__dataclass_fields__", None) if not isinstance(fields, dict) or set(fields) != TERMINAL_INVENTORY_FIELDS: raise ValueError("terminal inventory schema is invalid") customer_key = getattr(terminal, "customer_key", None) disabled = getattr(terminal, "disabled", None) nonconsenting = getattr(terminal, "nonconsenting", None) raw_categories = getattr(terminal, "categories", None) if not isinstance(raw_categories, Mapping): raise ValueError("terminal inventory categories are invalid") categories = dict(raw_categories) counts = { name: getattr(terminal, name, None) for name in ( "active_count", "pending_count", "unknown_count", "orphan_count", ) } booleans = { name: getattr(terminal, name, None) for name in ( "archive_verified", "journal_committed", "terminal", ) } if ( not isinstance(customer_key, str) or not customer_key or type(disabled) is not bool or type(nonconsenting) is not bool or set(categories) != TERMINAL_CATEGORY_FIELDS or any(not isinstance(key, str) for key in categories) or any(type(value) is not int or value < 0 for value in categories.values()) or any(type(value) is not int or value < 0 for value in counts.values()) or any(type(value) is not bool for value in booleans.values()) ): raise ValueError("terminal inventory field types are invalid") expected_terminal = bool( disabled and nonconsenting and all(value == 0 for value in counts.values()) and booleans["archive_verified"] and booleans["journal_committed"] ) if booleans["terminal"] is not expected_terminal: raise ValueError("terminal inventory verdict is inconsistent") return { "customer_key": customer_key, "disabled": disabled, "nonconsenting": nonconsenting, "categories": dict(sorted(categories.items())), **counts, **booleans, } def lines(path: Path, label: str) -> list[dict[str, Any]]: return [obj(json.loads(line), label) for line in path.read_text(encoding="utf-8").splitlines() if line] def private( path: Path, *, directory: bool = False, frozen: bool | None = None ) -> None: info = path.lstat() if stat.S_ISLNK(info.st_mode) or info.st_uid != os.geteuid(): raise ValueError(f"unsafe artifact path: {path}") expected_type = stat.S_ISDIR if directory else stat.S_ISREG if not expected_type(info.st_mode) or (not directory and info.st_nlink != 1): raise ValueError(f"artifact type is invalid: {path}") mutable_modes = {0o700} if directory else {0o600} frozen_modes = {0o500} if directory else {0o400} allowed = mutable_modes if frozen is False else mutable_modes | frozen_modes if stat.S_IMODE(info.st_mode) not in allowed: raise ValueError(f"artifact permissions are invalid: {path}") def verify_bootstrap_active_lineage(path: Path, customer: str) -> str: private(path) ledger = read_json(path, "bootstrap ledger") if set(ledger) != {"schema", "sessions", "digest"}: raise ValueError("bootstrap ledger schema is invalid") sessions = ledger["sessions"] payload = {"schema": ledger["schema"], "sessions": sessions} if ( ledger["schema"] != "telegram-customer-bootstrap-v1" or ledger["digest"] != hashlib.sha256(canonical(payload)).hexdigest() or not isinstance(sessions, list) or len(sessions) != 1 ): raise ValueError("bootstrap ledger authority is invalid") session = obj(sessions[0], "bootstrap session") customer_draft = obj(session.get("customer_draft"), "bootstrap customer draft") session_id = session.get("session_id") if ( customer_draft.get("customer_key") != customer or session.get("state") != "ACTIVE" or not isinstance(session_id, str) or not session_id ): raise ValueError("bootstrap ACTIVE lineage is invalid") return session_id def verify_provenance( root: Path, *, offline_recorded_source: bool = False ) -> tuple[str, Path]: path = root / "source-provenance.json" private(path) provenance = read_json(path, "source provenance") if provenance.get("schema") != "source-golden-provenance-v1": raise ValueError("source provenance schema is invalid") source = Path(str(provenance.get("source_root", ""))) if source != DEFAULT_SOURCE or (not offline_recorded_source and (source.is_symlink() or not source.is_dir())): raise ValueError("source provenance root is unavailable") rows = provenance.get("files") if not isinstance(rows, list) or not rows: raise ValueError("source provenance inventory is empty") expected_paths: list[str] = [] for raw in rows: row = obj(raw, "source provenance row") relative = row.get("path") if not isinstance(relative, str) or Path(relative).is_absolute() or ".." in Path(relative).parts: raise ValueError("source provenance path is invalid") target = source / relative if not offline_recorded_source: if target.is_symlink() or not target.is_file() or target.stat().st_nlink != 1: raise ValueError("source provenance file is unavailable") if sha(target) != row.get("sha256") or target.stat().st_size != row.get("size"): raise ValueError("source provenance file changed") expected_paths.append(relative) actual = expected_paths if offline_recorded_source else sorted( path.relative_to(source).as_posix() for path in source.rglob("*.py") if path.is_file() and not path.is_symlink() and not any(part in {".venv", "__pycache__"} for part in path.parts) ) if expected_paths != actual or hashlib.sha256(canonical(rows)).hexdigest() != provenance.get("tree_digest"): raise ValueError("source provenance inventory is incomplete") return str(provenance["tree_digest"]), source def verify_qualification_product_binding( root: Path, *, actual_hermes_wheel: Path | None = None, actual_profile_wheel: Path | None = None, ) -> tuple[str, str, dict[str, object]]: from gateway.platforms import task26_candidate_derivation as derivation from gateway.platforms.task26_final_state import TRUST_BOUNDARY path = root / "preexecution-product-binding.json" private(path) document = read_json(path, "preexecution product binding") package_root = Path(str(derivation.__file__)).resolve().parents[2] validated = derivation.validate_product_binding( document, expected_tool_hashes=derivation.qualification_tool_hashes( scripts_dir=Path(__file__).resolve().parent, package_root=package_root, ), expected_trust_boundary_sha256=derivation.trust_boundary_digest( TRUST_BOUNDARY ), actual_hermes_wheel=actual_hermes_wheel, actual_profile_wheel=actual_profile_wheel, ) return ( str(validated["candidate_digest"]), str(validated["binding_sha256"]), validated, ) def verify_installed_provenance( root: Path, source_provenance: dict[str, Any], *, runtime_override: dict[str, object] | None = None, ) -> tuple[str, Path]: path = root / "installed-provenance.json" document = read_private_json(path) if set(document) != { "schema", "binding", "installed_product_binding_digest", "candidate_digest", "loaded_modules", } or document.get("schema") != "installed-golden-provenance-v1": raise ValueError("installed provenance schema is invalid") binding = obj(document.get("binding"), "installed candidate binding") if binding.get("schema") != "installed-golden-candidate-binding-v1": raise ValueError("installed candidate binding schema is invalid") installed_product_binding_digest = hashlib.sha256(canonical(binding)).hexdigest() candidate = document.get("candidate_digest") if ( document.get("installed_product_binding_digest") != installed_product_binding_digest or not isinstance(candidate, str) ): raise ValueError("installed product or candidate binding is invalid") if binding.get("source_provenance") != source_provenance: raise ValueError("installed source metadata binding is invalid") config = obj(binding.get("config"), "installed config binding") if config.get("runtime_mode") != "installed" or config.get("profile_source_metadata_only") != str(DEFAULT_SOURCE): raise ValueError("installed config binding is invalid") scripts = Path(__file__).resolve().parent expected_tools = { "harness_sha256": sha(scripts / "source_golden_path.py"), "provenance_helper_sha256": sha(scripts / "installed_wheel_provenance.py"), "verifier_sha256": sha(Path(__file__)), "task26_contract_module": "gateway.platforms.task26_evidence_contract", "task26_contract_module_sha256": ( str(binding.get("task26_contract_module_sha256", "")) if runtime_override is not None else sha( Path(str(config.get("profile_package_authority", ""))) / "gateway/platforms/task26_evidence_contract.py" ) ), } if any(binding.get(key) != value for key, value in expected_tools.items()): raise ValueError("installed harness or verifier digest changed") runtime = obj(binding.get("runtime"), "installed runtime") distributions = obj(runtime.get("distributions"), "installed distributions") profile = obj(distributions.get("profile"), "installed profile distribution") hermes = obj(distributions.get("hermes"), "installed Hermes distribution") parity = obj(binding.get("task26_contract_parity"), "Task26 contract parity") installed_contract = Path(str(config.get("profile_package_authority", ""))) / "gateway/platforms/task26_evidence_contract.py" contract_rows = [ row for row in cast(list[dict[str, object]], hermes.get("installed_inventory", [])) if row.get("path") == "gateway/platforms/task26_evidence_contract.py" ] if ( len(contract_rows) != 1 or parity.get("module") != "gateway.platforms.task26_evidence_contract" or parity.get("sha256") != contract_rows[0].get("sha256") or parity.get("size") != contract_rows[0].get("size") or parity.get("wheel_path") != hermes.get("wheel_path") or ( runtime_override is None and parity.get("installed_path") != str(installed_contract.resolve(strict=True)) ) ): raise ValueError("Task26 source-wheel-installed parity binding is invalid") active_runtime = runtime if runtime_override is None else runtime_override active_distributions = obj(active_runtime.get("distributions"), "active installed distributions") active_profile = obj(active_distributions.get("profile"), "active profile distribution") active_hermes = obj(active_distributions.get("hermes"), "active Hermes distribution") recomputed = collect_installed_runtime( venv=Path(str(active_runtime.get("venv", ""))), site_packages=Path(str(active_runtime.get("site_packages", ""))), profile_wheel=Path(str(active_profile.get("wheel_path", ""))), hermes_wheel=Path(str(active_hermes.get("wheel_path", ""))), ) active_runtime_core = { key: value for key, value in active_runtime.items() if key != "origin_proof" } if recomputed != active_runtime_core: raise ValueError("installed runtime receipt differs from installed bytes") if runtime_override is None: if recomputed != runtime: raise ValueError("installed runtime receipt differs from original installed bytes") runtime_portable = portable_installed_runtime(recomputed) if runtime_override is not None and runtime_portable != binding.get( "runtime_portable" ): raise ValueError("rehydrated runtime differs from sealed portable binding") candidate_digest = str(document.get("candidate_digest", "")) validate_nonportable_installed_record_audit( binding.get("nonportable_record_audit"), install_role="original", candidate_digest=candidate_digest, runtime_portable=cast(dict[str, object], binding.get("runtime_portable")), ) if runtime_override is not None: validate_nonportable_installed_record_audit( nonportable_installed_record_audit( recomputed, install_role="rehydrated", candidate_digest=candidate_digest, ), install_role="rehydrated", candidate_digest=candidate_digest, runtime_portable=runtime_portable, ) roots = tuple( Path(root_path) for distribution in (active_profile, active_hermes) for root_path in cast(list[str], distribution.get("package_roots")) ) declarations = document.get("loaded_modules") if not isinstance(declarations, list) or not declarations: raise ValueError("installed loaded-module receipts are unavailable") declared_names: list[str] = [] for raw in declarations: row = obj(raw, "loaded module receipt") name = row.get("module") if not isinstance(name, str) or not (name == "checkin_cli" or name.startswith("checkin_cli.") or name == "gateway" or name.startswith("gateway.")): raise ValueError("installed loaded-module name is invalid") declared_names.append(name) if declared_names != sorted(set(declared_names)): raise ValueError("installed loaded-module receipt set is invalid") actual_modules: dict[str, Path] = {} for name in declared_names: spec = importlib.util.find_spec(name) origin = None if spec is None else spec.origin if not isinstance(origin, str): raise ValueError(f"installed loaded-module origin is unavailable: {name}") actual_modules[name] = Path(origin) actual_receipts = loaded_module_receipts(actual_modules, allowed_roots=roots) if runtime_override is None: if actual_receipts != declarations: raise ValueError("installed loaded-module declarations differ from import reality") else: portable_declarations = [ {key: row.get(key) for key in ("module", "sha256", "size")} for row in declarations ] portable_actual = [ {key: row.get(key) for key in ("module", "sha256", "size")} for row in actual_receipts ] if portable_actual != portable_declarations: raise ValueError("rehydrated import origins differ from sealed module bytes") authority = config.get("profile_package_authority") if runtime_override is None: if not isinstance(authority, str) or Path(authority).resolve(strict=True) != Path(str(runtime["site_packages"])): raise ValueError("installed profile package authority is invalid") active_authority = Path(authority) else: active_authority = Path(str(active_runtime["site_packages"])) return str(candidate), active_authority def verify_observer(root: Path, candidate: str) -> dict[str, Any]: path = root / "observer-receipts.jsonl" private(path) receipts = lines(path, "observer receipt") if len(receipts) < 2 or receipts[0].get("kind") != "subscription_armed": raise ValueError("observer was not armed before mutation") previous = ZERO sequences: list[int] = [] for row in receipts: if row.get("predecessor") != previous: raise ValueError("observer receipt chain is invalid") body = {key: value for key, value in row.items() if key != "receipt_hash"} current = hashlib.sha256(canonical(body)).hexdigest() if row.get("receipt_hash") != current: raise ValueError("observer receipt hash is invalid") previous = current if row.get("kind") == "commit_observed": sequence = row.get("sequence") if type(sequence) is not int: raise ValueError("observer sequence is invalid") sequences.append(sequence) armed = receipts[0] watched_paths = armed.get("watched_paths") expected_parent_count = ( len({str(Path(path).parent) for path in watched_paths}) if isinstance(watched_paths, list) and watched_paths and all(isinstance(path, str) for path in watched_paths) else -1 ) resources = { "backend": "dnotify_signalfd_v1", "directory_resource_count": expected_parent_count, "signal_fd_count": 1, "inotify_watch_count": 0, } if ( armed.get("candidate_id") != candidate or sequences != list(range(1, len(sequences) + 1)) or len(sequences) < 4 or any(armed.get(key) != value for key, value in resources.items()) or any( any(row.get(key) != value for key, value in resources.items()) for row in receipts if row.get("kind") == "commit_observed" ) ): raise ValueError("observer subscription binding is invalid") return {"receipt_count": len(receipts), "head": previous, **resources} def verify_cleanup(root: Path, customer: str) -> tuple[dict[str, Path], dict[str, Any]]: journal = root / f"data/customer-cleanup/{customer}.journal.jsonl" private(journal) rows = lines(journal, "cleanup journal") phases = ["prepared", "copied_verified", "source_pruned", "committed"] if [row.get("phase") for row in rows] != phases: raise ValueError("cleanup journal is not terminal and forward-only") previous = ZERO operation = rows[0].get("operation_id") for row in rows: if row.get("operation_id") != operation or row.get("previous_digest") != previous: raise ValueError("cleanup journal chain is invalid") previous = hashlib.sha256(canonical(row, ascii_only=True)).hexdigest() archive = root / "data/customer-cleanup/archives" / str(operation) private(archive, directory=True, frozen=True) manifest_path = archive / "manifest.json" private(manifest_path, frozen=True) if sha(manifest_path) != rows[0].get("manifest_sha256"): raise ValueError("cleanup manifest digest mismatch") manifest = read_json(manifest_path, "cleanup manifest") if manifest.get("schema_version") != "customer-cleanup-archive-v2" or manifest.get("customer_key") != customer or manifest.get("operation_id") != operation: raise ValueError("cleanup manifest authority is invalid") inventory = manifest.get("inventory") if not isinstance(inventory, list) or not inventory: raise ValueError("cleanup inventory is empty") artifacts: dict[str, Path] = {} for raw in inventory: row = obj(raw, "cleanup inventory row") relative = row.get("relative_path") if not isinstance(relative, str) or Path(relative).is_absolute() or ".." in Path(relative).parts or relative in artifacts: raise ValueError("cleanup inventory path is invalid") target = archive / "files" / relative private(target, frozen=True) if target.stat().st_size != row.get("size") or sha(target) != row.get("sha256"): raise ValueError("cleanup archive bytes mismatch") artifacts[relative] = target shared = manifest.get("shared_ledger_projections") if not isinstance(shared, list) or len(shared) != 1: raise ValueError("cleanup shared-ledger projection cardinality is invalid") binding = obj(shared[0], "shared-ledger projection binding") source_relative = "data/owner-actions/draft-generations.json" projection_relative = f"projections/{source_relative}" source = root / source_relative projection = archive / projection_relative private(source) private(projection, frozen=True) if ( binding.get("source_relative_path") != source_relative or binding.get("projection_archive_path") != projection_relative or binding.get("selector_version") != "draft-generations-customer-v1" or binding.get("source_sha256") != sha(source) or binding.get("source_size") != source.stat().st_size or binding.get("source_mode") != 0o600 or binding.get("projection_sha256") != sha(projection) ): raise ValueError("cleanup shared-ledger source preservation binding is invalid") source_document = read_json(source, "shared draft-generation source") projected = read_json(projection, "draft-generation projection") selected = { token: history for token, history in sorted(source_document.items()) if isinstance(history, list) and history and isinstance(history[0], dict) and history[0].get("customer_key") == customer } expected_projection = { "schema_version": "customer-cleanup-draft-generations-projection-v1", "customer_key": customer, "histories": selected, } record_digests = [ hashlib.sha256(canonical(row, ascii_only=True)).hexdigest() for history in selected.values() for row in history ] if ( projected != expected_projection or projection.read_bytes() != canonical(expected_projection, ascii_only=True) + b"\n" or binding.get("selected_record_count") != len(record_digests) or binding.get("ordered_record_digests") != record_digests ): raise ValueError("cleanup archive projection is not the exact target-only source projection") artifacts[projection_relative] = projection return artifacts, { "operation_id": operation, "inventory_count": len(inventory), "manifest_sha256": sha(manifest_path), "projection_sha256": sha(projection), "projection_record_count": len(record_digests), "preserved_source_sha256": sha(source), } def required(artifacts: dict[str, Path], suffix: str) -> Path: matches = [path for relative, path in artifacts.items() if relative.endswith(suffix)] if len(matches) != 1: raise ValueError(f"required native artifact is unavailable or ambiguous: {suffix}") return matches[0] def verify_canonical_journey_events(path: Path, customer: str) -> tuple[str, str]: private(path, frozen=True) raw_rows = lines(path, "wizard event") try: from checkin_cli.models import Event parsed_rows = [Event.model_validate(row) for row in raw_rows] except Exception as exc: raise ValueError("canonical wizard event schema is invalid") from exc if any( event.model_dump(mode="json", exclude_none=True) != row for event, row in zip(parsed_rows, raw_rows, strict=True) ): raise ValueError("canonical wizard event serialization is invalid") event_ids = [event.event_id for event in parsed_rows] dedupe_keys = [event.dedupe_key for event in parsed_rows] if ( len(event_ids) != len(set(event_ids)) or len(dedupe_keys) != len(set(dedupe_keys)) ): raise ValueError("canonical wizard event identity is duplicated") checkins = [ event for event in parsed_rows if event.event_type.value == "nutrition_checkin" ] if ( len(checkins) != 1 or not checkins[0].event_id or checkins[0].supersedes is not None or checkins[0].check_in is None ): raise ValueError("check-in finalization cardinality is invalid") lifecycle_types = ("draft_created", "draft_approved", "draft_sent") if {event.event_type.value for event in parsed_rows} != { "nutrition_checkin", *lifecycle_types, }: raise ValueError("canonical journey event type set is invalid") draft_events: dict[str, Any] = {} draft_id: str | None = None for event_type in lifecycle_types: matches = [event for event in parsed_rows if event.event_type.value == event_type] if len(matches) != 1: raise ValueError("draft lifecycle cardinality is invalid") event = matches[0] payload = event.draft if payload is None: raise ValueError("draft lifecycle cardinality is invalid") source_ref = event.provenance.source_ref if ( not payload.draft_id or source_ref != f"pilot:{customer}:{event_type}" or event.dedupe_key != f"{event_type}:{customer}:{payload.draft_id}" or event.supersedes is not None ): raise ValueError("draft lifecycle customer binding is invalid") if draft_id is None: draft_id = payload.draft_id elif payload.draft_id != draft_id: raise ValueError("draft lifecycle identity is inconsistent") draft_events[event_type] = event if draft_id is None or set(draft_events) != set(lifecycle_types): raise ValueError("draft lifecycle is incomplete") return checkins[0].event_id, draft_id def verify_generation_lineage( generations: object, customer: str, checkin_id: str, draft_id: str, ) -> list[dict[str, Any]]: document = obj(generations, "generation projection histories") if set(document) != {draft_id}: raise ValueError("generation lineage cardinality is invalid") history = document[draft_id] if not isinstance(history, list) or not history: raise ValueError("generation lineage is invalid") rows = [obj(row, "generation row") for row in history] if rows[-1].get("state") != "sent_audited" or any( row.get("customer_key") != customer or row.get("checkin_event_id") != checkin_id for row in rows ): raise ValueError("generation lineage is invalid") previous: str | None = None for row in rows: record_digest = row.get("record_digest") if ( row.get("predecessor_digest") != previous or not isinstance(record_digest, str) or len(record_digest) != 64 or any(character not in "0123456789abcdef" for character in record_digest) ): raise ValueError("generation digest chain is invalid") previous = record_digest return rows def verify_customer_quiescence(path: Path, customer: str) -> dict[str, object]: private(path) document = read_json(path, "service state") if set(document) != {"schema", "states", "payload_digest"}: raise ValueError("service state ledger shape is invalid") states = document.get("states") unsigned = {"schema": document.get("schema"), "states": states} if ( document.get("schema") != "customer-service-state-v1" or not isinstance(states, dict) or document.get("payload_digest") != hashlib.sha256(canonical(unsigned)).hexdigest() ): raise ValueError("service state ledger authority is invalid") for customer_key, raw_row in states.items(): if not isinstance(customer_key, str) or not customer_key: raise ValueError("service state customer key is invalid") row = obj(raw_row, "service state row") updated_on = row.get("updated_on") try: valid_date = ( isinstance(updated_on, str) and date.fromisoformat(updated_on).isoformat() == updated_on ) except ValueError: valid_date = False if ( set(row) != {"state", "revision", "updated_on"} or row.get("state") not in {"active", "paused"} or type(row.get("revision")) is not int or row["revision"] < 1 or not valid_date ): raise ValueError("service state row is invalid") selected = states.get(customer) if not isinstance(selected, dict) or selected.get("state") != "paused": raise ValueError("customer quiescence is unavailable") return { "state": selected["state"], "revision": selected["revision"], "updated_on": selected["updated_on"], } def verify_local_socket_transcript( root: Path, successful: Mapping[str, object], *, runtime_mode: str, candidate: str | None, ) -> dict[str, object]: relative = successful.get("socket_transcript_path") expected_relative = f"data/task26-local-http-telegram-api-qa-{runtime_mode}.json" if relative != expected_relative: raise ValueError("local socket transcript path is invalid") path = root / expected_relative private(path) transcript = read_json(path, "local socket transcript") expected_fields = { "schema", "runtime_mode", "candidate_digest", "actual_socket", "mock_network", "backend", "local_endpoint", "connection_count", "request_count", "methods", "getMe_observed", "getUpdates_observed", "revocation_committed", "disconnect", "post_revoke_updates", "server_cleanup", "socket_cleanup", "server_socket_closed", "active_connections", "max_active_connections", "watcher_cleanup", "watch_backend", "watch_directory_resource_count", "watch_inotify_count", "external_traffic", } methods = transcript.get("methods") endpoint = transcript.get("local_endpoint") transcript_hash = hashlib.sha256(canonical(transcript)).hexdigest() if ( set(transcript) != expected_fields or transcript.get("schema") != "task26-local-http-telegram-api-qa-v1" or transcript.get("runtime_mode") != runtime_mode or transcript.get("candidate_digest") != candidate or transcript.get("actual_socket") is not True or transcript.get("mock_network") is not False or transcript.get("backend") != "python-telegram-bot-httpx-polling" or endpoint != { "scheme": "http", "host": "127.0.0.1", "port_class": "ephemeral_loopback", "api_path": "/bot/{method}", } or type(transcript.get("connection_count")) is not int or transcript["connection_count"] < 1 or type(transcript.get("request_count")) is not int or not isinstance(methods, list) or any(not isinstance(method, str) for method in methods) or transcript["request_count"] != len(methods) or transcript.get("getMe_observed") is not True or transcript.get("getUpdates_observed") is not True or "getMe" not in methods or "getUpdates" not in methods or transcript.get("revocation_committed") is not True or transcript.get("disconnect") != "PASS" or transcript.get("post_revoke_updates") != 0 or transcript.get("server_cleanup") is not True or transcript.get("socket_cleanup") is not True or transcript.get("server_socket_closed") is not True or transcript.get("active_connections") != 0 or type(transcript.get("max_active_connections")) is not int or transcript["max_active_connections"] < 1 or transcript.get("watcher_cleanup") is not True or transcript.get("watch_backend") != "dnotify_signalfd_v1" or transcript.get("watch_directory_resource_count") != 1 or transcript.get("watch_inotify_count") != 0 or transcript.get("external_traffic") is not False or successful.get("socket_transcript_sha256") != transcript_hash or successful.get("actual_socket") != transcript.get("actual_socket") or successful.get("mock_network") != transcript.get("mock_network") or successful.get("socket_connection_count") != transcript.get("connection_count") or successful.get("socket_request_count") != transcript.get("request_count") or successful.get("socket_request_methods") != methods or successful.get("socket_disconnect") != transcript.get("disconnect") or successful.get("post_revoke_updates") != transcript.get("post_revoke_updates") or successful.get("server_cleanup") != transcript.get("server_cleanup") or successful.get("socket_cleanup") != transcript.get("socket_cleanup") or successful.get("server_socket_closed") != transcript.get("server_socket_closed") or successful.get("active_connections") != transcript.get("active_connections") or successful.get("authority_watcher_cleanup") != transcript.get("watcher_cleanup") or successful.get("external_traffic") != transcript.get("external_traffic") ): raise ValueError("local socket transcript binding is invalid") return { "schema": "task26-local-http-telegram-api-transcript-binding-v1", "runtime_mode": runtime_mode, "relative_path": expected_relative, "transcript_sha256": transcript_hash, } def verify_task26_contract( root: Path, *, runtime_mode: str, candidate: str | None = None, native: Mapping[str, object] | None = None, cleanup: Mapping[str, object] | None = None, deployment: Mapping[str, object] | None = None, ) -> dict[str, object]: """Derive every Task26 branch verdict from raw, hash-bound API outcomes.""" evidence_root = root / "task26-evidence" try: private(evidence_root, directory=True) chain_path = evidence_root / "receipts.jsonl" private(chain_path) receipts = lines(chain_path, "Task26 receipt") except OSError as exc: raise ValueError("Task26 evidence contract is unavailable") from exc if len(receipts) != len(TASK26_CLAUSES): raise ValueError("Task26 evidence contract is incomplete") expected_files = {"receipts.jsonl"} | { f"{index:02d}-{clause}.json" for index, clause in enumerate(TASK26_CLAUSES, 1) } if {path.name for path in evidence_root.iterdir()} != expected_files: raise ValueError("Task26 evidence inventory is not exact") previous = ZERO invocation_id: str | None = None artifacts: dict[str, dict[str, Any]] = {} for sequence, (clause, receipt) in enumerate( zip(TASK26_CLAUSES, receipts, strict=True), 1 ): if set(receipt) != { "schema", "sequence", "clause", "invocation_id", "candidate_digest", "previous_sha256", "artifact", "artifact_sha256", "receipt_sha256", }: raise ValueError("Task26 receipt schema is invalid") body = {key: value for key, value in receipt.items() if key != "receipt_sha256"} computed = hashlib.sha256(canonical(body)).hexdigest() current_invocation = receipt.get("invocation_id") if ( receipt.get("schema") != "task26-native-receipt-v1" or receipt.get("sequence") != sequence or receipt.get("clause") != clause or receipt.get("previous_sha256") != previous or receipt.get("receipt_sha256") != computed or not isinstance(current_invocation, str) or len(current_invocation) != 32 or any(character not in "0123456789abcdef" for character in current_invocation) or (invocation_id is not None and current_invocation != invocation_id) or (candidate is not None and receipt.get("candidate_digest") != candidate) ): raise ValueError("Task26 receipt chain or invocation binding is invalid") invocation_id = current_invocation artifact_name = f"{sequence:02d}-{clause}.json" if receipt.get("artifact") != artifact_name: raise ValueError("Task26 artifact binding is invalid") artifact_path = evidence_root / artifact_name private(artifact_path) if sha(artifact_path) != receipt.get("artifact_sha256"): raise ValueError("Task26 artifact hash binding is invalid") artifact = read_json(artifact_path, "Task26 artifact") if ( set(artifact) != {"schema", "invocation_id", "candidate_digest", "clause", "payload"} or artifact.get("schema") != "task26-native-artifact-v1" or artifact.get("invocation_id") != invocation_id or artifact.get("candidate_digest") != receipt.get("candidate_digest") or artifact.get("clause") != clause ): raise ValueError("Task26 native artifact envelope is invalid") artifacts[clause] = obj(artifact.get("payload"), f"Task26 {clause} payload") previous = computed from checkin_cli.nutrition_onboarding import QUESTION_FIELDS from checkin_cli.nutrition_onboarding_clarification_policy import compile_clarification_policy from checkin_cli.nutrition_onboarding_reconciliation import build_reconciliation from gateway.platforms.telegram_nutrition_onboarding_copy import parse_answer onboarding = artifacts["onboarding_22_clear"] observations = onboarding.get("observations") if ( onboarding.get("question_fields") != list(QUESTION_FIELDS) or not isinstance(observations, list) or len(observations) != 22 or len(TASK26_ANSWERS) != 22 ): raise ValueError("Task26 22-answer native evidence is incomplete") canonical_answers: dict[str, object] = {} for index, (field, raw, raw_observation) in enumerate( zip(QUESTION_FIELDS, TASK26_ANSWERS, observations, strict=True) ): observation = obj(raw_observation, "Task26 answer observation") parsed = parse_answer(field, raw) if observation != { "index": index, "field": field, "raw": raw, "canonical": parsed, "before_count": index, "after_count": index + 1, "handled": True, "reask_count": 0, }: raise ValueError("Task26 clear answer was re-asked or miscanonicalized") canonical_answers[field] = parsed if onboarding.get("production_answers") != canonical_answers: raise ValueError("Task26 production answers differ from raw canonical results") ambiguity = artifacts["deterministic_ambiguity_revision"] ambiguous_answers = obj(ambiguity.get("ambiguous_answers"), "ambiguous answers") revised_answers = obj(ambiguity.get("revised_answers"), "revised answers") try: reference_date = date.fromisoformat(str(ambiguity.get("reference_date"))) except ValueError as exc: raise ValueError("Task26 ambiguity reference date is invalid") from exc first = compile_clarification_policy(ambiguous_answers, reference_date=reference_date) second = compile_clarification_policy(ambiguous_answers, reference_date=reference_date) revised = compile_clarification_policy(revised_answers, reference_date=reference_date) first_record = build_reconciliation( answers=ambiguous_answers, answers_digest=first.answers_digest, advisory={"status": "unavailable"}, reference_date=reference_date, ) binding = obj(ambiguity.get("revision_binding"), "revision binding") native_branch = obj(ambiguity.get("native_branch"), "native ambiguity branch") workflow_relative = native_branch.get("workflow_path") if ( not isinstance(workflow_relative, str) or Path(workflow_relative).is_absolute() or ".." in Path(workflow_relative).parts ): raise ValueError("Task26 native ambiguity workflow path is invalid") workflow_path = root / workflow_relative private(workflow_path) workflow = read_json(workflow_path, "native ambiguity workflow") changed_fields = [ field for field in QUESTION_FIELDS if ambiguous_answers.get(field) != revised_answers.get(field) ] if ( ambiguity.get("first_result") != first.model_dump(mode="json") or ambiguity.get("second_result") != second.model_dump(mode="json") or first != second or len(first.issues) != 1 or first.issues[0].field != "schedule_constraints" or binding != { "issue_id": first.issues[0].issue_id, "field": "schedule_constraints", "field_index": QUESTION_FIELDS.index("schedule_constraints"), "answers_digest": first.answers_digest, "reconciliation_digest": first.result_digest, } or binding["field_index"] == 0 or changed_fields != ["schedule_constraints"] or ambiguity.get("revised_result") != revised.model_dump(mode="json") or revised.issues or native_branch.get("workflow_sha256") != sha(workflow_path) or native_branch.get("field_zero_rejection") != "reconciliation revision binding is stale" or obj(native_branch.get("status"), "native revised status").get("state") != "customer_attestation" or native_branch.get("initial_reconciliation") != first_record or native_branch.get("revised_reconciliation") != workflow.get("reconciliation") or obj(native_branch.get("revised_reconciliation"), "native revised reconciliation").get("state") != "resolved" or native_branch.get("answers") != revised_answers or workflow.get("answers") != revised_answers ): raise ValueError("Task26 deterministic ambiguity or revision binding is invalid") preview = artifacts["preview_isolation"] if ( preview.get("preview_phase") != "confirming" or preview.get("preview_clarifications") != [] or preview.get("production_answer_count") != 22 or preview.get("production_state") != "customer_attestation" or (native is not None and preview.get("production_session_id") != native.get("session_id")) ): raise ValueError("Task26 preview evidence was used as production acceptance") projection = artifacts["approved_card_projection_gate"] if ( projection.get("projection_fault_seen") is not True or projection.get("projection_error") != "source-golden-approved-card-projection-fault" or projection.get("delivery_rows_after_failure") != {} or not str(projection.get("capability_error", "")).startswith( "DraftGenerationTransitionError: delivery capability target is unavailable" ) ): raise ValueError("Task26 projection failure did not prevent capability issuance") unknown = artifacts["unknown_delivery_no_retry"] expected_unknown = { "accepted": True, "status": "unknown_provider_outcome", "transport_required": False, } unknown_ledger = obj(unknown.get("delivery_ledger"), "unknown delivery ledger") unknown_rows = list(unknown_ledger.values()) unknown_capability = ( obj(unknown_rows[0].get("delivery_capability"), "unknown capability") if len(unknown_rows) == 1 and isinstance(unknown_rows[0], dict) else {} ) unknown_bindings = obj( unknown_capability.get("bindings"), "unknown capability bindings" ) if unknown_capability else {} unknown_snapshot = validate_snapshot( unknown_bindings.get("runtime_authority_snapshot") ) if ( unknown.get("first") != expected_unknown or unknown.get("second") != expected_unknown or unknown.get("transport_calls_before_send") != 0 or len(unknown_rows) != 1 or not isinstance(unknown_rows[0], dict) or unknown_rows[0].get("status") != "unknown_provider_outcome" or unknown_capability.get("status") != "consumed" or deployment is None or unknown_bindings.get("candidate_digest") != deployment.get("candidate_digest") or unknown_snapshot.get("candidate_digest") != deployment.get("candidate_digest") or unknown_bindings.get("candidate_product_binding_sha256") != deployment.get("candidate_product_binding_sha256") or unknown_bindings.get("hermes_wheel_sha256") != deployment.get("hermes_wheel_sha256") or unknown_bindings.get("profile_wheel_sha256") != deployment.get("profile_wheel_sha256") or unknown_bindings.get("wheel_digest") != deployment.get("hermes_wheel_sha256") ): raise ValueError("Task26 unknown delivery outcome was retryable") successful = artifacts["successful_lifecycle"] transcript_binding = verify_local_socket_transcript( root, successful, runtime_mode=runtime_mode, candidate=candidate, ) successful_identity = obj( successful.get("capability_identity"), "successful capability identity" ) delivery_path = root / "data/owner-actions/draft-deliveries.json" successful_snapshot = validate_snapshot( successful.get("runtime_authority_snapshot") ) final_deliveries = read_json(delivery_path, "successful delivery ledger") final_delivery = obj( next(iter(final_deliveries.values())), "successful delivery" ) final_capability = obj( final_delivery.get("delivery_capability"), "successful capability" ) final_bindings = obj( final_capability.get("bindings"), "successful capability bindings" ) if ( successful.get("attestation_state") != "owner_review" or successful.get("owner_review_state") != "ready" or successful.get("activation_state") != "ACTIVE" or successful.get("checkin_finalized") is not True or successful.get("approval_status") != "approved" or successful.get("delivery_status") != "sent_audited" or successful.get("surface_schema") != "telegram-customer-surface-receipt-v2" or successful.get("transport_calls") != 1 or successful.get("duplicate_rejected") is not True or successful.get("capability_issue_path") != "dualcoach-controller-protected-v2" or successful.get("capability_issue_receipt_schema") != "draft-delivery-capability-issue-v2" or successful.get("live_revocation_disconnect") != "PASS" or successful.get("live_revocation_network_surface") != "local_http_telegram_api_v1" or successful.get("actual_socket") is not True or successful.get("mock_network") is not False or type(successful.get("socket_connection_count")) is not int or successful["socket_connection_count"] < 1 or type(successful.get("socket_request_count")) is not int or successful["socket_request_count"] < 1 or successful.get("socket_disconnect") != "PASS" or not isinstance(successful.get("socket_request_methods"), list) or "getMe" not in successful["socket_request_methods"] or "getUpdates" not in successful["socket_request_methods"] or successful.get("authority_watch_backend") != "dnotify_signalfd_v1" or successful.get("authority_watch_directory_resource_count") != 1 or successful.get("authority_watch_inotify_count") != 0 or successful.get("authority_watcher_cleanup") is not True or successful.get("live_advisory_store") is not False or successful.get("delivery_ledger_sha256") != sha(delivery_path) or deployment is None or successful_snapshot.get("candidate_digest") != deployment.get("candidate_digest") or successful_snapshot != final_bindings.get("runtime_authority_snapshot") or successful_identity != { "candidate_digest": deployment.get("candidate_digest"), "candidate_product_binding_sha256": deployment.get( "candidate_product_binding_sha256" ), "hermes_wheel_sha256": deployment.get("hermes_wheel_sha256"), "profile_wheel_sha256": deployment.get("profile_wheel_sha256"), "wheel_digest": deployment.get("hermes_wheel_sha256"), } ): raise ValueError("Task26 successful lifecycle native binding is invalid") terminal = artifacts["cleanup_resume_terminal"] journal = root / "data/customer-cleanup/client_001.journal.jsonl" if ( terminal.get("fault_phase") != "copied_verified" or terminal.get("fault_seen") is not True or terminal.get("terminal_phase") != "committed" or terminal.get("terminal") is not True or any(terminal.get(key) != 0 for key in ("active_count", "pending_count", "unknown_count", "orphan_count")) or terminal.get("journal_sha256") != sha(journal) or (cleanup is not None and terminal.get("operation_id") != cleanup.get("operation_id")) ): raise ValueError("Task26 cleanup did not resume forward to one terminal receipt") return { "clause_count": len(TASK26_CLAUSES), "chain_head": previous, "invocation_id": invocation_id, "local_socket_transcript": transcript_binding, } def verify_deployment_product_binding( root: Path, candidate: str, product_binding: dict[str, object], ) -> dict[str, object]: from gateway.platforms.task26_candidate_derivation import ( validate_deployment_receipt, ) deployment_path = root / "data/source-deployment-receipt.json" membership_path = root / "data/source-membership-evidence.json" private(deployment_path) private(membership_path) deployment = read_json(deployment_path, "source deployment receipt") membership = read_json(membership_path, "source membership evidence") if deployment.get("product_binding") != product_binding: raise ValueError("deployment product binding differs from qualification") validate_deployment_receipt(deployment, product_binding) fields = ( "candidate_digest", "candidate_product_binding_sha256", "hermes_wheel_sha256", "profile_wheel_sha256", ) if ( deployment.get("candidate_digest") != candidate or any(membership.get(field) != deployment.get(field) for field in fields) or membership.get("deployment_receipt_sha256") != sha(deployment_path) ): raise ValueError("activation deployment receipt binding differs") return { "schema": "task26-deployment-wheel-binding-v1", **{field: deployment[field] for field in fields}, "deployment_receipt_sha256": sha(deployment_path), } def verify_native( root: Path, artifacts: dict[str, Path], customer: str, candidate: str, deployment_wheels: Mapping[str, object], ) -> dict[str, Any]: if (root / "lifecycle-events.jsonl").exists(): raise ValueError("self-authored lifecycle/PASS declarations are forbidden") registry = read_json(root / "customers/registry.json", "registry") matches = [row for row in registry.get("customers", []) if isinstance(row, dict) and row.get("customer_key") == customer] if len(matches) != 1 or matches[0].get("enabled") is not False or obj(matches[0].get("ai_processing_consent"), "consent").get("granted") is not False: raise ValueError("terminal registry authority is not withdrawn") activation = read_json(root / "data/customer-activation-journal.json", "activation journal") if activation.get("state") != "committed" or activation.get("customer_id") != customer: raise ValueError("activation commit is unavailable") bootstrap_path = root / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" session_id = verify_bootstrap_active_lineage(bootstrap_path, customer) ready = read_json(required(artifacts, "nutrition-onboarding/ready.json"), "onboarding ready") baseline = read_json(required(artifacts, "nutrition-onboarding/baseline-v1.json"), "onboarding baseline") readiness = read_json(required(artifacts, "nutrition-onboarding/readiness-current.json"), "readiness pointer") receipt = read_json(required(artifacts, "nutrition-onboarding/readiness-receipt-v1.json"), "readiness receipt") if ready.get("state") != "ready" or ready.get("customer_key") != customer or receipt.get("delivery_enabled") is not False or receipt.get("activation_enabled") is not False: raise ValueError("native readiness state is invalid") if ( baseline.get("customer_key") != customer or baseline.get("equation_sex_basis") != "male" or baseline.get("height_cm") != 180.0 or baseline.get("weight_kg") != 80.0 or baseline.get("activity_category") != "moderate" or baseline.get("goal_type") != "maintain" or baseline.get("target_weight_kg") is not None or baseline.get("target_date") is not None or baseline.get("meal_count") != 3 or baseline.get("schedule_constraints") != TASK26_ANSWERS[-1] or any( baseline.get(field) != [] for field in ( "allergies", "intolerances", "religious_ethical_exclusions", "disliked_foods", "dietary_preferences", "conditions", "medications", ) ) or baseline.get("pregnancy_breastfeeding") is not False or not isinstance(baseline.get("customer_attested_at_kst"), str) or not isinstance(baseline.get("owner_review_receipt"), str) or len(str(baseline.get("owner_review_receipt"))) != 64 ): raise ValueError("native onboarding baseline does not prove the canonical answers and reviews") for document, label in ((baseline, "baseline"), (readiness, "readiness"), (receipt, "readiness receipt")): claimed = document.get("digest") if claimed != hashlib.sha256(canonical({key: value for key, value in document.items() if key != "digest"})).hexdigest(): raise ValueError(f"{label} digest is invalid") checkin_id, draft_id = verify_canonical_journey_events( required(artifacts, "wizard/events.jsonl"), customer ) generations = read_json(required(artifacts, "projections/data/owner-actions/draft-generations.json"), "generation projection") if generations.get("schema_version") != "customer-cleanup-draft-generations-projection-v1" or generations.get("customer_key") != customer: raise ValueError("generation projection schema is invalid") history = verify_generation_lineage( generations.get("histories"), customer, checkin_id, draft_id ) deliveries_path = root / "data/owner-actions/draft-deliveries.json" private(deliveries_path) deliveries = read_json(deliveries_path, "delivery ledger") if len(deliveries) != 1: raise ValueError("delivery cardinality is invalid") delivery = next(iter(deliveries.values())) capability = obj(delivery.get("delivery_capability"), "delivery capability") bindings = obj(capability.get("bindings"), "delivery capability bindings") if ( delivery.get("draft_id") != draft_id or delivery.get("customer_key") != customer or delivery.get("status") != "sent_audited" or capability.get("status") != "consumed" or bindings.get("candidate_digest") != candidate or bindings.get("candidate_product_binding_sha256") != deployment_wheels.get("candidate_product_binding_sha256") or bindings.get("hermes_wheel_sha256") != deployment_wheels.get("hermes_wheel_sha256") or bindings.get("profile_wheel_sha256") != deployment_wheels.get("profile_wheel_sha256") or bindings.get("wheel_digest") != deployment_wheels.get("hermes_wheel_sha256") or candidate in { bindings.get("wheel_digest"), bindings.get("hermes_wheel_sha256"), bindings.get("profile_wheel_sha256"), } ): raise ValueError("delivery capability identity lineage is invalid") raw_calls = lines(root / "raw-customer-transport-calls.jsonl", "raw customer call") if len(raw_calls) != 1 or raw_calls[0].get("provider_message_id") != delivery.get("message_id") or raw_calls[0].get("text_sha256") != hashlib.sha256(str(delivery.get("text", "")).encode()).hexdigest(): raise ValueError("raw transport cardinality or receipt binding is invalid") surface_path = root / "data/owner-actions/customer-surface-receipts-v2/receipts.jsonl" private(surface_path) surfaces = lines(surface_path, "surface receipt") surface = surfaces[0] if len(surfaces) == 1 else {} if ( surface.get("schema") != "telegram-customer-surface-receipt-v2" or surface.get("draft_id") != draft_id or surface.get("provider_message_id") != delivery.get("message_id") or surface.get("customer_key") != customer or surface.get("approved_generation") != delivery.get("generation") or surface.get("approved_generation_record_digest") != delivery.get("generation_record_digest") or surface.get("terminal_generation") != history[-1].get("generation") or surface.get("terminal_generation_record_digest") != history[-1].get("record_digest") ): raise ValueError("customer surface receipt descendant binding is invalid") service_path = root / "data/owner-actions/customer-service-state.json" verify_customer_quiescence(service_path, customer) outbox_path = root / "data/onboarding/telegram-publication-outbox-v1/ledger.json" private(outbox_path) outbox = read_json(outbox_path, "publication outbox") records = outbox.get("records") if not isinstance(records, list) or not records or any(row.get("session_id") != session_id or row.get("state") != "COMMITTED" for row in records): raise ValueError("publication receipt lineage is invalid") capability_identity: dict[str, object] = { "schema": "task26-delivery-capability-identity-receipt-v1", "candidate_digest": bindings["candidate_digest"], "candidate_product_binding_sha256": bindings[ "candidate_product_binding_sha256" ], "hermes_wheel_sha256": bindings["hermes_wheel_sha256"], "profile_wheel_sha256": bindings["profile_wheel_sha256"], "wheel_digest": bindings["wheel_digest"], } capability_identity["receipt_sha256"] = hashlib.sha256( canonical(capability_identity) ).hexdigest() return { "customer_key": customer, "session_id": session_id, "draft_id": draft_id, "publication_count": len(records), "generation_count": len(history), "capability_identity_receipt": capability_identity, "runtime_authority_snapshot": validate_snapshot( bindings.get("runtime_authority_snapshot") ), } def frozen_terminal_inventory( root: Path, customer: str, cleanup: Mapping[str, object] ) -> dict[str, object]: """Derive terminal state without asking production code to open write locks.""" categories = {field: 0 for field in TERMINAL_CATEGORY_FIELDS} categories["owner_action"] = 1 if cleanup.get("projection_record_count") else 0 activation_count = 0 for relative in ( "data/customer-activation-journal.json", "data/customer-activation-audit.jsonl", ): path = root / relative if not path.exists(): continue values = lines(path, "activation authority") if path.suffix == ".jsonl" else [read_json(path, "activation authority")] activation_count += sum( 1 for row in values if row.get("customer_id") == customer or row.get("customer_key") == customer ) categories["activation"] = activation_count return { "customer_key": customer, "disabled": True, "nonconsenting": True, "categories": dict(sorted(categories.items())), "active_count": 0, "pending_count": 0, "unknown_count": 0, "orphan_count": 0, "archive_verified": True, "journal_committed": True, "terminal": True, } def verify( root: Path, *, runtime_override: dict[str, object] | None = None ) -> dict[str, object]: root = root.absolute() private(root, directory=True) if (root / "lifecycle-events.jsonl").exists(): raise ValueError("self-authored lifecycle/PASS declarations are forbidden") source_tree_digest, source = verify_provenance( root, offline_recorded_source=runtime_override is not None ) candidate, candidate_binding_sha256, product_binding = ( verify_qualification_product_binding(root) ) if candidate == source_tree_digest: raise ValueError("candidate binding substituted the source tree digest") source_document = read_json(root / "source-provenance.json", "source provenance") input_doc = read_json(root / "driver-input.json", "driver input") runtime_mode = input_doc.get("runtime_mode", "source") if runtime_mode == "installed": installed_candidate, authority_root = verify_installed_provenance( root, source_document, runtime_override=runtime_override ) if installed_candidate != candidate: raise ValueError("installed and qualification candidate digests differ") installed_document = read_private_json(root / "installed-provenance.json") installed_binding = obj( installed_document.get("binding"), "installed candidate binding" ) active_runtime = ( runtime_override if runtime_override is not None else obj(installed_binding.get("runtime"), "installed runtime") ) active_distributions = obj( active_runtime.get("distributions"), "installed distributions" ) verified_candidate, _, _ = verify_qualification_product_binding( root, actual_hermes_wheel=Path( str(obj(active_distributions.get("hermes"), "Hermes distribution").get("wheel_path", "")) ), actual_profile_wheel=Path( str(obj(active_distributions.get("profile"), "profile distribution").get("wheel_path", "")) ), ) if verified_candidate != candidate: raise ValueError("installed wheel product candidate differs") elif runtime_mode == "source": authority_root = source source_text = str(source) if source_text not in sys.path: sys.path.insert(0, source_text) else: raise ValueError("driver runtime mode is invalid") observer = verify_observer(root, candidate) customer = input_doc.get("customer_key") if ( input_doc.get("candidate_digest") != candidate or input_doc.get("source_tree_digest") != source_tree_digest or input_doc.get("candidate_product_binding_sha256") != candidate_binding_sha256 or not isinstance(customer, str) ): raise ValueError("driver input binding is invalid") artifacts, cleanup = verify_cleanup(root, customer) if stat.S_IMODE(root.stat().st_mode) == 0o500: terminal_inventory = frozen_terminal_inventory(root, customer, cleanup) else: from checkin_cli.customer_cleanup import post_cleanup_authority_inventory terminal = post_cleanup_authority_inventory(root, customer) terminal_inventory = serialize_terminal_inventory(terminal) if not terminal.terminal or any( value != 0 for value in ( terminal.active_count, terminal.pending_count, terminal.unknown_count, terminal.orphan_count, ) ): raise ValueError("post-cleanup authority inventory is not terminal") cleanup["terminal_inventory"] = terminal_inventory deployment_wheels = verify_deployment_product_binding( root, candidate, product_binding ) native = verify_native( root, artifacts, customer, candidate, deployment_wheels ) task26_contract = verify_task26_contract( root, runtime_mode=str(runtime_mode), candidate=candidate, native=native, cleanup=cleanup, deployment=deployment_wheels, ) from gateway.platforms.task26_candidate_authority import verify_candidate_authority candidate_authority = verify_candidate_authority(root, candidate) chain_head = hashlib.sha256(canonical({ "candidate": candidate, "observer": observer["head"], "manifest": cleanup["manifest_sha256"], "native": native, "task26_contract_head": task26_contract["chain_head"], "authority_registry_head": candidate_authority["registry_head_sha256"], "authority_ledger_head": candidate_authority["ledger_head_sha256"], })).hexdigest() status = "ACTUAL_INSTALLED_GOLDEN_PATH_PASS" if runtime_mode == "installed" else "ACTUAL_SOURCE_GOLDEN_PATH_PASS" return {"status": status, "runtime_mode": runtime_mode, "candidate_digest": candidate, "source_tree_digest": source_tree_digest, "deployment_wheels": deployment_wheels, "chain_head": chain_head, "source": str(source), "package_authority": str(authority_root), "observer_receipt_count": observer["receipt_count"], "observer_backend": observer["backend"], "observer_directory_resource_count": observer["directory_resource_count"], "observer_inotify_watch_count": observer["inotify_watch_count"], "task26_contract": task26_contract, "candidate_authority": candidate_authority, **cleanup, **native} def verify_envelope(*_args: object, **_kwargs: object) -> tuple[dict[str, object], dict[str, str]]: raise ValueError("declaration, not a production artifact") def verify_phase_payload(phase: str, payload: dict[str, object]) -> None: calls = payload.get("transport_calls") if phase == "SENT_AUDITED" and ( payload.get("state") != "sent_audited" or not isinstance(calls, list) or len(calls) != 1 ): raise ValueError("sent audit is not bound to exactly one transport call") if phase == "CLEANUP_COMPLETE" and any(obj(payload.get("inventory"), "cleanup inventory").get(key) != 0 for key in ("pending", "unknown", "orphan", "residual_authorities")): raise ValueError("cleanup inventory is not empty") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("bundle", type=Path) parser.add_argument("--rehydrated-runtime", type=Path) args = parser.parse_args() try: runtime_override = ( read_private_json(args.rehydrated_runtime) if args.rehydrated_runtime is not None else None ) print(json.dumps(verify(args.bundle, runtime_override=runtime_override), sort_keys=True, separators=(",", ":"))) except (OSError, ValueError, json.JSONDecodeError) as exc: print(json.dumps({"status": "SOURCE_GOLDEN_PATH_FAIL", "reason": str(exc)}, sort_keys=True), file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())