From 2ba1d7116654d144499ba1ba61c0d755c6ca77a9 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 5 Aug 2026 05:44:25 +0200 Subject: [PATCH 1/3] refactor(devtools): prove lifecycle fast-forwards with raw replay Problem: the surviving v37 actuator had a per-version implementation and no source-backed semantic proof before promotion. What changed: lifecycle declarations now drive one generic actuator and executor. Preparation replays deterministic retained-raw samples through the production parser and writer, compares canonical rows, records sample IDs and parser, lowering, and materializer fingerprints, and activation refuses drift or incomplete proof. The v37 orphan repair is now a declared lifecycle operation. The obsolete actuator and focused test were removed. Compatibility/migration: semantic-reparse declarations remain ineligible for SQL fast-forward and continue through replay or rebuild. The command is now workspace index-fast-forward. --- devtools/command_catalog.py | 16 +- devtools/index_fast_forward.py | 590 ++++++++++++++++++ devtools/index_v37_fast_forward.py | 574 ----------------- docs/devtools.md | 2 +- docs/internals.md | 12 + polylogue/storage/index_generation.py | 13 +- .../index_fast_forward_executor.py | 17 + polylogue/storage/sqlite/lifecycle.py | 6 + .../unit/devtools/test_index_fast_forward.py | 187 ++++++ .../devtools/test_index_v37_fast_forward.py | 390 ------------ 10 files changed, 825 insertions(+), 982 deletions(-) create mode 100644 devtools/index_fast_forward.py delete mode 100644 devtools/index_v37_fast_forward.py create mode 100644 tests/unit/devtools/test_index_fast_forward.py delete mode 100644 tests/unit/devtools/test_index_v37_fast_forward.py diff --git a/devtools/command_catalog.py b/devtools/command_catalog.py index a41f6d6eed..761292cc95 100644 --- a/devtools/command_catalog.py +++ b/devtools/command_catalog.py @@ -589,18 +589,18 @@ def to_dict(self) -> dict[str, object]: ), ), CommandSpec( - "workspace index-v37-fast-forward", + "workspace index-fast-forward", "workspace", - "Clone-forward index v36 to v37 by retiring derived caches without raw replay.", - "devtools.index_v37_fast_forward", + "Plan and prove a declared index fast-forward against retained raw replay.", + "devtools.index_fast_forward", use_when=( - "Advance a stopped exact-shape v36 index to v37. The actuator reflink-clones the active generation, " - "drops only the three retired run-projection caches, proves surviving schema and row-count parity, " - "then separately atomically activates the proven generation." + "Advance a stopped index generation across a declared clone-safe schema gap. The actuator clones the " + "active generation, applies lifecycle operations, proves a deterministic retained-raw sample through " + "the production parser/materializer route, then atomically activates the proven generation." ), examples=( - "devtools workspace index-v37-fast-forward prepare --archive-root /path/to/archive --receipt /path/to/receipt.json", - "devtools workspace index-v37-fast-forward activate --receipt /path/to/receipt.json", + "devtools workspace index-fast-forward prepare --archive-root /path/to/archive --receipt /path/to/receipt.json", + "devtools workspace index-fast-forward activate --receipt /path/to/receipt.json", ), ), CommandSpec( diff --git a/devtools/index_fast_forward.py b/devtools/index_fast_forward.py new file mode 100644 index 0000000000..bbf77f1a69 --- /dev/null +++ b/devtools/index_fast_forward.py @@ -0,0 +1,590 @@ +"""Plan-driven, source-proven fast-forward for rebuildable index generations. + +The lifecycle declaration is the only version authority. This actuator owns +the clone, source-backed proof, receipt, and promotion boundary. The proof +replays a deterministic sample of retained raw evidence through the production +parser and session writer into a scratch current-schema index, then compares +canonical sessions/messages/blocks/FTS rows with the transformed clone. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sqlite3 +import time +import uuid +from contextlib import closing, suppress +from dataclasses import asdict +from pathlib import Path +from typing import cast + +from devtools.clone_support import reflink_clone +from polylogue.config import Config +from polylogue.maintenance.archive_verification import CORPUS_FIDELITY_CHECKS, verify_archive +from polylogue.maintenance.offline_guard import running_daemon_pid +from polylogue.pipeline.ids import session_content_hash +from polylogue.sources.origin_specs import lowering_fingerprint, parser_fingerprint_for_origin +from polylogue.storage.blob_publication import ArchiveBlobPublisher +from polylogue.storage.index_generation import IndexGenerationStore, RebuildLease, source_revision_snapshot +from polylogue.storage.runtime.store_constants import SESSION_INSIGHT_MATERIALIZER_VERSION +from polylogue.storage.sqlite.archive_tiers import ARCHIVE_DDL_BY_TIER +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.index import INDEX_DDL +from polylogue.storage.sqlite.archive_tiers.index_fast_forward_executor import apply_index_fast_forward +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive +from polylogue.storage.sqlite.connection_profile import open_readonly_connection +from polylogue.storage.sqlite.lifecycle import IndexFastForwardPlan, index_fast_forward_plan +from polylogue.storage.sqlite.runtime_indexes import ensure_runtime_indexes_sync + +RECEIPT_SCHEMA = "polylogue.index-fast-forward.v1" +DEFAULT_SAMPLE_SIZE = 8 + + +class IndexFastForwardError(RuntimeError): + """The declared index transition could not be proven safe to activate.""" + + +def _now_ms() -> int: + return int(time.time() * 1000) + + +def _normalize_ddl(sql: str) -> str: + """Normalize canonical DDL without collapsing literal or identifier boundaries.""" + tokens: list[str] = [] + current: list[str] = [] + quote: str | None = None + for char in sql: + if quote is not None: + current.append(char) + if char == quote: + quote = None + continue + if char in {"'", '"', "`"}: + if current: + tokens.append("".join(current).lower()) + current.clear() + quote = char + current.append(char) + elif char.isalnum() or char in {"_", "$", "."}: + current.append(char) + else: + if current: + tokens.append("".join(current).lower()) + current.clear() + if not char.isspace(): + tokens.append(char) + if current: + tokens.append("".join(current).lower()) + normalized: list[str] = [] + index = 0 + while index < len(tokens): + if tokens[index : index + 3] == ["if", "not", "exists"]: + index += 3 + continue + normalized.append(tokens[index]) + index += 1 + return json.dumps(normalized, separators=(",", ":")) + + +def _schema_objects(conn: sqlite3.Connection) -> dict[str, str]: + rows = conn.execute( + """ + SELECT type, name, sql FROM sqlite_master + WHERE type IN ('table', 'index', 'view', 'trigger') + AND name NOT LIKE 'sqlite_%' AND sql IS NOT NULL + ORDER BY type, name + """ + ) + return {f"{row[0]}:{row[1]}": _normalize_ddl(str(row[2])) for row in rows} + + +def _canonical_schema_objects() -> dict[str, str]: + with closing(sqlite3.connect(":memory:")) as conn: + conn.executescript(ARCHIVE_DDL_BY_TIER[ArchiveTier.INDEX]) + ensure_runtime_indexes_sync(conn) + return _schema_objects(conn) + + +def _canonical_schema_sha256(schema: dict[str, str]) -> str: + return hashlib.sha256(json.dumps(schema, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def _file_identity(path: Path) -> dict[str, object]: + resolved = path.resolve(strict=True) + stat = resolved.stat() + return { + "path": str(path), + "resolved_path": str(resolved), + "size_bytes": stat.st_size, + "allocated_bytes": stat.st_blocks * 512, + "inode": stat.st_ino, + "mtime_ns": stat.st_mtime_ns, + } + + +def _proven_clone_identity(path: Path) -> dict[str, object]: + identity = _file_identity(path) + digest = hashlib.sha256() + with path.resolve(strict=True).open("rb") as handle: + while chunk := handle.read(8 * 1024 * 1024): + digest.update(chunk) + identity["sha256"] = digest.hexdigest() + return identity + + +def _receipt_hash(payload: dict[str, object]) -> str: + body = {key: value for key, value in payload.items() if key != "receipt_sha256"} + return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def _write_receipt(path: Path, payload: dict[str, object]) -> None: + payload["receipt_sha256"] = _receipt_hash(payload) + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + with temporary.open("rb") as handle: + os.fsync(handle.fileno()) + os.replace(temporary, path) + descriptor = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _load_receipt(path: Path) -> dict[str, object]: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict) or payload.get("schema") != RECEIPT_SCHEMA: + raise IndexFastForwardError(f"invalid fast-forward receipt: {path}") + typed = cast(dict[str, object], payload) + if typed.get("receipt_sha256") != _receipt_hash(typed): + raise IndexFastForwardError(f"fast-forward receipt hash mismatch: {path}") + return typed + + +def _config(archive_root: Path) -> Config: + return Config( + archive_root=archive_root, render_root=archive_root / "render", sources=[], db_path=archive_root / "index.db" + ) + + +def _require_daemon_stopped(archive_root: Path) -> None: + if (pid := running_daemon_pid(_config(archive_root))) is not None: + raise IndexFastForwardError(f"polylogued PID {pid} is still running") + + +def _require_receipt_destination_writable(path: Path) -> None: + probe = path.with_name(f".{path.name}.{uuid.uuid4().hex}.probe") + try: + path.parent.mkdir(parents=True, exist_ok=True) + if path.exists() and not path.is_file(): + raise OSError(f"receipt destination is not a regular file: {path}") + descriptor = os.open(probe, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + os.close(descriptor) + probe.unlink() + except OSError as exc: + with suppress(OSError): + probe.unlink(missing_ok=True) + raise IndexFastForwardError(f"receipt destination is not writable: {path}: {exc}") from exc + + +def _checkpoint_stopped_database(path: Path, *, label: str = "active index") -> None: + resolved = path.resolve(strict=True) + with closing(sqlite3.connect(resolved, timeout=120.0)) as conn: + checkpoint = conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() + if checkpoint is None or int(checkpoint[0]) != 0 or int(checkpoint[1]) != int(checkpoint[2]): + raise IndexFastForwardError(f"{label} WAL checkpoint failed: {checkpoint}") + for suffix in ("-wal", "-shm"): + sidecar = Path(f"{resolved}{suffix}") + if sidecar.exists(): + sidecar.unlink() + + +def _inspect_clean_database(path: Path) -> tuple[int, dict[str, str]]: + for suffix in ("-wal", "-shm", "-journal"): + sidecar = Path(f"{path.resolve(strict=True)}{suffix}") + if sidecar.exists() and sidecar.stat().st_size: + raise IndexFastForwardError(f"non-empty SQLite sidecar blocks fast-forward: {sidecar}") + with closing(open_readonly_connection(path.resolve(strict=True), immutable=True)) as conn: + return int(conn.execute("PRAGMA user_version").fetchone()[0]), _schema_objects(conn) + + +def _plan_for_database(version: int) -> IndexFastForwardPlan: + from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER + + target = int(ARCHIVE_VERSION_BY_TIER[ArchiveTier.INDEX]) + plan = index_fast_forward_plan(version, target) + if plan is None or not plan.eligible_for_sql_fast_forward: + raise IndexFastForwardError( + f"index v{version} to v{target} is not SQL-fast-forwardable; route semantic work to replay/rebuild" + ) + return plan + + +def _expected_surplus(plan: IndexFastForwardPlan) -> set[str]: + return { + f"{kind}:{name}" + for declaration in plan.declarations + for operation in declaration.operations + if operation.kind.value == "drop-table" + for kind, name in operation.objects + } + + +def _transform_clone(path: Path, *, plan: IndexFastForwardPlan, before_schema: dict[str, str]) -> dict[str, object]: + with closing(sqlite3.connect(path, timeout=120.0)) as conn: + apply_index_fast_forward(conn, plan) + version = int(conn.execute("PRAGMA user_version").fetchone()[0]) + checks = [str(row[0]) for row in conn.execute("PRAGMA quick_check")] + after_schema = _schema_objects(conn) + canonical = _canonical_schema_objects() + actual_surplus = set(after_schema) - set(canonical) + expected_surplus = _expected_surplus(plan) + before_surplus = set(before_schema) - set(canonical) + if version != plan.target_version or checks != ["ok"] or after_schema != canonical: + raise IndexFastForwardError( + f"fast-forward postflight failed: version={version}, checks={checks}, " + f"missing={sorted(set(canonical) - set(after_schema))}, surplus={sorted(actual_surplus)}" + ) + if before_surplus != expected_surplus: + raise IndexFastForwardError( + f"active schema does not match declared plan surplus: expected={sorted(expected_surplus)}, " + f"found={sorted(before_surplus)}" + ) + return { + "quick_check": checks, + "schema_object_count": len(after_schema), + "removed_objects": sorted(expected_surplus), + } + + +def _materializer_fingerprint() -> str: + root = Path(__file__).resolve().parents[1] + paths = ( + root / "polylogue/storage/sqlite/archive_tiers/write.py", + root / "polylogue/storage/insights/session/rebuild.py", + root / "polylogue/storage/runtime/store_constants.py", + ) + digest = hashlib.sha256() + for path in paths: + digest.update(path.relative_to(root).as_posix().encode()) + digest.update(path.read_bytes()) + digest.update(str(SESSION_INSIGHT_MATERIALIZER_VERSION).encode()) + return digest.hexdigest() + + +def _fingerprints(origins: tuple[str, ...]) -> dict[str, object]: + return { + "parser": {origin: parser_fingerprint_for_origin(origin) for origin in origins}, + "lowering": lowering_fingerprint(), + "materializer": _materializer_fingerprint(), + "materializer_version": SESSION_INSIGHT_MATERIALIZER_VERSION, + } + + +def _sample_manifest(archive_root: Path, index_path: Path, *, limit: int) -> list[dict[str, object]]: + with closing(sqlite3.connect(archive_root / "source.db")) as source_conn: + raw_ids = [str(row[0]) for row in source_conn.execute("SELECT raw_id FROM raw_sessions ORDER BY raw_id")] + if not raw_ids: + raise IndexFastForwardError("source-backed fast-forward proof requires retained raw evidence") + marks = ", ".join("?" for _ in raw_ids) + with closing(open_readonly_connection(index_path.resolve(strict=True), immutable=True)) as index_conn: + rows = index_conn.execute( + f"SELECT raw_id, session_id, origin FROM sessions WHERE raw_id IN ({marks}) ORDER BY raw_id, session_id", + raw_ids, + ).fetchall() + grouped: dict[str, dict[str, object]] = {} + for raw_id, session_id, origin in rows: + entry = grouped.setdefault(str(raw_id), {"raw_id": str(raw_id), "session_ids": [], "origins": []}) + cast(list[str], entry["session_ids"]).append(str(session_id)) + if str(origin) not in cast(list[str], entry["origins"]): + cast(list[str], entry["origins"]).append(str(origin)) + sample = [grouped[raw_id] for raw_id in sorted(grouped)[:limit]] + if not sample: + raise IndexFastForwardError("source-backed fast-forward proof requires a raw-backed indexed session") + return sample + + +def _json_value(value: object) -> object: + if isinstance(value, bytes): + return {"bytes": value.hex()} + return value + + +def _hash_query(conn: sqlite3.Connection, sql: str, params: tuple[object, ...]) -> str: + rows = conn.execute(sql, params).fetchall() + payload = [[_json_value(value) for value in row] for row in rows] + return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def _canonical_hashes(conn: sqlite3.Connection, session_ids: tuple[str, ...]) -> dict[str, object]: + if not session_ids: + return {"sessions": "", "messages": "", "blocks": "", "fts": "", "scoped": {}} + marks = ", ".join("?" for _ in session_ids) + messages = conn.execute(f"SELECT message_id FROM messages WHERE session_id IN ({marks})", session_ids).fetchall() + message_ids = tuple(str(row[0]) for row in messages) + message_marks = ", ".join("?" for _ in message_ids) or "NULL" + scoped: dict[str, str] = {} + for table_name, key_column in ( + (str(row[0]), key_column) + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name" + ) + for key_column in ("session_id", "src_session_id") + if key_column and key_column in {str(info[1]) for info in conn.execute(f'PRAGMA table_info("{row[0]}")')} + ): + columns = [str(info[1]) for info in conn.execute(f'PRAGMA table_info("{table_name}")')] + order_by = ", ".join(str(index) for index in range(1, len(columns) + 1)) + scoped[table_name] = _hash_query( + conn, + f'SELECT * FROM "{table_name}" WHERE "{key_column}" IN ({marks}) ORDER BY {order_by}', + session_ids, + ) + return { + "sessions": _hash_query( + conn, f"SELECT * FROM sessions WHERE session_id IN ({marks}) ORDER BY session_id", session_ids + ), + "messages": _hash_query( + conn, f"SELECT * FROM messages WHERE session_id IN ({marks}) ORDER BY message_id", session_ids + ), + "blocks": _hash_query( + conn, f"SELECT * FROM blocks WHERE message_id IN ({message_marks}) ORDER BY block_id", message_ids + ), + "fts": _hash_query( + conn, + f"SELECT block_id, message_id, session_id, block_type, text FROM messages_fts WHERE session_id IN ({marks}) ORDER BY rowid", + session_ids, + ), + "scoped": scoped, + } + + +def _replay_sample(archive_root: Path, candidate_index: Path, manifest: list[dict[str, object]]) -> dict[str, object]: + from polylogue.sources.revision_backfill import parse_retained_raw_sessions + + expected_ids = tuple(session_id for entry in manifest for session_id in cast(list[str], entry["session_ids"])) + with closing(sqlite3.connect(":memory:")) as replay_conn: + replay_conn.executescript(INDEX_DDL) + ensure_runtime_indexes_sync(replay_conn) + # ``parse_retained_raw_sessions`` is semantically read-only, but the + # shared raw-revision descriptor requires the production blob + # publisher to be present. Attach its read-only filesystem facade to + # the read-only archive store. No pending blob is queued or flushed. + with ArchiveStore.open_existing(archive_root, read_only=True) as archive: + archive._blob_publisher = ArchiveBlobPublisher(archive_root / "source.db", archive_root / "blob") + replayed_ids: list[str] = [] + for entry in manifest: + raw_id = str(entry["raw_id"]) + for session in parse_retained_raw_sessions(archive, raw_id): + replayed_ids.append( + write_parsed_session_to_archive( + replay_conn, + session, + content_hash=session_content_hash(session), + raw_id=raw_id, + force_replace=True, + ) + ) + actual_ids = tuple(sorted(set(replayed_ids))) + canonical = _canonical_hashes(replay_conn, tuple(sorted(set(expected_ids) | set(actual_ids)))) + with closing(open_readonly_connection(candidate_index.resolve(strict=True), immutable=True)) as candidate_conn: + fast_forward = _canonical_hashes(candidate_conn, tuple(sorted(set(expected_ids) | set(actual_ids)))) + mismatches = [key for key in canonical if canonical[key] != fast_forward[key]] + if tuple(sorted(set(expected_ids))) != actual_ids: + mismatches.append("sample_session_ids") + return { + "fast_forward_hashes": fast_forward, + "canonical_replay_hashes": canonical, + "replayed_session_ids": list(actual_ids), + "mismatch_details": mismatches, + "verdict": "equivalent" if not mismatches else "mismatch", + } + + +def _require_complete_proof(proof: dict[str, object]) -> None: + required_hashes = {"sessions", "messages", "blocks", "fts"} + if proof.get("verdict") != "equivalent": + raise IndexFastForwardError(f"source replay proof failed: {proof.get('mismatch_details')}") + if proof.get("mismatch_details"): + raise IndexFastForwardError(f"source replay proof has mismatches: {proof['mismatch_details']}") + if not proof.get("replayed_session_ids"): + raise IndexFastForwardError("source replay proof has no replayed session ids") + for key in ("fast_forward_hashes", "canonical_replay_hashes"): + hashes = proof.get(key) + if ( + not isinstance(hashes, dict) + or not required_hashes <= set(hashes) + or not all(hashes[key] for key in required_hashes) + ): + raise IndexFastForwardError(f"source replay proof is incomplete: {key}") + scoped = hashes.get("scoped") + if not isinstance(scoped, dict) or not scoped: + raise IndexFastForwardError(f"source replay proof is incomplete: {key}.scoped") + + +def _require_candidate_corpus_fidelity(archive_root: Path, candidate_index: Path) -> None: + report = verify_archive(archive_root, checks=CORPUS_FIDELITY_CHECKS, index_path_override=candidate_index) + if report.blocking: + failing = "; ".join( + f"{check.name}: {check.summary}" for check in report.checks if check.status.value == "error" + ) + raise IndexFastForwardError(f"candidate corpus fidelity gate failed: {failing}") + + +def prepare_forward( + *, archive_root: Path, receipt_path: Path, sample_size: int = DEFAULT_SAMPLE_SIZE +) -> dict[str, object]: + """Create an inactive plan-driven candidate and prove it against retained raw replay.""" + if sample_size <= 0: + raise ValueError("sample_size must be positive") + archive_root = archive_root.resolve(strict=True) + _require_daemon_stopped(archive_root) + _require_receipt_destination_writable(receipt_path) + store = IndexGenerationStore.for_archive_root(archive_root) + with RebuildLease(archive_root): + _require_daemon_stopped(archive_root) + active_pointer = store.active_pointer + _checkpoint_stopped_database(active_pointer) + source_snapshot = source_revision_snapshot(archive_root) + active_identity = _file_identity(active_pointer) + source_version, before_schema = _inspect_clean_database(active_pointer) + plan = _plan_for_database(source_version) + manifest = _sample_manifest(archive_root, active_pointer, limit=sample_size) + origins = tuple(sorted({str(origin) for entry in manifest for origin in cast(list[str], entry["origins"])})) + fingerprints = _fingerprints(origins) + generation = store.create(source_snapshot=source_snapshot) + clone = Path(generation.index_path) + try: + clone.unlink() + reflink_clone(active_pointer, clone) + if _file_identity(active_pointer) != active_identity: + raise IndexFastForwardError("active index changed while its clone was created") + postflight = _transform_clone(clone, plan=plan, before_schema=before_schema) + proof = _replay_sample(archive_root, clone, manifest) + _require_complete_proof(proof) + if source_revision_snapshot(archive_root) != source_snapshot: + raise IndexFastForwardError("source evidence changed while preparing fast-forward") + receipt: dict[str, object] = { + "schema": RECEIPT_SCHEMA, + "status": "prepared", + "prepared_at_ms": _now_ms(), + "archive_root": str(archive_root), + "generation": asdict(generation), + "source_snapshot": source_snapshot, + "active_identity": active_identity, + "clone_identity": _proven_clone_identity(clone), + "source_version": source_version, + "target_version": plan.target_version, + "stage_names": list(plan.stage_names), + "canonical_schema_sha256": _canonical_schema_sha256(_canonical_schema_objects()), + "sample_manifest": manifest, + "fingerprints": fingerprints, + "proof": proof, + "postflight": postflight, + "raw_reparse": False, + } + _write_receipt(receipt_path, receipt) + return receipt + except Exception: + store.discard_if_inactive(generation) + raise + + +def activate_forward(*, receipt_path: Path) -> dict[str, object]: + """Verify the prepared proof and atomically promote its inactive generation.""" + receipt = _load_receipt(receipt_path) + status = receipt.get("status") + if status not in {"prepared", "activating", "activated"}: + raise IndexFastForwardError(f"receipt is not prepared: {status}") + if status == "activated": + return receipt + archive_root = Path(str(receipt["archive_root"])).resolve(strict=True) + _require_daemon_stopped(archive_root) + store = IndexGenerationStore.for_archive_root(archive_root) + generation_payload = cast(dict[str, object], receipt["generation"]) + generation = store.load(str(generation_payload["generation_id"])) + with RebuildLease(archive_root): + _require_daemon_stopped(archive_root) + if status == "activating": + generation = store.recover_promotion(generation.generation_id) + if generation.owner_id != generation_payload["owner_id"]: + raise IndexFastForwardError("prepared generation ownership changed") + clone = Path(generation.index_path) + if generation.state == "active": + if store.active_pointer.resolve(strict=True) != clone.resolve(strict=True): + raise IndexFastForwardError("active generation does not own the active index pointer") + receipt.update( + { + "status": "activated", + "activated_at_ms": receipt.get("activated_at_ms", _now_ms()), + "generation": asdict(generation), + } + ) + _write_receipt(receipt_path, receipt) + return receipt + if generation.state != "inactive": + raise IndexFastForwardError(f"prepared generation has unrecoverable state {generation.state}") + if source_revision_snapshot(archive_root) != receipt["source_snapshot"]: + raise IndexFastForwardError("source evidence changed since fast-forward preparation") + if _file_identity(store.active_pointer) != receipt["active_identity"]: + raise IndexFastForwardError("active index changed since fast-forward preparation") + canonical_sha = _canonical_schema_sha256(_canonical_schema_objects()) + if canonical_sha != receipt.get("canonical_schema_sha256"): + raise IndexFastForwardError("canonical index schema changed since preparation") + if _proven_clone_identity(clone) != receipt.get("clone_identity"): + raise IndexFastForwardError("prepared clone bytes changed before activation") + proof = cast(dict[str, object], receipt.get("proof", {})) + _require_complete_proof(proof) + manifest = cast(list[dict[str, object]], receipt.get("sample_manifest", [])) + origins = tuple(sorted({str(origin) for entry in manifest for origin in cast(list[str], entry["origins"])})) + if _fingerprints(origins) != receipt.get("fingerprints"): + raise IndexFastForwardError("parser/materializer fingerprints changed since preparation") + _require_candidate_corpus_fidelity(archive_root, clone) + if source_revision_snapshot(archive_root) != receipt["source_snapshot"]: + raise IndexFastForwardError("source evidence changed immediately before promotion") + if _proven_clone_identity(clone) != receipt.get("clone_identity"): + raise IndexFastForwardError("prepared clone bytes changed immediately before promotion") + if status == "prepared": + receipt.update({"status": "activating", "activation_started_at_ms": _now_ms()}) + _write_receipt(receipt_path, receipt) + promoted = store.promote(generation) + receipt.update( + { + "status": "activated", + "activated_at_ms": _now_ms(), + "generation": asdict(promoted), + "active_identity_after": _file_identity(store.active_pointer), + } + ) + _write_receipt(receipt_path, receipt) + return receipt + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + prepare = subparsers.add_parser("prepare") + prepare.add_argument("--archive-root", type=Path, required=True) + prepare.add_argument("--receipt", type=Path, required=True) + prepare.add_argument("--sample-size", type=int, default=DEFAULT_SAMPLE_SIZE) + activate = subparsers.add_parser("activate") + activate.add_argument("--receipt", type=Path, required=True) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + result = ( + prepare_forward(archive_root=args.archive_root, receipt_path=args.receipt, sample_size=args.sample_size) + if args.command == "prepare" + else activate_forward(receipt_path=args.receipt) + ) + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 + + +__all__ = ["IndexFastForwardError", "activate_forward", "main", "prepare_forward"] diff --git a/devtools/index_v37_fast_forward.py b/devtools/index_v37_fast_forward.py deleted file mode 100644 index 1c4fedbea1..0000000000 --- a/devtools/index_v37_fast_forward.py +++ /dev/null @@ -1,574 +0,0 @@ -"""Proof-gated clone-first index v36 -> v37 fast-forward. - -Index v37 removes three derived run-projection cache tables and changes no -surviving schema object. Replaying every raw blob is unnecessary for this -exact transition: clone the stopped active generation, prove that its only -schema surplus is the retired cache family, remove that family transactionally, -and promote through :class:`IndexGenerationStore` only after full postflight. -""" - -from __future__ import annotations - -import argparse -import hashlib -import json -import os -import re -import sqlite3 -import time -import uuid -from contextlib import closing, suppress -from dataclasses import asdict -from pathlib import Path -from typing import cast - -from devtools.clone_support import reflink_clone -from polylogue.config import Config -from polylogue.maintenance.archive_verification import CORPUS_FIDELITY_CHECKS, verify_archive -from polylogue.maintenance.offline_guard import running_daemon_pid -from polylogue.storage.index_generation import IndexGenerationStore, RebuildLease, source_revision_snapshot -from polylogue.storage.sqlite.archive_tiers.index import INDEX_DDL -from polylogue.storage.sqlite.connection_profile import open_readonly_connection -from polylogue.storage.sqlite.runtime_indexes import ensure_runtime_indexes_sync - -FROM_VERSION = 36 -TO_VERSION = 37 -RECEIPT_SCHEMA = "polylogue.index-v37-fast-forward.v1" -_DDL_NUMBER = re.compile(r"(?:0[xX][0-9A-Fa-f]+|(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?)") -RETIRED_TABLES = ( - "session_observed_events", - "session_context_snapshots", - "session_runs", -) - - -class IndexV37FastForwardError(RuntimeError): - """The v36 clone could not be proven safe for v37 promotion.""" - - -def _now_ms() -> int: - return int(time.time() * 1000) - - -def _ddl_tokens(sql: str) -> list[tuple[str, str]]: - """Tokenize SQLite DDL without erasing literal or token boundaries.""" - tokens: list[tuple[str, str]] = [] - index = 0 - length = len(sql) - while index < length: - char = sql[index] - if char.isspace(): - index += 1 - continue - if sql.startswith("--", index): - newline = sql.find("\n", index + 2) - index = length if newline < 0 else newline + 1 - continue - if sql.startswith("/*", index): - end = sql.find("*/", index + 2) - if end < 0: - raise IndexV37FastForwardError("unterminated comment in schema DDL") - index = end + 2 - continue - if char == "'": - index += 1 - value: list[str] = [] - while index < length: - if sql[index] == "'": - if index + 1 < length and sql[index + 1] == "'": - value.append("'") - index += 2 - continue - index += 1 - break - value.append(sql[index]) - index += 1 - else: - raise IndexV37FastForwardError("unterminated string literal in schema DDL") - tokens.append(("string", "".join(value))) - continue - if char in {'"', "`", "["}: - closing = "]" if char == "[" else char - index += 1 - value = [] - while index < length: - if sql[index] == closing: - if index + 1 < length and sql[index + 1] == closing: - value.append(closing) - index += 2 - continue - index += 1 - break - value.append(sql[index]) - index += 1 - else: - raise IndexV37FastForwardError("unterminated quoted identifier in schema DDL") - tokens.append(("word", "".join(value).casefold())) - continue - if char.isalpha() or char in {"_", "$"}: - end = index + 1 - while end < length and (sql[end].isalnum() or sql[end] in {"_", "$"}): - end += 1 - tokens.append(("word", sql[index:end].casefold())) - index = end - continue - if char.isdigit() or (char == "." and index + 1 < length and sql[index + 1].isdigit()): - match = _DDL_NUMBER.match(sql, index) - if match is None: - raise AssertionError("numeric DDL token did not match") - tokens.append(("number", match.group(0).casefold())) - index = match.end() - continue - operator = next( - ( - candidate - for candidate in ("->>", "||", "<=", ">=", "<>", "!=", "==", "<<", ">>", "->") - if sql.startswith(candidate, index) - ), - char, - ) - tokens.append(("symbol", operator)) - index += len(operator) - return tokens - - -def _normalize_ddl(sql: str) -> str: - tokens = _ddl_tokens(sql) - normalized: list[tuple[str, str]] = [] - index = 0 - while index < len(tokens): - if tokens[index : index + 3] == [ - ("word", "if"), - ("word", "not"), - ("word", "exists"), - ]: - index += 3 - continue - normalized.append(tokens[index]) - index += 1 - return json.dumps(normalized, ensure_ascii=False, separators=(",", ":")) - - -def _schema_objects(conn: sqlite3.Connection) -> dict[str, str]: - rows = conn.execute( - """ - SELECT type, name, sql - FROM sqlite_master - WHERE type IN ('table', 'index', 'view', 'trigger') - AND name NOT LIKE 'sqlite_%' - AND sql IS NOT NULL - ORDER BY type, name - """ - ).fetchall() - return {f"{row[0]}:{row[1]}": _normalize_ddl(str(row[2])) for row in rows} - - -def _canonical_schema_objects() -> dict[str, str]: - with closing(sqlite3.connect(":memory:")) as conn: - conn.executescript(INDEX_DDL) - ensure_runtime_indexes_sync(conn) - return _schema_objects(conn) - - -def _schema_rootpages(conn: sqlite3.Connection) -> dict[str, int]: - rows = conn.execute( - """ - SELECT type, name, rootpage - FROM sqlite_master - WHERE type IN ('table', 'index') - AND name NOT LIKE 'sqlite_%' - AND rootpage > 0 - ORDER BY type, name - """ - ).fetchall() - return {f"{row[0]}:{row[1]}": int(row[2]) for row in rows} - - -def _checks(conn: sqlite3.Connection) -> dict[str, object]: - quick_check = [str(row[0]) for row in conn.execute("PRAGMA quick_check")] - attachment_native_ids_foreign_keys = [ - tuple(row) for row in conn.execute("PRAGMA foreign_key_check(attachment_native_ids)") - ] - return { - "quick_check": quick_check, - "attachment_native_ids_foreign_key_check": attachment_native_ids_foreign_keys, - } - - -def _require_retired_tables_unreferenced(conn: sqlite3.Connection) -> None: - """Prove dropping the caches cannot remove a surviving FK parent.""" - tables = [str(row[0]) for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")] - references = { - (table, str(row[2])) - for table in tables - if table not in RETIRED_TABLES - for row in conn.execute(f'PRAGMA foreign_key_list("{table}")') - if str(row[2]) in RETIRED_TABLES - } - if references: - raise IndexV37FastForwardError(f"surviving tables reference retired cache parents: {sorted(references)}") - - -def _file_identity(path: Path) -> dict[str, object]: - resolved = path.resolve(strict=True) - stat = resolved.stat() - return { - "path": str(path), - "resolved_path": str(resolved), - "size_bytes": stat.st_size, - "allocated_bytes": stat.st_blocks * 512, - "inode": stat.st_ino, - "mtime_ns": stat.st_mtime_ns, - } - - -def _proven_clone_identity(path: Path) -> dict[str, object]: - """Bind a prepared proof to exact clone bytes, not only row counts.""" - identity = _file_identity(path) - digest = hashlib.sha256() - with path.resolve(strict=True).open("rb") as handle: - while chunk := handle.read(8 * 1024 * 1024): - digest.update(chunk) - identity["sha256"] = digest.hexdigest() - return identity - - -def _canonical_schema_sha256(schema: dict[str, str]) -> str: - return hashlib.sha256(json.dumps(schema, sort_keys=True, separators=(",", ":")).encode()).hexdigest() - - -def _receipt_hash(payload: dict[str, object]) -> str: - body = {key: value for key, value in payload.items() if key != "receipt_sha256"} - return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest() - - -def _require_receipt_destination_writable(path: Path) -> None: - """Fail before expensive preparation when an atomic receipt cannot land.""" - probe = path.with_name(f".{path.name}.{uuid.uuid4().hex}.probe") - try: - path.parent.mkdir(parents=True, exist_ok=True) - if path.exists() and not path.is_file(): - raise OSError(f"receipt destination is not a regular file: {path}") - descriptor = os.open(probe, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) - try: - os.fsync(descriptor) - finally: - os.close(descriptor) - probe.unlink() - except OSError as exc: - with suppress(OSError): - probe.unlink(missing_ok=True) - raise IndexV37FastForwardError(f"receipt destination is not writable: {path}: {exc}") from exc - - -def _write_receipt(path: Path, payload: dict[str, object]) -> None: - payload["receipt_sha256"] = _receipt_hash(payload) - path.parent.mkdir(parents=True, exist_ok=True) - temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") - temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") - with temporary.open("rb") as handle: - os.fsync(handle.fileno()) - os.replace(temporary, path) - descriptor = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(descriptor) - finally: - os.close(descriptor) - - -def _load_receipt(path: Path) -> dict[str, object]: - payload = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(payload, dict) or payload.get("schema") != RECEIPT_SCHEMA: - raise IndexV37FastForwardError(f"invalid v37 fast-forward receipt: {path}") - typed = cast(dict[str, object], payload) - if typed.get("receipt_sha256") != _receipt_hash(typed): - raise IndexV37FastForwardError(f"v37 fast-forward receipt hash mismatch: {path}") - return typed - - -def _config(archive_root: Path) -> Config: - return Config( - archive_root=archive_root, - render_root=archive_root / "render", - sources=[], - db_path=archive_root / "index.db", - ) - - -def _require_daemon_stopped(archive_root: Path) -> None: - if (pid := running_daemon_pid(_config(archive_root))) is not None: - raise IndexV37FastForwardError(f"polylogued PID {pid} is still running") - - -def _require_candidate_corpus_fidelity(archive_root: Path, candidate_index: Path) -> None: - """Require the managed-rebuild corpus gate before this candidate activates.""" - report = verify_archive( - archive_root, - checks=CORPUS_FIDELITY_CHECKS, - index_path_override=candidate_index, - ) - if report.blocking: - failing = "; ".join( - f"{check.name}: {check.summary}" for check in report.checks if check.status.value == "error" - ) - raise IndexV37FastForwardError(f"candidate corpus fidelity gate failed: {failing}") - - -def _checkpoint_stopped_database(path: Path, *, label: str = "active index") -> None: - """Consolidate a stopped writer's committed WAL before clone evidence.""" - resolved = path.resolve(strict=True) - with closing(sqlite3.connect(resolved, timeout=120.0)) as conn: - checkpoint = conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() - if checkpoint is None or int(checkpoint[0]) != 0 or len(checkpoint) < 3 or int(checkpoint[1]) != int(checkpoint[2]): - raise IndexV37FastForwardError(f"{label} WAL checkpoint failed: {checkpoint}") - for suffix in ("-wal", "-shm"): - # A successful checkpoint under the stopped-daemon rebuild lease makes - # both coordination files disposable. SQLite commonly leaves a - # non-empty SHM file behind after the final writer exits. - sidecar = Path(f"{resolved}{suffix}") - if sidecar.exists(): - sidecar.unlink() - - -def _inspect_clean_database(path: Path, *, expected_version: int) -> tuple[dict[str, str], dict[str, int]]: - for suffix in ("-wal", "-shm", "-journal"): - sidecar = Path(f"{path.resolve(strict=True)}{suffix}") - if sidecar.exists() and sidecar.stat().st_size: - raise IndexV37FastForwardError(f"non-empty SQLite sidecar blocks fast-forward: {sidecar}") - with closing(open_readonly_connection(path.resolve(strict=True), immutable=True)) as conn: - version = int(conn.execute("PRAGMA user_version").fetchone()[0]) - if version != expected_version: - raise IndexV37FastForwardError(f"expected index v{expected_version}, found v{version}") - return _schema_objects(conn), _schema_rootpages(conn) - - -def _prove_v36_delta(schema: dict[str, str]) -> dict[str, str]: - canonical = _canonical_schema_objects() - missing = sorted(set(canonical) - set(schema)) - surplus = {key: value for key, value in schema.items() if key not in canonical} - expected_surplus = { - key: value - for key, value in schema.items() - if key.split(":", 1)[1] in RETIRED_TABLES - or any(key.split(":", 1)[1].startswith(f"idx_{table}") for table in RETIRED_TABLES) - } - changed = sorted(key for key in canonical.keys() & schema.keys() if canonical[key] != schema[key]) - if missing or surplus != expected_surplus or changed: - raise IndexV37FastForwardError( - f"v36 schema is not the exact v37-plus-retired-caches shape: " - f"missing={missing}, unexpected_surplus={sorted(set(surplus) - set(expected_surplus))}, changed={changed}" - ) - retired_tables = {f"table:{table}" for table in RETIRED_TABLES} - if not retired_tables <= set(expected_surplus): - raise IndexV37FastForwardError("v36 index is missing one or more retired cache tables") - return canonical - - -def _transform_clone(path: Path, *, before_rootpages: dict[str, int], canonical: dict[str, str]) -> dict[str, object]: - with closing(sqlite3.connect(path, timeout=120.0)) as conn: - conn.execute("PRAGMA foreign_keys = OFF") - if int(conn.execute("PRAGMA user_version").fetchone()[0]) != FROM_VERSION: - raise IndexV37FastForwardError("clone version changed before transformation") - _require_retired_tables_unreferenced(conn) - changes_before = conn.total_changes - conn.execute("BEGIN IMMEDIATE") - try: - repaired_orphan_native_ids = conn.execute( - """ - DELETE FROM attachment_native_ids - WHERE NOT EXISTS ( - SELECT 1 FROM attachment_refs WHERE attachment_refs.ref_id = attachment_native_ids.ref_id - ) - """ - ).rowcount - for table in RETIRED_TABLES: - conn.execute(f'DROP TABLE "{table}"') - conn.execute(f"PRAGMA user_version = {TO_VERSION}") - conn.commit() - except Exception: - conn.rollback() - raise - if conn.total_changes - changes_before != repaired_orphan_native_ids: - raise IndexV37FastForwardError("v37 clone transformation changed rows outside the declared repair") - _checkpoint_stopped_database(path, label="prepared clone") - with closing(open_readonly_connection(path, immutable=True)) as conn: - checks = _checks(conn) - after_schema = _schema_objects(conn) - after_rootpages = _schema_rootpages(conn) - version = int(conn.execute("PRAGMA user_version").fetchone()[0]) - expected_rootpages = { - key: rootpage - for key, rootpage in before_rootpages.items() - if key.split(":", 1)[1] not in RETIRED_TABLES - and not any(key.split(":", 1)[1].startswith(f"idx_{table}") for table in RETIRED_TABLES) - } - if version != TO_VERSION: - raise IndexV37FastForwardError(f"clone ended at unexpected index version {version}") - if checks["quick_check"] != ["ok"] or checks["attachment_native_ids_foreign_key_check"]: - raise IndexV37FastForwardError(f"clone postflight failed: {checks}") - if after_schema != canonical: - raise IndexV37FastForwardError("clone schema does not exactly match canonical v37 DDL") - if after_rootpages != expected_rootpages: - raise IndexV37FastForwardError("one or more surviving schema root pages changed in the clone") - return { - "checks": checks, - "repaired_orphan_attachment_native_ids": repaired_orphan_native_ids, - "schema_rootpages": after_rootpages, - "schema_object_count": len(after_schema), - } - - -def prepare_forward(*, archive_root: Path, receipt_path: Path) -> dict[str, object]: - """Create and prove an owned inactive v37 generation.""" - prepare_started_ns = time.monotonic_ns() - phase_timings_ms: dict[str, int] = {} - archive_root = archive_root.resolve(strict=True) - _require_daemon_stopped(archive_root) - _require_receipt_destination_writable(receipt_path) - store = IndexGenerationStore.for_archive_root(archive_root) - active_pointer = store.active_pointer - with RebuildLease(archive_root): - _require_daemon_stopped(archive_root) - _checkpoint_stopped_database(active_pointer) - phase_started_ns = time.monotonic_ns() - source_snapshot = source_revision_snapshot(archive_root) - active_identity = _file_identity(active_pointer) - before_schema, before_rootpages = _inspect_clean_database(active_pointer, expected_version=FROM_VERSION) - canonical = _prove_v36_delta(before_schema) - phase_timings_ms["active_evidence"] = (time.monotonic_ns() - phase_started_ns) // 1_000_000 - generation = store.create(source_snapshot=source_snapshot) - clone = Path(generation.index_path) - try: - phase_started_ns = time.monotonic_ns() - clone.unlink() - reflink_clone(active_pointer, clone) - if _file_identity(active_pointer) != active_identity: - raise IndexV37FastForwardError("active index changed while its clone was created") - phase_timings_ms["reflink_clone"] = (time.monotonic_ns() - phase_started_ns) // 1_000_000 - phase_started_ns = time.monotonic_ns() - postflight = _transform_clone(clone, before_rootpages=before_rootpages, canonical=canonical) - phase_timings_ms["transform_and_postflight"] = (time.monotonic_ns() - phase_started_ns) // 1_000_000 - if source_revision_snapshot(archive_root) != source_snapshot: - raise IndexV37FastForwardError("source evidence changed while preparing v37 clone") - phase_started_ns = time.monotonic_ns() - clone_identity = _proven_clone_identity(clone) - phase_timings_ms["clone_sha256"] = (time.monotonic_ns() - phase_started_ns) // 1_000_000 - phase_timings_ms["total"] = (time.monotonic_ns() - prepare_started_ns) // 1_000_000 - receipt: dict[str, object] = { - "schema": RECEIPT_SCHEMA, - "status": "prepared", - "prepared_at_ms": _now_ms(), - "archive_root": str(archive_root), - "generation": asdict(generation), - "source_snapshot": source_snapshot, - "active_identity": active_identity, - "clone_identity": clone_identity, - "canonical_schema_sha256": _canonical_schema_sha256(canonical), - "before_schema_rootpages": before_rootpages, - "retired_tables": list(RETIRED_TABLES), - "postflight": postflight, - "phase_timings_ms": phase_timings_ms, - "raw_reparse": False, - } - _write_receipt(receipt_path, receipt) - return receipt - except Exception: - store.discard_if_inactive(generation) - raise - - -def activate_forward(*, receipt_path: Path) -> dict[str, object]: - """Reconcile or atomically promote one prepared v37 generation.""" - receipt = _load_receipt(receipt_path) - status = receipt.get("status") - if status not in {"prepared", "activating", "activated"}: - raise IndexV37FastForwardError(f"receipt is not prepared: {receipt.get('status')}") - if status == "activated": - return receipt - archive_root = Path(str(receipt["archive_root"])).resolve(strict=True) - _require_daemon_stopped(archive_root) - store = IndexGenerationStore.for_archive_root(archive_root) - generation_payload = cast(dict[str, object], receipt["generation"]) - generation = store.load(str(generation_payload["generation_id"])) - with RebuildLease(archive_root): - _require_daemon_stopped(archive_root) - if status == "activating": - generation = store.recover_promotion(generation.generation_id) - if generation.owner_id != generation_payload["owner_id"]: - raise IndexV37FastForwardError("prepared generation ownership changed") - clone = Path(generation.index_path) - if generation.state == "active": - if store.active_pointer.resolve(strict=True) != clone.resolve(strict=True): - raise IndexV37FastForwardError("active generation does not own the active index pointer") - if status != "activating": - raise IndexV37FastForwardError(f"active generation has incompatible receipt status {status}") - receipt.update( - { - "status": "activated", - "activated_at_ms": receipt.get("activated_at_ms", _now_ms()), - "generation": asdict(generation), - "active_identity_after": _file_identity(store.active_pointer), - } - ) - _write_receipt(receipt_path, receipt) - return receipt - if generation.state != "inactive": - raise IndexV37FastForwardError(f"prepared generation has unrecoverable state {generation.state}") - if source_revision_snapshot(archive_root) != receipt["source_snapshot"]: - raise IndexV37FastForwardError("source evidence changed since v37 preparation") - if _file_identity(store.active_pointer) != receipt["active_identity"]: - raise IndexV37FastForwardError("active index changed since v37 preparation") - canonical = _canonical_schema_objects() - if _canonical_schema_sha256(canonical) != receipt.get("canonical_schema_sha256"): - raise IndexV37FastForwardError("canonical v37 schema changed since clone preparation") - if _proven_clone_identity(clone) != receipt.get("clone_identity"): - raise IndexV37FastForwardError("prepared clone bytes changed before activation") - _require_candidate_corpus_fidelity(archive_root, clone) - if status == "prepared": - receipt.update({"status": "activating", "activation_started_at_ms": _now_ms()}) - _write_receipt(receipt_path, receipt) - promoted = store.promote(generation) - receipt.update( - { - "status": "activated", - "activated_at_ms": _now_ms(), - "generation": asdict(promoted), - "active_identity_after": _file_identity(store.active_pointer), - } - ) - _write_receipt(receipt_path, receipt) - return receipt - - -def _parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description=__doc__) - subparsers = parser.add_subparsers(dest="command", required=True) - prepare = subparsers.add_parser("prepare") - prepare.add_argument("--archive-root", type=Path, required=True) - prepare.add_argument("--receipt", type=Path, required=True) - activate = subparsers.add_parser("activate") - activate.add_argument("--receipt", type=Path, required=True) - return parser - - -def main(argv: list[str] | None = None) -> int: - args = _parser().parse_args(argv) - result = ( - prepare_forward(archive_root=args.archive_root, receipt_path=args.receipt) - if args.command == "prepare" - else activate_forward(receipt_path=args.receipt) - ) - print(json.dumps(result, indent=2, sort_keys=True)) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) - - -__all__ = [ - "IndexV37FastForwardError", - "activate_forward", - "main", - "prepare_forward", -] diff --git a/docs/devtools.md b/docs/devtools.md index 762432f14e..19d8e80768 100644 --- a/docs/devtools.md +++ b/docs/devtools.md @@ -244,7 +244,7 @@ These are the commands worth remembering during normal repo work: | `devtools workspace dev-loop` | Preflight branch-local daemon, web-shell, and browser-capture development loops. | | `devtools workspace failure-context` | Join testmon, git history, and fixtures for a pytest failure ID into a JSON envelope. | | `devtools workspace frontier` | Classify ready and in-progress Beads into devloop batches. | -| `devtools workspace index-v37-fast-forward` | Clone-forward index v36 to v37 by retiring derived caches without raw replay. | +| `devtools workspace index-fast-forward` | Plan and prove a declared index fast-forward against retained raw replay. | | `devtools workspace lane-brief` | Generate a dispatch brief for a bead lane with live footprint/prior-art evidence. | | `devtools workspace lane-init` | Provision a fanout lane worktree: branch, isolated venv, guard check, ledger record. | | `devtools workspace lineage-validation` | Validate lineage-count evidence before citing archive counts externally. | diff --git a/docs/internals.md b/docs/internals.md index 1155f67dce..4385249524 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -149,6 +149,18 @@ Polylogue has two schema-evolution regimes, keyed by tier durability. `messages_fts_identity` declaration existed but had no consumer: a freshly promoted v42 archive could not be opened by v43 code at all before this executor existed. +- **Index transition ownership map** (polylogue-9rw0 / polylogue-b5l.3): + `storage/sqlite/lifecycle.py` owns version declarations and generated + operations; `storage/sqlite/archive_tiers/index_fast_forward_executor.py` + owns generic operation execution; `devtools/index_fast_forward.py` owns + stopped-root clone, retained-raw sample replay, receipt, and atomic + promotion. The actuator has no per-version branch. Every eligible plan + records raw and session sample IDs, parser/lowering/materializer + fingerprints, structural hashes, canonical replay hashes, mismatch details, + and an equivalent verdict. Activation refuses absent proof, changed source, + changed fingerprints, changed schema, or changed clone bytes. The old + former version-specific actuator was the only surviving duplicate and was + removed after its v37 cleanup became a declared lifecycle operation. - **Disposable tiers** (`ops.db`) may keep narrow bootstrap-time `ALTER TABLE` helpers for daemon telemetry because the tier is disposable. - **Index-tier benign-DDL convergence** (polylogue-jc1b): a registered set of diff --git a/polylogue/storage/index_generation.py b/polylogue/storage/index_generation.py index a4c7a97cc3..ff4f66d9a0 100644 --- a/polylogue/storage/index_generation.py +++ b/polylogue/storage/index_generation.py @@ -869,15 +869,10 @@ def source_revision_snapshot(archive_root: Path) -> str: digest = hashlib.sha256() with closing(sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True)) as conn: - for raw_id, acquired_at_ms, blob_hash, blob_size, validation_status in conn.execute( - """ - SELECT raw_id, acquired_at_ms, blob_hash, blob_size, validation_status - FROM raw_sessions - ORDER BY acquired_at_ms, raw_id - """ - ): - for value in (raw_id, acquired_at_ms, bytes(blob_hash).hex(), blob_size, validation_status): - digest.update(str(value).encode()) + for row in conn.execute("SELECT * FROM raw_sessions ORDER BY raw_id"): + for value in row: + encoded = value.hex() if isinstance(value, bytes) else str(value) + digest.update(encoded.encode()) digest.update(b"\0") digest.update(b"\n") return digest.hexdigest() diff --git a/polylogue/storage/sqlite/archive_tiers/index_fast_forward_executor.py b/polylogue/storage/sqlite/archive_tiers/index_fast_forward_executor.py index e7d1fce6ad..19068f4ea0 100644 --- a/polylogue/storage/sqlite/archive_tiers/index_fast_forward_executor.py +++ b/polylogue/storage/sqlite/archive_tiers/index_fast_forward_executor.py @@ -95,6 +95,18 @@ def _apply_drop_table(conn: sqlite3.Connection, name: str) -> None: conn.execute(f'DROP TABLE IF EXISTS "{name}"') +def _apply_repair_orphan_attachment_native_ids(conn: sqlite3.Connection) -> None: + """Apply the retained v37 cleanup as a declared generic operation.""" + conn.execute( + """ + DELETE FROM attachment_native_ids + WHERE NOT EXISTS ( + SELECT 1 FROM attachment_refs WHERE attachment_refs.ref_id = attachment_native_ids.ref_id + ) + """ + ) + + def _apply_drop_trigger(conn: sqlite3.Connection, name: str) -> None: conn.execute(f'DROP TRIGGER IF EXISTS "{name}"') @@ -213,6 +225,11 @@ def _apply_operation( if operation.kind is FastForwardOperationKind.REBUILD_FTS: _apply_rebuild_fts(conn, operation) return + if operation.kind is FastForwardOperationKind.REPAIR_ORPHAN_ATTACHMENT_NATIVE_IDS: + if operation.objects != (("table", "attachment_native_ids"),): + raise RuntimeError("orphan attachment repair operation has an unexpected object set") + _apply_repair_orphan_attachment_native_ids(conn) + return for object_type, name in operation.objects: if operation.kind is FastForwardOperationKind.DROP_TABLE: # A DROP_TABLE operation may bundle the table's own triggers as diff --git a/polylogue/storage/sqlite/lifecycle.py b/polylogue/storage/sqlite/lifecycle.py index 647ba22337..94423382c7 100644 --- a/polylogue/storage/sqlite/lifecycle.py +++ b/polylogue/storage/sqlite/lifecycle.py @@ -85,6 +85,7 @@ class FastForwardOperationKind(StrEnum): CREATE_INDEX = "create-index" REBUILD_FTS = "rebuild-fts" DROP_TABLE = "drop-table" + REPAIR_ORPHAN_ATTACHMENT_NATIVE_IDS = "repair-orphan-attachment-native-ids" @dataclass(frozen=True, slots=True) @@ -290,6 +291,11 @@ class IndexDeltaDeclarationReport(TypedDict): ("table", "session_context_snapshots"), ), ), + FastForwardOperation( + name="v37-repair-orphan-attachment-native-ids", + kind=FastForwardOperationKind.REPAIR_ORPHAN_ATTACHMENT_NATIVE_IDS, + objects=(("table", "attachment_native_ids"),), + ), ), ), IndexDeltaDeclaration( diff --git a/tests/unit/devtools/test_index_fast_forward.py b/tests/unit/devtools/test_index_fast_forward.py new file mode 100644 index 0000000000..83e9ca7490 --- /dev/null +++ b/tests/unit/devtools/test_index_fast_forward.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +import json +import sqlite3 +from dataclasses import dataclass +from pathlib import Path + +import pytest + +import devtools.index_fast_forward as forward +from polylogue.core.enums import Provider +from polylogue.sources.dispatch import parse_payload +from polylogue.storage.index_generation import IndexGenerationStore +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + +@dataclass(frozen=True) +class _Report: + blocking: bool = False + checks: tuple[object, ...] = () + + +def _archive(tmp_path: Path) -> Path: + root = tmp_path / "archive" + root.mkdir() + for tier in (ArchiveTier.SOURCE, ArchiveTier.USER, ArchiveTier.EMBEDDINGS, ArchiveTier.OPS): + initialize_archive_database(root / f"{tier.value}.db", tier) + storage = tmp_path / "storage" + active_root = storage / ".index-generations" / "v36" + active_root.mkdir(parents=True) + active = active_root / "index.db" + initialize_archive_database(active, ArchiveTier.INDEX) + (storage / "index.db").symlink_to(active) + (root / "index.db").symlink_to(storage / "index.db") + + payload_object = { + "id": "source-backed-session", + "conversation_id": "source-backed-session", + "title": "source-backed proof", + "mapping": { + "message": { + "id": "message", + "parent": None, + "children": [], + "message": { + "id": "message", + "author": {"role": "user"}, + "content": {"content_type": "text", "parts": ["retained raw proof"]}, + "create_time": 1_700_000_000, + }, + } + }, + } + payload = json.dumps(payload_object, sort_keys=True).encode() + parsed = parse_payload( + Provider.CHATGPT, + payload_object, + "source-backed-session", + source_path="source-backed-session.json", + ) + assert len(parsed) == 1 + with ArchiveStore.open_existing(root, read_only=False) as archive: + archive.write_raw_and_parsed( + parsed[0], + payload=payload, + source_path="source-backed-session.json", + acquired_at_ms=1, + ) + with sqlite3.connect(active) as conn: + conn.execute("CREATE TABLE session_runs(id TEXT)") + conn.execute("CREATE TABLE session_observed_events(id TEXT)") + conn.execute("CREATE TABLE session_context_snapshots(id TEXT)") + conn.execute( + "INSERT INTO attachment_native_ids(ref_id, id_kind, native_id) VALUES ('orphan-ref', 'url', 'orphan')" + ) + conn.execute("PRAGMA user_version = 36") + conn.commit() + return root + + +@pytest.fixture +def _patch_v37(monkeypatch: pytest.MonkeyPatch) -> None: + import polylogue.storage.sqlite.lifecycle as lifecycle + from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER + + monkeypatch.setattr( + lifecycle, + "INDEX_DELTA_DECLARATIONS", + tuple(declaration for declaration in lifecycle.INDEX_DELTA_DECLARATIONS if 36 <= declaration.version <= 37), + ) + monkeypatch.setitem(ARCHIVE_VERSION_BY_TIER, ArchiveTier.INDEX, 37) + + +def _no_corpus_failure(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(forward, "verify_archive", lambda *args, **kwargs: _Report()) + + +def test_prepare_receipt_proves_source_replay_equivalence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_v37: None +) -> None: + root = _archive(tmp_path) + monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) + + receipt_path = tmp_path / "transition.json" + prepared = forward.prepare_forward(archive_root=root, receipt_path=receipt_path) + + proof = prepared["proof"] + assert isinstance(proof, dict) + assert proof["verdict"] == "equivalent" + assert proof["mismatch_details"] == [] + assert proof["replayed_session_ids"] + assert {"sessions", "messages", "blocks", "fts", "scoped"} <= set(proof["fast_forward_hashes"]) + assert "session_links" in proof["fast_forward_hashes"]["scoped"] + assert prepared["sample_manifest"] + fingerprints = prepared["fingerprints"] + assert isinstance(fingerprints, dict) + assert fingerprints["parser"] + assert fingerprints["lowering"] + assert fingerprints["materializer"] + + _no_corpus_failure(monkeypatch) + activated = forward.activate_forward(receipt_path=receipt_path) + assert activated["status"] == "activated" + assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().parent.name.startswith("gen-") + with sqlite3.connect(IndexGenerationStore.for_archive_root(root).active_pointer.resolve()) as conn: + assert conn.execute("SELECT 1 FROM attachment_native_ids WHERE ref_id = 'orphan-ref'").fetchone() is None + + +def test_activation_refuses_parser_or_materializer_fingerprint_drift( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_v37: None +) -> None: + root = _archive(tmp_path) + monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) + receipt_path = tmp_path / "transition.json" + forward.prepare_forward(archive_root=root, receipt_path=receipt_path) + monkeypatch.setattr(forward, "_materializer_fingerprint", lambda: "changed-materializer") + _no_corpus_failure(monkeypatch) + + with pytest.raises(forward.IndexFastForwardError, match="fingerprints changed"): + forward.activate_forward(receipt_path=receipt_path) + + assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().parent.name == "v36" + + +def test_activation_refuses_source_metadata_mutation_after_candidate_gate( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_v37: None +) -> None: + root = _archive(tmp_path) + monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) + receipt_path = tmp_path / "transition.json" + forward.prepare_forward(archive_root=root, receipt_path=receipt_path) + + def mutate_source(*args: object, **kwargs: object) -> None: + with sqlite3.connect(root / "source.db") as conn: + conn.execute("UPDATE raw_sessions SET source_path = 'mutated-after-proof'") + + monkeypatch.setattr(forward, "_require_candidate_corpus_fidelity", mutate_source) + with pytest.raises(forward.IndexFastForwardError, match="immediately before promotion"): + forward.activate_forward(receipt_path=receipt_path) + + assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().parent.name == "v36" + + +def test_bypassing_replay_cannot_create_an_activatable_receipt( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_v37: None +) -> None: + root = _archive(tmp_path) + monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) + + def removed_replay(*args: object, **kwargs: object) -> dict[str, object]: + # Mutation: deleting the production replay/canonical comparison would + # otherwise be paper-covered by a receipt that only says equivalent. + return { + "fast_forward_hashes": {}, + "canonical_replay_hashes": {}, + "replayed_session_ids": [], + "mismatch_details": [], + "verdict": "equivalent", + } + + monkeypatch.setattr(forward, "_replay_sample", removed_replay) + with pytest.raises(forward.IndexFastForwardError, match="proof"): + forward.prepare_forward(archive_root=root, receipt_path=tmp_path / "transition.json") + + assert not list(IndexGenerationStore.for_archive_root(root).generations_root.glob("gen-*/index.db")) diff --git a/tests/unit/devtools/test_index_v37_fast_forward.py b/tests/unit/devtools/test_index_v37_fast_forward.py deleted file mode 100644 index 2023134eab..0000000000 --- a/tests/unit/devtools/test_index_v37_fast_forward.py +++ /dev/null @@ -1,390 +0,0 @@ -from __future__ import annotations - -import os -import sqlite3 -from dataclasses import replace -from pathlib import Path - -import pytest - -import devtools.index_v37_fast_forward as forward -from devtools.index_v37_fast_forward import IndexV37FastForwardError, activate_forward, prepare_forward -from polylogue.storage.index_generation import IndexGeneration, IndexGenerationStore -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database, initialize_archive_tier -from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier -from polylogue.storage.sqlite.runtime_indexes import ensure_runtime_indexes_sync - - -def _archive(tmp_path: Path) -> tuple[Path, int]: - """Create a test archive with a v36 index database for devtools testing. - - The devtools index_v37_fast_forward module is specifically for v36→v37 migration. - These tests verify the fast-forward executor works correctly on this legacy path. - V36 had three cache tables that v37 drops (session_runs, session_observed_events, - session_context_snapshots), so we create them here to simulate v36. - - Returns: (archive_root, created_index_version) - """ - ffw_version = 36 # devtools index_v37_fast_forward tests v36→v37 migration - - root = tmp_path / "archive" - root.mkdir() - for tier in (ArchiveTier.SOURCE, ArchiveTier.USER, ArchiveTier.EMBEDDINGS, ArchiveTier.OPS): - initialize_archive_database(root / f"{tier.value}.db", tier) - storage = tmp_path / "storage" - active_root = storage / ".index-generations" / f"v{ffw_version}" - active_root.mkdir(parents=True) - active = active_root / "index.db" - with sqlite3.connect(active) as conn: - # Use current schema but set version to v36 for devtools testing. - # This tests the fast-forward executor without relying on historical DDL. - initialize_archive_tier(conn, ArchiveTier.INDEX) - ensure_runtime_indexes_sync(conn) - - # Add v36 cache tables that v37 removes (CACHE_REMOVAL delta). - # These are simple placeholder tables needed for devtools v36→v37 testing. - conn.execute("CREATE TABLE IF NOT EXISTS session_runs(id TEXT)") - conn.execute("CREATE TABLE IF NOT EXISTS session_observed_events(id TEXT)") - conn.execute("CREATE TABLE IF NOT EXISTS session_context_snapshots(id TEXT)") - - conn.execute(f"PRAGMA user_version = {ffw_version}") - # Add test data that will be preserved by fast-forward - conn.execute( - "INSERT INTO sessions(native_id, origin, content_hash) VALUES ('session', 'chatgpt-export', ?)", - (b"s" * 32,), - ) - conn.execute( - "INSERT INTO messages(session_id, native_id, position, role, content_hash) VALUES (?, 'message', 0, 'user', ?)", - ("chatgpt-export:session", b"m" * 32), - ) - conn.commit() - (storage / "index.db").symlink_to(active) - (root / "index.db").symlink_to(storage / "index.db") - return root, ffw_version - - -@pytest.fixture -def _patch_lifecycle_for_v36_upgrade(monkeypatch: pytest.MonkeyPatch) -> None: - """Patch lifecycle declarations to allow v36→v37 fast-forward only. - - The devtools v36→v37 forward tool is specifically designed to upgrade from v36 - to v37 using fast-forward (dropping cache tables). We patch the declarations - to stop at v37 instead of extending to v45, so the tool can complete its - specific migration without hitting SEMANTIC_REPARSE blocks from later versions. - """ - import polylogue.storage.sqlite.lifecycle as lifecycle - from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER - - # Keep only v36→v37 declarations. - # The real lifecycle has v37 as CACHE_REMOVAL (fast-forwardable). - # We exclude v38+ to avoid SEMANTIC_REPARSE blocks at v39+. - synthetic_decls = tuple(d for d in lifecycle.INDEX_DELTA_DECLARATIONS if 36 <= d.version <= 37) - - monkeypatch.setattr( - lifecycle, - "INDEX_DELTA_DECLARATIONS", - synthetic_decls, - ) - # Patch the expected version to v37 (devtools v36→v37 destination). - monkeypatch.setitem(ARCHIVE_VERSION_BY_TIER, ArchiveTier.INDEX, 37) - - -def test_prepare_and_activate_preserve_surviving_rows_without_raw_replay( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, ffw_version = _archive(tmp_path) - receipt = tmp_path / "receipt.json" - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - - prepared = prepare_forward(archive_root=root, receipt_path=receipt) - - assert prepared["status"] == "prepared" - assert prepared["raw_reparse"] is False - generation = prepared["generation"] - assert isinstance(generation, dict) - clone = Path(str(generation["index_path"])) - assert (root / "index.db").resolve().parent.name == f"v{ffw_version}" - with sqlite3.connect(clone) as conn: - # With synthetic declarations, the target is v37 (devtools v36→v37). - assert conn.execute("PRAGMA user_version").fetchone()[0] == 37 - assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 1 - for table in forward.RETIRED_TABLES: - assert conn.execute("SELECT 1 FROM sqlite_master WHERE name = ?", (table,)).fetchone() is None - - activated = activate_forward(receipt_path=receipt) - - assert activated["status"] == "activated" - assert (root / "index.db").resolve() == clone.resolve() - retired = list(IndexGenerationStore.for_archive_root(root).generations_root.glob("retired-*/index.db")) - assert retired - with sqlite3.connect(retired[0]) as conn: - assert conn.execute("PRAGMA user_version").fetchone()[0] == ffw_version - - -def test_ddl_normalization_preserves_token_and_literal_boundaries() -> None: - assert forward._normalize_ddl("CREATE TABLE t(a TEXT)") == forward._normalize_ddl( - 'create table IF NOT EXISTS "t" ( "a" text )' - ) - assert forward._normalize_ddl("CREATE TABLE t(a b)") != forward._normalize_ddl("CREATE TABLE t(ab)") - assert forward._normalize_ddl("CREATE TABLE t(a DEFAULT 'a b')") != forward._normalize_ddl( - "CREATE TABLE t(a DEFAULT 'ab')" - ) - assert forward._normalize_ddl("CREATE TABLE t(a CHECK(a = 1-2))") != forward._normalize_ddl( - "CREATE TABLE t(a CHECK(a = 1e-2))" - ) - - -def test_activate_keeps_completed_receipt_byte_for_byte( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, _ = _archive(tmp_path) - receipt = tmp_path / "receipt.json" - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - prepare_forward(archive_root=root, receipt_path=receipt) - activated = activate_forward(receipt_path=receipt) - before = receipt.read_bytes() - active = root / "index.db" - with sqlite3.connect(active) as conn: - conn.execute("UPDATE sessions SET title = 'later write'") - conn.commit() - - repeated = activate_forward(receipt_path=receipt) - - assert repeated == activated - assert receipt.read_bytes() == before - - -def test_prepare_refuses_unexpected_schema_surplus( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, ffw_version = _archive(tmp_path) - with sqlite3.connect(root / "index.db") as conn: - conn.execute("CREATE TABLE unexpected_cache(value TEXT)") - conn.commit() - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - - with pytest.raises(IndexV37FastForwardError, match="unexpected_surplus"): - prepare_forward(archive_root=root, receipt_path=tmp_path / "receipt.json") - - generations = IndexGenerationStore.for_archive_root(root).generations_root - assert not [path for path in generations.iterdir() if path.name != f"v{ffw_version}"] - - -def test_prepare_repairs_preexisting_orphan_attachment_native_ids( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, _ = _archive(tmp_path) - with sqlite3.connect(root / "index.db") as conn: - conn.execute( - "INSERT INTO attachment_native_ids(ref_id, id_kind, native_id) VALUES ('missing', 'file', 'stale')" - ) - conn.commit() - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - - prepared = prepare_forward(archive_root=root, receipt_path=tmp_path / "receipt.json") - - postflight = prepared["postflight"] - assert isinstance(postflight, dict) - assert postflight["repaired_orphan_attachment_native_ids"] == 1 - generation = prepared["generation"] - assert isinstance(generation, dict) - with sqlite3.connect(str(generation["index_path"])) as conn: - assert conn.execute("PRAGMA foreign_key_check").fetchall() == [] - - -def test_prepare_refuses_running_daemon_before_clone( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, ffw_version = _archive(tmp_path) - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: 1234) - - with pytest.raises(IndexV37FastForwardError, match="1234"): - prepare_forward(archive_root=root, receipt_path=tmp_path / "receipt.json") - - generations = IndexGenerationStore.for_archive_root(root).generations_root - assert list(generations.iterdir()) == [generations / f"v{ffw_version}"] - - -def test_prepare_refuses_unwritable_receipt_before_checkpoint_or_clone( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, ffw_version = _archive(tmp_path) - blocker = tmp_path / "not-a-directory" - blocker.write_text("block receipt parent creation", encoding="utf-8") - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - - def unexpected_checkpoint(_path: Path, *, label: str = "active index") -> None: - pytest.fail(f"receipt preflight ran after {label} checkpoint") - - monkeypatch.setattr(forward, "_checkpoint_stopped_database", unexpected_checkpoint) - - with pytest.raises(IndexV37FastForwardError, match="receipt destination is not writable"): - prepare_forward(archive_root=root, receipt_path=blocker / "receipt.json") - - generations = IndexGenerationStore.for_archive_root(root).generations_root - assert list(generations.iterdir()) == [generations / f"v{ffw_version}"] - - -def test_prepare_checkpoints_stopped_active_index_before_census( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, _ = _archive(tmp_path) - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - active = IndexGenerationStore.for_archive_root(root).active_pointer.resolve() - Path(f"{active}-shm").write_bytes(b"stopped-writer-residue") - observed: list[tuple[Path, str]] = [] - original = forward._checkpoint_stopped_database - - def checkpoint(path: Path, *, label: str = "active index") -> None: - observed.append((path, label)) - original(path, label=label) - - monkeypatch.setattr(forward, "_checkpoint_stopped_database", checkpoint) - - prepare_forward(archive_root=root, receipt_path=tmp_path / "receipt.json") - - assert observed[0] == (IndexGenerationStore.for_archive_root(root).active_pointer, "active index") - assert observed[1][1] == "prepared clone" - assert not Path(f"{active}-shm").exists() - - -def test_activate_refuses_changed_source_snapshot( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, ffw_version = _archive(tmp_path) - receipt = tmp_path / "receipt.json" - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - prepare_forward(archive_root=root, receipt_path=receipt) - monkeypatch.setattr(forward, "source_revision_snapshot", lambda _root: "changed") - - with pytest.raises(IndexV37FastForwardError, match="source evidence changed"): - activate_forward(receipt_path=receipt) - - assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().parent.name == f"v{ffw_version}" - - -def test_activate_refuses_in_place_clone_mutation_with_preserved_stat_identity( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, ffw_version = _archive(tmp_path) - receipt = tmp_path / "receipt.json" - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - prepared = prepare_forward(archive_root=root, receipt_path=receipt) - generation = prepared["generation"] - assert isinstance(generation, dict) - clone = Path(str(generation["index_path"])) - before = clone.stat() - with sqlite3.connect(clone) as conn: - conn.execute("UPDATE sessions SET content_hash = ?", (b"x" * 32,)) - conn.commit() - forward._checkpoint_stopped_database(clone, label="mutated test clone") - os.utime(clone, ns=(before.st_atime_ns, before.st_mtime_ns)) - - with pytest.raises(IndexV37FastForwardError, match="clone bytes changed"): - activate_forward(receipt_path=receipt) - - assert clone.stat().st_size == before.st_size - assert clone.stat().st_mtime_ns == before.st_mtime_ns - assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().parent.name == f"v{ffw_version}" - - -def _add_unfetched_candidate_attachment(candidate_index: Path) -> None: - with sqlite3.connect(candidate_index) as conn: - session_row = conn.execute("SELECT session_id FROM sessions LIMIT 1").fetchone() - assert session_row is not None - message_row = conn.execute( - "SELECT message_id FROM messages WHERE session_id = ? LIMIT 1", (session_row[0],) - ).fetchone() - assert message_row is not None - conn.execute( - "INSERT INTO attachments(attachment_id, acquisition_status) VALUES ('candidate-unfetched', 'unfetched')" - ) - conn.execute( - "INSERT INTO attachment_refs(attachment_id, session_id, message_id, position, upload_origin) " - "VALUES ('candidate-unfetched', ?, ?, 99, 'drive')", - (session_row[0], message_row[0]), - ) - - -def _refresh_receipt_clone_identity(receipt_path: Path, clone: Path) -> None: - receipt = forward._load_receipt(receipt_path) - receipt["clone_identity"] = forward._proven_clone_identity(clone) - forward._write_receipt(receipt_path, receipt) - - -def test_activate_refuses_corpus_invalid_inactive_candidate_before_pointer_swap( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, ffw_version = _archive(tmp_path) - receipt = tmp_path / "receipt.json" - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - prepared = prepare_forward(archive_root=root, receipt_path=receipt) - generation = prepared["generation"] - assert isinstance(generation, dict) - clone = Path(str(generation["index_path"])) - _add_unfetched_candidate_attachment(clone) - _refresh_receipt_clone_identity(receipt, clone) - - with pytest.raises( - IndexV37FastForwardError, match="candidate corpus fidelity gate failed.*corpus-attachment-fidelity" - ): - activate_forward(receipt_path=receipt) - - assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().parent.name == f"v{ffw_version}" - - -def test_activate_recovery_refuses_corpus_invalid_inactive_candidate_before_pointer_swap( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, ffw_version = _archive(tmp_path) - receipt = tmp_path / "receipt.json" - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - prepared = prepare_forward(archive_root=root, receipt_path=receipt) - generation_payload = prepared["generation"] - assert isinstance(generation_payload, dict) - clone = Path(str(generation_payload["index_path"])) - _add_unfetched_candidate_attachment(clone) - _refresh_receipt_clone_identity(receipt, clone) - activating = forward._load_receipt(receipt) - activating["status"] = "activating" - forward._write_receipt(receipt, activating) - store = IndexGenerationStore.for_archive_root(root) - generation = store.load(str(generation_payload["generation_id"])) - store._write(replace(generation, state="promoting")) - - with pytest.raises( - IndexV37FastForwardError, match="candidate corpus fidelity gate failed.*corpus-attachment-fidelity" - ): - activate_forward(receipt_path=receipt) - - assert store.active_pointer.resolve().parent.name == f"v{ffw_version}" - assert store.load(generation.generation_id).state == "inactive" - - -def test_activate_recovers_after_pointer_swap_before_final_receipt( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_lifecycle_for_v36_upgrade: None -) -> None: - root, _ = _archive(tmp_path) - receipt = tmp_path / "receipt.json" - monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) - prepared = prepare_forward(archive_root=root, receipt_path=receipt) - generation_payload = prepared["generation"] - assert isinstance(generation_payload, dict) - clone = Path(str(generation_payload["index_path"])) - original_promote = IndexGenerationStore.promote - - def promote_then_crash(store: IndexGenerationStore, generation: IndexGeneration) -> IndexGeneration: - original_promote(store, generation) - raise RuntimeError("simulated crash after pointer swap") - - monkeypatch.setattr(IndexGenerationStore, "promote", promote_then_crash) - with pytest.raises(RuntimeError, match="simulated crash"): - activate_forward(receipt_path=receipt) - assert forward._load_receipt(receipt)["status"] == "activating" - assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve() == clone.resolve() - - monkeypatch.setattr(IndexGenerationStore, "promote", original_promote) - activated = activate_forward(receipt_path=receipt) - - assert activated["status"] == "activated" - assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve() == clone.resolve() From 9e9862884b63c5561898bec7b910e17584f70acc Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 5 Aug 2026 06:15:05 +0200 Subject: [PATCH 2/3] fix(devtools): harden fast-forward replay receipts The retained-raw proof must work for archive-scale identifier sets and must reject a receipt whose declared equivalence no longer matches its stored hashes. Chunk SQLite IN queries, normalize FTS comparison ordering, and make activation validate the two hash dictionaries directly. The regressions exercise chunked archive reads, forged proof data, and a receipt modified without updating its integrity hash. Co-Authored-By: Codex --- devtools/index_fast_forward.py | 89 ++++++++++++++----- .../unit/devtools/test_index_fast_forward.py | 84 ++++++++++++++++- 2 files changed, 150 insertions(+), 23 deletions(-) diff --git a/devtools/index_fast_forward.py b/devtools/index_fast_forward.py index bbf77f1a69..9da3a583bc 100644 --- a/devtools/index_fast_forward.py +++ b/devtools/index_fast_forward.py @@ -16,6 +16,7 @@ import sqlite3 import time import uuid +from collections.abc import Callable, Iterator from contextlib import closing, suppress from dataclasses import asdict from pathlib import Path @@ -37,11 +38,12 @@ from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive from polylogue.storage.sqlite.connection_profile import open_readonly_connection -from polylogue.storage.sqlite.lifecycle import IndexFastForwardPlan, index_fast_forward_plan +from polylogue.storage.sqlite.lifecycle import FastForwardOperationKind, IndexFastForwardPlan, index_fast_forward_plan from polylogue.storage.sqlite.runtime_indexes import ensure_runtime_indexes_sync RECEIPT_SCHEMA = "polylogue.index-fast-forward.v1" DEFAULT_SAMPLE_SIZE = 8 +IN_QUERY_CHUNK_SIZE = 500 class IndexFastForwardError(RuntimeError): @@ -230,7 +232,7 @@ def _expected_surplus(plan: IndexFastForwardPlan) -> set[str]: f"{kind}:{name}" for declaration in plan.declarations for operation in declaration.operations - if operation.kind.value == "drop-table" + if operation.kind is FastForwardOperationKind.DROP_TABLE for kind, name in operation.objects } @@ -286,17 +288,39 @@ def _fingerprints(origins: tuple[str, ...]) -> dict[str, object]: } +def _chunks(values: tuple[str, ...] | list[str], *, size: int | None = None) -> Iterator[tuple[str, ...]]: + effective_size = IN_QUERY_CHUNK_SIZE if size is None else size + if effective_size <= 0: + raise ValueError("SQLite IN-query chunk size must be positive") + for offset in range(0, len(values), effective_size): + yield tuple(values[offset : offset + effective_size]) + + +def _query_rows_in_chunks( + conn: sqlite3.Connection, + sql_for_marks: Callable[[str], str], + values: tuple[str, ...] | list[str], +) -> list[tuple[object, ...]]: + rows: list[tuple[object, ...]] = [] + for chunk in _chunks(values): + marks = ", ".join("?" for _ in chunk) + rows.extend(conn.execute(sql_for_marks(marks), chunk).fetchall()) + return rows + + def _sample_manifest(archive_root: Path, index_path: Path, *, limit: int) -> list[dict[str, object]]: with closing(sqlite3.connect(archive_root / "source.db")) as source_conn: raw_ids = [str(row[0]) for row in source_conn.execute("SELECT raw_id FROM raw_sessions ORDER BY raw_id")] if not raw_ids: raise IndexFastForwardError("source-backed fast-forward proof requires retained raw evidence") - marks = ", ".join("?" for _ in raw_ids) with closing(open_readonly_connection(index_path.resolve(strict=True), immutable=True)) as index_conn: - rows = index_conn.execute( - f"SELECT raw_id, session_id, origin FROM sessions WHERE raw_id IN ({marks}) ORDER BY raw_id, session_id", + rows = _query_rows_in_chunks( + index_conn, + lambda marks: ( + f"SELECT raw_id, session_id, origin FROM sessions WHERE raw_id IN ({marks}) ORDER BY raw_id, session_id" + ), raw_ids, - ).fetchall() + ) grouped: dict[str, dict[str, object]] = {} for raw_id, session_id, origin in rows: entry = grouped.setdefault(str(raw_id), {"raw_id": str(raw_id), "session_ids": [], "origins": []}) @@ -315,19 +339,29 @@ def _json_value(value: object) -> object: return value -def _hash_query(conn: sqlite3.Connection, sql: str, params: tuple[object, ...]) -> str: - rows = conn.execute(sql, params).fetchall() +def _hash_rows(rows: list[tuple[object, ...]]) -> str: payload = [[_json_value(value) for value in row] for row in rows] + payload.sort(key=lambda row: json.dumps(row, sort_keys=True, separators=(",", ":"))) return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() +def _hash_query_in_chunks( + conn: sqlite3.Connection, + sql_for_marks: Callable[[str], str], + values: tuple[str, ...], +) -> str: + return _hash_rows(_query_rows_in_chunks(conn, sql_for_marks, values)) + + def _canonical_hashes(conn: sqlite3.Connection, session_ids: tuple[str, ...]) -> dict[str, object]: if not session_ids: return {"sessions": "", "messages": "", "blocks": "", "fts": "", "scoped": {}} - marks = ", ".join("?" for _ in session_ids) - messages = conn.execute(f"SELECT message_id FROM messages WHERE session_id IN ({marks})", session_ids).fetchall() - message_ids = tuple(str(row[0]) for row in messages) - message_marks = ", ".join("?" for _ in message_ids) or "NULL" + messages = _query_rows_in_chunks( + conn, + lambda marks: f"SELECT message_id FROM messages WHERE session_id IN ({marks}) ORDER BY message_id", + session_ids, + ) + message_ids = tuple(sorted(str(row[0]) for row in messages)) scoped: dict[str, str] = {} for table_name, key_column in ( (str(row[0]), key_column) @@ -339,24 +373,33 @@ def _canonical_hashes(conn: sqlite3.Connection, session_ids: tuple[str, ...]) -> ): columns = [str(info[1]) for info in conn.execute(f'PRAGMA table_info("{table_name}")')] order_by = ", ".join(str(index) for index in range(1, len(columns) + 1)) - scoped[table_name] = _hash_query( + scoped[table_name] = _hash_query_in_chunks( conn, - f'SELECT * FROM "{table_name}" WHERE "{key_column}" IN ({marks}) ORDER BY {order_by}', + lambda marks, table_name=table_name, key_column=key_column, order_by=order_by: ( + f'SELECT * FROM "{table_name}" WHERE "{key_column}" IN ({marks}) ORDER BY {order_by}' + ), session_ids, ) return { - "sessions": _hash_query( - conn, f"SELECT * FROM sessions WHERE session_id IN ({marks}) ORDER BY session_id", session_ids + "sessions": _hash_query_in_chunks( + conn, lambda marks: f"SELECT * FROM sessions WHERE session_id IN ({marks}) ORDER BY session_id", session_ids ), - "messages": _hash_query( - conn, f"SELECT * FROM messages WHERE session_id IN ({marks}) ORDER BY message_id", session_ids + "messages": _hash_query_in_chunks( + conn, + lambda marks: f"SELECT * FROM messages WHERE session_id IN ({marks}) ORDER BY message_id", + session_ids, ), - "blocks": _hash_query( - conn, f"SELECT * FROM blocks WHERE message_id IN ({message_marks}) ORDER BY block_id", message_ids + "blocks": _hash_query_in_chunks( + conn, + lambda marks: f"SELECT * FROM blocks WHERE message_id IN ({marks}) ORDER BY block_id", + message_ids, ), - "fts": _hash_query( + "fts": _hash_query_in_chunks( conn, - f"SELECT block_id, message_id, session_id, block_type, text FROM messages_fts WHERE session_id IN ({marks}) ORDER BY rowid", + lambda marks: ( + "SELECT block_id, message_id, session_id, block_type, text " + f"FROM messages_fts WHERE session_id IN ({marks}) ORDER BY session_id, message_id, block_id" + ), session_ids, ), "scoped": scoped, @@ -424,6 +467,8 @@ def _require_complete_proof(proof: dict[str, object]) -> None: scoped = hashes.get("scoped") if not isinstance(scoped, dict) or not scoped: raise IndexFastForwardError(f"source replay proof is incomplete: {key}.scoped") + if proof["fast_forward_hashes"] != proof["canonical_replay_hashes"]: + raise IndexFastForwardError("source replay proof hashes disagree between clone and canonical replay") def _require_candidate_corpus_fidelity(archive_root: Path, candidate_index: Path) -> None: diff --git a/tests/unit/devtools/test_index_fast_forward.py b/tests/unit/devtools/test_index_fast_forward.py index 83e9ca7490..9bfaf02656 100644 --- a/tests/unit/devtools/test_index_fast_forward.py +++ b/tests/unit/devtools/test_index_fast_forward.py @@ -22,7 +22,7 @@ class _Report: checks: tuple[object, ...] = () -def _archive(tmp_path: Path) -> Path: +def _archive(tmp_path: Path, *, extra_native_ids: tuple[str, ...] = ()) -> Path: root = tmp_path / "archive" root.mkdir() for tier in (ArchiveTier.SOURCE, ArchiveTier.USER, ArchiveTier.EMBEDDINGS, ArchiveTier.OPS): @@ -68,6 +68,8 @@ def _archive(tmp_path: Path) -> Path: source_path="source-backed-session.json", acquired_at_ms=1, ) + for native_id in extra_native_ids: + _write_raw_backed_session(root, native_id) with sqlite3.connect(active) as conn: conn.execute("CREATE TABLE session_runs(id TEXT)") conn.execute("CREATE TABLE session_observed_events(id TEXT)") @@ -97,6 +99,35 @@ def _no_corpus_failure(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(forward, "verify_archive", lambda *args, **kwargs: _Report()) +def _write_raw_backed_session(root: Path, native_id: str) -> None: + payload_object = { + "id": native_id, + "conversation_id": native_id, + "title": f"source-backed proof {native_id}", + "mapping": { + "message": { + "id": f"message-{native_id}", + "parent": None, + "children": [], + "message": { + "id": f"message-{native_id}", + "author": {"role": "user"}, + "content": {"content_type": "text", "parts": [f"retained raw proof {native_id}"]}, + "create_time": 1_700_000_000, + }, + } + }, + } + parsed = parse_payload(Provider.CHATGPT, payload_object, native_id, source_path=f"{native_id}.json") + with ArchiveStore.open_existing(root, read_only=False) as archive: + archive.write_raw_and_parsed( + parsed[0], + payload=json.dumps(payload_object, sort_keys=True).encode(), + source_path=f"{native_id}.json", + acquired_at_ms=1, + ) + + def test_prepare_receipt_proves_source_replay_equivalence( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_v37: None ) -> None: @@ -185,3 +216,54 @@ def removed_replay(*args: object, **kwargs: object) -> dict[str, object]: forward.prepare_forward(archive_root=root, receipt_path=tmp_path / "transition.json") assert not list(IndexGenerationStore.for_archive_root(root).generations_root.glob("gen-*/index.db")) + + +def test_chunked_proof_queries_cover_archive_scale_ids(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = _archive(tmp_path, extra_native_ids=("source-backed-session-2", "source-backed-session-3")) + monkeypatch.setattr(forward, "IN_QUERY_CHUNK_SIZE", 1) + active = IndexGenerationStore.for_archive_root(root).active_pointer.resolve() + + manifest = forward._sample_manifest(root, active, limit=3) + session_ids = tuple(session_id for entry in manifest for session_id in entry["session_ids"]) + with sqlite3.connect(active) as conn: + hashes = forward._canonical_hashes(conn, session_ids) + + assert len(manifest) == 3 + assert hashes["sessions"] and hashes["messages"] and hashes["blocks"] and hashes["fts"] + + +def test_activation_refuses_forged_equivalence_hashes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_v37: None +) -> None: + root = _archive(tmp_path) + monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) + receipt_path = tmp_path / "transition.json" + receipt = forward.prepare_forward(archive_root=root, receipt_path=receipt_path) + proof = receipt["proof"] + assert isinstance(proof, dict) + canonical = proof["canonical_replay_hashes"] + assert isinstance(canonical, dict) + canonical["fts"] = "forged" + forward._write_receipt(receipt_path, receipt) + _no_corpus_failure(monkeypatch) + + with pytest.raises(forward.IndexFastForwardError, match="hashes disagree"): + forward.activate_forward(receipt_path=receipt_path) + + assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().parent.name == "v36" + + +def test_activation_refuses_tampered_receipt(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _patch_v37: None) -> None: + root = _archive(tmp_path) + monkeypatch.setattr(forward, "running_daemon_pid", lambda _config: None) + receipt_path = tmp_path / "transition.json" + forward.prepare_forward(archive_root=root, receipt_path=receipt_path) + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + receipt["source_snapshot"] = "tampered" + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + _no_corpus_failure(monkeypatch) + + with pytest.raises(forward.IndexFastForwardError, match="hash mismatch"): + forward.activate_forward(receipt_path=receipt_path) + + assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().parent.name == "v36" From 42f4cfdf92701c0a6490760d913751a3d97dfc8c Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 5 Aug 2026 06:17:42 +0200 Subject: [PATCH 3/3] fix(devtools): type fast-forward proof queries Bind scoped proof query parameters explicitly and preserve typed manifest session identifiers in the archive-scale regression. Co-Authored-By: Codex --- devtools/index_fast_forward.py | 10 +++++++--- tests/unit/devtools/test_index_fast_forward.py | 3 ++- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/devtools/index_fast_forward.py b/devtools/index_fast_forward.py index 9da3a583bc..f27604e39a 100644 --- a/devtools/index_fast_forward.py +++ b/devtools/index_fast_forward.py @@ -19,6 +19,7 @@ from collections.abc import Callable, Iterator from contextlib import closing, suppress from dataclasses import asdict +from functools import partial from pathlib import Path from typing import cast @@ -353,6 +354,10 @@ def _hash_query_in_chunks( return _hash_rows(_query_rows_in_chunks(conn, sql_for_marks, values)) +def _scoped_table_sql(marks: str, *, table_name: str, key_column: str, order_by: str) -> str: + return f'SELECT * FROM "{table_name}" WHERE "{key_column}" IN ({marks}) ORDER BY {order_by}' + + def _canonical_hashes(conn: sqlite3.Connection, session_ids: tuple[str, ...]) -> dict[str, object]: if not session_ids: return {"sessions": "", "messages": "", "blocks": "", "fts": "", "scoped": {}} @@ -373,11 +378,10 @@ def _canonical_hashes(conn: sqlite3.Connection, session_ids: tuple[str, ...]) -> ): columns = [str(info[1]) for info in conn.execute(f'PRAGMA table_info("{table_name}")')] order_by = ", ".join(str(index) for index in range(1, len(columns) + 1)) + scoped[table_name] = _hash_query_in_chunks( conn, - lambda marks, table_name=table_name, key_column=key_column, order_by=order_by: ( - f'SELECT * FROM "{table_name}" WHERE "{key_column}" IN ({marks}) ORDER BY {order_by}' - ), + partial(_scoped_table_sql, table_name=table_name, key_column=key_column, order_by=order_by), session_ids, ) return { diff --git a/tests/unit/devtools/test_index_fast_forward.py b/tests/unit/devtools/test_index_fast_forward.py index 9bfaf02656..9b4aab6372 100644 --- a/tests/unit/devtools/test_index_fast_forward.py +++ b/tests/unit/devtools/test_index_fast_forward.py @@ -4,6 +4,7 @@ import sqlite3 from dataclasses import dataclass from pathlib import Path +from typing import cast import pytest @@ -224,7 +225,7 @@ def test_chunked_proof_queries_cover_archive_scale_ids(tmp_path: Path, monkeypat active = IndexGenerationStore.for_archive_root(root).active_pointer.resolve() manifest = forward._sample_manifest(root, active, limit=3) - session_ids = tuple(session_id for entry in manifest for session_id in entry["session_ids"]) + session_ids = tuple(session_id for entry in manifest for session_id in cast(list[str], entry["session_ids"])) with sqlite3.connect(active) as conn: hashes = forward._canonical_hashes(conn, session_ids)