diff --git a/docs/devtools.md b/docs/devtools.md index 297f693b67..b0552b574f 100644 --- a/docs/devtools.md +++ b/docs/devtools.md @@ -307,6 +307,38 @@ Catalog bypass audit sites are machine-checked across workflow runs, CI-owned np +## Cursor-authority reconciliation + +`polylogue ops maintenance cursor-authority-reconcile` is a dry-run-by-default +repair route for exactly one proven cursor-ahead source. It reads the fixed +`/realm/db/polylogue` archive root, requires the daemon to be stopped, and +writes a plan containing path and raw identifiers only as digests. Apply +requires that immutable plan, a freshly verified `full_evidence` backup +manifest with blob rollback evidence, and a new receipt path. The apply route +uses the normal live full-ingest/replay path under one single-use exact path +and frontier authorization. Receipts distinguish a performed ingest from an +observed recovery, leave cursor row counts null when the before/after state did +not prove them, and record typed deferred or failed post-ingest evidence. It +never accepts a global cursor bypass or writes `ingest_cursor` or accepted-head +rows directly. + +The dry-run form is: + +```text +polylogue ops maintenance cursor-authority-reconcile \ + --source-path-file /private/path-file \ + --output-plan /private/reconciliation-plan.json +``` + +The apply form is: + +```text +polylogue ops maintenance cursor-authority-reconcile --apply \ + --plan /private/reconciliation-plan.json \ + --backup-manifest /private/full-evidence-backup \ + --receipt /private/reconciliation-receipt.json +``` + ## Validation and Evidence When changing semantics, validation, or surfaces: diff --git a/polylogue/cli/commands/maintenance/__init__.py b/polylogue/cli/commands/maintenance/__init__.py index 3ddd56970f..6c29815aef 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.", ), + ( + "cursor-authority-reconcile", + "_cursor_authority", + "cursor_authority_reconcile_command", + "Plan or apply one backup-gated cursor-authority reconciliation.", + ), ) diff --git a/polylogue/cli/commands/maintenance/_cursor_authority.py b/polylogue/cli/commands/maintenance/_cursor_authority.py new file mode 100644 index 0000000000..3ea86b7866 --- /dev/null +++ b/polylogue/cli/commands/maintenance/_cursor_authority.py @@ -0,0 +1,52 @@ +"""``maintenance cursor-authority-reconcile`` command.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import click + + +@click.command("cursor-authority-reconcile") +@click.option("--source-path-file", type=click.Path(path_type=Path, dir_okay=False), default=None) +@click.option("--output-plan", type=click.Path(path_type=Path, dir_okay=False), default=None) +@click.option("--plan", "plan_path", type=click.Path(path_type=Path, dir_okay=False), default=None) +@click.option("--backup-manifest", type=click.Path(path_type=Path, file_okay=True, dir_okay=True), default=None) +@click.option("--receipt", type=click.Path(path_type=Path, dir_okay=False), default=None) +@click.option("--apply", "apply_changes", is_flag=True, help="Apply one previously written reconciliation plan.") +def cursor_authority_reconcile_command( + source_path_file: Path | None, + output_plan: Path | None, + plan_path: Path | None, + backup_manifest: Path | None, + receipt: Path | None, + apply_changes: bool, +) -> None: + """Plan or apply one backup-gated cursor-authority reconciliation.""" + + from polylogue.maintenance.cursor_authority_reconcile import ( + CursorAuthorityReconciliationError, + apply_reconciliation, + build_reconciliation_plan, + ) + + try: + if apply_changes: + if plan_path is None or backup_manifest is None or receipt is None: + raise click.UsageError("--apply requires --plan, --backup-manifest, and --receipt") + if source_path_file is not None or output_plan is not None: + raise click.UsageError("--apply does not accept --source-path-file or --output-plan") + result = apply_reconciliation(plan_path=plan_path, backup_manifest=backup_manifest, receipt=receipt) + else: + if source_path_file is None or output_plan is None: + raise click.UsageError("dry-run requires --source-path-file and --output-plan") + if plan_path is not None or backup_manifest is not None or receipt is not None: + raise click.UsageError("dry-run accepts only --source-path-file and --output-plan") + result = build_reconciliation_plan(source_path_file=source_path_file, output_plan=output_plan) + except CursorAuthorityReconciliationError as exc: + raise click.ClickException(str(exc)) from exc + click.echo(json.dumps(result, indent=2, sort_keys=True)) + + +__all__ = ["cursor_authority_reconcile_command"] diff --git a/polylogue/maintenance/cursor_authority_reconcile.py b/polylogue/maintenance/cursor_authority_reconcile.py new file mode 100644 index 0000000000..01ee7a0fdd --- /dev/null +++ b/polylogue/maintenance/cursor_authority_reconcile.py @@ -0,0 +1,939 @@ +"""Backup-gated reconciliation of one live cursor-authority violation. + +The command is intentionally narrow. It proves one source path from the +canonical raw-frontier projection, then runs that path through +``LiveBatchProcessor.ingest_files``. It never edits a cursor or accepted head +itself. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +import sqlite3 +import stat +import subprocess +import tempfile +import time +from collections.abc import Mapping +from contextlib import closing +from pathlib import Path + +from polylogue.api import Polylogue +from polylogue.config import Config +from polylogue.core.enums import IngestOutcome +from polylogue.operations.durable_change_train import acquire_durable_archive_ownership +from polylogue.pipeline.ingest_outcomes import IngestAttemptDisposition +from polylogue.sources.live.batch import ( + LiveBatchProcessor, + cursor_authority_path_digest, + scoped_cursor_authority_authorization, +) +from polylogue.sources.live.batch_support import sha256_range_from_path +from polylogue.sources.live.cursor import CursorStore +from polylogue.sources.live.metrics import LiveBatchMetrics +from polylogue.sources.live.watcher import WatchSource +from polylogue.storage.archive_identity import ArchiveLocation +from polylogue.storage.backup_attestation import BackupAttestationError, verify_verification_receipt +from polylogue.storage.raw_retention import RawFrontierIntegrityProjection, raw_frontier_integrity_projection + +PLAN_FORMAT = "polylogue.cursor-authority-reconciliation-plan.v1" +RECEIPT_FORMAT = "polylogue.cursor-authority-reconciliation-receipt.v1" +ARCHIVE_ROOT = Path("/realm/db/polylogue") +_REQUIRED_TIERS = ("source", "index", "ops", "audit") + + +class CursorAuthorityReconciliationError(RuntimeError): + """A reconciliation precondition or postcondition was not proven.""" + + +def _canonical_digest(payload: object) -> str: + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _file_fingerprint(path: Path) -> tuple[int, str]: + """Return size and digest from one descriptor observation.""" + + digest = hashlib.sha256() + try: + with path.open("rb") as handle: + size = os.fstat(handle.fileno()).st_size + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + except OSError as exc: + raise CursorAuthorityReconciliationError(f"required archive file is unreadable: {path}") from exc + return size, digest.hexdigest() + + +def _identity_digest(value: object) -> str: + return hashlib.sha256(str(value).encode("utf-8")).hexdigest() + + +def _path_identity(path: Path) -> dict[str, str]: + return { + "path_digest": cursor_authority_path_digest(path), + "basename": path.name, + } + + +def _stat_observation(path: Path) -> tuple[int, int, int, int, int]: + value = path.stat() + return value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns + + +def _archive_root() -> Path: + """Return the fixed archive root for this command. + + This deliberately does not call ``polylogue.paths.archive_root`` or read + any ambient archive-root environment variable. + """ + + return ARCHIVE_ROOT + + +def _read_private_source_path(path_file: Path) -> Path: + descriptor: int | None = None + try: + descriptor = os.open(path_file, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) + metadata = os.fstat(descriptor) + except OSError as exc: + raise CursorAuthorityReconciliationError(f"source path file is unreadable: {path_file}") from exc + try: + if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1: + raise CursorAuthorityReconciliationError("source path file must be a regular single-linked file") + if metadata.st_uid != os.geteuid() or stat.S_IMODE(metadata.st_mode) != 0o600: + raise CursorAuthorityReconciliationError("source path file must be owned by the operator and mode 0600") + chunks: list[bytes] = [] + while True: + chunk = os.read(descriptor, 64 * 1024) + if not chunk: + break + chunks.append(chunk) + except OSError as exc: + raise CursorAuthorityReconciliationError(f"source path file is unreadable: {path_file}") from exc + finally: + if descriptor is not None: + os.close(descriptor) + try: + lines = b"".join(chunks).decode("utf-8").splitlines() + except UnicodeDecodeError as exc: + raise CursorAuthorityReconciliationError("source path file must be UTF-8 text") from exc + if len(lines) != 1 or not lines[0].strip(): + raise CursorAuthorityReconciliationError("source path file must contain exactly one non-empty path") + candidate = Path(lines[0]) + if not candidate.is_absolute(): + raise CursorAuthorityReconciliationError("selected source path must be absolute") + try: + resolved = candidate.resolve(strict=True) + except OSError as exc: + raise CursorAuthorityReconciliationError("selected source path does not resolve") from exc + if not resolved.is_file(): + raise CursorAuthorityReconciliationError("selected source path must be a regular file") + return resolved + + +def _sqlite_snapshot(path: Path) -> dict[str, object]: + if not path.is_file(): + raise CursorAuthorityReconciliationError(f"required archive tier is missing: {path}") + try: + with tempfile.TemporaryDirectory(prefix="polylogue-sqlite-snapshot-") as temporary_dir: + snapshot_path = Path(temporary_dir) / "snapshot.db" + with ( + closing(sqlite3.connect(f"file:{path.resolve()}?mode=ro", uri=True)) as source_conn, + closing(sqlite3.connect(snapshot_path)) as snapshot_conn, + ): + source_conn.execute("PRAGMA query_only = ON") + source_conn.backup(snapshot_conn) + snapshot_conn.commit() + size_bytes, sha256 = _file_fingerprint(snapshot_path) + with closing(sqlite3.connect(f"file:{snapshot_path.resolve()}?mode=ro", uri=True)) as conn: + conn.execute("PRAGMA query_only = ON") + user_version = int(conn.execute("PRAGMA user_version").fetchone()[0] or 0) + schema_version = int(conn.execute("PRAGMA schema_version").fetchone()[0] or 0) + schema_rows = conn.execute( + "SELECT type, name, tbl_name, sql FROM sqlite_schema " + "WHERE name NOT LIKE 'sqlite_%' ORDER BY type, name, tbl_name" + ).fetchall() + quick_check = tuple(str(row[0]) for row in conn.execute("PRAGMA quick_check")) + except (OSError, sqlite3.Error) as exc: + raise CursorAuthorityReconciliationError(f"could not read SQLite tier {path}: {exc}") from exc + schema_digest = _canonical_digest( + [[str(value) if value is not None else None for value in row] for row in schema_rows] + ) + return { + "size_bytes": size_bytes, + "sha256": sha256, + "user_version": user_version, + "schema_version": schema_version, + "schema_sha256": schema_digest, + "quick_check": list(quick_check), + } + + +def _tier_snapshots(root: Path) -> dict[str, dict[str, object]]: + location = ArchiveLocation.resolve(root) + snapshots: dict[str, dict[str, object]] = {} + for tier in _REQUIRED_TIERS: + tier_path = location.active_index_path if tier == "index" else root / f"{tier}.db" + snapshots[tier] = _sqlite_snapshot(tier_path) + return snapshots + + +def _active_index_binding(root: Path) -> dict[str, object]: + location = ArchiveLocation.resolve(root) + return { + "path": _path_identity(location.active_index_path), + "generation_digest": _identity_digest(location.active_generation), + } + + +def _code_sha() -> str: + try: + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + cwd=Path(__file__).parents[2], + ) + except (OSError, subprocess.CalledProcessError): + return "unknown" + return result.stdout.strip() + + +def _deployed_package_sha() -> str: + package_root = Path(__file__).parents[1] + digest = hashlib.sha256() + for path in sorted(package_root.rglob("*.py")): + if any(part == "__pycache__" for part in path.parts): + continue + digest.update(str(path.relative_to(package_root)).encode("utf-8")) + digest.update(path.read_bytes()) + return digest.hexdigest() + + +def _projection_for(root: Path) -> RawFrontierIntegrityProjection: + from polylogue.storage.archive_readiness import raw_materialization_readiness_snapshot + + return raw_frontier_integrity_projection( + root, + raw_materialization_readiness_snapshot(root), + sample_limit=100, + ) + + +def _private_projection(projection: RawFrontierIntegrityProjection) -> dict[str, object]: + def redact(value: object) -> object: + if isinstance(value, dict): + return { + key: ( + _identity_digest(item) + if key in {"source_path", "logical_source_key", "accepted_raw_id", "raw_id", "session_id"} + and isinstance(item, str) + else None + if key in {"source_path", "logical_source_key", "accepted_raw_id", "raw_id", "session_id"} + and item is not None + else redact(item) + ) + for key, item in value.items() + } + if isinstance(value, list): + return [redact(item) for item in value] + return value + + redacted = redact(projection.to_dict()) + if not isinstance(redacted, dict): + raise CursorAuthorityReconciliationError("raw-frontier projection did not produce a mapping") + return redacted + + +def _cursor_rows(root: Path) -> list[tuple[str, int]]: + with closing(sqlite3.connect(f"file:{(root / 'ops.db').resolve()}?mode=ro", uri=True)) as conn: + rows = conn.execute( + "SELECT source_path, byte_offset FROM ingest_cursor " + "WHERE COALESCE(excluded, 0) = 0 AND byte_offset IS NOT NULL" + ).fetchall() + result: list[tuple[str, int]] = [] + for row in rows: + if not isinstance(row[0], str) or not row[0]: + raise CursorAuthorityReconciliationError("ingest cursor has an invalid source path") + result.append((row[0], _required_nonnegative_int(row[1], "ingest cursor byte_offset"))) + return result + + +def _find_path_by_digest(root: Path, digest: str) -> Path: + matches = [ + Path(path).resolve() + for path, _offset in _cursor_rows(root) + if cursor_authority_path_digest(Path(path)) == digest + ] + if len(matches) != 1: + raise CursorAuthorityReconciliationError("plan path digest does not identify exactly one current cursor path") + return matches[0] + + +def _head_details(root: Path, source_path: Path, projection: RawFrontierIntegrityProjection) -> dict[str, object]: + if projection.cursor_ahead_count != 1 or len(projection.cursor_ahead_samples) != 1: + raise CursorAuthorityReconciliationError("reconciliation requires exactly one true cursor-ahead row") + sample = projection.cursor_ahead_samples[0] + if Path(sample.source_path).resolve() != source_path.resolve(): + raise CursorAuthorityReconciliationError("selected source path is not the sole cursor-ahead path") + index_path = ArchiveLocation.resolve(root).active_index_path + with closing(sqlite3.connect(f"file:{index_path.resolve()}?mode=ro", uri=True)) as conn: + head = conn.execute( + "SELECT logical_source_key, accepted_raw_id, accepted_source_revision, " + "accepted_content_hash, accepted_frontier_kind, accepted_frontier, " + "acquisition_generation, append_end_offset " + "FROM raw_revision_heads WHERE logical_source_key = ?", + (sample.logical_source_key,), + ).fetchone() + if head is None: + raise CursorAuthorityReconciliationError("accepted head is missing for the selected path") + with closing(sqlite3.connect(f"file:{(root / 'source.db').resolve()}?mode=ro", uri=True)) as conn: + raw = conn.execute( + "SELECT raw_id, source_path, blob_hash, blob_size, revision_authority FROM raw_sessions WHERE raw_id = ?", + (str(head[1]),), + ).fetchone() + if raw is None or Path(str(raw[1])).resolve() != source_path.resolve(): + raise CursorAuthorityReconciliationError("accepted head does not match the recorded source path") + if str(head[4]) != "byte" or str(raw[4]) != "byte_proven": + raise CursorAuthorityReconciliationError("accepted head is not byte-authoritative") + logical_source_key = head[0] + if not isinstance(logical_source_key, str) or not logical_source_key: + raise CursorAuthorityReconciliationError("accepted head has an invalid logical source key") + frontier = _required_nonnegative_int(head[5], "accepted head frontier") + blob_hash = bytes(raw[2]).hex() if isinstance(raw[2], bytes) else str(raw[2]).lower() + blob_size = _required_nonnegative_int(raw[3], "accepted raw blob size") + try: + bytes.fromhex(blob_hash) + except ValueError as exc: + raise CursorAuthorityReconciliationError("accepted raw has an invalid blob hash") from exc + if blob_size != frontier or len(blob_hash) != 64: + raise CursorAuthorityReconciliationError("accepted raw does not bind a complete byte frontier") + cursor_matches = [offset for path, offset in _cursor_rows(root) if Path(path).resolve() == source_path.resolve()] + if len(cursor_matches) != 1: + raise CursorAuthorityReconciliationError("selected source path has no unique current cursor row") + cursor_offset = cursor_matches[0] + before_stat = _stat_observation(source_path) + prefix_digest, bytes_read = sha256_range_from_path(source_path, start_offset=0, end_offset=frontier) + after_stat = _stat_observation(source_path) + if before_stat != after_stat: + raise CursorAuthorityReconciliationError("source mutated during accepted-frontier hashing") + if prefix_digest != blob_hash: + raise CursorAuthorityReconciliationError("source prefix does not match the accepted raw blob hash") + return { + "logical_source_key": cursor_authority_path_digest(Path(logical_source_key)), + "cursor_byte_offset": cursor_offset, + "accepted_frontier": frontier, + "accepted_raw_id_digest": _canonical_digest(str(head[1])), + "accepted_blob_hash_digest": _canonical_digest(blob_hash), + "source_prefix_digest": prefix_digest, + "source_prefix_bytes": bytes_read, + "source_stat": list(after_stat), + } + + +def _require_healthy_projection_siblings(projection: RawFrontierIntegrityProjection) -> None: + if not projection.available: + raise CursorAuthorityReconciliationError("raw-frontier projection is unavailable") + if projection.broken_head_status != "healthy" or projection.missing_source_raw_status != "healthy": + raise CursorAuthorityReconciliationError("raw-frontier sibling projections are not healthy") + + +def _build_plan(root: Path, source_path: Path, *, require_candidate: bool = True) -> dict[str, object]: + tiers = _tier_snapshots(root) + projection = _projection_for(root) + _require_healthy_projection_siblings(projection) + path_digest = cursor_authority_path_digest(source_path) + if projection.cursor_ahead_count == 0: + if require_candidate and projection.cursor_authority_gap_count == 0 and projection.overall_status == "healthy": + not_applicable_plan: dict[str, object] = { + "format": PLAN_FORMAT, + "archive_identity": _path_identity(root), + "active_index": _active_index_binding(root), + "code_sha": _code_sha(), + "deployed_package_sha": _deployed_package_sha(), + "tier_fingerprints": tiers, + "source_schema_versions": {tier: tiers[tier]["user_version"] for tier in _REQUIRED_TIERS}, + "selected_path_digest": path_digest, + "observed_at_ms": int(time.time() * 1000), + "status": "not_applicable", + "cursor_byte_offset": None, + "accepted_frontier": None, + "accepted_raw_id_digest": None, + "source_prefix_digest": None, + "before_projection": _private_projection(projection), + } + not_applicable_plan["plan_digest"] = _canonical_digest(not_applicable_plan) + return not_applicable_plan + raise CursorAuthorityReconciliationError("cursor authority is incomparable or has no selected violation") + if projection.cursor_ahead_count != 1: + raise CursorAuthorityReconciliationError("refusing to guess among multiple cursor-ahead rows") + if projection.broken_head_count or projection.missing_source_raw_count: + raise CursorAuthorityReconciliationError( + "global raw-frontier violation set is not exactly one cursor-ahead row" + ) + details = _head_details(root, source_path, projection) + plan: dict[str, object] = { + "format": PLAN_FORMAT, + "archive_identity": _path_identity(root), + "active_index": _active_index_binding(root), + "code_sha": _code_sha(), + "deployed_package_sha": _deployed_package_sha(), + "tier_fingerprints": tiers, + "source_schema_versions": {tier: tiers[tier]["user_version"] for tier in _REQUIRED_TIERS}, + "selected_path_digest": path_digest, + "observed_at_ms": int(time.time() * 1000), + "status": "planned", + "cursor_byte_offset": details["cursor_byte_offset"], + "accepted_frontier": details["accepted_frontier"], + "accepted_raw_id_digest": details["accepted_raw_id_digest"], + "accepted_blob_hash_digest": details["accepted_blob_hash_digest"], + "source_prefix_digest": details["source_prefix_digest"], + "before_projection": _private_projection(projection), + } + plan["plan_digest"] = _canonical_digest(plan) + return plan + + +def _load_plan(path: Path) -> dict[str, object]: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise CursorAuthorityReconciliationError(f"invalid reconciliation plan: {path}") from exc + if not isinstance(payload, dict) or payload.get("format") != PLAN_FORMAT: + raise CursorAuthorityReconciliationError("unsupported reconciliation plan format") + digest = payload.get("plan_digest") + unsigned = dict(payload) + unsigned.pop("plan_digest", None) + if not isinstance(digest, str) or _canonical_digest(unsigned) != digest: + raise CursorAuthorityReconciliationError("reconciliation plan digest mismatch") + return payload + + +def _backup_root(manifest_path: Path) -> Path: + if manifest_path.is_dir(): + root = manifest_path + elif manifest_path.is_file() and manifest_path.name == "manifest.json": + root = manifest_path.parent + else: + raise CursorAuthorityReconciliationError( + "backup manifest must be manifest.json or a verified full-evidence backup directory" + ) + if not root.is_dir() or not (root / "manifest.json").is_file(): + raise CursorAuthorityReconciliationError("backup manifest must be a verified full-evidence backup directory") + return root + + +def _validate_backup(manifest_path: Path, plan: Mapping[str, object]) -> dict[str, object]: + root = _backup_root(manifest_path) + try: + manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8")) + receipt = json.loads((root / "verification-receipt.json").read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise CursorAuthorityReconciliationError("backup lacks a readable verification receipt") from exc + if not isinstance(manifest, dict) or manifest.get("profile") != "full_evidence": + raise CursorAuthorityReconciliationError("apply requires a full_evidence backup") + if not isinstance(receipt, dict) or receipt.get("verdict") != "success": + raise CursorAuthorityReconciliationError("backup verification receipt is not successful") + included = {str(item) for item in manifest.get("included_tiers", []) if isinstance(item, str)} + if {f"{tier}.db" for tier in _REQUIRED_TIERS} - included: + raise CursorAuthorityReconciliationError("full-evidence backup lacks source/index/ops/audit rollback evidence") + verification = receipt.get("verification") + required_verification = ("source_blobs_resolved", "index_attachment_blobs_resolved", "blob_inventory_exact") + if not isinstance(verification, dict) or any(verification.get(key) is not True for key in required_verification): + raise CursorAuthorityReconciliationError("backup lacks complete blob rollback evidence") + if not (root / "blob").is_dir() or not (root / "blob-inventory.json").is_file(): + raise CursorAuthorityReconciliationError("backup lacks blob rollback evidence") + archive_root = _archive_root() + location = ArchiveLocation.resolve(archive_root) + expected_active_index = plan.get("active_index") + if expected_active_index is not None and expected_active_index != _active_index_binding(archive_root): + raise CursorAuthorityReconciliationError("active index generation changed since planning") + try: + verify_verification_receipt( + receipt, + tier="source", + live_tier_path=location.configured_tier("source").configured_path, + ) + verify_verification_receipt( + receipt, + tier="user", + live_tier_path=location.configured_tier("user").configured_path, + ) + except BackupAttestationError as exc: + raise CursorAuthorityReconciliationError("backup verification receipt attestation is invalid") from exc + declared = manifest.get("tier_source_fingerprints") + expected = plan.get("tier_fingerprints") + if not isinstance(declared, dict) or not isinstance(expected, dict): + raise CursorAuthorityReconciliationError("plan or backup lacks tier fingerprints") + for tier in _REQUIRED_TIERS: + artifact = declared.get(f"{tier}.db") + expected_tier = expected.get(tier) + if not isinstance(artifact, dict) or not isinstance(expected_tier, dict): + raise CursorAuthorityReconciliationError(f"backup lacks {tier} fingerprint") + for key in ("size_bytes", "sha256", "user_version"): + if artifact.get(key) != expected_tier.get(key): + raise CursorAuthorityReconciliationError(f"backup {tier} fingerprint does not match the plan") + backup_tier = root / f"{tier}.db" + if not backup_tier.is_file(): + raise CursorAuthorityReconciliationError(f"backup {tier} tier is missing") + actual = _sqlite_snapshot(backup_tier) + if actual.get("sha256") != expected_tier.get("sha256") or actual.get("size_bytes") != expected_tier.get( + "size_bytes" + ): + raise CursorAuthorityReconciliationError(f"backup {tier} image does not match the plan fingerprint") + if tier == "index" and expected_active_index is not None: + source_fingerprint = artifact.get("path") + if ( + not isinstance(source_fingerprint, str) + or Path(source_fingerprint).resolve() != location.active_index_path.resolve() + ): + raise CursorAuthorityReconciliationError( + "backup index fingerprint does not bind the active index generation" + ) + return {"root": _path_identity(root), "manifest_sha256": _sha256_file(root / "manifest.json")} + + +def _quick_checks(root: Path) -> dict[str, list[str]]: + checks: dict[str, list[str]] = {} + location = ArchiveLocation.resolve(root) + for tier in ("source", "index", "ops", "audit"): + tier_path = location.active_index_path if tier == "index" else root / f"{tier}.db" + with closing(sqlite3.connect(f"file:{tier_path.resolve()}?mode=ro", uri=True)) as conn: + checks[tier] = [str(row[0]) for row in conn.execute("PRAGMA quick_check")] + if checks[tier] != ["ok"]: + raise CursorAuthorityReconciliationError(f"{tier}.db quick_check failed: {checks[tier]}") + return checks + + +def _write_atomic_json(path: Path, payload: Mapping[str, object], *, refuse_existing: bool) -> None: + if refuse_existing and path.exists(): + raise CursorAuthorityReconciliationError(f"output path already exists: {path}") + path.parent.mkdir(parents=True, exist_ok=True) + encoded = json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=False) + "\n" + descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent, text=True) + temporary_path = Path(temporary) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + handle.write(encoded) + handle.flush() + os.fsync(handle.fileno()) + if refuse_existing: + try: + os.link(temporary_path, path) + except FileExistsError as exc: + raise CursorAuthorityReconciliationError(f"output path already exists: {path}") from exc + temporary_path.unlink() + else: + os.replace(temporary_path, path) + directory_fd = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + finally: + temporary_path.unlink(missing_ok=True) + + +def _require_daemon_stopped(root: Path) -> None: + config = Config( + archive_root=root, + render_root=root / "render", + sources=[], + db_path=ArchiveLocation.resolve(root).active_index_path, + ) + from polylogue.maintenance.offline_guard import running_daemon_pid + + if running_daemon_pid(config) is not None: + raise CursorAuthorityReconciliationError("daemon must be stopped for cursor-authority reconciliation") + + +def build_reconciliation_plan(*, source_path_file: Path, output_plan: Path) -> dict[str, object]: + root = _archive_root() + _require_daemon_stopped(root) + source_path = _read_private_source_path(source_path_file) + plan = _build_plan(root, source_path) + _write_atomic_json(output_plan, plan, refuse_existing=True) + return plan + + +def _find_recovery_attempt( + root: Path, + source_path: Path, + plan_observed_at_ms: int, + *, + plan_digest: str, + path_digest: str, +) -> dict[str, object] | None: + with closing(sqlite3.connect(f"file:{(root / 'ops.db').resolve()}?mode=ro", uri=True)) as conn: + rows = conn.execute( + "SELECT attempt_id, status, source_path, source_paths_json, finished_at_ms, " + "outcome_code, retryable, diagnostic, remediation FROM ingest_attempts " + "ORDER BY COALESCE(finished_at_ms, heartbeat_at_ms, started_at_ms) DESC LIMIT 50" + ).fetchall() + event_rows = conn.execute( + "SELECT attempt_id, payload_json FROM daemon_stage_events " + "WHERE stage = 'planning' ORDER BY observed_at_ms DESC LIMIT 200" + ).fetchall() + planning_bindings: dict[str, bool] = {} + for attempt_id, payload_json in event_rows: + try: + payload = json.loads(str(payload_json)) + except (TypeError, ValueError): + continue + if not isinstance(payload, dict): + continue + planning_bindings[str(attempt_id)] = ( + payload.get("cursor_authority_plan_digest") == plan_digest + and payload.get("cursor_authority_path_digest") == path_digest + ) + for ( + attempt_id, + status, + single_path, + paths_json, + finished_at_ms, + outcome_code, + retryable, + diagnostic, + remediation, + ) in rows: + if str(status) not in {"completed", "completed_with_failures"}: + continue + if not isinstance(finished_at_ms, int) or finished_at_ms <= plan_observed_at_ms: + continue + if not planning_bindings.get(str(attempt_id), False): + continue + values: list[str] = [] + if isinstance(paths_json, str): + try: + decoded = json.loads(paths_json) + except ValueError: + decoded = [] + if isinstance(decoded, list): + values.extend(str(value) for value in decoded if isinstance(value, str)) + if not values and single_path: + values.append(str(single_path)) + if any(Path(value).resolve() == source_path.resolve() for value in values): + return { + "attempt_id": str(attempt_id), + "status": str(status), + "finished_at_ms": finished_at_ms, + "outcome_code": None if outcome_code is None else str(outcome_code), + "retryable": None if retryable is None else bool(retryable), + "diagnostic": None if diagnostic is None else str(diagnostic), + "remediation": None if remediation is None else str(remediation), + } + return None + + +async def _normal_ingest( + root: Path, source_path: Path, plan: Mapping[str, object] +) -> tuple[LiveBatchMetrics, dict[str, object]]: + from polylogue.sources.live import watcher as live_watcher + + async with Polylogue(archive_root=root, db_path=ArchiveLocation.resolve(root).active_index_path) as polylogue: + cursor = CursorStore(root / "ops.db", initialize=False, ops_db_path=root / "ops.db") + processor = LiveBatchProcessor( + polylogue, + (WatchSource(name=source_path.parent.name, root=source_path.parent),), + cursor=cursor, + parser_fingerprint=lambda: live_watcher._PARSER_FINGERPRINT, + ) + with scoped_cursor_authority_authorization( + source_path_digest=str(plan["selected_path_digest"]), + cursor_byte_offset=_plan_int(plan, "cursor_byte_offset"), + accepted_frontier=_plan_int(plan, "accepted_frontier"), + plan_digest=str(plan["plan_digest"]), + force_full_ingest=True, + ): + metrics = await processor.ingest_files([source_path], emit_event=False) + attempt = _find_recovery_attempt( + root, + source_path, + _plan_int(plan, "observed_at_ms"), + plan_digest=str(plan["plan_digest"]), + path_digest=str(plan["selected_path_digest"]), + ) + if attempt is None: + raise CursorAuthorityReconciliationError("completed ingest attempt lacks reconciliation planning binding") + return metrics, attempt + + +def _plan_int(plan: Mapping[str, object], key: str) -> int: + value = plan.get(key) + if isinstance(value, bool) or not isinstance(value, int): + raise CursorAuthorityReconciliationError(f"reconciliation plan field {key} is not an integer") + return value + + +def _required_nonnegative_int(value: object, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise CursorAuthorityReconciliationError(f"{field} is missing or invalid") + return value + + +def _before_projection(plan: Mapping[str, object]) -> dict[str, object]: + value = plan.get("before_projection") + if not isinstance(value, dict): + raise CursorAuthorityReconciliationError("plan lacks the before-projection authority census") + return value + + +def _same_plan_bindings(left: Mapping[str, object], right: Mapping[str, object]) -> bool: + def comparable(plan: Mapping[str, object]) -> dict[str, object]: + value = dict(plan) + value.pop("observed_at_ms", None) + value.pop("plan_digest", None) + return value + + return comparable(left) == comparable(right) + + +def _changed_rows() -> dict[str, int | None]: + return {"cursor": None, "accepted_head_direct_writes": 0} + + +def _typed_retryable_attempt(attempt: Mapping[str, object]) -> bool: + outcome_code = attempt.get("outcome_code") + retryable = attempt.get("retryable") + if not isinstance(outcome_code, str) or retryable is not True: + return False + try: + outcome = IngestOutcome(outcome_code) + except ValueError: + return False + return IngestAttemptDisposition(outcome=outcome).retryable is True + + +def _redacted_metrics(metrics: LiveBatchMetrics | None) -> dict[str, object] | None: + if metrics is None: + return None + payload = metrics.to_payload() + payload["failed_paths"] = [_path_identity(Path(str(path))) for path in metrics.failed_paths] + payload["new_sessions"] = [ + {"source_name_digest": _identity_digest(source_name), "session_id_digest": _identity_digest(session_id)} + for source_name, session_id in metrics.new_sessions + ] + payload["updated_sessions"] = [ + {"source_name_digest": _identity_digest(source_name), "session_id_digest": _identity_digest(session_id)} + for source_name, session_id in metrics.updated_sessions + ] + return payload + + +def _receipt_payload( + *, + plan: Mapping[str, object], + backup: Mapping[str, object], + root: Path, + verdict: str, + before_projection: Mapping[str, object], + after_projection: RawFrontierIntegrityProjection | None, + metrics: LiveBatchMetrics | None, + attempt_id: str | None, + attempt_observation: str, + evidence: Mapping[str, object], + tolerate_state_errors: bool = False, +) -> dict[str, object]: + try: + tier_fingerprints: object = _tier_snapshots(root) + quick_check: object = _quick_checks(root) + except Exception: + if not tolerate_state_errors: + raise + tier_fingerprints = None + quick_check = None + return { + "format": RECEIPT_FORMAT, + "verdict": verdict, + "archive_identity": { + "root": _path_identity(root), + "active_index": plan.get("active_index"), + }, + "plan_digest": plan["plan_digest"], + "backup": dict(backup), + "before_projection": dict(before_projection), + "after_projection": _private_projection(after_projection) if after_projection is not None else None, + "metrics": _redacted_metrics(metrics), + "changed_rows": _changed_rows(), + "ingest_attempt_id": attempt_id, + "ingest_attempt_observation": attempt_observation, + "operation": attempt_observation, + "code_sha": plan.get("code_sha"), + "deployed_package_sha": plan.get("deployed_package_sha"), + "tier_fingerprints": tier_fingerprints, + "quick_check": quick_check, + "evidence": dict(evidence), + } + + +def apply_reconciliation(*, plan_path: Path, backup_manifest: Path, receipt: Path) -> dict[str, object]: + plan = _load_plan(plan_path) + if plan.get("status") != "planned": + raise CursorAuthorityReconciliationError("only a planned one-path reconciliation can be applied") + root = _archive_root() + _require_daemon_stopped(root) + if receipt.exists(): + raise CursorAuthorityReconciliationError(f"output path already exists: {receipt}") + before_projection = _before_projection(plan) + owner = acquire_durable_archive_ownership(root, owner_id=f"cursor-authority-reconcile:{os.getpid()}") + with owner: + backup_evidence = _validate_backup(backup_manifest, plan) + current_path = _find_path_by_digest(root, str(plan["selected_path_digest"])) + try: + current_plan = _build_plan(root, current_path) + except CursorAuthorityReconciliationError: + current_plan = None + if current_plan is None or not _same_plan_bindings(current_plan, plan): + recovery_projection = _projection_for(root) + _require_healthy_projection_siblings(recovery_projection) + recovery_attempt = _find_recovery_attempt( + root, + current_path, + _plan_int(plan, "observed_at_ms"), + plan_digest=str(plan["plan_digest"]), + path_digest=str(plan["selected_path_digest"]), + ) + if recovery_attempt is None or recovery_projection.cursor_ahead_count != 0: + raise CursorAuthorityReconciliationError("plan bindings changed before archive ownership") + if recovery_projection.cursor_ahead_status != "healthy": + raise CursorAuthorityReconciliationError("recovery did not prove a healthy cursor frontier") + before_gap_count = before_projection.get("cursor_authority_gap_count") + if ( + not isinstance(before_gap_count, int) + or recovery_projection.cursor_authority_gap_count != before_gap_count + ): + raise CursorAuthorityReconciliationError( + "recovered ingest changed the pre-existing incomparable cursor population" + ) + recovered_receipt_payload = _receipt_payload( + plan=plan, + backup=backup_evidence, + root=root, + verdict="reconciled", + before_projection=before_projection, + after_projection=recovery_projection, + metrics=None, + attempt_id=str(recovery_attempt["attempt_id"]), + attempt_observation="observed", + evidence={ + "raw_frontier_worsening": False, + "invalid_ahead_reconciliation": False, + "changed_pre_existing_populations": False, + "attempt_outcome_code": recovery_attempt.get("outcome_code"), + }, + tolerate_state_errors=False, + ) + recovered_receipt_payload["receipt_digest"] = _canonical_digest(recovered_receipt_payload) + _write_atomic_json(receipt, recovered_receipt_payload, refuse_existing=True) + return recovered_receipt_payload + current_plan = _build_plan(root, current_path) + if not _same_plan_bindings(current_plan, plan): + raise CursorAuthorityReconciliationError("plan bindings changed after archive ownership") + metrics: LiveBatchMetrics | None = None + attempt: dict[str, object] | None = None + attempt_id = "unknown" + after_projection: RawFrontierIntegrityProjection | None = None + evidence: dict[str, object] = { + "raw_frontier_worsening": False, + "invalid_ahead_reconciliation": False, + "changed_pre_existing_populations": False, + } + try: + metrics, attempt = asyncio.run(_normal_ingest(root, current_path, plan)) + attempt_id = str(attempt["attempt_id"]) + after_projection = _projection_for(root) + _require_healthy_projection_siblings(after_projection) + if after_projection.broken_head_count or after_projection.missing_source_raw_count: + evidence["raw_frontier_worsening"] = True + raise CursorAuthorityReconciliationError("reconciliation introduced unrelated raw-frontier worsening") + if after_projection.cursor_ahead_count: + if ( + metrics.succeeded_file_count != 0 + or str(current_path) not in metrics.failed_paths + or metrics.time_budget_exceeded + or not _typed_retryable_attempt(attempt) + ): + evidence["invalid_ahead_reconciliation"] = True + raise CursorAuthorityReconciliationError( + "cursor-ahead postcondition lacks a typed retryable deferral outcome" + ) + verdict = "typed_deferred" + else: + if after_projection.cursor_ahead_status != "healthy": + raise CursorAuthorityReconciliationError("reconciliation did not prove a healthy cursor frontier") + verdict = "reconciled" + before_gap_count = before_projection.get("cursor_authority_gap_count") + if not isinstance(before_gap_count, int) or after_projection.cursor_authority_gap_count != before_gap_count: + evidence["changed_pre_existing_populations"] = True + raise CursorAuthorityReconciliationError( + "reconciliation changed the pre-existing incomparable cursor population" + ) + receipt_payload = _receipt_payload( + plan=plan, + backup=backup_evidence, + root=root, + verdict=verdict, + before_projection=before_projection, + after_projection=after_projection, + metrics=metrics, + attempt_id=attempt_id, + attempt_observation="performed", + evidence={**evidence, "attempt_outcome_code": attempt.get("outcome_code") if attempt else None}, + tolerate_state_errors=False, + ) + except Exception as exc: + if after_projection is None: + try: + after_projection = _projection_for(root) + except Exception: + after_projection = None + failure_payload = _receipt_payload( + plan=plan, + backup=backup_evidence, + root=root, + verdict="failed", + before_projection=before_projection, + after_projection=after_projection, + metrics=metrics, + attempt_id=attempt_id, + attempt_observation="performed", + evidence=evidence, + tolerate_state_errors=True, + ) + failure_message = str(exc).replace(str(root), "") + if "current_path" in locals(): + failure_message = failure_message.replace(str(current_path), f"") + failure_payload["error"] = {"type": type(exc).__name__, "message": failure_message} + failure_payload["receipt_digest"] = _canonical_digest(failure_payload) + _write_atomic_json(receipt, failure_payload, refuse_existing=True) + if isinstance(exc, CursorAuthorityReconciliationError): + raise + raise CursorAuthorityReconciliationError("cursor-authority reconciliation failed after ingest") from exc + receipt_payload["receipt_digest"] = _canonical_digest(receipt_payload) + _write_atomic_json(receipt, receipt_payload, refuse_existing=True) + return receipt_payload + + +__all__ = [ + "ARCHIVE_ROOT", + "PLAN_FORMAT", + "RECEIPT_FORMAT", + "CursorAuthorityReconciliationError", + "apply_reconciliation", + "build_reconciliation_plan", + "cursor_authority_path_digest", +] diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 51faf45cdc..cfd79ecb05 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -3,12 +3,13 @@ from __future__ import annotations import asyncio +import contextvars import os import sqlite3 import time import zipfile -from collections.abc import Awaitable, Callable, Iterable -from contextlib import closing +from collections.abc import Awaitable, Callable, Iterable, Iterator +from contextlib import closing, contextmanager from dataclasses import dataclass, field from datetime import UTC, datetime from hashlib import sha256 @@ -142,6 +143,7 @@ original_sqlite_source_path, snapshot_sqlite_to_blob, ) +from polylogue.storage.archive_identity import ArchiveLocation from polylogue.storage.blob_store import BlobStore from polylogue.storage.runtime import RawSessionRecord from polylogue.storage.sqlite.archive_tiers.archive import ActiveByteRevisionChainError @@ -163,6 +165,61 @@ class CursorAuthorityBlockedError(RuntimeError): """The canonical raw frontier proof did not authorize live source selection.""" +@dataclass(slots=True) +class CursorAuthorityAuthorization: + """Single-use, exact-use exception to the live cursor authority gate. + + This token is deliberately process-local and context-local. It is not a + switch, environment variable, or archive setting. The reconciliation + command creates one only after re-proving the selected path and frontier; + the first gate check consumes it. + """ + + source_path_digest: str + cursor_byte_offset: int + accepted_frontier: int + plan_digest: str + force_full_ingest: bool = False + consumed: bool = False + + +_CURSOR_AUTHORIZATION: contextvars.ContextVar[CursorAuthorityAuthorization | None] = contextvars.ContextVar( + "polylogue_cursor_authority_authorization", + default=None, +) + + +def cursor_authority_path_digest(path: Path) -> str: + """Digest one resolved source path without retaining its private text.""" + + return sha256(str(path.resolve()).encode("utf-8")).hexdigest() + + +@contextmanager +def scoped_cursor_authority_authorization( + *, + source_path_digest: str, + cursor_byte_offset: int, + accepted_frontier: int, + plan_digest: str, + force_full_ingest: bool = False, +) -> Iterator[None]: + """Install one exact-use authorization for the normal ingest route.""" + + authorization = CursorAuthorityAuthorization( + source_path_digest=source_path_digest, + cursor_byte_offset=cursor_byte_offset, + accepted_frontier=accepted_frontier, + plan_digest=plan_digest, + force_full_ingest=force_full_ingest, + ) + marker = _CURSOR_AUTHORIZATION.set(authorization) + try: + yield + finally: + _CURSOR_AUTHORIZATION.reset(marker) + + # polylogue-0jf4: known ~/.codex live SQLite state filenames, matched by name # first (cheap, no I/O) before the structural table-shape check in # ``codex_state.is_in_scope_codex_sqlite_path`` decides whether to acquire. @@ -507,16 +564,67 @@ def cursor_authority_block_reason(self) -> str | None: with raw convergence, recovery, and reindex. """ archive_root = Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent)) - if not (archive_root / "source.db").is_file() or not (archive_root / "index.db").is_file(): + if ( + not (archive_root / "source.db").is_file() + or not ArchiveLocation.resolve(archive_root).active_index_path.is_file() + ): return None from polylogue.readiness.capability import raw_frontier_source_selection_block_reason return raw_frontier_source_selection_block_reason(archive_root) - def require_cursor_authority(self) -> None: + def _consume_scoped_cursor_authority(self, paths: Iterable[Path]) -> CursorAuthorityAuthorization: + authorization = _CURSOR_AUTHORIZATION.get() + if authorization is None: + raise CursorAuthorityBlockedError("scoped cursor authority authorization is missing") + if authorization.consumed: + raise CursorAuthorityBlockedError("scoped cursor authority authorization was already consumed") + selected_paths = tuple(path.resolve() for path in paths) + if ( + len(selected_paths) != 1 + or cursor_authority_path_digest(selected_paths[0]) != authorization.source_path_digest + ): + raise CursorAuthorityBlockedError("scoped cursor authority authorization does not match the selected path") + archive_root = Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent)) + from polylogue.readiness.capability import raw_frontier_integrity_projection + from polylogue.storage.archive_readiness import raw_materialization_readiness_snapshot + + projection = raw_frontier_integrity_projection( + archive_root, + raw_materialization_readiness_snapshot(archive_root), + sample_limit=100, + ) + if ( + projection.overall_status != "violated" + or projection.broken_head_count + or projection.missing_source_raw_count + or projection.cursor_ahead_count != 1 + or len(projection.cursor_ahead_samples) != 1 + ): + raise CursorAuthorityBlockedError("scoped cursor authority no longer matches the global violation set") + sample = projection.cursor_ahead_samples[0] + if ( + cursor_authority_path_digest(Path(sample.source_path)) != authorization.source_path_digest + or sample.cursor_byte_offset != authorization.cursor_byte_offset + or sample.accepted_frontier != authorization.accepted_frontier + ): + raise CursorAuthorityBlockedError("scoped cursor authority frontier binding changed") + authorization.consumed = True + return authorization + + def require_cursor_authority(self, paths: Iterable[Path] | None = None) -> CursorAuthorityAuthorization | None: """Fail closed before a live batch can create attempts or write data.""" - if reason := self.cursor_authority_block_reason(): - raise CursorAuthorityBlockedError(f"live watcher source-selection gate blocked: {reason}") + reason = self.cursor_authority_block_reason() + authorization = _CURSOR_AUTHORIZATION.get() + if reason is None: + if authorization is not None: + raise CursorAuthorityBlockedError("scoped cursor authority authorization has no planned violation") + return None + if authorization is not None: + if paths is None: + raise CursorAuthorityBlockedError("scoped cursor authority requires an exact selected path") + return self._consume_scoped_cursor_authority(paths) + raise CursorAuthorityBlockedError(f"live watcher source-selection gate blocked: {reason}") async def ingest_files( self, @@ -528,7 +636,7 @@ async def ingest_files( max_pass_seconds: float | None = None, ) -> LiveBatchMetrics: """Ingest files in batch, run post-ingest convergence, and return metrics.""" - self.require_cursor_authority() + authorization = self.require_cursor_authority(paths) if is_fully_degraded(): # The daemon has been marked structurally unable to ingest (e.g. # schema mismatch detected at preflight or on the first batch). @@ -563,6 +671,14 @@ async def ingest_files( parse_time_s=0.0, convergence_time_s=0.0, total_time_s=0.0, + stage_payload=( + { + "cursor_authority_plan_digest": authorization.plan_digest, + "cursor_authority_path_digest": authorization.source_path_digest, + } + if authorization is not None + else None + ), ) source_payload_read_bytes = 0 cursor_fingerprint_read_bytes = 0 @@ -696,6 +812,9 @@ async def flush_append_plans() -> None: deferred_paths.append(plan.path) for path in paths: + if authorization is not None and authorization.force_full_ingest: + full_paths.append(path) + continue if is_fully_degraded(): full_paths.append(path) continue @@ -2110,7 +2229,9 @@ def _ingest_full_paths_sync( return result def _archive_active(self, archive_root: Path) -> bool: - return (archive_root / "index.db").exists() and (archive_root / "source.db").exists() + return ( + ArchiveLocation.resolve(archive_root).active_index_path.exists() and (archive_root / "source.db").exists() + ) def _archive_storage_probe_payload( self, @@ -2119,7 +2240,14 @@ def _archive_storage_probe_payload( archive_active: bool, archive_bootstrapped: bool, ) -> dict[str, object]: - tier_paths = {spec.tier.value: archive_root / spec.filename for spec in ARCHIVE_TIER_SPECS.values()} + tier_paths = { + spec.tier.value: ( + ArchiveLocation.resolve(archive_root).active_index_path + if spec.tier.value == "index" + else archive_root / spec.filename + ) + for spec in ARCHIVE_TIER_SPECS.values() + } present = [tier for tier, path in tier_paths.items() if path.exists()] missing = [tier for tier, path in tier_paths.items() if not path.exists()] user_versions: dict[str, int | None] = {} @@ -3417,7 +3545,7 @@ def _codex_session_meta_native_id(self, path: Path) -> str | None: def _archive_has_native_session(self, origin: str, native_id: str) -> bool: archive_root = Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent)) - index_db = archive_root / "index.db" + index_db = ArchiveLocation.resolve(archive_root).active_index_path if not index_db.exists(): return False try: @@ -3440,7 +3568,7 @@ def _archive_has_native_session(self, origin: str, native_id: str) -> bool: def _existing_archive_session_native_id(self, path: Path) -> str | None: archive_root = Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent)) - index_db = archive_root / "index.db" + index_db = ArchiveLocation.resolve(archive_root).active_index_path source_db = archive_root / "source.db" if not index_db.exists() or not source_db.exists(): return None @@ -3504,7 +3632,7 @@ def _compact_superseded_raw_snapshots(self, paths: list[Path]) -> None: archive_root = Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent)) source_db = archive_root / "source.db" - index_db = archive_root / "index.db" + index_db = ArchiveLocation.resolve(archive_root).active_index_path if not source_db.exists(): return with closing(sqlite3.connect(source_db)) as conn, conn: diff --git a/tests/unit/cli/test_archive_maintenance_cli.py b/tests/unit/cli/test_archive_maintenance_cli.py index b4ff8048a6..67b1397ede 100644 --- a/tests/unit/cli/test_archive_maintenance_cli.py +++ b/tests/unit/cli/test_archive_maintenance_cli.py @@ -1726,6 +1726,86 @@ def test_archive_maintenance_help_omits_copy_activation_surface(cli_runner: CliR assert removed not in result.output +def test_cursor_authority_reconcile_cli_exposes_only_scoped_inputs(cli_runner: CliRunner) -> None: + result = cli_runner.invoke( + cli, + ["--plain", "ops", "maintenance", "cursor-authority-reconcile", "--help"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "--source-path-file" in result.output + assert "--output-plan" in result.output + assert "--plan" in result.output + assert "--backup-manifest" in result.output + assert "--receipt" in result.output + assert "--apply" in result.output + assert "--force" not in result.output + assert "--bypass" not in result.output + + +def test_cursor_authority_reconcile_cli_accepts_verified_backup_directory( + cli_runner: CliRunner, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from polylogue.maintenance import cursor_authority_reconcile + + backup = tmp_path / "verified-backup" + backup.mkdir() + (backup / "manifest.json").write_text("{}", encoding="utf-8") + observed: dict[str, Path] = {} + + def fake_apply(*, plan_path: Path, backup_manifest: Path, receipt: Path) -> dict[str, object]: + observed["backup"] = backup_manifest + return {"verdict": "failed"} + + monkeypatch.setattr(cursor_authority_reconcile, "apply_reconciliation", fake_apply) + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "cursor-authority-reconcile", + "--apply", + "--plan", + str(tmp_path / "plan.json"), + "--backup-manifest", + str(backup), + "--receipt", + str(tmp_path / "receipt.json"), + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert observed["backup"] == backup + + +@pytest.mark.parametrize( + "args", + [ + ["--apply", "--backup-manifest", "backup", "--receipt", "receipt"], + ["--apply", "--plan", "plan", "--receipt", "receipt"], + ["--apply", "--plan", "plan", "--backup-manifest", "backup"], + ["--source-path-file", "source", "--output-plan", "plan", "--plan", "existing"], + ["--source-path-file", "source", "--output-plan", "plan", "--receipt", "receipt"], + ["--plan", "plan", "--backup-manifest", "backup", "--receipt", "receipt"], + ], +) +def test_cursor_authority_reconcile_cli_rejects_mixed_or_missing_mode_options( + cli_runner: CliRunner, + args: list[str], +) -> None: + result = cli_runner.invoke( + cli, + ["--plain", "ops", "maintenance", "cursor-authority-reconcile", *args], + catch_exceptions=False, + ) + + assert result.exit_code == 2 + assert "requires" in result.output or "accepts only" in result.output + + def test_raw_authority_frontier_cli_replaces_incident_specific_commands( cli_workspace: dict[str, Path], cli_runner: CliRunner, diff --git a/tests/unit/maintenance/test_cursor_authority_reconcile.py b/tests/unit/maintenance/test_cursor_authority_reconcile.py new file mode 100644 index 0000000000..4e5c16c843 --- /dev/null +++ b/tests/unit/maintenance/test_cursor_authority_reconcile.py @@ -0,0 +1,627 @@ +"""Real-route tests for cursor-authority reconciliation.""" + +from __future__ import annotations + +import json +import sqlite3 +from dataclasses import replace +from pathlib import Path + +import pytest + +from polylogue.maintenance import cursor_authority_reconcile as reconcile +from polylogue.sources.live.batch import CursorAuthorityBlockedError, scoped_cursor_authority_authorization +from polylogue.sources.live.metrics import LiveBatchMetrics + + +def _private_path_file(path: Path, source: Path) -> None: + path.write_text(f"{source}\n", encoding="utf-8") + path.chmod(0o600) + + +def test_dry_run_plan_is_deterministic_and_does_not_store_private_path( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from tests.unit.sources.test_live_watcher import _live_archive_snapshot, _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + monkeypatch.setattr(reconcile, "ARCHIVE_ROOT", tmp_path) + path_file = tmp_path / "selected-path" + _private_path_file(path_file, source_path) + before = _live_archive_snapshot(tmp_path) + + first = reconcile.build_reconciliation_plan(source_path_file=path_file, output_plan=tmp_path / "plan-1.json") + second = reconcile.build_reconciliation_plan(source_path_file=path_file, output_plan=tmp_path / "plan-2.json") + + assert first | {"observed_at_ms": None, "plan_digest": None} == second | { + "observed_at_ms": None, + "plan_digest": None, + } + assert isinstance(first["observed_at_ms"], int) + assert first["format"] == reconcile.PLAN_FORMAT + assert str(source_path) not in json.dumps(first, sort_keys=True) + assert _live_archive_snapshot(tmp_path) == before + watcher.stop() + + +@pytest.mark.asyncio +async def test_apply_authorization_invokes_normal_full_ingest_route( + tmp_path: Path, +) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path, force_full_fallback=True) + projection = reconcile._projection_for(tmp_path) + sample = projection.cursor_ahead_samples[0] + processor._append_plan = lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("append planning called")) # type: ignore[method-assign] + with scoped_cursor_authority_authorization( + source_path_digest=reconcile.cursor_authority_path_digest(source_path), + cursor_byte_offset=sample.cursor_byte_offset, + accepted_frontier=sample.accepted_frontier, + plan_digest="test-plan", + force_full_ingest=True, + ): + metrics = await processor.ingest_files([source_path], emit_event=False) + + assert metrics.full_file_count == 1 + assert metrics.succeeded_file_count == 1 + watcher.stop() + + +@pytest.mark.asyncio +async def test_scoped_authorization_rejects_a_different_path_without_mutation(tmp_path: Path) -> None: + from tests.unit.sources.test_live_watcher import _live_archive_snapshot, _seed_live_cursor_authority_case + + processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + other_path = source_path.with_name("other.jsonl") + other_path.write_bytes(source_path.read_bytes()) + projection = reconcile._projection_for(tmp_path) + sample = projection.cursor_ahead_samples[0] + before = _live_archive_snapshot(tmp_path) + + with scoped_cursor_authority_authorization( + source_path_digest=reconcile.cursor_authority_path_digest(source_path), + cursor_byte_offset=sample.cursor_byte_offset, + accepted_frontier=sample.accepted_frontier, + plan_digest="test-plan", + ): + with pytest.raises(CursorAuthorityBlockedError, match="selected path"): + await processor.ingest_files([other_path], emit_event=False) + + assert _live_archive_snapshot(tmp_path) == before + watcher.stop() + + +def test_plan_refuses_overwrite(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + monkeypatch.setattr(reconcile, "ARCHIVE_ROOT", tmp_path) + path_file = tmp_path / "selected-path" + _private_path_file(path_file, source_path) + output = tmp_path / "plan.json" + reconcile.build_reconciliation_plan(source_path_file=path_file, output_plan=output) + + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="already exists"): + reconcile.build_reconciliation_plan(source_path_file=path_file, output_plan=output) + watcher.stop() + + +def test_private_path_file_requires_exact_permissions_and_absolute_path(tmp_path: Path) -> None: + source = tmp_path / "source.jsonl" + source.write_text("{}\n", encoding="utf-8") + path_file = tmp_path / "selected-path" + path_file.write_text(str(source) + "\n", encoding="utf-8") + path_file.chmod(0o644) + + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="mode 0600"): + reconcile._read_private_source_path(path_file) + + path_file.chmod(0o600) + path_file.write_text("relative.jsonl\n", encoding="utf-8") + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="absolute"): + reconcile._read_private_source_path(path_file) + + +def test_plan_digest_tampering_is_rejected(tmp_path: Path) -> None: + plan_path = tmp_path / "plan.json" + plan_path.write_text( + json.dumps({"format": reconcile.PLAN_FORMAT, "status": "planned", "plan_digest": "wrong"}), + encoding="utf-8", + ) + + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="digest mismatch"): + reconcile._load_plan(plan_path) + + +@pytest.mark.asyncio +async def test_scoped_authorization_is_single_use(tmp_path: Path) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + projection = reconcile._projection_for(tmp_path) + sample = projection.cursor_ahead_samples[0] + authorization = scoped_cursor_authority_authorization( + source_path_digest=reconcile.cursor_authority_path_digest(source_path), + cursor_byte_offset=sample.cursor_byte_offset, + accepted_frontier=sample.accepted_frontier, + plan_digest="test-plan", + ) + with authorization: + processor.require_cursor_authority([source_path]) + with pytest.raises(CursorAuthorityBlockedError, match="already consumed"): + processor.require_cursor_authority([source_path]) + watcher.stop() + + +def test_planner_preserves_incomparable_population(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + projection = replace(reconcile._projection_for(tmp_path), cursor_authority_gap_count=727) + monkeypatch.setattr(reconcile, "_projection_for", lambda root: projection) + plan = reconcile._build_plan(tmp_path, source_path) + before_projection = plan["before_projection"] + assert isinstance(before_projection, dict) + assert before_projection["cursor_authority_gap_count"] == 727 + watcher.stop() + + +def test_planner_refuses_multiple_true_ahead_rows(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + projection = replace(reconcile._projection_for(tmp_path), cursor_ahead_count=2) + monkeypatch.setattr(reconcile, "_projection_for", lambda root: projection) + + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="multiple"): + reconcile._build_plan(tmp_path, source_path) + watcher.stop() + + +def test_planner_refuses_unavailable_healthy_sibling_projection( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + projection = replace(reconcile._projection_for(tmp_path), missing_source_raw_status="unknown") + monkeypatch.setattr(reconcile, "_projection_for", lambda root: projection) + + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="sibling"): + reconcile._build_plan(tmp_path, source_path) + watcher.stop() + + +def test_wal_effective_snapshot_matches_sqlite_backup(tmp_path: Path) -> None: + live = tmp_path / "live.db" + backup = tmp_path / "backup.db" + with sqlite3.connect(live) as conn: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("CREATE TABLE values_table (value TEXT)") + conn.execute("INSERT INTO values_table VALUES ('wal-frame')") + conn.commit() + live_snapshot = reconcile._sqlite_snapshot(live) + with sqlite3.connect(backup) as backup_conn: + conn.backup(backup_conn) + backup_conn.commit() + assert live_snapshot == reconcile._sqlite_snapshot(backup) + + +def test_planner_rejects_source_mutation_during_prefix_hash( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + observations = iter(((1, 2, 3, 4, 5), (1, 2, 3, 4, 6))) + monkeypatch.setattr(reconcile, "_stat_observation", lambda path: next(observations)) + + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="mutated"): + reconcile._build_plan(tmp_path, source_path) + watcher.stop() + + +@pytest.mark.asyncio +async def test_scoped_authorization_rejects_changed_cursor_frontier(tmp_path: Path) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + processor, watcher, cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + projection = reconcile._projection_for(tmp_path) + sample = projection.cursor_ahead_samples[0] + with scoped_cursor_authority_authorization( + source_path_digest=reconcile.cursor_authority_path_digest(source_path), + cursor_byte_offset=sample.cursor_byte_offset, + accepted_frontier=sample.accepted_frontier, + plan_digest="test-plan", + ): + with sqlite3.connect(cursor._db_path) as conn: + conn.execute( + "UPDATE ingest_cursor SET byte_offset = byte_offset + 1 WHERE source_path = ?", + (str(source_path),), + ) + with pytest.raises(CursorAuthorityBlockedError, match="frontier binding"): + processor.require_cursor_authority([source_path]) + watcher.stop() + + +@pytest.mark.asyncio +async def test_scoped_authorization_cannot_turn_a_healthy_archive_into_an_exception( + tmp_path: Path, +) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path, exact_frontier=True) + with scoped_cursor_authority_authorization( + source_path_digest=reconcile.cursor_authority_path_digest(source_path), + cursor_byte_offset=0, + accepted_frontier=0, + plan_digest="test-plan", + ): + with pytest.raises(CursorAuthorityBlockedError, match="no planned violation"): + processor.require_cursor_authority([source_path]) + watcher.stop() + + +def test_backup_validation_requires_blob_rollback_evidence(tmp_path: Path) -> None: + backup = tmp_path / "backup" + backup.mkdir() + (backup / "manifest.json").write_text( + json.dumps( + { + "profile": "full_evidence", + "included_tiers": ["source.db", "index.db", "ops.db", "audit.db"], + } + ), + encoding="utf-8", + ) + (backup / "verification-receipt.json").write_text(json.dumps({"verdict": "success"}), encoding="utf-8") + + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="blob rollback"): + reconcile._validate_backup(backup, {}) + + +def test_private_projection_redacts_paths_and_preserves_missing_sample_branches(tmp_path: Path) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + projection = reconcile._projection_for(tmp_path) + private = reconcile._private_projection(projection) + samples = private["cursor_ahead_samples"] + assert isinstance(samples, list) + sample = samples[0] + assert isinstance(sample, dict) + original = projection.cursor_ahead_samples[0] + assert sample["source_path"] == reconcile.cursor_authority_path_digest(source_path) + assert sample["logical_source_key"] == reconcile._identity_digest(original.logical_source_key) + watcher.stop() + + +def test_recovery_attempt_requires_a_later_completed_observation(tmp_path: Path) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + with sqlite3.connect(cursor._db_path) as conn: + conn.execute( + "INSERT INTO ingest_attempts " + "(attempt_id, status, started_at_ms, finished_at_ms, source_paths_json) " + "VALUES (?, 'completed', ?, ?, ?)", + ("attempt-old", 100, 150, json.dumps([str(source_path)])), + ) + conn.execute( + "INSERT INTO ingest_attempts " + "(attempt_id, status, started_at_ms, finished_at_ms, source_paths_json) " + "VALUES (?, 'completed', ?, ?, ?)", + ("attempt-new", 200, 250, json.dumps([str(source_path)])), + ) + plan_digest = "plan-digest" + path_digest = reconcile.cursor_authority_path_digest(source_path) + with sqlite3.connect(cursor._db_path) as conn: + conn.execute( + "INSERT INTO daemon_stage_events " + "(event_id, attempt_id, stage, status, observed_at_ms, payload_json) VALUES (?, ?, 'planning', 'completed', ?, ?)", + ( + "event-new-plan", + "attempt-new", + 210, + json.dumps( + { + "cursor_authority_plan_digest": plan_digest, + "cursor_authority_path_digest": path_digest, + } + ), + ), + ) + assert ( + reconcile._find_recovery_attempt( + tmp_path, source_path, 150, plan_digest="unrelated-plan", path_digest=path_digest + ) + is None + ) + assert ( + reconcile._find_recovery_attempt(tmp_path, source_path, 250, plan_digest=plan_digest, path_digest=path_digest) + is None + ) + observed = reconcile._find_recovery_attempt( + tmp_path, source_path, 150, plan_digest=plan_digest, path_digest=path_digest + ) + assert observed is not None + assert observed["attempt_id"] == "attempt-new" + watcher.stop() + + +def test_backup_validation_rehashes_and_rejects_mismatched_tier( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + backup = tmp_path / "backup" + backup.mkdir() + (backup / "blob").mkdir() + (backup / "blob-inventory.json").write_text("{}", encoding="utf-8") + tiers: dict[str, dict[str, object]] = {} + for tier in ("source", "index", "ops", "audit"): + path = backup / f"{tier}.db" + with sqlite3.connect(path) as conn: + conn.execute("CREATE TABLE marker (value TEXT)") + tiers[tier] = reconcile._sqlite_snapshot(path) + plan = {"tier_fingerprints": tiers} + manifest = { + "profile": "full_evidence", + "included_tiers": [f"{tier}.db" for tier in tiers], + "tier_source_fingerprints": {f"{tier}.db": value for tier, value in tiers.items()}, + } + (backup / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + (backup / "verification-receipt.json").write_text( + json.dumps( + { + "verdict": "success", + "verification": { + "source_blobs_resolved": True, + "index_attachment_blobs_resolved": True, + "blob_inventory_exact": True, + }, + } + ), + encoding="utf-8", + ) + monkeypatch.setattr(reconcile, "ARCHIVE_ROOT", tmp_path) + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="attestation"): + reconcile._validate_backup(backup, plan) + monkeypatch.setattr(reconcile, "verify_verification_receipt", lambda *args, **kwargs: None) + validated = reconcile._validate_backup(backup, plan) + assert isinstance(validated["root"], dict) + assert validated["root"]["basename"] == backup.name + (backup / "audit.db").unlink() + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="tier is missing"): + reconcile._validate_backup(backup, plan) + with sqlite3.connect(backup / "audit.db") as conn: + conn.execute("CREATE TABLE marker (value TEXT)") + with sqlite3.connect(backup / "source.db") as conn: + conn.execute("INSERT INTO marker VALUES ('tampered')") + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="image does not match"): + reconcile._validate_backup(backup, plan) + + +def _deferred_metrics(source_path: Path) -> LiveBatchMetrics: + return LiveBatchMetrics( + queued_file_count=1, + needed_file_count=1, + skipped_file_count=0, + succeeded_file_count=0, + failed_file_count=1, + source_group_count=1, + input_bytes=1, + source_payload_read_bytes=1, + cursor_fingerprint_read_bytes=1, + ingest_worker_count_max=1, + append_file_count=0, + full_file_count=1, + archive_bytes_before=1, + archive_bytes_after=1, + archive_write_bytes_delta=0, + parse_time_s=0.0, + convergence_time_s=0.0, + total_time_s=0.0, + failed_paths=[str(source_path)], + ) + + +def test_typed_deferred_apply_receipt_is_metric_backed_and_does_not_claim_cursor_change( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + plan = reconcile._build_plan(tmp_path, source_path) + projection = reconcile._projection_for(tmp_path) + receipt_path = tmp_path / "receipt.json" + plan_path = tmp_path / "plan.json" + plan_path.write_text(json.dumps(plan), encoding="utf-8") + monkeypatch.setattr(reconcile, "ARCHIVE_ROOT", tmp_path) + monkeypatch.setattr(reconcile, "_require_daemon_stopped", lambda root: None) + monkeypatch.setattr(reconcile, "_validate_backup", lambda manifest, plan: {"root": "backup"}) + monkeypatch.setattr(reconcile, "_find_path_by_digest", lambda root, digest: source_path) + monkeypatch.setattr(reconcile, "_build_plan", lambda root, path: plan) + + async def deferred_ingest( + root: Path, path: Path, plan_payload: dict[str, object] + ) -> tuple[LiveBatchMetrics, dict[str, object]]: + return _deferred_metrics(path), { + "attempt_id": "attempt-deferred", + "outcome_code": "transient_error", + "retryable": True, + } + + monkeypatch.setattr(reconcile, "_normal_ingest", deferred_ingest) + monkeypatch.setattr(reconcile, "_projection_for", lambda root: projection) + monkeypatch.setattr(reconcile, "_tier_snapshots", lambda root: {}) + monkeypatch.setattr(reconcile, "_quick_checks", lambda root: {}) + + result = reconcile.apply_reconciliation( + plan_path=plan_path, + backup_manifest=tmp_path / "backup", + receipt=receipt_path, + ) + + assert result["verdict"] == "typed_deferred" + metrics = result["metrics"] + assert isinstance(metrics, dict) + assert metrics["failed_paths"] == [ + {"path_digest": reconcile.cursor_authority_path_digest(source_path), "basename": source_path.name} + ] + changed_rows = result["changed_rows"] + assert isinstance(changed_rows, dict) + assert changed_rows["cursor"] is None + assert result["ingest_attempt_observation"] == "performed" + watcher.stop() + + +def test_observed_recovery_receipt_does_not_claim_local_cursor_mutation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + plan = reconcile._build_plan(tmp_path, source_path) + projection = replace( + reconcile._projection_for(tmp_path), + overall_status="healthy", + cursor_ahead_status="healthy", + cursor_ahead_count=0, + cursor_ahead_samples=(), + ) + receipt_path = tmp_path / "receipt.json" + plan_path = tmp_path / "plan.json" + plan_path.write_text(json.dumps(plan), encoding="utf-8") + monkeypatch.setattr(reconcile, "ARCHIVE_ROOT", tmp_path) + monkeypatch.setattr(reconcile, "_require_daemon_stopped", lambda root: None) + monkeypatch.setattr(reconcile, "_validate_backup", lambda manifest, plan: {"root": "backup"}) + monkeypatch.setattr(reconcile, "_find_path_by_digest", lambda root, digest: source_path) + monkeypatch.setattr(reconcile, "_build_plan", lambda root, path: None) + monkeypatch.setattr( + reconcile, + "_find_recovery_attempt", + lambda root, path, observed, **kwargs: { + "attempt_id": "external-attempt", + "outcome_code": "success", + "retryable": False, + }, + ) + monkeypatch.setattr(reconcile, "_projection_for", lambda root: projection) + monkeypatch.setattr(reconcile, "_tier_snapshots", lambda root: {}) + monkeypatch.setattr(reconcile, "_quick_checks", lambda root: {}) + + result = reconcile.apply_reconciliation( + plan_path=plan_path, + backup_manifest=tmp_path / "backup", + receipt=receipt_path, + ) + + assert result["verdict"] == "reconciled" + assert result["ingest_attempt_observation"] == "observed" + changed_rows = result["changed_rows"] + assert isinstance(changed_rows, dict) + assert changed_rows["cursor"] is None + assert result["metrics"] is None + watcher.stop() + + +def test_unexpected_post_ingest_failure_writes_typed_audit_receipt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + plan = reconcile._build_plan(tmp_path, source_path) + plan_path = tmp_path / "plan.json" + plan_path.write_text(json.dumps(plan), encoding="utf-8") + projection = replace(reconcile._projection_for(tmp_path), broken_head_count=1) + receipt_path = tmp_path / "receipt.json" + monkeypatch.setattr(reconcile, "ARCHIVE_ROOT", tmp_path) + monkeypatch.setattr(reconcile, "_require_daemon_stopped", lambda root: None) + monkeypatch.setattr(reconcile, "_validate_backup", lambda manifest, plan: {"root": "backup"}) + monkeypatch.setattr(reconcile, "_find_path_by_digest", lambda root, digest: source_path) + monkeypatch.setattr(reconcile, "_build_plan", lambda root, path: plan) + + async def ingest( + root: Path, path: Path, plan_payload: dict[str, object] + ) -> tuple[LiveBatchMetrics, dict[str, object]]: + return _deferred_metrics(path), { + "attempt_id": "attempt-failed", + "outcome_code": "legacy_unknown", + "retryable": None, + } + + monkeypatch.setattr(reconcile, "_normal_ingest", ingest) + monkeypatch.setattr(reconcile, "_projection_for", lambda root: projection) + monkeypatch.setattr(reconcile, "_tier_snapshots", lambda root: {}) + monkeypatch.setattr(reconcile, "_quick_checks", lambda root: {}) + + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="raw-frontier worsening"): + reconcile.apply_reconciliation(plan_path=plan_path, backup_manifest=tmp_path / "backup", receipt=receipt_path) + + payload = json.loads(receipt_path.read_text(encoding="utf-8")) + assert payload["verdict"] == "failed" + assert payload["metrics"]["failed_paths"] == [ + {"path_digest": reconcile.cursor_authority_path_digest(source_path), "basename": source_path.name} + ] + assert payload["evidence"]["raw_frontier_worsening"] is True + watcher.stop() + + +def test_receipt_integrity_evidence_is_required_for_success_but_failure_is_durable( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + plan = {"plan_digest": "plan", "active_index": None} + + def unavailable(root: Path) -> dict[str, dict[str, object]]: + raise reconcile.CursorAuthorityReconciliationError("state unavailable") + + monkeypatch.setattr(reconcile, "_tier_snapshots", unavailable) + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="state unavailable"): + reconcile._receipt_payload( + plan=plan, + backup={}, + root=tmp_path, + verdict="reconciled", + before_projection={}, + after_projection=None, + metrics=None, + attempt_id="attempt", + attempt_observation="performed", + evidence={}, + tolerate_state_errors=False, + ) + failed = reconcile._receipt_payload( + plan=plan, + backup={}, + root=tmp_path, + verdict="failed", + before_projection={}, + after_projection=None, + metrics=None, + attempt_id="attempt", + attempt_observation="performed", + evidence={}, + tolerate_state_errors=True, + ) + assert failed["tier_fingerprints"] is None + assert failed["quick_check"] is None + + +def test_head_details_reports_missing_cursor_as_typed_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from tests.unit.sources.test_live_watcher import _seed_live_cursor_authority_case + + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + projection = reconcile._projection_for(tmp_path) + monkeypatch.setattr(reconcile, "_cursor_rows", lambda root: []) + with pytest.raises(reconcile.CursorAuthorityReconciliationError, match="unique current cursor row"): + reconcile._head_details(tmp_path, source_path, projection) + watcher.stop() diff --git a/tests/unit/sources/test_live_watcher.py b/tests/unit/sources/test_live_watcher.py index 9eab5b55ba..7f508c3175 100644 --- a/tests/unit/sources/test_live_watcher.py +++ b/tests/unit/sources/test_live_watcher.py @@ -6,6 +6,7 @@ import asyncio import contextlib import json +import shutil import sqlite3 import time import zipfile @@ -341,6 +342,48 @@ async def test_live_watcher_allows_append_at_authoritative_frontier(tmp_path: Pa watcher.stop() +@pytest.mark.asyncio +async def test_active_index_pointer_keeps_shadow_index_unmodified(tmp_path: Path) -> None: + from polylogue.maintenance import cursor_authority_reconcile as reconcile + from polylogue.maintenance.cursor_authority_reconcile import cursor_authority_path_digest + from polylogue.sources.live.batch import scoped_cursor_authority_authorization + + processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + shadow_index = tmp_path / "index.db" + active_index = tmp_path / "generations" / "active" / "index.db" + active_index.parent.mkdir(parents=True) + shutil.copy2(shadow_index, active_index) + (tmp_path / ".index-active-pointer").write_text(f"{active_index}\n", encoding="utf-8") + projection = reconcile._projection_for(tmp_path) + sample = projection.cursor_ahead_samples[0] + shadow_before = shadow_index.read_bytes() + with scoped_cursor_authority_authorization( + source_path_digest=cursor_authority_path_digest(source_path), + cursor_byte_offset=sample.cursor_byte_offset, + accepted_frontier=sample.accepted_frontier, + plan_digest="active-index-test", + force_full_ingest=True, + ): + metrics = await processor.ingest_files([source_path], emit_event=False) + + assert metrics.full_file_count == 1 + assert shadow_index.read_bytes() == shadow_before + watcher.stop() + + +@pytest.mark.asyncio +async def test_cursor_authority_seam_blocks_normal_live_route_before_writes(tmp_path: Path) -> None: + """The exact selector exercises the production live authority seam.""" + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path) + before = _live_archive_snapshot(tmp_path) + + with pytest.raises(CursorAuthorityBlockedError, match="source-selection gate blocked"): + await watcher._ingest_files([source_path]) + + assert _live_archive_snapshot(tmp_path) == before + watcher.stop() + + def test_live_ingest_metrics_log_separates_read_bytes_from_candidate_size( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -2944,6 +2987,17 @@ def test_parser_fingerprint_change_triggers_reingest(tmp_path: Path, monkeypatch assert record.parser_fingerprint == "live-batched-v3" +def test_live_batch_processor_observes_dynamic_parser_fingerprint( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "src" + root.mkdir() + watcher, _parse_sources = _make_watcher(tmp_path, root) + monkeypatch.setattr(live_watcher, "_PARSER_FINGERPRINT", "live-batched-dynamic-test") + + assert watcher._batch_processor._current_parser_fingerprint() == "live-batched-dynamic-test" + + def test_truncate_rewrite_triggers_reingest(tmp_path: Path) -> None: root = tmp_path / "src" root.mkdir() diff --git a/tests/unit/storage/test_raw_retention.py b/tests/unit/storage/test_raw_retention.py index 558510a378..60011c82ca 100644 --- a/tests/unit/storage/test_raw_retention.py +++ b/tests/unit/storage/test_raw_retention.py @@ -3,6 +3,7 @@ from __future__ import annotations import sqlite3 +from contextlib import closing from pathlib import Path import pytest @@ -756,70 +757,71 @@ def test_superseded_raw_snapshot_cleanup_keeps_newest_per_source(tmp_path: Path) source.write_text('{"type":"message"}\n', encoding="utf-8") blob_store = BlobStore(tmp_path / "blob") - conn = sqlite3.connect(db_path) - conn.row_factory = sqlite3.Row - conn.execute("PRAGMA foreign_keys = ON") - _ensure_archive_source_schema(conn) - - full_old, full_old_size = _write_blob(blob_store, b"full-old") - full_new, full_new_size = _write_blob(blob_store, b"full-new") - append_old, append_old_size = _write_blob(blob_store, b"append-old") - append_current, append_current_size = _write_blob(blob_store, b"append-current") - leased_old, leased_old_size = _write_blob(blob_store, b"leased-old") - missing_old, missing_old_size = _write_blob(blob_store, b"missing-old") - missing_new, missing_new_size = _write_blob(blob_store, b"missing-new") - - # Archive file-set retention ranks snapshots by recency, but callers must - # protect raw rows still referenced by index.db sessions before deleting. - def _seed(raw_id: str, source_path: Path, source_index: int, blob_size: int, acquired_at_ms: int) -> None: - _insert_archive_raw_session( - conn, - raw_id=raw_id, - source_path=source_path, - source_index=source_index, - blob_hash=raw_id, - blob_size=blob_size, - acquired_at_ms=acquired_at_ms, - ) - - _seed(full_old, source, 0, full_old_size, 1_000) - _seed(full_new, source, 0, full_new_size, 2_000) - _seed(append_old, source, -1, append_old_size, 3_000) - _seed(append_current, source, -1, append_current_size, 4_000) - _seed(leased_old, source, -1, leased_old_size, 2_500) - _seed(missing_old, missing_source, 0, missing_old_size, 1_000) - _seed(missing_new, missing_source, 0, missing_new_size, 2_000) - conn.commit() - - # full_old (superseded by full_new) and append_old + leased_old (superseded - # by append_current). missing_old is superseded too, but its source file is - # gone, so it is excluded from candidates. - candidates = superseded_raw_snapshot_candidates(conn, limit=100) - assert {candidate.raw_id for candidate in candidates} == {full_old, append_old, leased_old} - - dry_run = cleanup_superseded_raw_snapshots(conn, dry_run=True, blob_store=blob_store) - assert dry_run.candidate_count == 3 - assert blob_store.exists(full_old) - assert blob_store.exists(append_old) - assert blob_store.exists(leased_old) + with closing(sqlite3.connect(db_path)) as conn: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + _ensure_archive_source_schema(conn) + full_old, full_old_size = _write_blob(blob_store, b"full-old") + full_new, full_new_size = _write_blob(blob_store, b"full-new") + append_old, append_old_size = _write_blob(blob_store, b"append-old") + append_current, append_current_size = _write_blob(blob_store, b"append-current") + leased_old, leased_old_size = _write_blob(blob_store, b"leased-old") + missing_old, missing_old_size = _write_blob(blob_store, b"missing-old") + missing_new, missing_new_size = _write_blob(blob_store, b"missing-new") + + # Archive file-set retention ranks snapshots by recency, but callers must + # protect raw rows still referenced by index.db sessions before deleting. + def _seed(raw_id: str, source_path: Path, source_index: int, blob_size: int, acquired_at_ms: int) -> None: + _insert_archive_raw_session( + conn, + raw_id=raw_id, + source_path=source_path, + source_index=source_index, + blob_hash=raw_id, + blob_size=blob_size, + acquired_at_ms=acquired_at_ms, + ) + + _seed(full_old, source, 0, full_old_size, 1_000) + _seed(full_new, source, 0, full_new_size, 2_000) + _seed(append_old, source, -1, append_old_size, 3_000) + _seed(append_current, source, -1, append_current_size, 4_000) + _seed(leased_old, source, -1, leased_old_size, 2_500) + _seed(missing_old, missing_source, 0, missing_old_size, 1_000) + _seed(missing_new, missing_source, 0, missing_new_size, 2_000) + conn.commit() - result = cleanup_superseded_raw_snapshots(conn, dry_run=False, blob_store=blob_store) - assert result.deleted_raw_count == 3 - assert result.deleted_blob_count == 3 - assert not blob_store.exists(full_old) - assert not blob_store.exists(append_old) - assert not blob_store.exists(leased_old) - assert blob_store.exists(full_new) - assert blob_store.exists(append_current) - assert blob_store.exists(missing_old) - assert blob_store.exists(missing_new) - - remaining_raw_ids = { - str(row[0]) for row in conn.execute("SELECT raw_id FROM raw_sessions ORDER BY raw_id").fetchall() - } - assert remaining_raw_ids == {full_new, append_current, missing_old, missing_new} - remaining_ref_ids = {str(row[0]) for row in conn.execute("SELECT ref_id FROM blob_refs ORDER BY ref_id").fetchall()} - assert remaining_ref_ids == {full_new, append_current, missing_old, missing_new} + # full_old (superseded by full_new) and append_old + leased_old (superseded + # by append_current). missing_old is superseded too, but its source file is + # gone, so it is excluded from candidates. + candidates = superseded_raw_snapshot_candidates(conn, limit=100) + assert {candidate.raw_id for candidate in candidates} == {full_old, append_old, leased_old} + + dry_run = cleanup_superseded_raw_snapshots(conn, dry_run=True, blob_store=blob_store) + assert dry_run.candidate_count == 3 + assert blob_store.exists(full_old) + assert blob_store.exists(append_old) + assert blob_store.exists(leased_old) + + result = cleanup_superseded_raw_snapshots(conn, dry_run=False, blob_store=blob_store) + assert result.deleted_raw_count == 3 + assert result.deleted_blob_count == 3 + assert not blob_store.exists(full_old) + assert not blob_store.exists(append_old) + assert not blob_store.exists(leased_old) + assert blob_store.exists(full_new) + assert blob_store.exists(append_current) + assert blob_store.exists(missing_old) + assert blob_store.exists(missing_new) + + remaining_raw_ids = { + str(row[0]) for row in conn.execute("SELECT raw_id FROM raw_sessions ORDER BY raw_id").fetchall() + } + assert remaining_raw_ids == {full_new, append_current, missing_old, missing_new} + remaining_ref_ids = { + str(row[0]) for row in conn.execute("SELECT ref_id FROM blob_refs ORDER BY ref_id").fetchall() + } + assert remaining_ref_ids == {full_new, append_current, missing_old, missing_new} def test_superseded_raw_snapshot_cleanup_preserves_index_referenced_raws(tmp_path: Path) -> None: