#!/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 _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 DEFAULT_SOURCE = Path("/home/cube/.cache/task26-strict-successor-1786976146/src-p") 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 = False) -> 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}") allowed = ({0o700, 0o500} if directory else {0o600, 0o400}) if frozen else ({0o700} if directory else {0o600}) 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) -> 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 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 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 = 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_installed_provenance( root: Path, source_provenance: dict[str, Any] ) -> tuple[str, Path]: path = root / "installed-provenance.json" document = read_private_json(path) if set(document) != {"schema", "binding", "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") expected_candidate = hashlib.sha256(canonical(binding)).hexdigest() if document.get("candidate_digest") != expected_candidate: raise ValueError("installed candidate digest 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__)), } 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") recomputed = collect_installed_runtime( venv=Path(str(runtime.get("venv", ""))), site_packages=Path(str(runtime.get("site_packages", ""))), profile_wheel=Path(str(profile.get("wheel_path", ""))), hermes_wheel=Path(str(hermes.get("wheel_path", ""))), ) if recomputed != runtime: raise ValueError("installed runtime receipt differs from installed bytes") roots = tuple( Path(root_path) for distribution in (profile, 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 actual_receipts != declarations: raise ValueError("installed loaded-module declarations differ from import reality") authority = config.get("profile_package_authority") if not isinstance(authority, str) or Path(authority).resolve(strict=True) != Path(str(runtime["site_packages"])): raise ValueError("installed profile package authority is invalid") return expected_candidate, Path(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] if armed.get("candidate_id") != candidate or sequences != list(range(1, len(sequences) + 1)) or len(sequences) < 4: raise ValueError("observer subscription binding is invalid") return {"receipt_count": len(receipts), "head": previous} 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_native(root: Path, artifacts: dict[str, Path], customer: str, candidate: str) -> 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") 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: raise ValueError("delivery capability 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") return {"customer_key": customer, "session_id": session_id, "draft_id": draft_id, "publication_count": len(records), "generation_count": len(history)} def verify(root: Path) -> 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_candidate, source = verify_provenance(root) 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": candidate, authority_root = verify_installed_provenance(root, source_document) elif runtime_mode == "source": candidate, authority_root = source_candidate, 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 not isinstance(customer, str): raise ValueError("driver input binding is invalid") artifacts, cleanup = verify_cleanup(root, customer) 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 native = verify_native(root, artifacts, customer, candidate) chain_head = hashlib.sha256(canonical({"candidate": candidate, "observer": observer["head"], "manifest": cleanup["manifest_sha256"], "native": native})).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, "chain_head": chain_head, "source": str(source), "package_authority": str(authority_root), "observer_receipt_count": observer["receipt_count"], **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) args = parser.parse_args() try: print(json.dumps(verify(args.bundle), 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())