diff --git a/docs/maintenance.md b/docs/maintenance.md index ff7e9dbd7e..26141aba65 100644 --- a/docs/maintenance.md +++ b/docs/maintenance.md @@ -480,6 +480,22 @@ Exit code is non-zero when any check reports `error` (or, with `--strict`, temporarily busy under a concurrent rebuild — never aborts the rest; each check independently reports its own outcome. +### `polylogue ops maintenance live-proof` — immutable campaign evidence + +Read-only evidence collection for the reindex campaign. The command accepts a fixed registered proof id and writes one new self-hashed JSON receipt outside the archive. It has no command-execution option and cannot apply a mutation, control the daemon, migrate a tier, or promote a generation. + +```bash +polylogue ops maintenance live-proof \ + --proof-id archive-verification \ + --output /path/to/new/live-proof.json +``` + +The registry currently has exactly three routes: `archive-verification` for a fixed read-only archive-check profile, `candidate-archive-verification` for that profile against one named inactive generation, and `existing-apply-receipt` for a pre-existing `polylogue.apply-receipt.v1` input. Candidate mode requires `--candidate-generation`; existing-apply mode requires `--apply-receipt` and the registered `known-source-remediation` operation id; every other combination is rejected. An arbitrary nonempty operation id is not accepted. + +Each `polylogue.live-proof-receipt.v1` binds the proof and Bead id, exact code SHA, archive identity, source snapshot, all six active archive-tier schema versions, parser and lowering fingerprints, the active SQLite file set, and the candidate generation, schema, and SQLite file set where applicable. SQLite bindings include the database and WAL/journal sidecars and refuse a file set that changes while it is captured. Candidate metadata must name the canonical inactive generation and the same source snapshot. Archives whose configured paths contain SQLite URI query characters are rejected before proof dependencies open them. Private local paths are represented only as a SHA-256 digest plus basename. + +The receipt keeps complete structured archive-verification evidence after redacting archive paths and any absolute paths emitted by checks. `archive-verification` runs the entire live archive profile. `candidate-archive-verification` runs both canonical candidate acceptance profiles: the index-candidate checks against the inactive generation and the cross-tier checks against that generation plus the durable archive. Existing-apply evidence embeds the validated input receipt, so consumers revalidate its self-hash, bindings, registered operation id, and match to the recorded input digest. Verification consumers validate the fixed profile membership, check outcomes, typed status/residue relationship, bindings, and input hashes again. Aggregate validation binds pre-promotion candidate evidence to the inactive generation. After promotion, it validates the retained generation and candidate file set directly while recapturing the current active binding for post-promotion receipts. Final proof consumption requires every registered route exactly once; `not_applicable` is accepted only with its typed residue. Output creation is exclusive and failure-atomic: a failed write removes its partial file and syncs the destination directory. + ### `--operation-id` and `--resume`: worked example Replay execution writes a small JSON state file under diff --git a/polylogue/cli/commands/maintenance/__init__.py b/polylogue/cli/commands/maintenance/__init__.py index 6c29815aef..6513730dbf 100644 --- a/polylogue/cli/commands/maintenance/__init__.py +++ b/polylogue/cli/commands/maintenance/__init__.py @@ -189,6 +189,12 @@ "verify_archive_command", "Prove the archive is coherent after a rebuild, restore, or promotion. Read-only.", ), + ( + "live-proof", + "_live_proof", + "live_proof_command", + "Collect one fixed, immutable live-proof receipt. Read-only.", + ), ( "cursor-authority-reconcile", "_cursor_authority", diff --git a/polylogue/cli/commands/maintenance/_live_proof.py b/polylogue/cli/commands/maintenance/_live_proof.py new file mode 100644 index 0000000000..0a18139bbb --- /dev/null +++ b/polylogue/cli/commands/maintenance/_live_proof.py @@ -0,0 +1,63 @@ +"""``maintenance live-proof``: collect one fixed, immutable evidence receipt.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import click + +from polylogue.paths import archive_root + + +def _is_under_any(path: Path, roots: tuple[Path, ...]) -> bool: + return any(path == root or root in path.parents for root in roots) + + +@click.command("live-proof") +@click.option("--proof-id", required=True, help="One registered live-proof id.") +@click.option("--candidate-generation", type=str, help="Inactive generation id for the candidate proof route only.") +@click.option( + "--apply-receipt", + type=click.Path(path_type=Path, file_okay=True, dir_okay=False, readable=True), + help="Existing immutable apply receipt for the existing-apply route only.", +) +@click.option( + "--output", + type=click.Path(path_type=Path, file_okay=True, dir_okay=False, writable=True), + required=True, + help="New receipt path outside the archive. Existing files are refused.", +) +def live_proof_command( + proof_id: str, + candidate_generation: str | None, + apply_receipt: Path | None, + output: Path, +) -> None: + """Collect one registered proof without mutating archive or daemon state.""" + + from polylogue.maintenance.live_proof import ( + LiveProofError, + archive_owned_storage_roots, + collect_live_proof, + write_live_proof_receipt, + ) + + root = archive_root().resolve() + target = output.expanduser().resolve() + try: + owned_roots = archive_owned_storage_roots(root) + if _is_under_any(target, owned_roots): + raise click.BadParameter("receipt output must be outside archive-owned storage", param_hint="--output") + receipt = collect_live_proof( + proof_id, + root, + candidate_generation_id=candidate_generation, + apply_receipt_path=apply_receipt, + ) + write_live_proof_receipt(target, receipt) + except LiveProofError as exc: + raise click.ClickException(str(exc)) from exc + except OSError as exc: + raise click.ClickException("live-proof receipt output could not be written") from exc + click.echo(json.dumps(receipt.to_document(), indent=2, sort_keys=True)) diff --git a/polylogue/maintenance/live_proof.py b/polylogue/maintenance/live_proof.py new file mode 100644 index 0000000000..ec53640aaf --- /dev/null +++ b/polylogue/maintenance/live_proof.py @@ -0,0 +1,1359 @@ +"""Static, read-only live-proof receipt protocol for the reindex campaign. + +This module collects immutable evidence only. Its fixed registry has no +runtime-registration seam, and the CLI adapter accepts neither commands nor +callables. A consumer can therefore validate a receipt without turning this +proof protocol into mutation authority. +""" + +from __future__ import annotations + +import json +import os +import re +import sqlite3 +import subprocess +import tempfile +import time +from collections.abc import Mapping, Sequence +from contextlib import closing +from dataclasses import dataclass, replace +from enum import StrEnum +from pathlib import Path +from typing import Final, Literal, cast + +from polylogue.core.hashing import hash_file, hash_payload, hash_text +from polylogue.core.json import JSONDocument, JSONValue, is_json_document, require_json_document +from polylogue.version import VERSION_INFO + +LIVE_PROOF_RECEIPT_SCHEMA: Final = "polylogue.live-proof-receipt.v1" +EXISTING_APPLY_RECEIPT_SCHEMA: Final = "polylogue.apply-receipt.v1" +LIVE_PROOF_REGISTRY_VERSION: Final = 1 + +_SHA256_RE: Final = re.compile(r"[0-9a-f]{64}") +_CODE_SHA_RE: Final = re.compile(r"[0-9a-f]{40,64}") +_GENERATION_ID_RE: Final = re.compile(r"gen-[A-Za-z0-9][A-Za-z0-9._-]{0,127}") +_RESIDUE_CODE_RE: Final = re.compile(r"[a-z][a-z0-9]*(?:[-_.:][a-z0-9]+)*") +_ABSOLUTE_PATH_RE: Final = re.compile(r"(? JSONDocument: + return {"kind": self.kind.value, "code": self.code} + + +@dataclass(frozen=True, slots=True) +class PrivatePathReference: + """A private local path represented only by its basename and opaque digest.""" + + basename: str + sha256: str + + @classmethod + def capture(cls, path: Path) -> PrivatePathReference: + resolved = path.expanduser().resolve() + return cls(basename=resolved.name, sha256=hash_text(str(resolved))) + + def to_document(self) -> JSONDocument: + return {"basename": self.basename, "sha256": self.sha256} + + +@dataclass(frozen=True, slots=True) +class LiveProofArchiveProfile: + """One fixed archive-verification profile for a proof route.""" + + name: str + checks: tuple[str, ...] + target: Literal["active", "candidate_index", "candidate_cross_tier"] + + +@dataclass(frozen=True, slots=True) +class LiveProofSpec: + """One compile-time proof route with no executable registration seam.""" + + proof_id: LiveProofId + bead_id: str + mode: LiveProofMode + producer: Literal["archive_verification", "existing_apply_receipt"] + archive_profiles: tuple[LiveProofArchiveProfile, ...] = () + + +@dataclass(frozen=True, slots=True) +class LiveProofBindings: + code_sha: str + archive_identity_digest: str + source_snapshot: str + schema_versions: tuple[tuple[str, int], ...] + tier_file_set_digests: tuple[tuple[str, str], ...] + candidate_index_schema_version: int | None + parser_fingerprints: tuple[tuple[str, str], ...] + lowering_fingerprint: str + active_index_sha256: str + candidate_generation_id: str | None + candidate_index_sha256: str | None + private_paths: tuple[tuple[str, PrivatePathReference], ...] + + def to_document(self) -> JSONDocument: + return { + "code_sha": self.code_sha, + "archive_identity_digest": self.archive_identity_digest, + "source_snapshot": self.source_snapshot, + "schema_versions": dict(self.schema_versions), + "tier_file_set_digests": dict(self.tier_file_set_digests), + "candidate_index_schema_version": self.candidate_index_schema_version, + "parser_fingerprints": dict(self.parser_fingerprints), + "lowering_fingerprint": self.lowering_fingerprint, + "active_index_sha256": self.active_index_sha256, + "candidate_generation_id": self.candidate_generation_id, + "candidate_index_sha256": self.candidate_index_sha256, + "private_paths": {name: ref.to_document() for name, ref in self.private_paths}, + } + + +@dataclass(frozen=True, slots=True) +class LiveProofReceipt: + proof_id: LiveProofId + bead_id: str + mode: LiveProofMode + registry_version: int + generated_at_ms: int + bindings: LiveProofBindings + result: JSONDocument + residues: tuple[LiveProofResidue, ...] + input_receipt_digests: tuple[str, ...] + + def payload(self) -> JSONDocument: + return { + "receipt_schema": LIVE_PROOF_RECEIPT_SCHEMA, + "proof_id": self.proof_id.value, + "bead_id": self.bead_id, + "mode": self.mode.value, + "registry_version": self.registry_version, + "generated_at_ms": self.generated_at_ms, + "bindings": self.bindings.to_document(), + "result": self.result, + "residues": [residue.to_document() for residue in self.residues], + "input_receipt_digests": list(self.input_receipt_digests), + } + + @property + def receipt_sha256(self) -> str: + return hash_payload(self.payload()) + + def to_document(self) -> JSONDocument: + return {**self.payload(), "receipt_sha256": self.receipt_sha256} + + +def _archive_profiles() -> tuple[LiveProofArchiveProfile, ...]: + from polylogue.maintenance.archive_verification import ( + ARCHIVE_VERIFICATION_CHECK_NAMES, + REINDEX_ACCEPTANCE_CHECKS, + REINDEX_CROSS_TIER_ACCEPTANCE_CHECKS, + ) + + return ( + LiveProofArchiveProfile("active-archive", ARCHIVE_VERIFICATION_CHECK_NAMES, "active"), + LiveProofArchiveProfile("candidate-index", REINDEX_ACCEPTANCE_CHECKS, "candidate_index"), + LiveProofArchiveProfile("candidate-cross-tier", REINDEX_CROSS_TIER_ACCEPTANCE_CHECKS, "candidate_cross_tier"), + ) + + +_ACTIVE_ARCHIVE_PROFILE, _CANDIDATE_INDEX_PROFILE, _CANDIDATE_CROSS_TIER_PROFILE = _archive_profiles() + +LIVE_PROOF_SPECS: Final[tuple[LiveProofSpec, ...]] = ( + LiveProofSpec( + proof_id=LiveProofId.ARCHIVE_VERIFICATION, + bead_id="polylogue-x97cf", + mode=LiveProofMode.READ_ONLY, + producer="archive_verification", + archive_profiles=(_ACTIVE_ARCHIVE_PROFILE,), + ), + LiveProofSpec( + proof_id=LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION, + bead_id="polylogue-x97cf", + mode=LiveProofMode.CANDIDATE, + producer="archive_verification", + archive_profiles=(_CANDIDATE_INDEX_PROFILE, _CANDIDATE_CROSS_TIER_PROFILE), + ), + LiveProofSpec( + proof_id=LiveProofId.EXISTING_APPLY_RECEIPT, + bead_id="polylogue-x97cf", + mode=LiveProofMode.EXISTING_APPLY_RECEIPT, + producer="existing_apply_receipt", + ), +) + + +def validate_live_proof_registry(specs: Sequence[LiveProofSpec] = LIVE_PROOF_SPECS) -> None: + """Require the complete fixed protocol registry and canonical profiles.""" + + from polylogue.maintenance.archive_verification import ARCHIVE_VERIFICATION_CHECKS + + expected = set(LiveProofId) + actual = {spec.proof_id for spec in specs} + if actual != expected or len(specs) != len(expected): + raise LiveProofError("live-proof registry must contain every fixed proof id exactly once") + for spec in specs: + if not spec.bead_id.startswith("polylogue-"): + raise LiveProofError("live-proof spec has an invalid bead id") + if spec.mode is LiveProofMode.EXISTING_APPLY_RECEIPT: + if spec.producer != "existing_apply_receipt" or spec.archive_profiles: + raise LiveProofError("existing-apply proof spec may only validate an input receipt") + continue + if spec.producer != "archive_verification" or not spec.archive_profiles: + raise LiveProofError("archive proof specs require registered archive checks") + for profile in spec.archive_profiles: + if not profile.checks or len(set(profile.checks)) != len(profile.checks): + raise LiveProofError("live-proof archive profile is incomplete") + for check_name in profile.checks: + if next((check for check in ARCHIVE_VERIFICATION_CHECKS if check.name == check_name), None) is None: + raise LiveProofError("live-proof spec references an unknown archive verification check") + by_id = {spec.proof_id: spec for spec in specs} + if by_id[LiveProofId.ARCHIVE_VERIFICATION].archive_profiles != (_ACTIVE_ARCHIVE_PROFILE,): + raise LiveProofError("read-only proof must use the complete active archive profile") + if by_id[LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION].archive_profiles != ( + _CANDIDATE_INDEX_PROFILE, + _CANDIDATE_CROSS_TIER_PROFILE, + ): + raise LiveProofError("candidate proof must use both canonical candidate acceptance profiles") + + +def live_proof_spec(proof_id: str) -> LiveProofSpec: + try: + parsed = LiveProofId(proof_id) + except ValueError as exc: + raise LiveProofError("unknown live-proof id") from exc + for spec in LIVE_PROOF_SPECS: + if spec.proof_id is parsed: + return spec + raise LiveProofError("live-proof registry is incomplete") + + +def _run_git(repository: Path, *arguments: str) -> subprocess.CompletedProcess[str]: + try: + return subprocess.run( + ("git", "-C", str(repository), *arguments), + check=False, + capture_output=True, + text=True, + timeout=_GIT_TIMEOUT_SECONDS, + ) + except (FileNotFoundError, OSError, subprocess.TimeoutExpired) as exc: + raise LiveProofError("exact code SHA is unavailable") from exc + + +def _installed_code_sha() -> str: + commit = (VERSION_INFO.commit or "").lower() + if not _CODE_SHA_RE.fullmatch(commit): + raise LiveProofError("installed package has no exact build commit") + return commit + + +def _code_sha() -> str: + configured = os.environ.get("POLYLOGUE_CODE_SHA", "").strip().lower() + repository = Path(__file__).resolve().parents[2] + if (repository / ".git").exists(): + dirty = _run_git(repository, "status", "--porcelain=v1", "--untracked-files=all") + if dirty.returncode != 0 or dirty.stdout.strip(): + raise LiveProofError("live proofs require a clean git worktree") + if configured: + if not _CODE_SHA_RE.fullmatch(configured): + raise LiveProofError("POLYLOGUE_CODE_SHA must be an exact git commit SHA") + return configured + completed = _run_git(repository, "rev-parse", "--verify", "HEAD") + sha = completed.stdout.strip().lower() + if completed.returncode != 0 or not _CODE_SHA_RE.fullmatch(sha): + raise LiveProofError("exact code SHA is unavailable") + return sha + if configured: + if not _CODE_SHA_RE.fullmatch(configured): + raise LiveProofError("POLYLOGUE_CODE_SHA must be an exact git commit SHA") + return configured + return _installed_code_sha() + + +def _readonly_uri(path: Path) -> str: + return f"{path.resolve(strict=True).as_uri()}?mode=ro" + + +def _open_readonly(path: Path) -> sqlite3.Connection: + return sqlite3.connect(_readonly_uri(path), uri=True, timeout=2) + + +def _require_uri_safe_location(location: object) -> None: + """Fail closed before dependencies that still interpolate SQLite URIs.""" + + from polylogue.storage.archive_identity import ArchiveLocation + + assert isinstance(location, ArchiveLocation) + paths = (location.configured_root, location.active_index_path) + tuple( + location.configured_tier(name).configured_path for name in ("audit", "source", "embeddings", "ops", "user") + ) + if any(any(character in str(path) for character in ("%", "?", "#")) for path in paths): + raise LiveProofError("live-proof archive paths cannot contain SQLite URI query characters") + + +def archive_owned_storage_roots(archive_root: Path) -> tuple[Path, ...]: + """Return every resolved root whose files belong to archive lifecycle state.""" + + from polylogue.storage.archive_identity import ArchiveLocation + + try: + location = ArchiveLocation.resolve(archive_root) + except RuntimeError as exc: + raise LiveProofError("archive-owned storage roots are unavailable") from exc + bases = {location.configured_root.resolve(), location.active_index_path.parent.resolve()} + if location.active_pointer is not None: + bases.add(location.active_pointer.parent.resolve()) + roots = set(bases) + for base in bases: + roots.add((base / ".index-generations").resolve()) + roots.add((base / ".index-rebuild-transactions").resolve()) + return tuple(sorted(roots)) + + +def _sqlite_file_state(path: Path, *, allow_symlink: bool) -> JSONDocument: + try: + metadata = path.lstat() + except OSError as exc: + raise LiveProofError("live-proof SQLite binding is unavailable") from exc + if path.is_symlink(): + if not allow_symlink: + raise LiveProofError("live-proof SQLite binding is not a regular file") + path = path.resolve(strict=True) + metadata = path.stat() + if not path.is_file(): + raise LiveProofError("live-proof SQLite binding is not a regular file") + files: dict[str, JSONValue] = { + "database": { + "size": metadata.st_size, + "sha256": hash_file(path), + } + } + for suffix in _SQLITE_SIDECARS: + sidecar = path.with_name(path.name + suffix) + if not sidecar.exists() and not sidecar.is_symlink(): + files[suffix] = {"exists": False} + continue + try: + sidecar_metadata = sidecar.lstat() + except OSError as exc: + raise LiveProofError("live-proof SQLite sidecar is unavailable") from exc + if sidecar.is_symlink() or not sidecar.is_file(): + raise LiveProofError("live-proof SQLite sidecar is not a regular file") + # SQLite may create an empty WAL for a read-only connection. It holds + # no logical pages and is equivalent to an absent WAL, so normalize it + # before comparing the pre/post quiescence snapshots. + if sidecar_metadata.st_size == 0: + files[suffix] = {"exists": False} + continue + files[suffix] = { + "exists": True, + "size": sidecar_metadata.st_size, + "sha256": hash_file(sidecar), + } + return files + + +def _sqlite_file_set_digest(path: Path, *, allow_symlink: bool = False) -> str: + """Bind a stable SQLite database plus WAL/journal sidecars as one file set.""" + + before = _sqlite_file_state(path, allow_symlink=allow_symlink) + try: + with closing(_open_readonly(path)) as connection: + connection.execute("PRAGMA query_only = ON") + connection.execute("BEGIN") + connection.execute("PRAGMA schema_version").fetchone() + except (OSError, sqlite3.Error) as exc: + raise LiveProofError("live-proof SQLite binding is unavailable") from exc + after = _sqlite_file_state(path, allow_symlink=allow_symlink) + if after != before: + raise LiveProofError("live-proof SQLite file set changed while capturing binding") + return hash_payload(before) + + +def _schema_version(path: Path) -> int: + try: + with closing(_open_readonly(path)) as connection: + row = connection.execute("PRAGMA user_version").fetchone() + except (OSError, sqlite3.Error) as exc: + raise LiveProofError("live-proof schema binding is unavailable") from exc + return int(row[0]) if row is not None else 0 + + +def _active_tier_paths(location: object) -> tuple[tuple[str, Path], ...]: + from polylogue.storage.archive_identity import ArchiveLocation + + assert isinstance(location, ArchiveLocation) + return tuple( + (name, location.active_index_path if name == "index" else location.active_tier(name).configured_path) + for name in _ARCHIVE_TIER_NAMES + ) + + +def _schema_versions(location: object) -> tuple[tuple[str, int], ...]: + return tuple((name, _schema_version(path)) for name, path in _active_tier_paths(location)) + + +def _candidate_index(location: object, generation_id: str, *, source_snapshot: str) -> Path: + """Resolve one inactive generation through the lifecycle store's canonical root.""" + + from polylogue.storage.archive_identity import ArchiveLocation, TierFileIdentity + from polylogue.storage.index_generation import IndexGenerationStore + + assert isinstance(location, ArchiveLocation) + if not _GENERATION_ID_RE.fullmatch(generation_id): + raise LiveProofError("candidate generation id is invalid") + # The store bootstraps an absent pointer, which would violate this module's + # read-only contract. Real generations can only exist after that lifecycle + # anchor exists, so fail closed instead of asking the store to create it. + if location.active_pointer is None: + raise LiveProofError("candidate generation requires an archive lifecycle pointer") + if ".index-generations" in location.active_pointer.parts: + raise LiveProofError("candidate generation requires a canonical active-index pointer") + store = IndexGenerationStore(location) + try: + generation = store.load(generation_id) + index_path = Path(generation.index_path) + expected_root = store.generations_root / generation_id + index_resolved = index_path.resolve(strict=True) + expected_resolved = expected_root.resolve(strict=True) / "index.db" + transactions = [] + for transaction_path in store.transactions_root.glob("*.json"): + try: + transaction = store.load_transaction(transaction_path.stem) + except (OSError, RuntimeError, TypeError, ValueError, json.JSONDecodeError): + continue + if transaction.generation_id == generation_id: + transactions.append(transaction) + except (OSError, RuntimeError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise LiveProofError("candidate generation metadata is unavailable") from exc + if ( + generation.generation_id != generation_id + or generation.state != "inactive" + or Path(generation.archive_root).resolve() != location.configured_root.resolve() + or generation.source_snapshot != source_snapshot + or len(transactions) != 1 + or transactions[0].generation_owner_id != generation.owner_id + or transactions[0].source_snapshot != source_snapshot + or transactions[0].status != "ready" + or index_path != expected_root / "index.db" + or index_path.is_symlink() + or not index_path.is_file() + or index_resolved != expected_resolved + or location.active_index.same_file(TierFileIdentity.resolve("index", index_path)) + ): + raise LiveProofError("candidate generation binding is stale or invalid") + return index_path + + +def capture_live_proof_bindings(archive_root: Path, *, candidate_generation_id: str | None = None) -> LiveProofBindings: + """Capture one coherent, read-only binding snapshot for a proof receipt.""" + + from polylogue.maintenance.schema_inference_gate import rebuild_source_revision_snapshot + from polylogue.sources.origin_specs import lowering_fingerprint, parser_fingerprint_for_origin + from polylogue.storage.archive_identity import ArchiveIdentity, ArchiveLocation + + root = Path(archive_root).expanduser().resolve() + try: + location = ArchiveLocation.resolve(root) + _require_uri_safe_location(location) + source_snapshot = rebuild_source_revision_snapshot(root) + with closing(_open_readonly(location.configured_tier("source").configured_path)) as source: + origins = sorted(str(row[0]) for row in source.execute("SELECT DISTINCT origin FROM raw_sessions")) + candidate_index = ( + _candidate_index(location, candidate_generation_id, source_snapshot=source_snapshot) + if candidate_generation_id is not None + else None + ) + identity = ArchiveIdentity.resolve_location(location) + active_index_sha256 = _sqlite_file_set_digest(location.active_index_path, allow_symlink=True) + candidate_index_sha256 = _sqlite_file_set_digest(candidate_index) if candidate_index is not None else None + schema_versions = _schema_versions(location) + tier_file_set_digests = tuple( + (name, _sqlite_file_set_digest(path, allow_symlink=name == "index")) + for name, path in _active_tier_paths(location) + ) + candidate_index_schema_version = _schema_version(candidate_index) if candidate_index is not None else None + parser_fingerprints = tuple((origin, parser_fingerprint_for_origin(origin)) for origin in origins) + lowering = lowering_fingerprint() + except LiveProofError: + raise + except (OSError, RuntimeError, sqlite3.Error, ValueError) as exc: + raise LiveProofError("live-proof archive binding is unavailable") from exc + private_paths: list[tuple[str, PrivatePathReference]] = [ + ("archive_root", PrivatePathReference.capture(root)), + ("active_index", PrivatePathReference.capture(location.active_index_path)), + ] + if candidate_index is not None: + private_paths.append(("candidate_index", PrivatePathReference.capture(candidate_index))) + return LiveProofBindings( + code_sha=_code_sha(), + archive_identity_digest=identity.authority_identity_digest, + source_snapshot=source_snapshot, + schema_versions=schema_versions, + tier_file_set_digests=tier_file_set_digests, + candidate_index_schema_version=candidate_index_schema_version, + parser_fingerprints=parser_fingerprints, + lowering_fingerprint=lowering, + active_index_sha256=active_index_sha256, + candidate_generation_id=candidate_generation_id, + candidate_index_sha256=candidate_index_sha256, + private_paths=tuple(private_paths), + ) + + +def _redacted_report(report: object, *, roots: Sequence[Path]) -> JSONDocument: + if not hasattr(report, "to_json"): + raise LiveProofError("archive verification report is unavailable") + document = cast(JSONDocument, report.to_json()) + if not isinstance(document, Mapping): + raise LiveProofError("archive verification report is malformed") + replacements = {str(path.resolve()): f"[private-path:{hash_text(str(path.resolve()))}]" for path in roots} + + def redact(value: object) -> JSONValue: + if isinstance(value, Mapping): + return {str(key): redact(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [redact(item) for item in value] + if isinstance(value, str): + for raw, replacement in sorted(replacements.items(), key=lambda item: len(item[0]), reverse=True): + value = value.replace(raw, replacement) + if _ABSOLUTE_PATH_RE.search(value): + return f"[private-text:{hash_text(value)}]" + return value + if value is None or isinstance(value, bool | int | float): + return value + raise LiveProofError("archive verification emitted non-JSON evidence") + + evidence = {key: value for key, value in document.items() if key not in {"archive_root", "generated_at"}} + redacted = redact(evidence) + if not isinstance(redacted, dict): + raise LiveProofError("archive verification emitted malformed evidence") + return redacted + + +def _archive_verification_result( + spec: LiveProofSpec, + archive_root: Path, + *, + candidate_generation_id: str | None, + bindings: LiveProofBindings, +) -> tuple[JSONDocument, tuple[LiveProofResidue, ...]]: + from polylogue.maintenance.archive_verification import verify_archive + from polylogue.storage.archive_identity import ArchiveLocation + + location = ArchiveLocation.resolve(archive_root) + archive_paths = (archive_root, location.active_index_path) + candidate_index: Path | None = None + candidate_root: Path | None = None + if candidate_generation_id is not None: + candidate_index = _candidate_index(location, candidate_generation_id, source_snapshot=bindings.source_snapshot) + candidate_root = candidate_index.parent + profiles: dict[str, JSONValue] = {} + residues: list[LiveProofResidue] = [] + for profile in spec.archive_profiles: + if profile.target == "active": + report = verify_archive(archive_root, checks=profile.checks) + roots: tuple[Path, ...] = archive_paths + elif profile.target == "candidate_index": + assert candidate_root is not None + report = verify_archive(candidate_root, checks=profile.checks) + roots = (*archive_paths, candidate_root) + else: + assert candidate_index is not None + report = verify_archive(archive_root, checks=profile.checks, index_path_override=candidate_index) + roots = (*archive_paths, candidate_index.parent) + profiles[profile.name] = _redacted_report(report, roots=roots) + for check in report.checks: + if check.status.value != "ok": + residues.append(LiveProofResidue(LiveProofResidueKind.CHECK_FAILED, f"{profile.name}:{check.name}")) + status = LiveProofStatus.PASSED if not residues else LiveProofStatus.FAILED + return cast(JSONDocument, {"status": status.value, "archive_verification": {"profiles": profiles}}), tuple(residues) + + +def _validate_private_path_references(value: object) -> None: + if not isinstance(value, Mapping): + raise LiveProofError("input receipt private paths are malformed") + for reference in value.values(): + if not isinstance(reference, Mapping): + raise LiveProofError("input receipt private paths are malformed") + basename = reference.get("basename") + digest = reference.get("sha256") + if ( + not isinstance(basename, str) + or Path(basename).name != basename + or not isinstance(digest, str) + or not _SHA256_RE.fullmatch(digest) + ): + raise LiveProofError("input receipt private paths are malformed") + + +def _parse_bindings_document(value: object) -> LiveProofBindings: + if not isinstance(value, Mapping): + raise LiveProofError("live-proof receipt bindings are malformed") + + def string_field(name: str) -> str: + field = value.get(name) + if not isinstance(field, str) or not field: + raise LiveProofError("live-proof receipt bindings are malformed") + return field + + def mapping_field(name: str) -> Mapping[object, object]: + field = value.get(name) + if not isinstance(field, Mapping): + raise LiveProofError("live-proof receipt bindings are malformed") + return field + + def digest_mapping(name: str) -> tuple[tuple[str, str], ...]: + field = mapping_field(name) + if set(field) != set(_ARCHIVE_TIER_NAMES) and name == "tier_file_set_digests": + raise LiveProofError("live-proof receipt bindings are malformed") + result: list[tuple[str, str]] = [] + keys: tuple[str, ...] + if name == "tier_file_set_digests": + keys = _ARCHIVE_TIER_NAMES + else: + if any(not isinstance(key, str) for key in field): + raise LiveProofError("live-proof receipt bindings are malformed") + keys = tuple(sorted(cast(str, key) for key in field)) + for key in keys: + digest = field.get(key) + if not isinstance(digest, str) or not _SHA256_RE.fullmatch(digest): + raise LiveProofError("live-proof receipt bindings are malformed") + result.append((key, digest)) + return tuple(result) + + schema_mapping = mapping_field("schema_versions") + if set(schema_mapping) != set(_ARCHIVE_TIER_NAMES): + raise LiveProofError("live-proof receipt bindings are malformed") + schema_versions: list[tuple[str, int]] = [] + for name in _ARCHIVE_TIER_NAMES: + version = schema_mapping.get(name) + if isinstance(version, bool) or not isinstance(version, int): + raise LiveProofError("live-proof receipt bindings are malformed") + schema_versions.append((name, version)) + + parser_mapping = mapping_field("parser_fingerprints") + parser_items: list[tuple[str, str]] = [] + for raw_name, raw_fingerprint in parser_mapping.items(): + if not isinstance(raw_name, str) or not isinstance(raw_fingerprint, str) or not raw_fingerprint: + raise LiveProofError("live-proof receipt bindings are malformed") + parser_items.append((raw_name, raw_fingerprint)) + parser_fingerprints = tuple(sorted(parser_items)) + if len(parser_fingerprints) != len(parser_mapping): + raise LiveProofError("live-proof receipt bindings are malformed") + + candidate_generation_id = value.get("candidate_generation_id") + if candidate_generation_id is not None and ( + not isinstance(candidate_generation_id, str) or not _GENERATION_ID_RE.fullmatch(candidate_generation_id) + ): + raise LiveProofError("live-proof receipt bindings are malformed") + candidate_schema = value.get("candidate_index_schema_version") + if candidate_schema is not None and (isinstance(candidate_schema, bool) or not isinstance(candidate_schema, int)): + raise LiveProofError("live-proof receipt bindings are malformed") + candidate_digest = value.get("candidate_index_sha256") + if candidate_digest is not None and ( + not isinstance(candidate_digest, str) or not _SHA256_RE.fullmatch(candidate_digest) + ): + raise LiveProofError("live-proof receipt bindings are malformed") + + private_paths = mapping_field("private_paths") + parsed_private_paths: list[tuple[str, PrivatePathReference]] = [] + for raw_name, raw_reference in private_paths.items(): + if not isinstance(raw_name, str) or not isinstance(raw_reference, Mapping): + raise LiveProofError("live-proof receipt bindings are malformed") + name = raw_name + reference = raw_reference + basename = reference.get("basename") + digest = reference.get("sha256") + if ( + not isinstance(basename, str) + or Path(basename).name != basename + or not isinstance(digest, str) + or not _SHA256_RE.fullmatch(digest) + ): + raise LiveProofError("live-proof receipt bindings are malformed") + parsed_private_paths.append((name, PrivatePathReference(basename, digest))) + + active_digest = string_field("active_index_sha256") + if not _SHA256_RE.fullmatch(active_digest): + raise LiveProofError("live-proof receipt bindings are malformed") + lowering = string_field("lowering_fingerprint") + return LiveProofBindings( + code_sha=string_field("code_sha"), + archive_identity_digest=string_field("archive_identity_digest"), + source_snapshot=string_field("source_snapshot"), + schema_versions=tuple(schema_versions), + tier_file_set_digests=digest_mapping("tier_file_set_digests"), + candidate_index_schema_version=candidate_schema, + parser_fingerprints=parser_fingerprints, + lowering_fingerprint=lowering, + active_index_sha256=active_digest, + candidate_generation_id=candidate_generation_id, + candidate_index_sha256=candidate_digest, + private_paths=tuple(parsed_private_paths), + ) + + +def _apply_proof_status(status: str) -> LiveProofStatus: + if status in {"applied", "already_satisfied"}: + return LiveProofStatus.PASSED + if status == "not_applicable": + return LiveProofStatus.NOT_APPLICABLE + if status == "blocked": + return LiveProofStatus.BLOCKED + if status == "failed": + return LiveProofStatus.FAILED + raise LiveProofError("existing apply receipt result status is invalid") + + +def _validate_existing_apply_document(document: object, bindings: LiveProofBindings) -> tuple[JSONDocument, str]: + if not is_json_document(document): + raise LiveProofError("existing apply receipt is malformed") + payload = dict(document) + digest = payload.pop("receipt_sha256", None) + if not isinstance(digest, str) or not _SHA256_RE.fullmatch(digest) or hash_payload(payload) != digest: + raise LiveProofError("existing apply receipt self-hash is invalid") + if payload.get("receipt_schema") != EXISTING_APPLY_RECEIPT_SCHEMA: + raise LiveProofError("existing apply receipt schema is not accepted") + if payload.get("operation_id") not in _REGISTERED_APPLY_OPERATION_IDS: + raise LiveProofError("existing apply receipt operation binding is invalid") + receipt_bindings = payload.get("bindings") + if not isinstance(receipt_bindings, Mapping) or receipt_bindings != bindings.to_document(): + raise LiveProofError("existing apply receipt bindings are stale or mismatched") + result = payload.get("result") + if not isinstance(result, Mapping) or not is_json_document(result): + raise LiveProofError("existing apply receipt result is malformed") + status = result.get("status") + if not isinstance(status, str) or status not in _APPLY_RESULT_STATUSES: + raise LiveProofError("existing apply receipt result status is not recognized") + _validate_private_path_references(receipt_bindings.get("private_paths")) + return require_json_document(document, context="existing apply receipt"), digest + + +def _validated_existing_apply_receipt(path: Path, bindings: LiveProofBindings) -> tuple[JSONDocument, str]: + try: + document = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise LiveProofError("existing apply receipt is unavailable") from exc + return _validate_existing_apply_document(document, bindings) + + +def _apply_residues(status: LiveProofStatus) -> tuple[LiveProofResidue, ...]: + if status is LiveProofStatus.NOT_APPLICABLE: + return (LiveProofResidue(LiveProofResidueKind.NOT_APPLICABLE, "apply-result-not-applicable"),) + if status is LiveProofStatus.BLOCKED: + return (LiveProofResidue(LiveProofResidueKind.BLOCKED, "apply-result-blocked"),) + if status is LiveProofStatus.FAILED: + return (LiveProofResidue(LiveProofResidueKind.CHECK_FAILED, "apply-result-failed"),) + return () + + +def collect_live_proof( + proof_id: str, + archive_root: Path, + *, + candidate_generation_id: str | None = None, + apply_receipt_path: Path | None = None, +) -> LiveProofReceipt: + """Collect exactly one fixed proof route without operating on the archive.""" + + validate_live_proof_registry() + resolved_archive_root = Path(archive_root).expanduser().resolve() + spec = live_proof_spec(proof_id) + if spec.mode is LiveProofMode.CANDIDATE: + if candidate_generation_id is None or apply_receipt_path is not None: + raise LiveProofError("candidate proof requires only an inactive candidate generation id") + elif spec.mode is LiveProofMode.EXISTING_APPLY_RECEIPT: + if apply_receipt_path is None or candidate_generation_id is not None: + raise LiveProofError("existing-apply proof requires only an existing apply receipt") + elif candidate_generation_id is not None or apply_receipt_path is not None: + raise LiveProofError("read-only proof accepts no candidate or apply receipt input") + + bindings = capture_live_proof_bindings(resolved_archive_root, candidate_generation_id=candidate_generation_id) + if spec.producer == "archive_verification": + result, residues = _archive_verification_result( + spec, + resolved_archive_root, + candidate_generation_id=candidate_generation_id, + bindings=bindings, + ) + input_digests: tuple[str, ...] = () + else: + assert apply_receipt_path is not None + apply_receipt, digest = _validated_existing_apply_receipt(apply_receipt_path, bindings) + apply_result = require_json_document(apply_receipt["result"], context="existing apply receipt result") + proof_status = _apply_proof_status(cast(str, apply_result["status"])) + result = {"status": proof_status.value, "apply_receipt": apply_receipt} + input_digests = (digest,) + residues = _apply_residues(proof_status) + return LiveProofReceipt( + proof_id=spec.proof_id, + bead_id=spec.bead_id, + mode=spec.mode, + registry_version=LIVE_PROOF_REGISTRY_VERSION, + generated_at_ms=int(time.time() * 1000), + bindings=bindings, + result=result, + residues=residues, + input_receipt_digests=input_digests, + ) + + +def _parse_residues(value: object) -> tuple[LiveProofResidue, ...]: + if not isinstance(value, list): + raise LiveProofError("live-proof receipt residues are malformed") + parsed: list[LiveProofResidue] = [] + seen: set[tuple[LiveProofResidueKind, str]] = set() + for residue in value: + if not isinstance(residue, Mapping): + raise LiveProofError("live-proof receipt residues are malformed") + kind = residue.get("kind") + code = residue.get("code") + if not isinstance(kind, str) or not isinstance(code, str) or not code: + raise LiveProofError("live-proof receipt residues are malformed") + try: + parsed_residue = LiveProofResidue(LiveProofResidueKind(kind), code) + except (KeyError, TypeError, ValueError) as exc: + raise LiveProofError("live-proof receipt residues are malformed") from exc + residue_key = (parsed_residue.kind, parsed_residue.code) + if not _RESIDUE_CODE_RE.fullmatch(parsed_residue.code) or residue_key in seen: + raise LiveProofError("live-proof receipt residues are malformed") + seen.add(residue_key) + parsed.append(parsed_residue) + return tuple(parsed) + + +def _validate_status_residues(status: LiveProofStatus, residues: tuple[LiveProofResidue, ...]) -> None: + expected = { + LiveProofStatus.PASSED: frozenset(), + LiveProofStatus.FAILED: frozenset({LiveProofResidueKind.CHECK_FAILED, LiveProofResidueKind.UNVERIFIED}), + LiveProofStatus.BLOCKED: frozenset({LiveProofResidueKind.BLOCKED}), + LiveProofStatus.NOT_APPLICABLE: frozenset({LiveProofResidueKind.NOT_APPLICABLE}), + }[status] + if (not residues and status is not LiveProofStatus.PASSED) or any( + residue.kind not in expected for residue in residues + ): + raise LiveProofError("live-proof receipt status and residues are inconsistent") + if status is LiveProofStatus.PASSED and residues: + raise LiveProofError("live-proof receipt status and residues are inconsistent") + + +def _validate_archive_result( + spec: LiveProofSpec, result: JSONDocument +) -> tuple[LiveProofStatus, tuple[LiveProofResidue, ...]]: + check_keys = frozenset( + {"name", "status", "summary", "count", "details", "breakdown", "evidence", "check_class", "waived_bead_id"} + ) + + def validate_check(check: object) -> tuple[str, str]: + if not isinstance(check, Mapping) or frozenset(check) != check_keys: + raise LiveProofError("live-proof receipt archive verification evidence is malformed") + name = check.get("name") + status = check.get("status") + summary = check.get("summary") + count = check.get("count") + details = check.get("details") + breakdown = check.get("breakdown") + evidence = check.get("evidence") + check_class = check.get("check_class") + waived_bead_id = check.get("waived_bead_id") + if ( + not isinstance(name, str) + or not isinstance(status, str) + or status not in {"ok", "warning", "error", "skip"} + or not isinstance(summary, str) + or isinstance(count, bool) + or not isinstance(count, int) + or not isinstance(details, list) + or any(not isinstance(detail, str) for detail in details) + or not isinstance(breakdown, Mapping) + or any( + not isinstance(key, str) or isinstance(value, bool) or not isinstance(value, int) + for key, value in breakdown.items() + ) + or not is_json_document(evidence) + or not isinstance(check_class, str) + or (waived_bead_id is not None and not isinstance(waived_bead_id, str)) + ): + raise LiveProofError("live-proof receipt archive verification evidence is malformed") + return name, status + + archive_verification = result.get("archive_verification") + if not isinstance(archive_verification, Mapping): + raise LiveProofError("live-proof receipt archive verification evidence is malformed") + profiles = archive_verification.get("profiles") + if not isinstance(profiles, Mapping) or set(profiles) != {profile.name for profile in spec.archive_profiles}: + raise LiveProofError("live-proof receipt archive verification evidence is incomplete") + residues: list[LiveProofResidue] = [] + for profile in spec.archive_profiles: + evidence = profiles.get(profile.name) + if not isinstance(evidence, Mapping): + raise LiveProofError("live-proof receipt archive verification evidence is malformed") + summary = evidence.get("summary") + blocking = evidence.get("blocking") + checks = evidence.get("checks") + if ( + frozenset(evidence) != frozenset({"summary", "blocking", "checks"}) + or not isinstance(summary, Mapping) + or frozenset(summary) != frozenset({"ok", "warning", "error", "skip"}) + or any(isinstance(value, bool) or not isinstance(value, int) for value in summary.values()) + or not isinstance(blocking, bool) + or not isinstance(checks, list) + ): + raise LiveProofError("live-proof receipt archive verification evidence is malformed") + names: list[str] = [] + statuses: list[str] = [] + for check in checks: + name, status = validate_check(check) + names.append(name) + statuses.append(status) + if tuple(names) != profile.checks: + raise LiveProofError("live-proof receipt archive verification profile is not canonical") + residues.extend( + LiveProofResidue(LiveProofResidueKind.CHECK_FAILED, f"{profile.name}:{name}") + for name, status in zip(names, statuses, strict=True) + if status != "ok" + ) + return (LiveProofStatus.FAILED, tuple(residues)) if residues else (LiveProofStatus.PASSED, ()) + + +def _validate_route_result( + spec: LiveProofSpec, result: JSONDocument +) -> tuple[LiveProofStatus, tuple[LiveProofResidue, ...]]: + status = result.get("status") + try: + parsed_status = LiveProofStatus(cast(str, status)) + except (TypeError, ValueError) as exc: + raise LiveProofError("live-proof receipt result status is malformed") from exc + if spec.producer == "archive_verification": + expected, residues = _validate_archive_result(spec, result) + if parsed_status is not expected: + raise LiveProofError("live-proof receipt archive verification status is inconsistent") + else: + apply_receipt = result.get("apply_receipt") + if not isinstance(apply_receipt, Mapping) or not is_json_document(apply_receipt): + raise LiveProofError("live-proof receipt apply evidence is malformed") + apply_result = apply_receipt.get("result") + if not isinstance(apply_result, Mapping) or not is_json_document(apply_result): + raise LiveProofError("live-proof receipt apply evidence is malformed") + raw_status = apply_result.get("status") + if not isinstance(raw_status, str) or raw_status not in _APPLY_RESULT_STATUSES: + raise LiveProofError("live-proof receipt apply evidence is malformed") + expected = _apply_proof_status(raw_status) + if parsed_status is not expected: + raise LiveProofError("live-proof receipt apply status is inconsistent") + residues = _apply_residues(expected) + return parsed_status, residues + + +def _capture_expected_bindings(archive_root: Path, candidate_generation_id: str | None) -> LiveProofBindings: + try: + return capture_live_proof_bindings(archive_root, candidate_generation_id=candidate_generation_id) + except LiveProofError as exc: + raise LiveProofError("live-proof receipt bindings are stale or mismatched") from exc + + +def _capture_promoted_candidate_bindings( + document: object, archive_root: Path, candidate_generation_id: str +) -> LiveProofBindings: + """Validate a candidate receipt against a promoted or retained generation. + + The receipt's active-index fields intentionally describe the pre-promotion + state. Promotion changes that state, so validation instead rechecks the + retained candidate content and the durable source-side bindings while + preserving the historical receipt document as the expected binding. + """ + + from polylogue.maintenance.schema_inference_gate import rebuild_source_revision_snapshot + from polylogue.storage.archive_identity import ArchiveLocation + from polylogue.storage.index_generation import IndexGenerationStore + + if not isinstance(document, Mapping): + raise LiveProofError("candidate proof receipt is malformed") + recorded = _parse_bindings_document(document.get("bindings")) + if recorded.candidate_generation_id != candidate_generation_id: + raise LiveProofError("candidate proof receipt targets a different generation") + if recorded.candidate_index_sha256 is None or recorded.candidate_index_schema_version is None: + raise LiveProofError("candidate proof receipt has incomplete candidate binding") + + try: + location = ArchiveLocation.resolve(archive_root) + store = IndexGenerationStore(location) + generation = store.load(candidate_generation_id) + candidate_path = Path(generation.index_path) + expected_path = store.generations_root / candidate_generation_id / "index.db" + generation_checks = { + "state": generation.state in {"active", "retained"}, + "archive_root": Path(generation.archive_root).resolve() == location.configured_root.resolve(), + "candidate_path": candidate_path == expected_path, + "regular_file": not candidate_path.is_symlink() and candidate_path.is_file(), + "resolved_path": candidate_path.resolve(strict=True) == expected_path.resolve(strict=True), + "source_snapshot": rebuild_source_revision_snapshot(archive_root) == generation.source_snapshot, + } + if not all(generation_checks.values()): + raise LiveProofError("promoted candidate generation binding is stale or invalid") + current = capture_live_proof_bindings(archive_root) + current_schema = dict(current.schema_versions) + recorded_schema = dict(recorded.schema_versions) + current_tiers = dict(current.tier_file_set_digests) + recorded_tiers = dict(recorded.tier_file_set_digests) + binding_checks = { + "code_sha": recorded.code_sha == current.code_sha, + "source_snapshot": recorded.source_snapshot == current.source_snapshot, + "parser_fingerprints": recorded.parser_fingerprints == current.parser_fingerprints, + "lowering": recorded.lowering_fingerprint == current.lowering_fingerprint, + "schema_versions": all( + recorded_schema[name] == current_schema[name] for name in _ARCHIVE_TIER_NAMES if name != "index" + ), + "tier_digests": all( + recorded_tiers[name] == current_tiers[name] for name in _ARCHIVE_TIER_NAMES if name != "index" + ), + "candidate_schema": recorded_schema["index"] == recorded.candidate_index_schema_version, + "candidate_schema_current": _schema_version(candidate_path) == recorded.candidate_index_schema_version, + "candidate_digest": _sqlite_file_set_digest(candidate_path) == recorded.candidate_index_sha256, + "archive_private_path": dict(recorded.private_paths).get("archive_root") + == dict(current.private_paths).get("archive_root"), + "candidate_private_path": dict(recorded.private_paths).get("candidate_index") + == PrivatePathReference.capture(candidate_path), + } + if not all(binding_checks.values()): + raise LiveProofError("promoted candidate generation binding is stale or invalid") + except LiveProofError: + raise + except (OSError, RuntimeError, sqlite3.Error, ValueError, json.JSONDecodeError) as exc: + raise LiveProofError("promoted candidate generation binding is unavailable") from exc + return recorded + + +def _validate_live_proof_receipt( + document: object, + archive_root: Path, + *, + candidate_generation_id: str | None = None, + expected_bindings: LiveProofBindings | None = None, +) -> LiveProofReceipt: + if not is_json_document(document): + raise LiveProofError("live-proof receipt is malformed") + payload = dict(document) + digest = payload.pop("receipt_sha256", None) + if not isinstance(digest, str) or not _SHA256_RE.fullmatch(digest) or hash_payload(payload) != digest: + raise LiveProofError("live-proof receipt self-hash is invalid") + try: + proof_id = LiveProofId(cast(str, payload["proof_id"])) + mode = LiveProofMode(cast(str, payload["mode"])) + except (KeyError, TypeError, ValueError) as exc: + raise LiveProofError("live-proof receipt route is invalid") from exc + spec = live_proof_spec(proof_id.value) + if payload.get("receipt_schema") != LIVE_PROOF_RECEIPT_SCHEMA or payload.get("bead_id") != spec.bead_id: + raise LiveProofError("live-proof receipt protocol identity is invalid") + if payload.get("registry_version") != LIVE_PROOF_REGISTRY_VERSION or mode is not spec.mode: + raise LiveProofError("live-proof receipt registry binding is stale") + generated_at_ms = payload.get("generated_at_ms") + now_ms = int(time.time() * 1000) + if ( + isinstance(generated_at_ms, bool) + or not isinstance(generated_at_ms, int) + or generated_at_ms > now_ms + 5_000 + or now_ms - generated_at_ms > _LIVE_PROOF_MAX_AGE_MS + ): + raise LiveProofError("live-proof receipt is stale or has an invalid generation time") + receipt_bindings = payload.get("bindings") + if not isinstance(receipt_bindings, Mapping): + raise LiveProofError("live-proof receipt bindings are malformed") + if mode is LiveProofMode.CANDIDATE: + recorded_candidate = receipt_bindings.get("candidate_generation_id") + if not isinstance(recorded_candidate, str): + raise LiveProofError("candidate proof receipt has no candidate binding") + if candidate_generation_id is None: + candidate_generation_id = recorded_candidate + elif candidate_generation_id != recorded_candidate: + raise LiveProofError("candidate proof receipt targets a different generation") + elif candidate_generation_id is not None: + raise LiveProofError("candidate validation requires a candidate proof receipt") + expected = expected_bindings or _capture_expected_bindings(archive_root, candidate_generation_id) + if receipt_bindings != expected.to_document(): + raise LiveProofError("live-proof receipt bindings are stale or mismatched") + result = payload.get("result") + input_digests = payload.get("input_receipt_digests") + if not is_json_document(result) or not isinstance(input_digests, list): + raise LiveProofError("live-proof receipt evidence is malformed") + residues = _parse_residues(payload.get("residues")) + status, expected_residues = _validate_route_result(spec, result) + if residues != expected_residues: + raise LiveProofError("live-proof receipt status and residues are inconsistent") + _validate_status_residues(status, residues) + if any(not isinstance(value, str) or not _SHA256_RE.fullmatch(value) for value in input_digests): + raise LiveProofError("live-proof receipt input digests are malformed") + if spec.mode is LiveProofMode.EXISTING_APPLY_RECEIPT: + apply_receipt = result.get("apply_receipt") + _apply_result, embedded_digest = _validate_existing_apply_document(apply_receipt, expected) + if input_digests != [embedded_digest]: + raise LiveProofError("live-proof receipt apply evidence is not bound to its input digest") + if spec.mode is not LiveProofMode.EXISTING_APPLY_RECEIPT and input_digests: + raise LiveProofError("live-proof receipt input digests are malformed") + return LiveProofReceipt( + proof_id=proof_id, + bead_id=spec.bead_id, + mode=mode, + registry_version=LIVE_PROOF_REGISTRY_VERSION, + generated_at_ms=generated_at_ms, + bindings=expected, + result=result, + residues=residues, + input_receipt_digests=tuple(cast(list[str], input_digests)), + ) + + +def validate_live_proof_receipt( + document: object, + archive_root: Path, + *, + candidate_generation_id: str | None = None, +) -> LiveProofReceipt: + """Validate a self-hashed proof receipt against current archive bindings.""" + + return _validate_live_proof_receipt(document, archive_root, candidate_generation_id=candidate_generation_id) + + +def _require_acceptable_result(receipt: LiveProofReceipt) -> None: + status, expected_residues = _validate_route_result(live_proof_spec(receipt.proof_id.value), receipt.result) + if receipt.residues != expected_residues: + raise LiveProofError("live-proof receipt status and residues are inconsistent") + _validate_status_residues(status, receipt.residues) + if status is LiveProofStatus.PASSED: + return + if status is LiveProofStatus.NOT_APPLICABLE: + return + raise LiveProofError("live-proof receipt result is not acceptable to an aggregate") + + +def _candidate_id_from_documents(receipts: Sequence[object]) -> str | None: + candidates: set[str] = set() + for receipt in receipts: + if not isinstance(receipt, Mapping) or receipt.get("mode") != LiveProofMode.CANDIDATE.value: + continue + bindings = receipt.get("bindings") + if not isinstance(bindings, Mapping) or not isinstance(bindings.get("candidate_generation_id"), str): + raise LiveProofError("candidate proof receipt has no candidate binding") + candidates.add(cast(str, bindings["candidate_generation_id"])) + if len(candidates) > 1: + raise LiveProofError("proof aggregate targets multiple candidate generations") + return next(iter(candidates), None) + + +def _active_bindings(bindings: LiveProofBindings) -> LiveProofBindings: + return replace( + bindings, + candidate_generation_id=None, + candidate_index_sha256=None, + candidate_index_schema_version=None, + private_paths=tuple((name, path) for name, path in bindings.private_paths if name != "candidate_index"), + ) + + +def _validate_aggregate(receipts: Sequence[object], archive_root: Path) -> tuple[LiveProofReceipt, ...]: + if not receipts: + raise LiveProofError("live-operation aggregate requires at least one proof receipt") + candidate_id = _candidate_id_from_documents(receipts) + candidate_document = ( + next( + receipt + for receipt in receipts + if isinstance(receipt, Mapping) and receipt.get("mode") == LiveProofMode.CANDIDATE.value + ) + if candidate_id is not None + else None + ) + promoted_candidate = False + if candidate_id is not None: + try: + candidate_bindings = _capture_expected_bindings(archive_root, candidate_id) + except LiveProofError: + candidate_bindings = _capture_promoted_candidate_bindings(candidate_document, archive_root, candidate_id) + promoted_candidate = True + else: + candidate_bindings = None + active_bindings = ( + _capture_expected_bindings(archive_root, None) + if promoted_candidate + else _active_bindings(candidate_bindings) + if candidate_bindings is not None + else _capture_expected_bindings(archive_root, None) + ) + validated: list[LiveProofReceipt] = [] + for receipt in receipts: + mode = receipt.get("mode") if isinstance(receipt, Mapping) else None + expected = candidate_bindings if mode == LiveProofMode.CANDIDATE.value else active_bindings + validated.append(_validate_live_proof_receipt(receipt, archive_root, expected_bindings=expected)) + return tuple(validated) + + +def validate_live_operation_aggregate(receipts: Sequence[object], archive_root: Path) -> tuple[LiveProofReceipt, ...]: + """Consumer seam for a validated aggregate, without scheduling work.""" + + validated = _validate_aggregate(receipts, archive_root) + for receipt in validated: + _require_acceptable_result(receipt) + return validated + + +def validate_candidate_proof_receipts( + receipts: Sequence[object], archive_root: Path, *, candidate_generation_id: str +) -> tuple[LiveProofReceipt, ...]: + """Accept only the canonical candidate route for one inactive generation.""" + + candidate_bindings = _capture_expected_bindings(archive_root, candidate_generation_id) + validated = tuple( + _validate_live_proof_receipt( + receipt, + archive_root, + candidate_generation_id=candidate_generation_id, + expected_bindings=candidate_bindings, + ) + for receipt in receipts + ) + if not validated: + raise LiveProofError("candidate proof consumer requires at least one proof receipt") + if any(receipt.proof_id is not LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION for receipt in validated): + raise LiveProofError("candidate proof consumer requires candidate route evidence") + for receipt in validated: + _require_acceptable_result(receipt) + return validated + + +def validate_final_proof_receipts(receipts: Sequence[object], archive_root: Path) -> tuple[LiveProofReceipt, ...]: + """Require every fixed proof route before treating a campaign proof as final.""" + + validated = validate_live_operation_aggregate(receipts, archive_root) + proof_ids = [receipt.proof_id for receipt in validated] + if len(proof_ids) != len(LiveProofId) or set(proof_ids) != set(LiveProofId): + raise LiveProofError("final proof requires complete route coverage exactly once") + return validated + + +def _fsync_directory(path: Path) -> None: + descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def write_live_proof_receipt(path: Path, receipt: LiveProofReceipt) -> None: + """Atomically publish a fully durable receipt without replacing an existing one.""" + + target = Path(path).expanduser().resolve() + encoded = ( + json.dumps(receipt.to_document(), sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n" + ).encode("utf-8") + descriptor: int | None = None + temporary: Path | None = None + published = False + try: + target.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent) + temporary = Path(temporary_name) + os.fchmod(descriptor, 0o600) + offset = 0 + while offset < len(encoded): + offset += os.write(descriptor, encoded[offset:]) + os.fsync(descriptor) + os.close(descriptor) + descriptor = None + os.link(temporary, target) + published = True + _fsync_directory(target.parent) + temporary.unlink() + temporary = None + _fsync_directory(target.parent) + except OSError as exc: + if descriptor is not None: + os.close(descriptor) + descriptor = None + if published: + try: + target.unlink() + _fsync_directory(target.parent) + except OSError: + pass + if temporary is not None: + try: + temporary.unlink() + _fsync_directory(target.parent) + except OSError: + pass + if isinstance(exc, FileExistsError): + raise LiveProofError("live-proof receipt output already exists") from exc + raise LiveProofError("live-proof receipt output could not be written") from exc + + +validate_live_proof_registry() + +__all__ = [ + "EXISTING_APPLY_RECEIPT_SCHEMA", + "LIVE_PROOF_RECEIPT_SCHEMA", + "LIVE_PROOF_REGISTRY_VERSION", + "LIVE_PROOF_SPECS", + "LiveProofArchiveProfile", + "LiveProofBindings", + "LiveProofError", + "LiveProofId", + "LiveProofMode", + "LiveProofReceipt", + "LiveProofResidue", + "LiveProofResidueKind", + "LiveProofSpec", + "capture_live_proof_bindings", + "collect_live_proof", + "live_proof_spec", + "validate_candidate_proof_receipts", + "validate_final_proof_receipts", + "validate_live_operation_aggregate", + "validate_live_proof_receipt", + "validate_live_proof_registry", + "write_live_proof_receipt", +] diff --git a/tests/unit/cli/test_maintenance_live_proof_cli.py b/tests/unit/cli/test_maintenance_live_proof_cli.py new file mode 100644 index 0000000000..e205473c45 --- /dev/null +++ b/tests/unit/cli/test_maintenance_live_proof_cli.py @@ -0,0 +1,139 @@ +"""Real Click dispatch tests for the fixed maintenance live-proof command.""" + +from __future__ import annotations + +import json +import os +import subprocess +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from polylogue.cli.click_app import cli +from polylogue.maintenance import live_proof + + +@pytest.fixture(autouse=True) +def clean_git_identity(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + live_proof, + "_run_git", + lambda repository, *arguments: subprocess.CompletedProcess( + args=("git", str(repository), *arguments), + returncode=0, + stdout="" if arguments[0] == "status" else "a" * 40, + stderr="", + ), + ) + + +def test_live_proof_cli_dispatches_registered_read_only_proof( + cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("POLYLOGUE_CODE_SHA", "b" * 40) + output = cli_workspace["archive_root"].parent / "live-proof.json" + + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "live-proof", + "--proof-id", + "archive-verification", + "--output", + str(output), + ], + ) + + assert result.exit_code == 0, result.output + receipt = json.loads(output.read_text(encoding="utf-8")) + assert receipt["proof_id"] == "archive-verification" + assert receipt["mode"] == "read_only" + assert str(cli_workspace["archive_root"]) not in output.read_text(encoding="utf-8") + + +def test_live_proof_cli_rejects_unknown_route_without_creating_output( + cli_workspace: dict[str, Path], cli_runner: CliRunner +) -> None: + output = cli_workspace["archive_root"].parent / "unknown-live-proof.json" + + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "live-proof", + "--proof-id", + "arbitrary-shell-command", + "--output", + str(output), + ], + ) + + assert result.exit_code != 0 + assert "unknown live-proof id" in result.output + assert not output.exists() + + +def test_live_proof_cli_rejects_output_under_external_active_generation( + cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("POLYLOGUE_CODE_SHA", "d" * 40) + archive_root = cli_workspace["archive_root"] + external_root = archive_root.parent / "external-active-generation" + external_root.mkdir() + external_index = external_root / "index.db" + os.link(archive_root / "index.db", external_index) + (archive_root / ".index-active-pointer").write_text(str(external_index), encoding="utf-8") + output = external_root / "live-proof.json" + + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "live-proof", + "--proof-id", + "archive-verification", + "--output", + str(output), + ], + ) + + assert result.exit_code != 0 + assert "archive-owned storage" in result.output + assert not output.exists() + + +def test_live_proof_cli_translates_output_os_error( + cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("POLYLOGUE_CODE_SHA", "c" * 40) + output = cli_workspace["archive_root"].parent / "live-proof-output-error.json" + + def fail_write(_path: Path, _receipt: object) -> None: + raise OSError("read-only filesystem") + + monkeypatch.setattr(live_proof, "write_live_proof_receipt", fail_write) + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "live-proof", + "--proof-id", + "archive-verification", + "--output", + str(output), + ], + ) + + assert result.exit_code != 0 + assert "live-proof receipt output could not be written" in result.output + assert not output.exists() diff --git a/tests/unit/maintenance/test_live_proof.py b/tests/unit/maintenance/test_live_proof.py new file mode 100644 index 0000000000..616bd3c151 --- /dev/null +++ b/tests/unit/maintenance/test_live_proof.py @@ -0,0 +1,656 @@ +"""Tests for the fixed live-proof protocol. + +The production dependencies exercised here are the archive-verification +profiles and archive generation store. Red mutations alter a captured binding, +generation record, result status, or output write so consumers must reject +evidence a receipt-only serializer would otherwise accept. +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import subprocess +from dataclasses import replace +from pathlib import Path + +import pytest + +from polylogue.core.hashing import hash_payload +from polylogue.core.json import JSONDocument +from polylogue.maintenance import live_proof +from polylogue.maintenance.live_proof import ( + EXISTING_APPLY_RECEIPT_SCHEMA, + LIVE_PROOF_SPECS, + LiveProofBindings, + LiveProofError, + LiveProofId, + LiveProofMode, + capture_live_proof_bindings, + collect_live_proof, + validate_candidate_proof_receipts, + validate_final_proof_receipts, + validate_live_operation_aggregate, + validate_live_proof_receipt, + validate_live_proof_registry, + write_live_proof_receipt, +) +from polylogue.maintenance.schema_inference_gate import rebuild_source_revision_snapshot +from polylogue.storage.archive_identity import ArchiveLocation +from polylogue.storage.index_generation import IndexGenerationStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root +from polylogue.version import VERSION_INFO + + +@pytest.fixture +def archive_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + root = (tmp_path / "private-archive").resolve() + initialize_active_archive_root(root) + monkeypatch.setenv("POLYLOGUE_CODE_SHA", "a" * 40) + monkeypatch.setattr( + live_proof, + "_run_git", + lambda repository, *arguments: subprocess.CompletedProcess( + args=("git", str(repository), *arguments), + returncode=0, + stdout="" if arguments[0] == "status" else "a" * 40, + stderr="", + ), + ) + return root + + +def _apply_receipt( + bindings: LiveProofBindings, *, status: str = "applied", operation_id: str = "known-source-remediation" +) -> dict[str, object]: + document = { + "receipt_schema": EXISTING_APPLY_RECEIPT_SCHEMA, + "operation_id": operation_id, + "bindings": bindings.to_document(), + "result": {"status": status, "changed_count": 1}, + } + return {**document, "receipt_sha256": hash_payload(document)} + + +def _candidate(root: Path) -> str: + store = IndexGenerationStore(ArchiveLocation.resolve(root)) + transaction = store.create_transaction(source_snapshot=rebuild_source_revision_snapshot(root)) + store.save_transaction(replace(transaction, status="ready")) + return transaction.generation_id + + +def _rehash(document: JSONDocument) -> JSONDocument: + unsigned = dict(document) + unsigned.pop("receipt_sha256") + document["receipt_sha256"] = hash_payload(unsigned) + return document + + +def test_fixed_registry_has_exactly_the_three_supported_modes() -> None: + validate_live_proof_registry() + + assert {spec.mode for spec in LIVE_PROOF_SPECS} == set(LiveProofMode) + assert {spec.proof_id for spec in LIVE_PROOF_SPECS} == set(LiveProofId) + assert {spec.producer for spec in LIVE_PROOF_SPECS} == { + "archive_verification", + "existing_apply_receipt", + } + + +def test_read_only_receipt_preserves_full_redacted_canonical_evidence(archive_root: Path) -> None: + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + repeated = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + + first_document = receipt.to_document() + repeated_document = repeated.to_document() + first_document.pop("generated_at_ms") + repeated_document.pop("generated_at_ms") + first_document.pop("receipt_sha256") + repeated_document.pop("receipt_sha256") + assert first_document == repeated_document + assert repeated.generated_at_ms >= receipt.generated_at_ms + evidence = receipt.result["archive_verification"] + assert isinstance(evidence, dict) + profiles = evidence["profiles"] + assert isinstance(profiles, dict) + assert list(profiles) == ["active-archive"] + active = profiles["active-archive"] + assert isinstance(active, dict) + checks = active["checks"] + assert isinstance(checks, list) + assert len(checks) > 2 + assert str(archive_root) not in json.dumps(receipt.to_document()) + assert receipt.to_document()["receipt_sha256"] == receipt.receipt_sha256 + assert {name for name, _version in receipt.bindings.schema_versions} == { + "audit", + "embeddings", + "index", + "ops", + "source", + "user", + } + assert {name for name, _digest in receipt.bindings.tier_file_set_digests} == { + "audit", + "embeddings", + "index", + "ops", + "source", + "user", + } + assert receipt.bindings.active_index_sha256 + assert validate_live_proof_receipt(receipt.to_document(), archive_root) == receipt + + +def test_read_only_receipt_normalizes_archive_root_alias(archive_root: Path, tmp_path: Path) -> None: + alias = tmp_path / "archive-alias" + alias.symlink_to(archive_root, target_is_directory=True) + + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, alias) + + private_paths = dict(receipt.bindings.private_paths) + assert private_paths["archive_root"] == live_proof.PrivatePathReference.capture(archive_root) + assert str(alias) not in json.dumps(receipt.to_document()) + assert validate_live_proof_receipt(receipt.to_document(), alias) == receipt + + +def test_read_only_receipt_redacts_an_external_active_index_path(archive_root: Path) -> None: + external_root = archive_root.parent / "private-active-index" + external_root.mkdir() + external_index = external_root / "index.db" + os.link(archive_root / "index.db", external_index) + (archive_root / ".index-active-pointer").write_text(str(external_index), encoding="utf-8") + + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + + assert str(external_root) not in json.dumps(receipt.to_document()) + + +def test_read_only_receipt_redacts_external_verification_evidence(archive_root: Path) -> None: + external_source = archive_root.parent / "private source" / "session.jsonl" + with sqlite3.connect(archive_root / "ops.db") as connection: + connection.execute( + """ + INSERT INTO ingest_cursor(source_path, excluded, stat_size, byte_offset, updated_at_ms) + VALUES (?, 0, 100000000, 5000000, 0) + """, + (str(external_source),), + ) + + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + + serialized = json.dumps(receipt.to_document()) + assert str(external_source) not in serialized + assert "[private-path:" in serialized + + +def test_receipt_rejects_expired_generation_time(archive_root: Path) -> None: + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + mutated = receipt.to_document() + mutated["generated_at_ms"] = receipt.generated_at_ms - live_proof._LIVE_PROOF_MAX_AGE_MS - 1 + + with pytest.raises(LiveProofError, match="stale or has an invalid generation time"): + validate_live_proof_receipt(_rehash(mutated), archive_root) + + +def test_capture_accepts_empty_checkpointed_wal_sidecar(archive_root: Path) -> None: + (archive_root / "index.db-wal").write_bytes(b"") + + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + + assert receipt.bindings.active_index_sha256 + + +def test_mode_inputs_are_isolated(archive_root: Path, tmp_path: Path) -> None: + apply_path = tmp_path / "apply.json" + apply_path.write_text(json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root))), encoding="utf-8") + + with pytest.raises(LiveProofError, match="read-only proof"): + collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root, apply_receipt_path=apply_path) + with pytest.raises(LiveProofError, match="candidate proof"): + collect_live_proof(LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION.value, archive_root) + with pytest.raises(LiveProofError, match="existing-apply proof"): + collect_live_proof(LiveProofId.EXISTING_APPLY_RECEIPT.value, archive_root) + + +def test_existing_apply_receipt_is_bound_and_rejects_controlled_binding_mutation( + archive_root: Path, tmp_path: Path +) -> None: + apply_path = tmp_path / "private-apply-receipt.json" + apply_path.write_text(json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root))), encoding="utf-8") + + receipt = collect_live_proof( + LiveProofId.EXISTING_APPLY_RECEIPT.value, + archive_root, + apply_receipt_path=apply_path, + ) + + assert receipt.input_receipt_digests + assert str(apply_path) not in json.dumps(receipt.to_document()) + mutated = receipt.to_document() + binding_value = mutated["bindings"] + assert isinstance(binding_value, dict) + bindings = dict(binding_value) + bindings["source_snapshot"] = "0" * 64 + mutated["bindings"] = bindings + with pytest.raises(LiveProofError, match="bindings are stale"): + validate_live_proof_receipt(_rehash(mutated), archive_root) + + +@pytest.mark.parametrize( + ("apply_status", "proof_status"), + [("applied", "passed"), ("already_satisfied", "passed"), ("not_applicable", "not_applicable")], +) +def test_existing_apply_receipt_preserves_terminal_result_status( + archive_root: Path, tmp_path: Path, apply_status: str, proof_status: str +) -> None: + apply_path = tmp_path / "apply-receipt.json" + apply_path.write_text( + json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root), status=apply_status)), encoding="utf-8" + ) + + receipt = collect_live_proof(LiveProofId.EXISTING_APPLY_RECEIPT.value, archive_root, apply_receipt_path=apply_path) + + assert receipt.result["status"] == proof_status + apply_receipt = receipt.result["apply_receipt"] + assert isinstance(apply_receipt, dict) + apply_result = apply_receipt["result"] + assert isinstance(apply_result, dict) + assert apply_result["status"] == apply_status + assert validate_live_operation_aggregate((receipt.to_document(),), archive_root) == (receipt,) + + +def test_existing_apply_receipt_embeds_the_digest_bound_input(archive_root: Path, tmp_path: Path) -> None: + apply_path = tmp_path / "apply-receipt.json" + apply_path.write_text(json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root))), encoding="utf-8") + receipt = collect_live_proof(LiveProofId.EXISTING_APPLY_RECEIPT.value, archive_root, apply_receipt_path=apply_path) + mutated = receipt.to_document() + result = mutated["result"] + assert isinstance(result, dict) + embedded = result["apply_receipt"] + assert isinstance(embedded, dict) + embedded_result = embedded["result"] + assert isinstance(embedded_result, dict) + embedded_result["status"] = "failed" + _rehash(embedded) + result["status"] = "failed" + mutated["residues"] = [{"kind": "check_failed", "code": "apply-result-failed"}] + + with pytest.raises(LiveProofError, match="input digest"): + validate_live_proof_receipt(_rehash(mutated), archive_root) + + +def test_existing_apply_receipt_keeps_failure_as_failed_evidence(archive_root: Path, tmp_path: Path) -> None: + apply_path = tmp_path / "failed-apply-receipt.json" + apply_path.write_text( + json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root), status="failed")), encoding="utf-8" + ) + + receipt = collect_live_proof(LiveProofId.EXISTING_APPLY_RECEIPT.value, archive_root, apply_receipt_path=apply_path) + + assert receipt.result["status"] == "failed" + with pytest.raises(LiveProofError, match="not acceptable"): + validate_live_operation_aggregate((receipt.to_document(),), archive_root) + + +def test_existing_apply_receipt_requires_not_applicable_residue(archive_root: Path, tmp_path: Path) -> None: + apply_path = tmp_path / "not-applicable-apply-receipt.json" + apply_path.write_text( + json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root), status="not_applicable")), + encoding="utf-8", + ) + + receipt = collect_live_proof(LiveProofId.EXISTING_APPLY_RECEIPT.value, archive_root, apply_receipt_path=apply_path) + + assert validate_live_operation_aggregate((receipt.to_document(),), archive_root) == (receipt,) + mutated = receipt.to_document() + mutated["residues"] = [] + with pytest.raises(LiveProofError, match="status and residues are inconsistent"): + validate_live_operation_aggregate((_rehash(mutated),), archive_root) + + +def test_existing_apply_receipt_rejects_unknown_result(archive_root: Path, tmp_path: Path) -> None: + apply_path = tmp_path / "unknown-apply-receipt.json" + apply_path.write_text( + json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root), status="unknown")), encoding="utf-8" + ) + + with pytest.raises(LiveProofError, match="result status is not recognized"): + collect_live_proof(LiveProofId.EXISTING_APPLY_RECEIPT.value, archive_root, apply_receipt_path=apply_path) + + +def test_existing_apply_receipt_requires_registered_operation(archive_root: Path, tmp_path: Path) -> None: + apply_path = tmp_path / "unregistered-apply-receipt.json" + apply_path.write_text( + json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root), operation_id="invented-operation")), + encoding="utf-8", + ) + + with pytest.raises(LiveProofError, match="operation binding is invalid"): + collect_live_proof(LiveProofId.EXISTING_APPLY_RECEIPT.value, archive_root, apply_receipt_path=apply_path) + + +def test_candidate_receipt_binds_canonical_generation_and_all_profiles(archive_root: Path) -> None: + generation_id = _candidate(archive_root) + receipt = collect_live_proof( + LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION.value, + archive_root, + candidate_generation_id=generation_id, + ) + + assert receipt.bindings.candidate_generation_id == generation_id + assert receipt.bindings.candidate_index_sha256 is not None + verification = receipt.result["archive_verification"] + assert isinstance(verification, dict) + profiles = verification["profiles"] + assert isinstance(profiles, dict) + assert set(profiles) == {"candidate-index", "candidate-cross-tier"} + candidate_path = Path(IndexGenerationStore(ArchiveLocation.resolve(archive_root)).load(generation_id).index_path) + with sqlite3.connect(candidate_path) as connection: + connection.execute("CREATE TABLE proof_binding_mutation (marker TEXT NOT NULL)") + with pytest.raises(LiveProofError, match="bindings are stale"): + validate_candidate_proof_receipts((receipt.to_document(),), archive_root, candidate_generation_id=generation_id) + + +def test_candidate_receipt_requires_ready_rebuild_transaction(archive_root: Path) -> None: + generation_id = _candidate(archive_root) + store = IndexGenerationStore(ArchiveLocation.resolve(archive_root)) + transaction_path = next(store.transactions_root.glob("*.json")) + transaction = store.load_transaction(transaction_path.stem) + store.save_transaction(replace(transaction, status="running")) + + with pytest.raises(LiveProofError, match="candidate generation binding"): + collect_live_proof( + LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION.value, + archive_root, + candidate_generation_id=generation_id, + ) + + +def test_candidate_rejects_source_snapshot_drift(archive_root: Path) -> None: + generation_id = _candidate(archive_root) + store = IndexGenerationStore(ArchiveLocation.resolve(archive_root)) + generation_path = store.generations_root / generation_id / "generation.json" + payload = json.loads(generation_path.read_text(encoding="utf-8")) + payload["source_snapshot"] = "outdated" + generation_path.write_text(json.dumps(payload), encoding="utf-8") + + with pytest.raises(LiveProofError, match="candidate generation binding"): + collect_live_proof( + LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION.value, + archive_root, + candidate_generation_id=generation_id, + ) + + +def test_candidate_rejects_symlink_index(archive_root: Path) -> None: + generation_id = _candidate(archive_root) + store = IndexGenerationStore(ArchiveLocation.resolve(archive_root)) + candidate_path = Path(store.load(generation_id).index_path) + candidate_path.unlink() + candidate_path.symlink_to(archive_root / "index.db") + + with pytest.raises(LiveProofError, match="candidate generation binding"): + collect_live_proof( + LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION.value, + archive_root, + candidate_generation_id=generation_id, + ) + + +def test_candidate_rejects_poisoned_active_pointer_without_repairing_it(archive_root: Path) -> None: + generation_id = _candidate(archive_root) + store = IndexGenerationStore(ArchiveLocation.resolve(archive_root)) + candidate_path = Path(store.load(generation_id).index_path) + pointer = archive_root / ".index-active-pointer" + pointer.write_text(str(candidate_path), encoding="utf-8") + + with pytest.raises(LiveProofError, match="canonical active-index pointer"): + collect_live_proof( + LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION.value, + archive_root, + candidate_generation_id=generation_id, + ) + + assert pointer.read_text(encoding="utf-8") == str(candidate_path) + + +def test_receipt_rejects_invalid_residue_code(archive_root: Path) -> None: + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + mutated = receipt.to_document() + residues = mutated["residues"] + assert isinstance(residues, list) and residues + residue = residues[0] + assert isinstance(residue, dict) + residue["code"] = "not a valid code" + + with pytest.raises(LiveProofError, match="residues are malformed"): + validate_live_proof_receipt(_rehash(mutated), archive_root) + + +def test_receipt_rejects_residues_mismatched_to_check_evidence(archive_root: Path) -> None: + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + mutated = receipt.to_document() + result = mutated["result"] + assert isinstance(result, dict) + result["status"] = "failed" + verification = result["archive_verification"] + assert isinstance(verification, dict) + profiles = verification["profiles"] + assert isinstance(profiles, dict) + profile = profiles["active-archive"] + assert isinstance(profile, dict) + checks = profile["checks"] + assert isinstance(checks, list) and checks + check = checks[0] + assert isinstance(check, dict) + check["status"] = "error" + + with pytest.raises(LiveProofError, match="status and residues are inconsistent"): + validate_live_proof_receipt(_rehash(mutated), archive_root) + + +def test_receipt_rejects_incomplete_archive_check_evidence(archive_root: Path) -> None: + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + mutated = receipt.to_document() + result = mutated["result"] + assert isinstance(result, dict) + verification = result["archive_verification"] + assert isinstance(verification, dict) + profiles = verification["profiles"] + assert isinstance(profiles, dict) + profile = profiles["active-archive"] + assert isinstance(profile, dict) + checks = profile["checks"] + assert isinstance(checks, list) and checks + check = checks[0] + assert isinstance(check, dict) + profile["checks"] = [{"name": check["name"], "status": "ok"}, *checks[1:]] + + with pytest.raises(LiveProofError, match="archive verification evidence is malformed"): + validate_live_proof_receipt(_rehash(mutated), archive_root) + + +def test_aggregate_rejects_a_self_hashed_failed_proof_result(archive_root: Path) -> None: + bindings = capture_live_proof_bindings(archive_root) + document = _apply_receipt(bindings, status="failed") + apply_path = archive_root.parent / "apply.json" + apply_path.write_text(json.dumps(document), encoding="utf-8") + failed = collect_live_proof(LiveProofId.EXISTING_APPLY_RECEIPT.value, archive_root, apply_receipt_path=apply_path) + + with pytest.raises(LiveProofError, match="not acceptable"): + validate_live_operation_aggregate((failed.to_document(),), archive_root) + + +def test_final_proof_requires_every_route_exactly_once(archive_root: Path, tmp_path: Path) -> None: + apply_path = tmp_path / "apply.json" + apply_path.write_text(json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root))), encoding="utf-8") + receipt = collect_live_proof(LiveProofId.EXISTING_APPLY_RECEIPT.value, archive_root, apply_receipt_path=apply_path) + + with pytest.raises(LiveProofError, match="complete route coverage"): + validate_final_proof_receipts((receipt.to_document(),), archive_root) + + +def test_aggregate_captures_candidate_and_active_bindings_once( + archive_root: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + generation_id = _candidate(archive_root) + apply_path = tmp_path / "apply.json" + apply_path.write_text(json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root))), encoding="utf-8") + receipts = ( + collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root).to_document(), + collect_live_proof( + LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION.value, + archive_root, + candidate_generation_id=generation_id, + ).to_document(), + collect_live_proof( + LiveProofId.EXISTING_APPLY_RECEIPT.value, + archive_root, + apply_receipt_path=apply_path, + ).to_document(), + ) + original = live_proof._capture_expected_bindings + calls = 0 + + def capture_once(root: Path, candidate_id: str | None) -> LiveProofBindings: + nonlocal calls + calls += 1 + if calls > 1: + raise AssertionError("aggregate captured another active snapshot") + return original(root, candidate_id) + + monkeypatch.setattr(live_proof, "_capture_expected_bindings", capture_once) + + assert live_proof._validate_aggregate(receipts, archive_root) + assert calls == 1 + + +def test_aggregate_preserves_candidate_receipt_after_promotion(archive_root: Path, tmp_path: Path) -> None: + generation_id = _candidate(archive_root) + candidate_receipt = collect_live_proof( + LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION.value, + archive_root, + candidate_generation_id=generation_id, + ) + store = IndexGenerationStore(ArchiveLocation.resolve(archive_root)) + store.promote(store.load(generation_id)) + + apply_path = tmp_path / "post-promotion-apply.json" + apply_path.write_text(json.dumps(_apply_receipt(capture_live_proof_bindings(archive_root))), encoding="utf-8") + active_receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + apply_receipt = collect_live_proof( + LiveProofId.EXISTING_APPLY_RECEIPT.value, + archive_root, + apply_receipt_path=apply_path, + ) + + validated = live_proof._validate_aggregate( + (active_receipt.to_document(), candidate_receipt.to_document(), apply_receipt.to_document()), archive_root + ) + assert [receipt.proof_id for receipt in validated] == [ + LiveProofId.ARCHIVE_VERIFICATION, + LiveProofId.CANDIDATE_ARCHIVE_VERIFICATION, + LiveProofId.EXISTING_APPLY_RECEIPT, + ] + + +def test_readonly_uri_encodes_sqlite_metacharacters(tmp_path: Path) -> None: + database = tmp_path / "archive%2Fname?name#fragment.db" + sqlite3.connect(database).close() + + uri = live_proof._readonly_uri(database) + + assert "%25" in uri and "%3F" in uri and "%23" in uri + with sqlite3.connect(uri, uri=True) as connection: + assert connection.execute("PRAGMA user_version").fetchone() == (0,) + + +def test_capture_rejects_archive_paths_unsafe_for_dependency_uris( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "archive?name#fragment" + initialize_active_archive_root(root) + monkeypatch.setenv("POLYLOGUE_CODE_SHA", "a" * 40) + + with pytest.raises(LiveProofError, match="SQLite URI query characters"): + capture_live_proof_bindings(root) + + +def test_git_fallback_rejects_dirty_worktree(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("POLYLOGUE_CODE_SHA", raising=False) + monkeypatch.setattr( + live_proof, + "_run_git", + lambda *_args: subprocess.CompletedProcess(args=(), returncode=0, stdout=" M live_proof.py\n", stderr=""), + ) + + with pytest.raises(LiveProofError, match="clean git worktree"): + live_proof._code_sha() + + +def test_git_sha_override_still_rejects_dirty_worktree(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("POLYLOGUE_CODE_SHA", "a" * 40) + monkeypatch.setattr( + live_proof, + "_run_git", + lambda *_args: subprocess.CompletedProcess(args=(), returncode=0, stdout=" M live_proof.py\n", stderr=""), + ) + + with pytest.raises(LiveProofError, match="clean git worktree"): + live_proof._code_sha() + + +def test_installed_code_identity_uses_version_info_commit(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("POLYLOGUE_CODE_SHA", raising=False) + monkeypatch.setattr(VERSION_INFO, "commit", "B" * 40) + monkeypatch.setattr(live_proof, "__file__", "/installed/polylogue/maintenance/live_proof.py") + + assert live_proof._code_sha() == "b" * 40 + + +def test_receipt_write_removes_partial_output_after_os_error( + archive_root: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + target = tmp_path / "new-receipt.json" + + def fail_write(_descriptor: int, _payload: bytes) -> int: + raise OSError("disk full") + + monkeypatch.setattr(os, "write", fail_write) + with pytest.raises(LiveProofError, match="could not be written"): + write_live_proof_receipt(target, receipt) + + assert not target.exists() + + +def test_receipt_write_removes_published_output_when_directory_sync_fails( + archive_root: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + target = tmp_path / "new-receipt.json" + calls = 0 + + def fail_after_publish(_path: Path) -> None: + nonlocal calls + calls += 1 + if calls == 2: + raise OSError("directory sync failed") + + monkeypatch.setattr(live_proof, "_fsync_directory", fail_after_publish) + with pytest.raises(LiveProofError, match="could not be written"): + write_live_proof_receipt(target, receipt) + + assert not target.exists() + + +def test_receipt_write_cleans_temporary_file_when_output_exists(archive_root: Path, tmp_path: Path) -> None: + receipt = collect_live_proof(LiveProofId.ARCHIVE_VERIFICATION.value, archive_root) + target = tmp_path / "existing-receipt.json" + write_live_proof_receipt(target, receipt) + + with pytest.raises(LiveProofError, match="output already exists"): + write_live_proof_receipt(target, receipt) + + assert list(tmp_path.glob(f".{target.name}.*")) == []