diff --git a/devtools/command_catalog.py b/devtools/command_catalog.py index d35f8d1bcb..d27548f5e4 100644 --- a/devtools/command_catalog.py +++ b/devtools/command_catalog.py @@ -1511,6 +1511,29 @@ def to_dict(self) -> dict[str, object]: "--reason 'tracked in polylogue-xxxx, not fixed inline' --ref polylogue-xxxx", ), ), + CommandSpec( + "lab policy raw-authority-frontier-executability", + "verification lab", + "Verify every raw-authority frontier state has a reachable actuator.", + "devtools.verify_raw_authority_frontier_executability", + use_when=( + "polylogue-w32w / polylogue-lb39z (Phase 1, item 4): " + "RawAuthorityFrontierItem.__post_init__ raises if a CONSTRUCTED item pairs a " + "dispatched actuator (_APPLY_DISPATCHED_ACTUATORS) with a non-executable state " + "(_EXECUTABLE_STATES) -- but that only fires when a test or live classification " + "actually builds one; a new frontier state or re-paired actuator can ship an " + "unexercised branch that stays silent until it accumulates against real archive " + "data (the original defect: 4,174 blockers demanding an unreachable actuator, " + "undetected for weeks). This lint statically enumerates every literal " + "(state, actuator) construction site in polylogue/storage/raw_reconciler.py " + "(_item(...) and _StrategyOverride(...) calls) and re-checks the same invariant " + "at review time, independent of test coverage." + ), + examples=( + "devtools lab policy raw-authority-frontier-executability", + "devtools lab policy raw-authority-frontier-executability --json", + ), + ), CommandSpec( "lab policy backlog-hygiene", "verification lab", diff --git a/devtools/raw_append_chain_backfill_apply.py b/devtools/raw_append_chain_backfill_apply.py new file mode 100644 index 0000000000..90bc66c801 --- /dev/null +++ b/devtools/raw_append_chain_backfill_apply.py @@ -0,0 +1,112 @@ +"""Actuator: promote membershipless append raws proven correct by live-source verification. + +polylogue-lb39z (Phase 1, item 3): 2,712 ``raw_sessions`` rows (measured +2026-08-02) are ``revision_kind='append'``, ``revision_authority='quarantined'``, +and have no ``raw_session_memberships`` row at all -- a genuine fixed point, +because the only mechanism that ever promotes an append raw +(``_promote_contiguous_append_evidence``) requires its byte-contiguous +predecessor to already be ``byte_proven``. This proves each such row's own +claimed byte range directly against its live source file's current bytes, +independent of any ancestor's authority. + +Default mode is dry-run (report only, zero mutation). Pass ``--apply`` to +actually promote rows, which additionally requires ``--backup-manifest`` +pointing at a verified backup manifest for the ``source`` tier (see +``polylogue backup --output-dir --verify``). This never runs blob GC or +``VACUUM`` -- that is a separate, later, operator-invoked step. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import TextIO + +from polylogue.maintenance.raw_append_chain_backfill_apply import ( + RawAppendChainBackfillApplyError, + apply_raw_append_chain_backfill, +) +from polylogue.paths import archive_root as default_archive_root + + +def main(argv: list[str] | None = None, *, stdout: TextIO | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--archive-root", + type=Path, + default=None, + help="Archive root to operate on; defaults to the configured active archive root.", + ) + parser.add_argument( + "--limit", + type=int, + default=None, + help="Cap the number of membershipless append rows classified/promoted (unbounded by default).", + ) + parser.add_argument( + "--apply", + action="store_true", + help="Actually promote rows. Without this flag, nothing is mutated.", + ) + parser.add_argument( + "--backup-manifest", + type=Path, + default=None, + help="Verified backup manifest for the source tier. Required with --apply.", + ) + parser.add_argument("--json", action="store_true", help="Emit the report as JSON.") + args = parser.parse_args(argv) + + root = args.archive_root if args.archive_root is not None else default_archive_root() + + try: + report = apply_raw_append_chain_backfill( + root, + backup_manifest=args.backup_manifest, + limit=args.limit, + dry_run=not args.apply, + ) + except (RawAppendChainBackfillApplyError, FileNotFoundError) as exc: + print(f"refused: {exc}", file=stdout) + return 1 + + if args.json: + payload = { + "applied": report.applied, + "scanned_count": report.scanned_count, + "promoted_count": report.promoted_count, + "promoted_bytes": report.promoted_bytes, + "diverged_count": report.diverged_count, + "source_missing_count": report.source_missing_count, + "promoted_raw_ids": list(report.promoted_raw_ids), + "backup_manifest": str(report.backup_manifest) if report.backup_manifest is not None else None, + } + print(json.dumps(payload, indent=2, sort_keys=True), file=stdout) + return 0 + + def _mb(byte_count: int) -> str: + return f"{byte_count / (1024 * 1024):.1f} MB" + + mode = "APPLIED" if report.applied else "dry-run (no mutation performed -- pass --apply to promote)" + print(f"mode: {mode}", file=stdout) + print(f"membershipless quarantined append rows scanned: {report.scanned_count}", file=stdout) + print( + f"{'promoted' if report.applied else 'promotable'}: {report.promoted_count:>7} ({_mb(report.promoted_bytes)})", + file=stdout, + ) + print(f"diverged (left untouched): {report.diverged_count}", file=stdout) + print(f"source missing (left untouched): {report.source_missing_count}", file=stdout) + if report.applied: + print(f"backup manifest used: {report.backup_manifest}", file=stdout) + print( + "Each promoted row has an immutable receipt in " + "raw_append_chain_backfill_receipts. No blob GC or VACUUM was run -- " + "that is a separate, later step.", + file=stdout, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/devtools/verify.py b/devtools/verify.py index 6e7c89a1c3..e4be180d72 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -1810,6 +1810,20 @@ def build_verify_steps( # that reorder entries) -- polylogue-gysk3 found the same # hazard still live for provider_message_id. ("lab policy position-derived-identity", _devtools_cmd("lab policy position-derived-identity")), + # Static, archive-independent, sub-second: forbids a NEW + # unreachable (frontier state, dispatched actuator) pairing in + # polylogue/storage/raw_reconciler.py -- polylogue-w32w found + # UNRESOLVED_PROVENANCE paired with the dispatched + # REFINE_QUARANTINE actuator, an actuator no path could ever + # select, and 4,174 blockers accumulated behind it for weeks + # before anyone noticed. The runtime constructor guard + # (RawAuthorityFrontierItem.__post_init__) only fires when + # something actually constructs the bad combination; this + # lint re-checks every literal pairing at review time. + ( + "lab policy raw-authority-frontier-executability", + _devtools_cmd("lab policy raw-authority-frontier-executability"), + ), # Publication gate. Committed provider schema packages are # public artifacts; this blocks local provenance # (bundle_scopes/representative_paths) and scans for secrets. diff --git a/devtools/verify_raw_authority_frontier_executability.py b/devtools/verify_raw_authority_frontier_executability.py new file mode 100644 index 0000000000..1d51f03396 --- /dev/null +++ b/devtools/verify_raw_authority_frontier_executability.py @@ -0,0 +1,261 @@ +"""Statically verify every raw-authority frontier state has a reachable actuator. + +Background +---------- + +``polylogue.storage.raw_reconciler`` classifies every accepted raw-authority +head into one of a small closed set of ``RawAuthorityFrontierState`` values, +each paired with a ``RawAuthorityActuator``. Only actuators with a real +``apply()`` dispatch branch (``_APPLY_DISPATCHED_ACTUATORS``) promise +"something automatically executes this"; only states in ``_EXECUTABLE_STATES`` +are ever selected by the daemon or the operator break-glass path +(``item.executable``). polylogue-w32w found a state (``UNRESOLVED_PROVENANCE``) +paired with a dispatched actuator (``REFINE_QUARANTINE``) that was NOT in +``_EXECUTABLE_STATES`` -- 4,174 blockers demanded an actuator no path could +ever select, and the gap accumulated silently for weeks because nothing +checked the pairing except live production data eventually noticing the +backlog never drained. + +``RawAuthorityFrontierItem.__post_init__`` now raises if a *constructed* +instance has this shape (PR #3466) -- but that is a runtime assertion: it +only fires on whichever (state, actuator) pairs a test happens to construct. +A future contributor adding a new frontier state, or re-pairing an existing +one, can ship a classification branch that is never exercised by any test +fixture; the constructor guard stays silent until that branch runs against +real archive data, which is exactly the failure mode that let the original +defect accumulate for weeks undetected. This lint closes that gap: it +statically enumerates every (state, actuator) pair +``polylogue/storage/raw_reconciler.py``'s classification code can literally +construct -- independent of whether any test ever exercises that branch -- +and re-validates the same invariant the constructor enforces, so the CHECK +fails at review time, not months later against live data. + +What this lint checks +---------------------- + +Parses ``polylogue/storage/raw_reconciler.py`` and finds every call site that +constructs a frontier item's ``state``/``actuator`` pairing with **literal** +enum-attribute arguments: + +* ``_item(state=RawAuthorityFrontierState.X, actuator=RawAuthorityActuator.Y, ...)`` + -- the sole ``RawAuthorityFrontierItem`` builder. +* ``_StrategyOverride(state=RawAuthorityFrontierState.X, + actuator=RawAuthorityActuator.Y, ...)`` -- overrides that later flow into + ``_item`` via ``_item(state=strategy_override.state, + actuator=strategy_override.actuator, ...)``; that forwarding call site's + arguments are not literal (they read a variable), so this lint checks the + override's own literal construction instead -- the same (state, actuator) + pair reaches ``RawAuthorityFrontierItem.__post_init__`` either way. + +For each literal pair found, re-checks the exact invariant +``RawAuthorityFrontierItem.__post_init__`` enforces at runtime: an actuator in +``_APPLY_DISPATCHED_ACTUATORS`` must only ever be paired with a state in +``_EXECUTABLE_STATES``. Both sets are imported directly from +``polylogue.storage.raw_reconciler`` (not re-declared here), so this lint +never drifts out of sync with the real executability gate. + +A ``state=``/``actuator=`` argument that is not a literal +``RawAuthorityFrontierState.X`` / ``RawAuthorityActuator.Y`` attribute access +(e.g. a bare variable) cannot be resolved statically and is reported +separately as "dynamic" -- informational only, never a failure, since every +current dynamic pairing (the ``_item(state=strategy_override.state, ...)`` +forwarding call) is already covered by checking its override's own literal +construction site. A future dynamic pairing with no literal source anywhere +in this file would not be caught by this lint; it would still be caught by +the runtime constructor guard the first time a test or live classification +constructs it. + +Wired standalone via ``devtools lab policy raw-authority-frontier-executability`` +(like ``schema-versioning``): static, archive-independent, sub-second. +""" + +from __future__ import annotations + +import argparse +import ast +import json +import sys +from dataclasses import dataclass +from pathlib import Path + +from devtools import repo_root as _get_root +from polylogue.storage.raw_reconciler import ( + _APPLY_DISPATCHED_ACTUATORS, + _EXECUTABLE_STATES, + RawAuthorityActuator, + RawAuthorityFrontierState, +) + +ROOT = _get_root() +RECONCILER_PATH = ROOT / "polylogue" / "storage" / "raw_reconciler.py" + +_TARGET_CALLEES: tuple[str, ...] = ("_item", "_StrategyOverride") + +_EXECUTABLE_STATE_NAMES = {state.name for state in _EXECUTABLE_STATES} +_APPLY_DISPATCHED_ACTUATOR_NAMES = {actuator.name for actuator in _APPLY_DISPATCHED_ACTUATORS} + + +@dataclass(frozen=True, slots=True) +class FrontierPair: + callee: str + lineno: int + state: str + actuator: str + + +@dataclass(frozen=True, slots=True) +class DynamicSite: + callee: str + lineno: int + detail: str + + +@dataclass(frozen=True, slots=True) +class ExecutabilityReport: + pairs: tuple[FrontierPair, ...] + dynamic_sites: tuple[DynamicSite, ...] + violations: tuple[FrontierPair, ...] + + @property + def ok(self) -> bool: + return not self.violations + + +def _literal_enum_attr(node: ast.expr, *, enum_name: str) -> str | None: + """Return ``X`` for an ``EnumName.X`` attribute access node, else ``None``.""" + if not isinstance(node, ast.Attribute): + return None + value = node.value + if not isinstance(value, ast.Name) or value.id != enum_name: + return None + return node.attr + + +def collect_frontier_pairs(path: Path = RECONCILER_PATH) -> tuple[tuple[FrontierPair, ...], tuple[DynamicSite, ...]]: + """Statically enumerate every literal (state, actuator) construction pair.""" + tree = ast.parse(path.read_text(encoding="utf-8")) + pairs: list[FrontierPair] = [] + dynamic: list[DynamicSite] = [] + + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func = node.func + if not isinstance(func, ast.Name) or func.id not in _TARGET_CALLEES: + continue + state_arg: ast.expr | None = None + actuator_arg: ast.expr | None = None + for keyword in node.keywords: + if keyword.arg == "state": + state_arg = keyword.value + elif keyword.arg == "actuator": + actuator_arg = keyword.value + if state_arg is None or actuator_arg is None: + # Every real call site names both explicitly by keyword; a call + # missing either is not this lint's concern (it would fail at + # import/call time as a TypeError against _item's/StrategyOverride's + # own required signature). + continue + state_name = _literal_enum_attr(state_arg, enum_name="RawAuthorityFrontierState") + actuator_name = _literal_enum_attr(actuator_arg, enum_name="RawAuthorityActuator") + if state_name is None or actuator_name is None: + dynamic.append( + DynamicSite( + callee=func.id, + lineno=node.lineno, + detail="state/actuator argument is not a literal EnumName.MEMBER attribute access", + ) + ) + continue + pairs.append(FrontierPair(callee=func.id, lineno=node.lineno, state=state_name, actuator=actuator_name)) + + return tuple(pairs), tuple(dynamic) + + +def compute_executability_report(path: Path = RECONCILER_PATH) -> ExecutabilityReport: + pairs, dynamic_sites = collect_frontier_pairs(path) + # Fail closed on an unknown name: a rename that outpaces this lint's own + # enum imports must not silently pass as "no violation found". + for pair in pairs: + if pair.state not in {state.name for state in RawAuthorityFrontierState}: + raise ValueError(f"{path}:{pair.lineno}: unknown RawAuthorityFrontierState member {pair.state!r}") + if pair.actuator not in {actuator.name for actuator in RawAuthorityActuator}: + raise ValueError(f"{path}:{pair.lineno}: unknown RawAuthorityActuator member {pair.actuator!r}") + violations = tuple( + pair + for pair in pairs + if pair.actuator in _APPLY_DISPATCHED_ACTUATOR_NAMES and pair.state not in _EXECUTABLE_STATE_NAMES + ) + return ExecutabilityReport(pairs=pairs, dynamic_sites=dynamic_sites, violations=violations) + + +def _format_report(report: ExecutabilityReport, *, path: Path) -> str: + rel = path.relative_to(ROOT) if path.is_absolute() else path + lines = [ + f"frontier (state, actuator) construction sites checked: {len(report.pairs)}", + f"dynamic (unresolvable) sites, informational only: {len(report.dynamic_sites)}", + f"unreachable-actuator violations: {len(report.violations)}", + ] + if report.violations: + lines.append("") + lines.append( + "Frontier states pairing a dispatched actuator with a non-executable " + "state -- no path (daemon or operator) would ever select these:" + ) + for pair in report.violations: + lines.append( + f" {rel}:{pair.lineno}: {pair.callee}(state={pair.state}, actuator={pair.actuator}) -- " + f"{pair.actuator} has an apply() dispatch branch but {pair.state} is not in _EXECUTABLE_STATES" + ) + lines.append( + " Fix: either add the state to _EXECUTABLE_STATES (and prove the daemon/operator " + "path can safely select it), or pair this classification with a non-dispatched " + "actuator (RawAuthorityActuator.NONE, REACQUIRE, or REQUEST_JUDGMENT)." + ) + if report.dynamic_sites: + lines.append("") + lines.append("Dynamic sites (state/actuator not a literal enum attribute -- not checked here):") + for site in report.dynamic_sites: + lines.append(f" {rel}:{site.lineno}: {site.callee}(...) -- {site.detail}") + if report.ok: + lines.append("") + lines.append("Raw-authority frontier executability policy intact.") + return "\n".join(lines) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("--json", action="store_true", help="emit machine-readable JSON") + args = parser.parse_args(argv) + + report = compute_executability_report() + + if args.json: + print( + json.dumps( + { + "pairs_checked": len(report.pairs), + "dynamic_sites": [ + {"callee": site.callee, "lineno": site.lineno, "detail": site.detail} + for site in report.dynamic_sites + ], + "violations": [ + {"callee": pair.callee, "lineno": pair.lineno, "state": pair.state, "actuator": pair.actuator} + for pair in report.violations + ], + "ok": report.ok, + }, + indent=2, + ) + ) + else: + print(_format_report(report, path=RECONCILER_PATH)) + + return 0 if report.ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/docs/devtools.md b/docs/devtools.md index 0b16ba515b..1115913dfb 100644 --- a/docs/devtools.md +++ b/docs/devtools.md @@ -148,6 +148,7 @@ These are the commands worth remembering during normal repo work: | `devtools lab policy docs-drift` | Verify checkable factual claims in the Reference-docs table against current source. | | `devtools lab policy insight-honesty` | Verify every registered insight product is rigor-contracted or exempt. | | `devtools lab policy position-derived-identity` | Verify no parser mints cross-revision comparison identity from positional/index data. | +| `devtools lab policy raw-authority-frontier-executability` | Verify every raw-authority frontier state has a reachable actuator. | | `devtools lab policy raw-payload-hash-purity` | Verify no raw-capture write path splices a synthesized literal onto captured bytes before hashing. | | `devtools lab policy schema-versioning` | Verify durable-tier migration and derived-tier rebuild boundaries. | | `devtools lab policy timestamp-doctrine` | Verify durable-tier DDL never stores a timestamp column as TEXT. | diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index 85adc69518..ee63d54288 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -2223,6 +2223,10 @@ files: loc: 505 target: polylogue/maintenance/preview.py owner: stable + - path: polylogue/maintenance/raw_append_chain_backfill_apply.py + loc: 244 + target: polylogue/maintenance/raw_append_chain_backfill_apply.py + owner: stable - path: polylogue/maintenance/raw_authority_reset.py loc: 109 target: polylogue/maintenance/raw_authority_reset.py @@ -3960,6 +3964,10 @@ files: loc: 137 target: polylogue/storage/raw/models.py owner: stable + - path: polylogue/storage/raw_append_chain_backfill.py + loc: 214 + target: TBD + owner: storage-domain - path: polylogue/storage/raw_authority.py loc: 2295 target: TBD @@ -4282,7 +4290,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/session_annotations_write.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/source.py - loc: 538 + loc: 569 target: polylogue/storage/sqlite/archive_tiers/source.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/source_write.py diff --git a/polylogue/maintenance/raw_append_chain_backfill_apply.py b/polylogue/maintenance/raw_append_chain_backfill_apply.py new file mode 100644 index 0000000000..e0c2a25331 --- /dev/null +++ b/polylogue/maintenance/raw_append_chain_backfill_apply.py @@ -0,0 +1,244 @@ +"""Promote membershipless append raws proven correct by live-source verification. + +polylogue-lb39z (Phase 1, item 3): 2,712 ``raw_sessions`` rows are +``revision_kind='append'``, ``revision_authority='quarantined'``, and have no +``raw_session_memberships`` row at all -- a genuine fixed point, because the +only mechanism that ever promotes an append raw +(``_promote_contiguous_append_evidence``, +``storage/sqlite/archive_tiers/revision_governance.py``) requires its +byte-contiguous predecessor to already be ``byte_proven``. When the +predecessor itself is stuck quarantined, no amount of re-running that cascade +resolves the child. :mod:`polylogue.storage.raw_append_chain_backfill` +classifies each such row against its live source file's *current* bytes at +its own recorded ``[append_start_offset:append_end_offset)`` window -- a proof +that does not depend on any ancestor's authority. This module is the "act" +half, following the identical safety pattern as the already-merged +``raw_live_source_reconciliation_apply`` (polylogue-u19l) and +``raw_membership_writeback_apply`` (this bead's item 2): + +* Dry-run by default. ``dry_run=False`` requires a verified backup manifest + for the ``source`` tier, validated with the same gate durable-tier schema + migrations use. +* Classification is re-run *live*, inside the same write transaction the + promotion runs in -- never trusts a previously computed plan. +* Every promoted row gets an immutable receipt in + ``raw_append_chain_backfill_receipts`` in the same transaction as its + ``revision_authority`` update. +* Deliberately never touches ``predecessor_raw_id`` / ``baseline_raw_id`` / + ``acquisition_generation`` -- exactly like polylogue-u19l's actuator, that + revision-graph linkage remains ``classify_raw_revision_cohort``'s / + ``_promote_contiguous_append_evidence``'s concern; the existing cascade + picks a newly proven row up for free on the next normal convergence pass, + either resolving it as its true predecessor's child (once that ancestor is + itself proven) or using it as a newly eligible parent for whatever append + fragment sits downstream -- unblocking a stuck chain one proven link at a + time. +* Records ``revision_authority_evidence='live_source_verification_v1'`` -- + the identical evidence value polylogue-u19l's actuator uses, because the + proof mechanism (byte-window comparison against the live source file) is + the same mechanism; only the target population (membershipless append + rows, specifically) differs, and that population distinction is what this + module's own dedicated receipt table records. +* Never performs blob GC or ``VACUUM``. +""" + +from __future__ import annotations + +import sqlite3 +import time +from dataclasses import dataclass +from pathlib import Path + +from polylogue.config import Config +from polylogue.maintenance.offline_guard import offline_maintenance_block_reason +from polylogue.paths import render_root +from polylogue.storage.blob_store import BlobStore +from polylogue.storage.live_source_reconciliation import LIVE_SOURCE_VERIFICATION_EVIDENCE +from polylogue.storage.raw_append_chain_backfill import ( + AppendChainBackfillPlan, + plan_append_chain_backfill, +) +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.migration_runner import validate_migration_backup_manifest + +TOOL_VERSION = "raw-append-chain-backfill-apply-v1" + + +class RawAppendChainBackfillApplyError(RuntimeError): + """Raised when applying an append-chain backfill promotion is refused.""" + + +@dataclass(frozen=True, slots=True) +class RawAppendChainBackfillApplyReport: + scanned_count: int + promoted_count: int + promoted_bytes: int + diverged_count: int + diverged_bytes: int + source_missing_count: int + source_missing_bytes: int + promoted_raw_ids: tuple[str, ...] + applied: bool + backup_manifest: Path | None = None + + @classmethod + def from_plan( + cls, + plan: AppendChainBackfillPlan, + *, + applied: bool, + promoted_raw_ids: tuple[str, ...] = (), + backup_manifest: Path | None = None, + ) -> RawAppendChainBackfillApplyReport: + promoted_bytes = ( + sum(c.blob_size for c in plan.exact_match if c.raw_id in set(promoted_raw_ids)) + if applied + else plan.exact_match_bytes + ) + return cls( + scanned_count=plan.scanned_count, + promoted_count=len(promoted_raw_ids) if applied else len(plan.exact_match), + promoted_bytes=promoted_bytes, + diverged_count=len(plan.diverged), + diverged_bytes=plan.diverged_bytes, + source_missing_count=len(plan.source_missing), + source_missing_bytes=plan.source_missing_bytes, + promoted_raw_ids=promoted_raw_ids, + applied=applied, + backup_manifest=backup_manifest, + ) + + +def _offline_config(archive_root: Path) -> Config: + return Config(archive_root=archive_root, render_root=render_root(), sources=[]) + + +def _checkpoint_live_tier(conn: sqlite3.Connection) -> None: + try: + row = conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() + except sqlite3.Error as exc: + raise RawAppendChainBackfillApplyError("could not checkpoint source.db before backup validation") from exc + if row is None: + raise RawAppendChainBackfillApplyError("could not checkpoint source.db before backup validation") + + +def apply_raw_append_chain_backfill( + archive_root: Path, + *, + backup_manifest: Path | None = None, + limit: int | None = None, + dry_run: bool = True, +) -> RawAppendChainBackfillApplyReport: + """Classify membershipless append raws live, then promote the exact matches. + + ``dry_run=True`` (the default) never opens a write transaction. It runs + the same classifier a real apply would and reports what it would do. + + ``dry_run=False`` requires ``backup_manifest`` and re-runs classification + live, inside the same ``BEGIN IMMEDIATE`` write transaction the + promotion UPDATE/INSERT pair runs in, so nothing acted on can be stale + relative to what gets promoted. + """ + source_db = archive_root / "source.db" + if not source_db.exists(): + raise FileNotFoundError(f"no source.db at {source_db}") + blob_store = BlobStore(archive_root / "blob") + + if dry_run: + conn = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True) + try: + plan = plan_append_chain_backfill(conn, blob_store=blob_store, limit=limit) + finally: + conn.close() + return RawAppendChainBackfillApplyReport.from_plan(plan, applied=False) + + if backup_manifest is None: + raise RawAppendChainBackfillApplyError( + "applying raw-append-chain-backfill requires a verified backup manifest (--backup-manifest)" + ) + if reason := offline_maintenance_block_reason(_offline_config(archive_root), active=True, dry_run=False): + raise RawAppendChainBackfillApplyError(reason) + + conn = sqlite3.connect(source_db) + promoted: list[str] = [] + try: + _checkpoint_live_tier(conn) + validate_migration_backup_manifest(backup_manifest, ArchiveTier.SOURCE, connection=conn) + + conn.execute("BEGIN IMMEDIATE") + try: + validate_migration_backup_manifest(backup_manifest, ArchiveTier.SOURCE, connection=conn) + + plan = plan_append_chain_backfill(conn, blob_store=blob_store, limit=limit) + compared_at_ms = int(time.time() * 1000) + + for candidate in plan.exact_match: + assert candidate.source_path is not None # exact_match rows always have a source_path + cursor = conn.execute( + """ + UPDATE raw_sessions + SET revision_authority = 'byte_proven', + revision_authority_evidence = ? + WHERE raw_id = ? AND revision_authority = 'quarantined' + """, + (LIVE_SOURCE_VERIFICATION_EVIDENCE, candidate.raw_id), + ) + if cursor.rowcount != 1: + # Defensive: no longer quarantined under this same locked + # transaction's own read -- should not happen given the + # classification above ran on this connection under this + # same lock, but skipping instead of asserting keeps this + # pass conservative. + continue + conn.execute( + """ + INSERT INTO raw_append_chain_backfill_receipts ( + raw_id, logical_source_key, source_path, blob_hash, blob_size, + append_start_offset, append_end_offset, matched_after_codex_header_strip, + previous_revision_authority, compared_at_ms, tool_version, + backup_manifest_path, detail + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'quarantined', ?, ?, ?, ?) + """, + ( + candidate.raw_id, + candidate.logical_source_key, + candidate.source_path, + candidate.blob_hash, + candidate.blob_size, + candidate.append_start_offset, + candidate.append_end_offset, + 1 if candidate.comparison.matched_after_codex_header_strip else 0, + compared_at_ms, + TOOL_VERSION, + str(backup_manifest), + candidate.comparison.detail or "", + ), + ) + promoted.append(candidate.raw_id) + + quick_check = conn.execute("PRAGMA quick_check").fetchone() + if quick_check is None or str(quick_check[0]).lower() != "ok": + raise RawAppendChainBackfillApplyError(f"source.db quick_check failed after promotion: {quick_check!r}") + except Exception: + if conn.in_transaction: + conn.rollback() + raise + else: + conn.commit() + finally: + conn.close() + + return RawAppendChainBackfillApplyReport.from_plan( + plan, + applied=True, + promoted_raw_ids=tuple(promoted), + backup_manifest=backup_manifest, + ) + + +__all__ = [ + "TOOL_VERSION", + "RawAppendChainBackfillApplyError", + "RawAppendChainBackfillApplyReport", + "apply_raw_append_chain_backfill", +] diff --git a/polylogue/storage/raw_append_chain_backfill.py b/polylogue/storage/raw_append_chain_backfill.py new file mode 100644 index 0000000000..b5f7f752b3 --- /dev/null +++ b/polylogue/storage/raw_append_chain_backfill.py @@ -0,0 +1,214 @@ +"""Read-only classification: membershipless append raws provable against live sources. + +polylogue-lb39z (Phase 1, item 3): a live, read-only audit found 2,712 +``raw_sessions`` rows that are ``revision_kind='append'``, +``revision_authority='quarantined'``, and have NO corresponding +``raw_session_memberships`` row at all -- not ``ambiguous``, not +``deferred``, genuinely absent. This is a distinct population from the two +already-landed Phase-1 actuators: + +* :mod:`polylogue.storage.raw_membership_writeback` acts on quarantined raws + whose membership row already carries a *decided* verdict. +* :mod:`polylogue.storage.live_source_reconciliation` (polylogue-u19l) acts on + every quarantined raw regardless of membership presence, using a pure + byte/prefix/offset-range comparison against the live source file. + +The membershipless population never reaches either mechanism because +``_promote_contiguous_append_evidence`` +(``storage/sqlite/archive_tiers/revision_governance.py``) -- the *only* +mechanism that ever promotes an append raw's ``revision_authority`` -- requires +its byte-contiguous *predecessor* row to already be ``revision_authority= +'byte_proven'``. When the predecessor itself is stuck ``quarantined`` (a +chain of quarantined dominoes), the child can never be promoted through that +path, no matter how many times it runs: it is a genuine fixed point, not a +transient backlog. + +This module proves a stuck append raw's *own* claimed byte range +(``[append_start_offset:append_end_offset)``) directly against its live +source file's *current* bytes -- a proof that does not depend on any +ancestor's authority at all, reusing the identical byte-window comparison +:mod:`polylogue.storage.live_source_reconciliation` already implements for +``revision_kind == RawRevisionKind.APPEND`` (see that module's +``compare_raw_against_live_source`` docstring). Promoting this raw's own +``revision_authority`` to ``byte_proven`` does not by itself establish +``predecessor_raw_id``/``baseline_raw_id``/``acquisition_generation`` -- +exactly like polylogue-u19l's actuator, that revision-graph linkage remains +``classify_raw_revision_cohort``'s/``_promote_contiguous_append_evidence``'s +concern, and the existing cascade picks it up for free on the next normal +convergence pass, either as the resolved child of its own true predecessor +(once that ancestor is itself proven) or as a newly eligible *parent* for +whatever append fragment sits downstream of it -- unblocking the rest of an +otherwise permanently stuck chain one proven link at a time. + +Scope is deliberately narrower than polylogue-u19l's: only rows with +``revision_kind='append'`` AND zero ``raw_session_memberships`` rows. A +membershipless row already covered by u19l's broader classifier is still +independently discoverable here (the two populations overlap), but this +module's own query filters explicitly to the population named by this bead +item, "append-chain backfill" -- it is not a replacement for the broader +live-source reconciliation actuator. +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass +from pathlib import Path + +from polylogue.archive.revision_authority import RawRevisionKind +from polylogue.core.enums import Origin, Provider +from polylogue.core.sources import provider_from_origin +from polylogue.storage.blob_store import BlobStore +from polylogue.storage.live_source_reconciliation import ( + LiveSourceComparison, + LiveSourceVerdict, + compare_raw_against_live_source, +) + + +@dataclass(frozen=True, slots=True) +class AppendChainBackfillCandidate: + """One membershipless append raw's live-source classification.""" + + raw_id: str + logical_source_key: str | None + provider: Provider + source_path: str | None + blob_hash: bytes + blob_size: int + append_start_offset: int + append_end_offset: int + comparison: LiveSourceComparison + + +@dataclass(frozen=True, slots=True) +class AppendChainBackfillPlan: + """Read-only projection: how much of the membershipless population is provable. + + Mirrors ``LiveSourceReconciliationPlan``'s report-first shape -- nothing + here marks a row's authority or drops a blob. + """ + + scanned_count: int + exact_match: tuple[AppendChainBackfillCandidate, ...] + diverged: tuple[AppendChainBackfillCandidate, ...] + source_missing: tuple[AppendChainBackfillCandidate, ...] + + @property + def exact_match_bytes(self) -> int: + return sum(candidate.blob_size for candidate in self.exact_match) + + @property + def diverged_bytes(self) -> int: + return sum(candidate.blob_size for candidate in self.diverged) + + @property + def source_missing_bytes(self) -> int: + return sum(candidate.blob_size for candidate in self.source_missing) + + +def plan_append_chain_backfill( + source_conn: sqlite3.Connection, + *, + blob_store: BlobStore, + limit: int | None = None, +) -> AppendChainBackfillPlan: + """Read-only: classify every membershipless quarantined append row. + + Never mutates ``source.db`` and never touches the blob store beyond + reading. Safe to run against a live archive opened read-only + (``file:...?mode=ro``). + """ + original_row_factory = source_conn.row_factory + source_conn.row_factory = sqlite3.Row + try: + query = """ + SELECT r.raw_id, r.origin, r.capture_mode, r.logical_source_key, + r.source_path, r.blob_size, r.blob_hash, + lower(hex(r.blob_hash)) AS blob_hash_hex, + r.append_start_offset, r.append_end_offset + FROM raw_sessions AS r + WHERE r.revision_authority = 'quarantined' + AND r.revision_kind = 'append' + AND r.append_start_offset IS NOT NULL + AND r.append_end_offset IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM raw_session_memberships AS m WHERE m.raw_id = r.raw_id + ) + ORDER BY r.raw_id + """ + params: tuple[object, ...] = () + if limit is not None: + query += " LIMIT ?" + params = (limit,) + rows = source_conn.execute(query, params).fetchall() + + exact: list[AppendChainBackfillCandidate] = [] + diverged: list[AppendChainBackfillCandidate] = [] + missing: list[AppendChainBackfillCandidate] = [] + for row in rows: + provider = provider_from_origin( + Origin.from_string(str(row["origin"])), + family_hint=row["capture_mode"], + ) + source_path_value = row["source_path"] + blob_size = int(row["blob_size"] or 0) + append_start_offset = int(row["append_start_offset"]) + append_end_offset = int(row["append_end_offset"]) + + if source_path_value is None: + comparison = LiveSourceComparison( + verdict=LiveSourceVerdict.SOURCE_MISSING, + detail="no recorded source_path", + ) + else: + blob_hash_hex = row["blob_hash_hex"] + if blob_hash_hex is None or not blob_store.exists(blob_hash_hex): + comparison = LiveSourceComparison( + verdict=LiveSourceVerdict.SOURCE_MISSING, + detail="archived blob unavailable", + ) + else: + raw_payload = blob_store.read_all(blob_hash_hex) + comparison = compare_raw_against_live_source( + raw_payload=raw_payload, + provider=provider, + revision_kind=RawRevisionKind.APPEND, + source_path=Path(source_path_value), + append_start_offset=append_start_offset, + append_end_offset=append_end_offset, + ) + + candidate = AppendChainBackfillCandidate( + raw_id=str(row["raw_id"]), + logical_source_key=row["logical_source_key"], + provider=provider, + source_path=source_path_value, + blob_hash=bytes(row["blob_hash"]), + blob_size=blob_size, + append_start_offset=append_start_offset, + append_end_offset=append_end_offset, + comparison=comparison, + ) + if comparison.verdict == LiveSourceVerdict.EXACT_MATCH: + exact.append(candidate) + elif comparison.verdict == LiveSourceVerdict.SOURCE_MISSING: + missing.append(candidate) + else: + diverged.append(candidate) + + return AppendChainBackfillPlan( + scanned_count=len(rows), + exact_match=tuple(exact), + diverged=tuple(diverged), + source_missing=tuple(missing), + ) + finally: + source_conn.row_factory = original_row_factory + + +__all__ = [ + "AppendChainBackfillCandidate", + "AppendChainBackfillPlan", + "plan_append_chain_backfill", +] diff --git a/polylogue/storage/sqlite/archive_tiers/source.py b/polylogue/storage/sqlite/archive_tiers/source.py index dd3f98fa4f..2c1cb1f1f6 100644 --- a/polylogue/storage/sqlite/archive_tiers/source.py +++ b/polylogue/storage/sqlite/archive_tiers/source.py @@ -9,7 +9,7 @@ from polylogue.core.enums import ArtifactSupportStatus, Origin, Provider, ValidationMode, ValidationStatus from polylogue.storage.sqlite.archive_tiers.common import check, nullable_check -SOURCE_SCHEMA_VERSION = 19 +SOURCE_SCHEMA_VERSION = 20 SOURCE_DDL = f""" CREATE TABLE IF NOT EXISTS raw_sessions ( @@ -186,6 +186,37 @@ CREATE INDEX IF NOT EXISTS idx_raw_membership_writeback_receipts_promoted_at ON raw_membership_writeback_receipts(promoted_at_ms); +-- v20 (polylogue-lb39z, Phase 1 item 3): one immutable receipt per +-- raw_sessions row promoted out of quarantine because its own claimed +-- [append_start_offset:append_end_offset) byte range was proven directly +-- against its live source file's current bytes -- the membershipless +-- append-chain-backfill population (a row stuck quarantined with no +-- raw_session_memberships row at all because its predecessor is itself +-- unresolved, so the normal _promote_contiguous_append_evidence cascade can +-- never reach it). Reuses revision_authority_evidence= +-- 'live_source_verification_v1' (the proof mechanism is identical to +-- polylogue-u19l's); this table's own existence records the distinct target +-- population. See polylogue.storage.raw_append_chain_backfill + +-- polylogue.maintenance.raw_append_chain_backfill_apply. +CREATE TABLE IF NOT EXISTS raw_append_chain_backfill_receipts ( + raw_id TEXT PRIMARY KEY REFERENCES raw_sessions(raw_id) ON DELETE CASCADE, + logical_source_key TEXT, + source_path TEXT NOT NULL, + blob_hash BLOB NOT NULL CHECK(length(blob_hash) = 32), + blob_size INTEGER NOT NULL CHECK(blob_size >= 0), + append_start_offset INTEGER NOT NULL CHECK(append_start_offset >= 0), + append_end_offset INTEGER NOT NULL CHECK(append_end_offset > append_start_offset), + matched_after_codex_header_strip INTEGER NOT NULL CHECK(matched_after_codex_header_strip IN (0, 1)), + previous_revision_authority TEXT NOT NULL CHECK(previous_revision_authority IN ('asserted', 'byte_proven', 'quarantined')), + compared_at_ms INTEGER NOT NULL CHECK(compared_at_ms >= 0), + tool_version TEXT NOT NULL, + backup_manifest_path TEXT NOT NULL, + detail TEXT NOT NULL DEFAULT '' +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_raw_append_chain_backfill_receipts_compared_at +ON raw_append_chain_backfill_receipts(compared_at_ms); + -- Durable authority reconciliation ledger. The source tier owns this -- evidence because index.db and ops.db are rebuildable/disposable: neither -- can be the authority for whether an accepted replay plan was conserved. diff --git a/polylogue/storage/sqlite/migrations/source/020_raw_append_chain_backfill_receipts.sql b/polylogue/storage/sqlite/migrations/source/020_raw_append_chain_backfill_receipts.sql new file mode 100644 index 0000000000..6533497a86 --- /dev/null +++ b/polylogue/storage/sqlite/migrations/source/020_raw_append_chain_backfill_receipts.sql @@ -0,0 +1,35 @@ +-- polylogue-lb39z (Phase 1, item 3): 2,712 raw_sessions rows are +-- revision_kind='append', revision_authority='quarantined', and have no +-- raw_session_memberships row at all -- a genuine fixed point, because the +-- only mechanism that ever promotes an append raw +-- (_promote_contiguous_append_evidence) requires its byte-contiguous +-- predecessor to already be byte_proven. A new, explicitly operator-invoked +-- actuator (devtools workspace raw-append-chain-backfill-apply) proves each +-- such row's own claimed [append_start_offset:append_end_offset) byte range +-- directly against its live source file's current bytes -- a proof that does +-- not depend on any ancestor's authority -- and promotes exact matches to +-- revision_authority='byte_proven'. +-- +-- Reuses the existing revision_authority_evidence='live_source_verification_v1' +-- value (migration 018): the proof mechanism (byte-window comparison against +-- the live source file) is identical to polylogue-u19l's actuator; only the +-- target population (membershipless append rows specifically) differs, and +-- that distinction is what this dedicated receipt table records. +CREATE TABLE IF NOT EXISTS raw_append_chain_backfill_receipts ( + raw_id TEXT PRIMARY KEY REFERENCES raw_sessions(raw_id) ON DELETE CASCADE, + logical_source_key TEXT, + source_path TEXT NOT NULL, + blob_hash BLOB NOT NULL CHECK(length(blob_hash) = 32), + blob_size INTEGER NOT NULL CHECK(blob_size >= 0), + append_start_offset INTEGER NOT NULL CHECK(append_start_offset >= 0), + append_end_offset INTEGER NOT NULL CHECK(append_end_offset > append_start_offset), + matched_after_codex_header_strip INTEGER NOT NULL CHECK(matched_after_codex_header_strip IN (0, 1)), + previous_revision_authority TEXT NOT NULL CHECK(previous_revision_authority IN ('asserted', 'byte_proven', 'quarantined')), + compared_at_ms INTEGER NOT NULL CHECK(compared_at_ms >= 0), + tool_version TEXT NOT NULL, + backup_manifest_path TEXT NOT NULL, + detail TEXT NOT NULL DEFAULT '' +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_raw_append_chain_backfill_receipts_compared_at +ON raw_append_chain_backfill_receipts(compared_at_ms); diff --git a/tests/unit/devtools/test_verify_raw_authority_frontier_executability.py b/tests/unit/devtools/test_verify_raw_authority_frontier_executability.py new file mode 100644 index 0000000000..fc4118d4b9 --- /dev/null +++ b/tests/unit/devtools/test_verify_raw_authority_frontier_executability.py @@ -0,0 +1,166 @@ +"""polylogue-lb39z (Phase 1, item 4): static frontier-state executability lint. + +Proves this lint catches the exact defect class polylogue-w32w found (a +dispatched actuator paired with a non-executable state) statically, from +source alone -- without ever constructing a ``RawAuthorityFrontierItem`` -- +so it fails at review time even for a branch no test exercises. Also proves +the live repo's real ``raw_reconciler.py`` currently passes (anti-regression: +the fixed state post-#3466). +""" + +from __future__ import annotations + +from pathlib import Path + +from devtools import verify_raw_authority_frontier_executability as lint + + +def test_live_repo_raw_reconciler_has_no_unreachable_actuator_pairing() -> None: + """Anti-regression: the real, current raw_reconciler.py passes clean.""" + report = lint.compute_executability_report() + assert report.ok + assert report.violations == () + # Sanity: this lint actually found real construction sites, not zero + # (a lint that silently matches nothing would trivially "pass"). + assert len(report.pairs) > 10 + + +def test_detects_dispatched_actuator_paired_with_non_executable_state(tmp_path: Path) -> None: + """Anti-vacuity: reproduce the exact polylogue-w32w defect shape in a fixture. + + A synthetic module pairing REFINE_QUARANTINE (a dispatched actuator) with + UNRESOLVED_PROVENANCE (not in _EXECUTABLE_STATES) -- the precise + pre-#3466 shape -- must be flagged as a violation. + """ + fixture = tmp_path / "fixture_reconciler.py" + fixture.write_text( + "\n".join( + [ + "from polylogue.storage.raw_reconciler import RawAuthorityActuator, RawAuthorityFrontierState", + "", + "def _classify(row):", + " return _item(", + " state=RawAuthorityFrontierState.UNRESOLVED_PROVENANCE,", + " actuator=RawAuthorityActuator.REFINE_QUARANTINE,", + " row=row,", + " reason='pre-w32w regression shape',", + " )", + "", + ] + ), + encoding="utf-8", + ) + report = lint.compute_executability_report(fixture) + assert not report.ok + assert len(report.violations) == 1 + violation = report.violations[0] + assert violation.state == "UNRESOLVED_PROVENANCE" + assert violation.actuator == "REFINE_QUARANTINE" + assert violation.callee == "_item" + + +def test_safely_rekeyable_pairing_with_same_actuator_passes(tmp_path: Path) -> None: + """Control: the SAME dispatched actuator paired with an executable state is fine.""" + fixture = tmp_path / "fixture_reconciler_ok.py" + fixture.write_text( + "\n".join( + [ + "from polylogue.storage.raw_reconciler import RawAuthorityActuator, RawAuthorityFrontierState", + "", + "def _classify(row):", + " return _item(", + " state=RawAuthorityFrontierState.SAFELY_REKEYABLE,", + " actuator=RawAuthorityActuator.REFINE_QUARANTINE,", + " row=row,", + " reason='fixed shape',", + " )", + "", + ] + ), + encoding="utf-8", + ) + report = lint.compute_executability_report(fixture) + assert report.ok + assert len(report.pairs) == 1 + + +def test_non_dispatched_actuator_paired_with_non_executable_state_is_fine(tmp_path: Path) -> None: + """Control: RawAuthorityActuator.NONE (never dispatched) is always safe to pair + with a non-executable state -- most terminal/informational states use exactly + this shape (CORRUPT, PROVEN_CURRENT, SUPERSEDED, UNRESOLVED_PROVENANCE).""" + fixture = tmp_path / "fixture_reconciler_none.py" + fixture.write_text( + "\n".join( + [ + "from polylogue.storage.raw_reconciler import RawAuthorityActuator, RawAuthorityFrontierState", + "", + "def _classify(row):", + " return _item(", + " state=RawAuthorityFrontierState.UNRESOLVED_PROVENANCE,", + " actuator=RawAuthorityActuator.NONE,", + " row=row,", + " reason='terminal, no actuator',", + " )", + "", + ] + ), + encoding="utf-8", + ) + report = lint.compute_executability_report(fixture) + assert report.ok + + +def test_dynamic_forwarding_site_is_reported_but_never_fails(tmp_path: Path) -> None: + """A state/actuator sourced from a variable (not a literal enum attribute) + cannot be statically resolved -- reported as informational, not a violation + (mirrors the real _item(state=strategy_override.state, ...) forwarding call).""" + fixture = tmp_path / "fixture_reconciler_dynamic.py" + fixture.write_text( + "\n".join( + [ + "def _classify(row, strategy_override):", + " return _item(", + " state=strategy_override.state,", + " actuator=strategy_override.actuator,", + " row=row,", + " reason='forwarded override',", + " )", + "", + ] + ), + encoding="utf-8", + ) + report = lint.compute_executability_report(fixture) + assert report.ok + assert report.pairs == () + assert len(report.dynamic_sites) == 1 + assert report.dynamic_sites[0].callee == "_item" + + +def test_unknown_enum_member_name_raises_instead_of_silently_passing(tmp_path: Path) -> None: + """Fail closed: an unrecognized state/actuator name (e.g. a rename this lint's + own imports haven't caught up with) must not be silently treated as 'no violation'.""" + fixture = tmp_path / "fixture_reconciler_unknown.py" + fixture.write_text( + "\n".join( + [ + "from polylogue.storage.raw_reconciler import RawAuthorityActuator, RawAuthorityFrontierState", + "", + "def _classify(row):", + " return _item(", + " state=RawAuthorityFrontierState.NOT_A_REAL_MEMBER,", + " actuator=RawAuthorityActuator.NONE,", + " row=row,", + " reason='typo',", + " )", + "", + ] + ), + encoding="utf-8", + ) + try: + lint.compute_executability_report(fixture) + except ValueError as exc: + assert "NOT_A_REAL_MEMBER" in str(exc) + else: + raise AssertionError("expected ValueError for an unknown enum member name") diff --git a/tests/unit/maintenance/test_raw_append_chain_backfill_apply.py b/tests/unit/maintenance/test_raw_append_chain_backfill_apply.py new file mode 100644 index 0000000000..74d9adb744 --- /dev/null +++ b/tests/unit/maintenance/test_raw_append_chain_backfill_apply.py @@ -0,0 +1,272 @@ +"""polylogue-lb39z (Phase 1, item 3): the actuator for membershipless append backfill. + +Builds a fixture archive with one row of each classification (exact_match / +diverged / source_missing) plus a control row that already has a membership +row, and proves: + +* dry-run (the default) never mutates anything; +* --apply promotes only the exact_match row to revision_authority= + 'byte_proven' with revision_authority_evidence='live_source_verification_v1', + and writes an immutable receipt; +* diverged and source_missing rows are left untouched, no exceptions; +* a row that already has a raw_session_memberships row is never touched here + even when its bytes match exactly -- proving the NOT EXISTS scope is + respected end-to-end through the actuator, not just the classifier; +* applying without a backup manifest is refused before anything is touched; +* predecessor_raw_id / baseline_raw_id / acquisition_generation are never + touched -- that revision-graph linkage belongs to classify_raw_revision_cohort + / _promote_contiguous_append_evidence, exactly like polylogue-u19l's actuator. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from polylogue.archive.revision_authority import RawRevisionAuthority, RawRevisionEnvelope, RawRevisionKind +from polylogue.core.enums import Provider +from polylogue.maintenance.raw_append_chain_backfill_apply import ( + TOOL_VERSION, + RawAppendChainBackfillApplyError, + apply_raw_append_chain_backfill, +) +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + +_PREFIX = b"PREFIX-BYTES" + + +def _write_append_raw( + archive: ArchiveStore, + *, + raw_id: str, + payload: bytes, + source_path: str, + logical_source_key: str, + start_offset: int, + end_offset: int, +) -> None: + archive.write_raw_payload( + provider=Provider.CLAUDE_CODE, + payload=payload, + source_path=source_path, + source_index=-1, + acquired_at_ms=1_700_000_000_000, + raw_id=raw_id, + revision=RawRevisionEnvelope( + logical_source_key=logical_source_key, + kind=RawRevisionKind.APPEND, + source_revision=f"{raw_id}-revision", + acquisition_generation=0, + predecessor_source_revision=f"{raw_id}-predecessor", + append_start_offset=start_offset, + append_end_offset=end_offset, + authority=RawRevisionAuthority.QUARANTINED, + ), + ) + + +def _write_membership(conn: sqlite3.Connection, *, raw_id: str, logical_source_key: str) -> None: + conn.execute( + """ + INSERT INTO raw_session_memberships ( + raw_id, logical_source_key, provider_session_id, source_revision, + normalized_content_hash, message_count, revision_authority + ) VALUES (?, ?, 'session-1', 'rev-1', ?, 1, 'quarantined') + """, + (raw_id, logical_source_key, b"\x00" * 32), + ) + + +def _build_fixture_archive(tmp_path: Path) -> Path: + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + + delta = b'{"delta":1}\n' + + exact_live = tmp_path / "exact.jsonl" + exact_live.write_bytes(_PREFIX + delta) + + diverged_live = tmp_path / "diverged.jsonl" + diverged_live.write_bytes(_PREFIX + b'{"CHANGED":true}\n') + + missing_live_path = str(tmp_path / "gone.jsonl") + + has_membership_live = tmp_path / "has-membership.jsonl" + has_membership_live.write_bytes(_PREFIX + delta) + + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + _write_append_raw( + archive, + raw_id="raw-exact", + payload=delta, + source_path=str(exact_live), + logical_source_key="claude-code:exact", + start_offset=len(_PREFIX), + end_offset=len(_PREFIX) + len(delta), + ) + _write_append_raw( + archive, + raw_id="raw-diverged", + payload=delta, + source_path=str(diverged_live), + logical_source_key="claude-code:diverged", + start_offset=len(_PREFIX), + end_offset=len(_PREFIX) + len(delta), + ) + _write_append_raw( + archive, + raw_id="raw-missing", + payload=delta, + source_path=missing_live_path, + logical_source_key="claude-code:missing", + start_offset=0, + end_offset=len(delta), + ) + _write_append_raw( + archive, + raw_id="raw-has-membership", + payload=delta, + source_path=str(has_membership_live), + logical_source_key="claude-code:has-membership", + start_offset=len(_PREFIX), + end_offset=len(_PREFIX) + len(delta), + ) + archive.commit() + + conn = sqlite3.connect(archive_root / "source.db") + try: + _write_membership(conn, raw_id="raw-has-membership", logical_source_key="claude-code:has-membership") + conn.commit() + finally: + conn.close() + + return archive_root + + +def _revision_authority_rows(archive_root: Path) -> dict[str, tuple[str, str | None]]: + conn = sqlite3.connect(archive_root / "source.db") + try: + rows = conn.execute( + "SELECT raw_id, revision_authority, revision_authority_evidence FROM raw_sessions" + ).fetchall() + finally: + conn.close() + return {raw_id: (authority, evidence) for raw_id, authority, evidence in rows} + + +def _linkage_rows(archive_root: Path) -> dict[str, tuple[object, object, object]]: + conn = sqlite3.connect(archive_root / "source.db") + try: + rows = conn.execute( + "SELECT raw_id, predecessor_raw_id, baseline_raw_id, acquisition_generation FROM raw_sessions" + ).fetchall() + finally: + conn.close() + return {raw_id: (pred, baseline, gen) for raw_id, pred, baseline, gen in rows} + + +def _receipt_rows(archive_root: Path) -> dict[str, str]: + conn = sqlite3.connect(archive_root / "source.db") + try: + rows = conn.execute("SELECT raw_id, tool_version FROM raw_append_chain_backfill_receipts").fetchall() + finally: + conn.close() + return dict(rows) + + +def test_dry_run_makes_zero_mutations(tmp_path: Path) -> None: + archive_root = _build_fixture_archive(tmp_path) + before = _revision_authority_rows(archive_root) + assert all(authority == "quarantined" for authority, _evidence in before.values()) + + report = apply_raw_append_chain_backfill(archive_root, dry_run=True) + + assert report.applied is False + assert report.scanned_count == 3 # exact, diverged, missing -- has-membership excluded + assert report.promoted_count == 1 + assert report.diverged_count == 1 + assert report.source_missing_count == 1 + assert report.promoted_raw_ids == () + + assert _revision_authority_rows(archive_root) == before + assert _receipt_rows(archive_root) == {} + + +def test_apply_promotes_only_membershipless_exact_match(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + archive_root = _build_fixture_archive(tmp_path) + + validated: list[tuple[Path, object]] = [] + + def _fake_validate(manifest: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + validated.append((manifest, tier)) + assert connection.execute("SELECT 1").fetchone() == (1,) + return manifest.with_name("verification-receipt.json") + + monkeypatch.setattr( + "polylogue.maintenance.raw_append_chain_backfill_apply.validate_migration_backup_manifest", + _fake_validate, + ) + + linkage_before = _linkage_rows(archive_root) + + manifest = tmp_path / "verified-backup" / "manifest.json" + report = apply_raw_append_chain_backfill(archive_root, backup_manifest=manifest, dry_run=False) + + assert report.applied is True + assert report.scanned_count == 3 + assert report.promoted_count == 1 + assert set(report.promoted_raw_ids) == {"raw-exact"} + assert report.backup_manifest == manifest + assert validated == [(manifest, ArchiveTier.SOURCE), (manifest, ArchiveTier.SOURCE)] + + rows = _revision_authority_rows(archive_root) + assert rows["raw-exact"] == ("byte_proven", "live_source_verification_v1") + # Never touched. + assert rows["raw-diverged"] == ("quarantined", None) + assert rows["raw-missing"] == ("quarantined", None) + # Has a membership row -- out of this actuator's scope even though its + # bytes match exactly. + assert rows["raw-has-membership"] == ("quarantined", None) + + receipts = _receipt_rows(archive_root) + assert receipts == {"raw-exact": TOOL_VERSION} + + # Revision-graph linkage is untouched -- classify_raw_revision_cohort / + # _promote_contiguous_append_evidence own that. + linkage_after = _linkage_rows(archive_root) + assert linkage_after["raw-exact"] == linkage_before["raw-exact"] + + +def test_apply_refuses_without_backup_manifest(tmp_path: Path) -> None: + archive_root = _build_fixture_archive(tmp_path) + before = _revision_authority_rows(archive_root) + + with pytest.raises(RawAppendChainBackfillApplyError, match="backup manifest"): + apply_raw_append_chain_backfill(archive_root, backup_manifest=None, dry_run=False) + + assert _revision_authority_rows(archive_root) == before + assert _receipt_rows(archive_root) == {} + + +def test_apply_refuses_when_backup_manifest_invalid(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + archive_root = _build_fixture_archive(tmp_path) + before = _revision_authority_rows(archive_root) + + def _reject(manifest: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + raise ValueError("backup manifest does not match live source.db") + + monkeypatch.setattr( + "polylogue.maintenance.raw_append_chain_backfill_apply.validate_migration_backup_manifest", + _reject, + ) + + manifest = tmp_path / "stale-backup" / "manifest.json" + with pytest.raises(ValueError, match="does not match"): + apply_raw_append_chain_backfill(archive_root, backup_manifest=manifest, dry_run=False) + + assert _revision_authority_rows(archive_root) == before + assert _receipt_rows(archive_root) == {} diff --git a/tests/unit/storage/test_durable_migrations.py b/tests/unit/storage/test_durable_migrations.py index 83a825a394..0e5482a6bb 100644 --- a/tests/unit/storage/test_durable_migrations.py +++ b/tests/unit/storage/test_durable_migrations.py @@ -489,7 +489,7 @@ def test_source_tier_v1_migrates_to_current_without_native_uniqueness( result = migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=manifest) assert result.from_version == 1 assert result.to_version == SOURCE_SCHEMA_VERSION - assert result.applied_versions == (2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19) + assert result.applied_versions == (2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20) assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == SOURCE_SCHEMA_VERSION columns = {str(row[1]) for row in conn.execute("PRAGMA table_info('raw_sessions')")} assert "predecessor_source_revision" in columns @@ -577,8 +577,8 @@ def test_source_publication_backfill_requires_verified_backup( result = migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=manifest) assert result.from_version == 9 - assert result.to_version == SOURCE_SCHEMA_VERSION == 19 - assert result.applied_versions == (10, 11, 12, 13, 14, 15, 16, 17, 18, 19) + assert result.to_version == SOURCE_SCHEMA_VERSION == 20 + assert result.applied_versions == (10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20) assert result.backup_receipt == manifest.with_name("verification-receipt.json") tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")} assert { @@ -732,8 +732,8 @@ def test_source_tier_v7_expands_origin_checks_with_verified_backup( with sqlite3.connect(db_path) as conn: result = migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=manifest) assert result.from_version == 7 - assert result.to_version == SOURCE_SCHEMA_VERSION == 19 - assert result.applied_versions == (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19) + assert result.to_version == SOURCE_SCHEMA_VERSION == 20 + assert result.applied_versions == (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20) assert conn.execute( """ SELECT predecessor_source_revision, predecessor_raw_id, baseline_raw_id, @@ -894,7 +894,7 @@ def test_source_tier_v2_migrates_to_v3_dropping_pending_blob_refs( assert result.from_version == 2 assert result.to_version == SOURCE_SCHEMA_VERSION - assert result.applied_versions == (3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19) + assert result.applied_versions == (3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20) assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == SOURCE_SCHEMA_VERSION assert not conn.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name='pending_blob_refs'" @@ -952,7 +952,7 @@ def test_source_tier_v3_adds_publication_reservations_with_verified_backup_recei conn = sqlite3.connect(db_path) try: result = migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=manifest) - assert result.applied_versions == (4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19) + assert result.applied_versions == (4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20) assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == SOURCE_SCHEMA_VERSION conn.execute( """ @@ -1021,14 +1021,14 @@ def test_source_tier_v13_adds_raw_sessions_blob_hash_index( result = migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=manifest) assert result.from_version == 13 - assert result.to_version == SOURCE_SCHEMA_VERSION == 19 + assert result.to_version == SOURCE_SCHEMA_VERSION == 20 # v13 -> current also picks up migrations 015 (polylogue-hord), 016 - # (polylogue-buns), 017 (polylogue-byw3y), 018 (polylogue-u19l), and - # 019 (polylogue-lb39z), all added after this test -- a v13 fixture - # migrating to "current" is exactly the shape a real archive frozen - # at v13 would go through. - assert result.applied_versions == (14, 15, 16, 17, 18, 19) - assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == 19 + # (polylogue-buns), 017 (polylogue-byw3y), 018 (polylogue-u19l), 019 + # (polylogue-lb39z item 2), and 020 (polylogue-lb39z item 3), all added + # after this test -- a v13 fixture migrating to "current" is exactly + # the shape a real archive frozen at v13 would go through. + assert result.applied_versions == (14, 15, 16, 17, 18, 19, 20) + assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == 20 indexes_after = {row[1] for row in conn.execute("PRAGMA index_list('raw_sessions')")} assert "idx_raw_sessions_blob_hash" in indexes_after assert "idx_raw_sessions_blob_hash_raw_id" in indexes_after @@ -1089,9 +1089,9 @@ def test_source_tier_v14_adds_raw_sessions_blob_hash_raw_id_index( result = migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=manifest) assert result.from_version == 14 - assert result.to_version == SOURCE_SCHEMA_VERSION == 19 - assert result.applied_versions == (15, 16, 17, 18, 19) - assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == 19 + assert result.to_version == SOURCE_SCHEMA_VERSION == 20 + assert result.applied_versions == (15, 16, 17, 18, 19, 20) + assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == 20 indexes_after = {row[1] for row in conn.execute("PRAGMA index_list('raw_sessions')")} assert "idx_raw_sessions_blob_hash_raw_id" in indexes_after plan = conn.execute( diff --git a/tests/unit/storage/test_raw_append_chain_backfill.py b/tests/unit/storage/test_raw_append_chain_backfill.py new file mode 100644 index 0000000000..8cfc043834 --- /dev/null +++ b/tests/unit/storage/test_raw_append_chain_backfill.py @@ -0,0 +1,159 @@ +"""polylogue-lb39z (Phase 1, item 3): membershipless append-chain backfill. + +Proves the read-only classifier scopes strictly to the named population: +quarantined, revision_kind='append', with zero raw_session_memberships rows +-- and that it reuses the identical live-source byte-window proof +polylogue-u19l already validated, including the Codex header-strip case. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +from polylogue.archive.revision_authority import RawRevisionAuthority, RawRevisionEnvelope, RawRevisionKind +from polylogue.core.enums import Provider +from polylogue.storage.blob_store import BlobStore +from polylogue.storage.live_source_reconciliation import LiveSourceVerdict +from polylogue.storage.raw_append_chain_backfill import plan_append_chain_backfill +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + +def _write_raw( + archive: ArchiveStore, + *, + raw_id: str, + payload: bytes, + source_path: str, + kind: RawRevisionKind, + logical_source_key: str, + append_start_offset: int | None = None, + append_end_offset: int | None = None, +) -> None: + predecessor_source_revision = f"{raw_id}-predecessor" if kind is RawRevisionKind.APPEND else None + archive.write_raw_payload( + provider=Provider.CLAUDE_CODE, + payload=payload, + source_path=source_path, + source_index=-1, + acquired_at_ms=1_700_000_000_000, + raw_id=raw_id, + revision=RawRevisionEnvelope( + logical_source_key=logical_source_key, + kind=kind, + source_revision=f"{raw_id}-revision", + acquisition_generation=0, + predecessor_source_revision=predecessor_source_revision, + append_start_offset=append_start_offset, + append_end_offset=append_end_offset, + authority=RawRevisionAuthority.QUARANTINED, + ), + ) + + +def _write_membership(conn: sqlite3.Connection, *, raw_id: str, logical_source_key: str) -> None: + conn.execute( + """ + INSERT INTO raw_session_memberships ( + raw_id, logical_source_key, provider_session_id, source_revision, + normalized_content_hash, message_count, revision_authority + ) VALUES (?, ?, 'session-1', 'rev-1', ?, 1, 'quarantined') + """, + (raw_id, logical_source_key, b"\x00" * 32), + ) + + +def test_scopes_to_quarantined_append_rows_with_no_membership_row(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + + exact_live = tmp_path / "exact.jsonl" + exact_live.write_bytes(b"PREFIX-BYTES" + b'{"delta":1}\n') + + diverged_live = tmp_path / "diverged.jsonl" + diverged_live.write_bytes(b"PREFIX-BYTES" + b'{"CHANGED":true}\n') + + missing_live_path = str(tmp_path / "gone.jsonl") + + has_membership_live = tmp_path / "has-membership.jsonl" + has_membership_live.write_bytes(b"PREFIX-BYTES" + b'{"delta":1}\n') + + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + # Membershipless, exact match at its own append window -- the target population. + _write_raw( + archive, + raw_id="raw-membershipless-exact", + payload=b'{"delta":1}\n', + source_path=str(exact_live), + kind=RawRevisionKind.APPEND, + logical_source_key="claude-code:exact", + append_start_offset=len(b"PREFIX-BYTES"), + append_end_offset=len(b"PREFIX-BYTES") + len(b'{"delta":1}\n'), + ) + # Membershipless, but its own window has diverged. + _write_raw( + archive, + raw_id="raw-membershipless-diverged", + payload=b'{"delta":1}\n', + source_path=str(diverged_live), + kind=RawRevisionKind.APPEND, + logical_source_key="claude-code:diverged", + append_start_offset=len(b"PREFIX-BYTES"), + append_end_offset=len(b"PREFIX-BYTES") + len(b'{"delta":1}\n'), + ) + # Membershipless, source file gone. + _write_raw( + archive, + raw_id="raw-membershipless-missing", + payload=b'{"delta":1}\n', + source_path=missing_live_path, + kind=RawRevisionKind.APPEND, + logical_source_key="claude-code:missing", + append_start_offset=0, + append_end_offset=len(b'{"delta":1}\n'), + ) + # HAS a membership row already -- must be excluded even though its + # bytes would otherwise match exactly (this is u19l's population, + # not item 3's). + _write_raw( + archive, + raw_id="raw-has-membership", + payload=b'{"delta":1}\n', + source_path=str(has_membership_live), + kind=RawRevisionKind.APPEND, + logical_source_key="claude-code:has-membership", + append_start_offset=len(b"PREFIX-BYTES"), + append_end_offset=len(b"PREFIX-BYTES") + len(b'{"delta":1}\n'), + ) + # A membershipless FULL row -- out of scope (title says "append-chain"). + _write_raw( + archive, + raw_id="raw-membershipless-full", + payload=b'{"a":1}\n', + source_path=str(tmp_path / "full.jsonl"), + kind=RawRevisionKind.FULL, + logical_source_key="claude-code:full", + ) + (tmp_path / "full.jsonl").write_bytes(b'{"a":1}\n') + archive.commit() + + conn = sqlite3.connect(archive_root / "source.db") + try: + _write_membership(conn, raw_id="raw-has-membership", logical_source_key="claude-code:has-membership") + conn.commit() + finally: + conn.close() + + conn = sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True) + try: + plan = plan_append_chain_backfill(conn, blob_store=BlobStore(archive_root / "blob")) + finally: + conn.close() + + assert plan.scanned_count == 3 # exact, diverged, missing -- has-membership and full excluded + assert {c.raw_id for c in plan.exact_match} == {"raw-membershipless-exact"} + assert {c.raw_id for c in plan.diverged} == {"raw-membershipless-diverged"} + assert {c.raw_id for c in plan.source_missing} == {"raw-membershipless-missing"} + for candidate in plan.exact_match: + assert candidate.comparison.verdict == LiveSourceVerdict.EXACT_MATCH