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..f27604e39a --- /dev/null +++ b/devtools/index_fast_forward.py @@ -0,0 +1,639 @@ +"""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 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 + +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 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): + """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 is FastForwardOperationKind.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 _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") + with closing(open_readonly_connection(index_path.resolve(strict=True), immutable=True)) as index_conn: + 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, + ) + 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_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 _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": {}} + 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) + 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_in_chunks( + conn, + partial(_scoped_table_sql, table_name=table_name, key_column=key_column, order_by=order_by), + session_ids, + ) + return { + "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_in_chunks( + conn, + lambda marks: f"SELECT * FROM messages WHERE session_id IN ({marks}) ORDER BY message_id", + session_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_in_chunks( + conn, + 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, + } + + +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") + 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: + 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..9b4aab6372 --- /dev/null +++ b/tests/unit/devtools/test_index_fast_forward.py @@ -0,0 +1,270 @@ +from __future__ import annotations + +import json +import sqlite3 +from dataclasses import dataclass +from pathlib import Path +from typing import cast + +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, *, extra_native_ids: tuple[str, ...] = ()) -> 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, + ) + 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)") + 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 _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: + 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")) + + +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 cast(list[str], 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" 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()