diff --git a/devtools/command_catalog.py b/devtools/command_catalog.py index 9a22cf3ab2..732f06fc4a 100644 --- a/devtools/command_catalog.py +++ b/devtools/command_catalog.py @@ -408,11 +408,12 @@ def to_dict(self) -> dict[str, object]: CommandSpec( "workspace index-fast-forward", "workspace", - "Clone-first index.db v32-to-v35 fast-forward with receipts and atomic rollback.", + "Apply a declared clone-first index.db fast-forward with receipts and rollback.", "devtools.index_fast_forward", use_when=( - "Upgrade a quiesced v32 derived index without raw replay: reflink an inactive generation, " - "apply exact canonical schema/FTS deltas, validate the clone, then separately activate or roll back." + "Upgrade a quiesced supported derived index without raw replay: reflink an inactive generation, " + "apply declared canonical schema/FTS deltas, validate structural equivalence on the clone, " + "then separately activate or roll back. Semantic-reparse deltas deliberately require rebuild/reprocess." ), examples=( "devtools workspace index-fast-forward plan --source /path/to/index.db", diff --git a/devtools/index_fast_forward.py b/devtools/index_fast_forward.py index 933d10d334..4ee5809133 100644 --- a/devtools/index_fast_forward.py +++ b/devtools/index_fast_forward.py @@ -33,6 +33,7 @@ ) from polylogue.storage.fts.pl_fold import pl_fold, pl_fold_sql_expr from polylogue.storage.sqlite.archive_tiers.index import INDEX_DDL +from polylogue.storage.sqlite.lifecycle import IndexFastForwardPlan, index_fast_forward_plan FAST_FORWARD_FROM_VERSION = 32 FAST_FORWARD_TO_VERSION = 35 @@ -41,23 +42,16 @@ DEFAULT_MAX_IO_FULL_AVG10 = 45.0 DEFAULT_MAX_MEMORY_FULL_AVG10 = 5.0 -_CANONICAL_OBJECTS = ( - ("table", "insight_materialization"), - ("index", "idx_web_constructs_message"), - ("view", "delegations"), - ("table", "messages_fts"), - ("trigger", "messages_fts_ai"), - ("trigger", "messages_fts_ad"), - ("trigger", "messages_fts_au"), - ("table", "session_work_events_fts"), - ("trigger", "session_work_events_fts_ai"), - ("trigger", "session_work_events_fts_ad"), - ("trigger", "session_work_events_fts_au"), - ("table", "threads_fts"), - ("trigger", "threads_fts_ai"), - ("trigger", "threads_fts_ad"), - ("trigger", "threads_fts_au"), +_EXECUTOR_STAGE_NAMES = ( + "v33-insight-check", + "v34-index-and-delegations", + "v35-messages-fts", + "v35-insight-fts", ) +_PLAN = index_fast_forward_plan(FAST_FORWARD_FROM_VERSION, FAST_FORWARD_TO_VERSION) +if _PLAN is None: + raise RuntimeError("the deployed index fast-forward executor lacks a declared clone-safe plan") +_CANONICAL_OBJECTS = _PLAN.canonical_objects _STRUCTURAL_TABLES = ( "sessions", @@ -262,6 +256,14 @@ def _database_metrics(conn: sqlite3.Connection) -> dict[str, int]: } +def _deployed_plan(source_version: int) -> IndexFastForwardPlan | None: + """Return a declared plan only when the deployed executor owns every stage.""" + plan = index_fast_forward_plan(source_version, FAST_FORWARD_TO_VERSION) + if plan is None or any(stage not in _EXECUTOR_STAGE_NAMES for stage in plan.stage_names): + return None + return plan + + def plan_index(path: Path) -> dict[str, object]: identity = file_identity(path) with sqlite3.connect(f"file:{identity.resolved_path}?mode=ro", uri=True, timeout=30.0) as conn: @@ -275,15 +277,31 @@ def plan_index(path: Path) -> dict[str, object]: "threads_source": int(conn.execute("SELECT COUNT(*) FROM threads").fetchone()[0]), "threads_indexed": int(conn.execute("SELECT COUNT(*) FROM threads_fts").fetchone()[0]), } + plan = _deployed_plan(metrics["user_version"]) + already_current = metrics["user_version"] == FAST_FORWARD_TO_VERSION return { "identity": asdict(identity), "metrics": metrics, "counts": counts, "fts": fts, "pressure": asdict(pressure_sample()), - "eligible": metrics["user_version"] == FAST_FORWARD_FROM_VERSION, + "eligible": plan is not None, + "already_current": already_current, "target_version": FAST_FORWARD_TO_VERSION, - "raw_reparse": False, + "delta_classes": ( + [delta_class.value for declaration in plan.declarations for delta_class in declaration.classes] + if plan is not None + else [] + ), + "declared_stages": list(plan.stage_names) if plan is not None else [], + "required_action": ( + "already_current" + if already_current + else "clone_sql_fast_forward" + if plan is not None + else "rebuild_or_reprocess" + ), + "raw_reparse": plan is None and not already_current, } @@ -532,25 +550,31 @@ def fast_forward_clone( batch_rows: int = DEFAULT_MESSAGE_BATCH_ROWS, fail_after_stage: str | None = None, ) -> dict[str, object]: - canonical = _canonical_schema() identity = file_identity(db_path) with sqlite3.connect(db_path, timeout=120.0) as conn: conn.execute("PRAGMA foreign_keys = ON") metrics = _database_metrics(conn) - if metrics["user_version"] != FAST_FORWARD_FROM_VERSION: - raise RuntimeError( - f"clone must start at index user_version {FAST_FORWARD_FROM_VERSION}, found {metrics['user_version']}" - ) + source_version = metrics["user_version"] before_counts = _table_counts(conn) + plan = _deployed_plan(source_version) + if plan is None: + raise RuntimeError( + f"no deployed clone-safe plan from index schema v{source_version}; rebuild or reprocess from source evidence" + ) + canonical = _canonical_schema() receipt: dict[str, object] = { "schema": RECEIPT_SCHEMA, "status": "upgrading", "started_at_ms": _now_ms(), - "source_version": FAST_FORWARD_FROM_VERSION, + "source_version": source_version, "target_version": FAST_FORWARD_TO_VERSION, "clone_identity_before": asdict(identity), "structural_counts_before": before_counts, "raw_reparse": False, + "delta_classes": [ + delta_class.value for declaration in plan.declarations for delta_class in declaration.classes + ], + "declared_stages": list(plan.stage_names), "stages": [], "pressure_samples": [], } @@ -576,84 +600,62 @@ def apply_v34(conn: sqlite3.Connection) -> dict[str, object]: conn.commit() return {"index": "idx_web_constructs_message", "view": "delegations"} + def apply_v35_messages(conn: sqlite3.Connection) -> dict[str, object]: + return _rebuild_messages_fts( + conn, + canonical, + batch_rows=batch_rows, + pressure_guard=guard, + ) + + def apply_v35_insights(conn: sqlite3.Connection) -> dict[str, object]: + result: dict[str, object] = { + "session_work_events_fts": _rebuild_small_fts( + conn, + canonical, + table_name="session_work_events_fts", + trigger_names=( + "session_work_events_fts_ai", + "session_work_events_fts_ad", + "session_work_events_fts_au", + ), + insert_sql=f""" + INSERT INTO session_work_events_fts (event_id, session_id, work_event_type, text) + SELECT event_id, session_id, work_event_type, {pl_fold_sql_expr("search_text")} + FROM session_work_events + """, + ), + "threads_fts": _rebuild_small_fts( + conn, + canonical, + table_name="threads_fts", + trigger_names=("threads_fts_ai", "threads_fts_ad", "threads_fts_au"), + insert_sql=f""" + INSERT INTO threads_fts (thread_id, root_id, text) + SELECT thread_id, thread_id, {pl_fold_sql_expr("search_text")} + FROM threads + """, + ), + } + record_fts_invariant_snapshot_sync(conn, fts_invariant_snapshot_sync(conn)) + conn.commit() + return result + try: guard() with sqlite3.connect(db_path, timeout=120.0) as conn: conn.execute("PRAGMA foreign_keys = ON") - - _run_stage( - receipt, - receipt_path, - "v33-insight-check", - lambda: apply_v33(conn), - ) - if fail_after_stage == "v33-insight-check": - raise RuntimeError("injected failure after v33-insight-check") - guard() - - _run_stage( - receipt, - receipt_path, - "v34-index-and-delegations", - lambda: apply_v34(conn), - ) - if fail_after_stage == "v34-index-and-delegations": - raise RuntimeError("injected failure after v34-index-and-delegations") - guard() - - _run_stage( - receipt, - receipt_path, - "v35-messages-fts", - lambda: _rebuild_messages_fts( - conn, - canonical, - batch_rows=batch_rows, - pressure_guard=guard, - ), - ) - if fail_after_stage == "v35-messages-fts": - raise RuntimeError("injected failure after v35-messages-fts") - guard() - - _run_stage( - receipt, - receipt_path, - "v35-insight-fts", - lambda: { - "session_work_events_fts": _rebuild_small_fts( - conn, - canonical, - table_name="session_work_events_fts", - trigger_names=( - "session_work_events_fts_ai", - "session_work_events_fts_ad", - "session_work_events_fts_au", - ), - insert_sql=f""" - INSERT INTO session_work_events_fts (event_id, session_id, work_event_type, text) - SELECT event_id, session_id, work_event_type, {pl_fold_sql_expr("search_text")} - FROM session_work_events - """, - ), - "threads_fts": _rebuild_small_fts( - conn, - canonical, - table_name="threads_fts", - trigger_names=("threads_fts_ai", "threads_fts_ad", "threads_fts_au"), - insert_sql=f""" - INSERT INTO threads_fts (thread_id, root_id, text) - SELECT thread_id, thread_id, {pl_fold_sql_expr("search_text")} - FROM threads - """, - ), - }, - ) - record_fts_invariant_snapshot_sync(conn, fts_invariant_snapshot_sync(conn)) - conn.commit() - if fail_after_stage == "v35-insight-fts": - raise RuntimeError("injected failure after v35-insight-fts") - guard() + stage_handlers: dict[str, Callable[[], dict[str, object]]] = { + "v33-insight-check": lambda: apply_v33(conn), + "v34-index-and-delegations": lambda: apply_v34(conn), + "v35-messages-fts": lambda: apply_v35_messages(conn), + "v35-insight-fts": lambda: apply_v35_insights(conn), + } + for stage_name in plan.stage_names: + _run_stage(receipt, receipt_path, stage_name, stage_handlers[stage_name]) + if fail_after_stage == stage_name: + raise RuntimeError(f"injected failure after {stage_name}") + guard() pre_version_validation = _run_stage( receipt, @@ -662,7 +664,7 @@ def apply_v34(conn: sqlite3.Connection) -> dict[str, object]: lambda: validate_clone( db_path, expected_counts=before_counts, - expected_version=FAST_FORWARD_FROM_VERSION, + expected_version=source_version, run_quick_check=True, ), ) @@ -966,16 +968,20 @@ def activate_generation(receipt_path: Path, *, service: str | None = None, resta "clone_identity": clone_identity, "receipt_validation_reused": True, } - _swap_active_symlink(source_link, clone, label="v35") + # Persist rollback metadata BEFORE the swap: a crash between the symlink + # swap and the receipt write must leave a receipt that rollback accepts. receipt.update( { - "status": "activated", - "activated_at_ms": _now_ms(), + "status": "activating", + "activation_started_at_ms": _now_ms(), "rollback_target": str(source_target), "activation_validation": validation, } ) _write_receipt(receipt_path, receipt) + _swap_active_symlink(source_link, clone, label="v35") + receipt.update({"status": "activated", "activated_at_ms": _now_ms()}) + _write_receipt(receipt_path, receipt) if service and restart: _restart_and_verify_or_rollback( receipt, @@ -989,8 +995,8 @@ def activate_generation(receipt_path: Path, *, service: str | None = None, resta def rollback_generation(receipt_path: Path, *, service: str | None = None, restart: bool = False) -> dict[str, object]: receipt = _load_receipt(receipt_path) - if receipt.get("status") != "activated": - raise RuntimeError("rollback requires an activated receipt") + if receipt.get("status") not in ("activated", "activating"): + raise RuntimeError("rollback requires an activated or interrupted-activation receipt") if service and _service_active(service): raise RuntimeError(f"service must be stopped before rollback: {service}") source_link = Path(str(receipt["source_link"])) @@ -1032,6 +1038,10 @@ def _parser() -> argparse.ArgumentParser: rollback.add_argument("--receipt", type=Path, required=True) rollback.add_argument("--service", default="polylogued.service") rollback.add_argument("--restart", action="store_true") + # devtools click_dispatch forwards a root/local --json flag into argv; + # output is already JSON, so accept it everywhere as a no-op. + for owner in (parser, plan, clone, validate, activate, rollback): + owner.add_argument("--json", action="store_true", help="JSON output (always on)") return parser diff --git a/devtools/render_quality_reference.py b/devtools/render_quality_reference.py index 1ad6fa3d9e..5b10be8c97 100644 --- a/devtools/render_quality_reference.py +++ b/devtools/render_quality_reference.py @@ -230,15 +230,19 @@ def build_document(registry: QualityRegistry, *, runtime_coverage: RuntimeScenar "Polylogue uses two schema-evolution regimes (see", "[Schema Versioning Model](internals.md#schema-versioning-model)):", "durable tiers use explicit additive migrations with a backup gate, while", - "derived tiers rebuild or blue-green replace from source evidence.", + "derived tiers rebuild or blue-green replace from source evidence, except for", + "declared, clone-validated SQL fast-forwards for non-semantic deltas; they", + "prove structural clone equivalence, not parser-content equivalence.", f"`{control_plane_command('lab policy schema-versioning')}` enforces that policy boundary:", "", "- It scans derived-tier storage modules for upgrade-shaped helpers", " (`build_vN_to_vM`, `_apply_version_upgrade_plan`, `upgrade_vN_to_vM`,", " `migrate_vN_*`, `ensure_schema_upgrades_vN`).", + "- It requires every index schema bump since the supported compatibility floor", + " to declare its delta class before it can ship.", "- It allows numbered SQL resources only under durable migration roots:", " `polylogue/storage/sqlite/migrations/{source,user}/`.", - "- If a derived upgrade helper or invalid migration resource is found,", + "- If a derived upgrade helper, undeclared index delta, or invalid migration resource is found,", " the lint fails.", "", "The lint runs as part of `devtools verify --lab`, not the fast default path:", diff --git a/devtools/verify_schema_upgrade_lane.py b/devtools/verify_schema_upgrade_lane.py index a941021e8e..edc4ad8b28 100644 --- a/devtools/verify_schema_upgrade_lane.py +++ b/devtools/verify_schema_upgrade_lane.py @@ -8,8 +8,8 @@ * Durable tiers (``source.db`` and ``user.db``) may use explicit additive SQL migrations with a backup gate. * Derived/rebuildable tiers (``index.db`` and ``embeddings.db``) do not use - in-place migration machinery; they are rebuilt or blue-green replaced from - durable source evidence. + migration chains. They are rebuilt or blue-green replaced from durable source + evidence, except for explicitly declared, clone-validated SQL plans. What this lint checks --------------------- @@ -20,8 +20,9 @@ out in ``docs/internals.md`` and in the witness archive (``.local/witnesses/new/*schema_upgrades*``). -2. Fail if any helper exists for a derived tier. Durable-tier migrations must - live under ``polylogue/storage/sqlite/migrations/{source,user}/`` as +2. Fail if any legacy helper exists for a derived tier, or when the current + index schema version lacks a delta-class declaration. Durable-tier migrations + must live under ``polylogue/storage/sqlite/migrations/{source,user}/`` as numbered SQL resources. The lint is intentionally narrow. It detects helper names associated @@ -43,6 +44,8 @@ from pathlib import Path from devtools import repo_root as _get_root +from polylogue.storage.sqlite.archive_tiers.index import INDEX_SCHEMA_VERSION +from polylogue.storage.sqlite.lifecycle import IndexDeltaDeclarationReport, index_delta_declaration_report ROOT = _get_root() STORAGE_SQLITE_DIR = ROOT / "polylogue" / "storage" / "sqlite" @@ -113,10 +116,13 @@ def _invalid_migration_paths() -> list[Path]: return invalid -def _format_report(*, helpers: list[HelperHit], invalid_migrations: list[Path]) -> str: +def _format_report( + *, helpers: list[HelperHit], invalid_migrations: list[Path], delta_report: IndexDeltaDeclarationReport +) -> str: lines = [ f"derived-tier upgrade helpers found: {len(helpers)}", f"invalid durable migration resources found: {len(invalid_migrations)}", + f"undeclared index schema deltas found: {len(delta_report['missing_versions'])}", ] if helpers: lines.append("") @@ -131,7 +137,16 @@ def _format_report(*, helpers: list[HelperHit], invalid_migrations: list[Path]) lines.append("Invalid migration resources:") for path in invalid_migrations: lines.append(f" {path.relative_to(ROOT)}") - if not helpers and not invalid_migrations: + if not bool(delta_report["ok"]): + lines.append("") + lines.append("Index fast-forward declaration drift:") + lines.append(f" compatibility floor: v{delta_report['compatibility_floor']}") + lines.append(f" missing: {list(delta_report['missing_versions'])}") + lines.append(f" duplicate: {list(delta_report['duplicate_versions'])}") + lines.append(f" invalid: {list(delta_report['invalid_versions'])}") + lines.append("") + lines.append("Policy violation: each index schema bump needs a declared delta class.") + if not helpers and not invalid_migrations and bool(delta_report["ok"]): lines.append("") lines.append("Schema evolution policy intact.") return "\n".join(lines) @@ -147,6 +162,7 @@ def main(argv: list[str] | None = None) -> int: helpers = _collect_upgrade_helpers() invalid_migrations = _invalid_migration_paths() + delta_report = index_delta_declaration_report(INDEX_SCHEMA_VERSION) if args.json: payload = { @@ -154,13 +170,14 @@ def main(argv: list[str] | None = None) -> int: {"name": hit.name, "path": str(hit.path.relative_to(ROOT)), "line": hit.lineno} for hit in helpers ], "invalid_migration_resources": [str(path.relative_to(ROOT)) for path in invalid_migrations], - "ok": not helpers and not invalid_migrations, + "index_delta_declarations": delta_report, + "ok": not helpers and not invalid_migrations and bool(delta_report["ok"]), } print(json.dumps(payload, indent=2)) else: - print(_format_report(helpers=helpers, invalid_migrations=invalid_migrations)) + print(_format_report(helpers=helpers, invalid_migrations=invalid_migrations, delta_report=delta_report)) - return 0 if not helpers and not invalid_migrations else 1 + return 0 if not helpers and not invalid_migrations and bool(delta_report["ok"]) else 1 if __name__ == "__main__": diff --git a/docs/devtools.md b/docs/devtools.md index 7d4f1146a1..4b937b5ea8 100644 --- a/docs/devtools.md +++ b/docs/devtools.md @@ -196,7 +196,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-fast-forward` | Clone-first index.db v32-to-v35 fast-forward with receipts and atomic rollback. | +| `devtools workspace index-fast-forward` | Apply a declared clone-first index.db fast-forward with receipts and rollback. | | `devtools workspace lineage-validation` | Validate lineage-count evidence before citing archive counts externally. | | `devtools workspace read-package` | Render a declarative package of Polylogue read artifacts. | | `devtools workspace scale-regression` | Run the seeded large-archive scale-regression probe. | diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index b76ef26891..4f0dc7e090 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -3577,6 +3577,10 @@ files: loc: 289 target: polylogue/storage/sqlite/connection_profile.py owner: stable + - path: polylogue/storage/sqlite/lifecycle.py + loc: 235 + target: polylogue/storage/sqlite/lifecycle.py + owner: stable - path: polylogue/storage/sqlite/maintenance.py loc: 100 target: polylogue/storage/sqlite/maintenance.py diff --git a/docs/test-quality-workflows.md b/docs/test-quality-workflows.md index 7e13e902c6..5b36400720 100644 --- a/docs/test-quality-workflows.md +++ b/docs/test-quality-workflows.md @@ -87,15 +87,19 @@ devtools lab lanes --lane live-archive-smoke --dry-run Polylogue uses two schema-evolution regimes (see [Schema Versioning Model](internals.md#schema-versioning-model)): durable tiers use explicit additive migrations with a backup gate, while -derived tiers rebuild or blue-green replace from source evidence. +derived tiers rebuild or blue-green replace from source evidence, except for +declared, clone-validated SQL fast-forwards for non-semantic deltas; they +prove structural clone equivalence, not parser-content equivalence. `devtools lab policy schema-versioning` enforces that policy boundary: - It scans derived-tier storage modules for upgrade-shaped helpers (`build_vN_to_vM`, `_apply_version_upgrade_plan`, `upgrade_vN_to_vM`, `migrate_vN_*`, `ensure_schema_upgrades_vN`). +- It requires every index schema bump since the supported compatibility floor + to declare its delta class before it can ship. - It allows numbered SQL resources only under durable migration roots: `polylogue/storage/sqlite/migrations/{source,user}/`. -- If a derived upgrade helper or invalid migration resource is found, +- If a derived upgrade helper, undeclared index delta, or invalid migration resource is found, the lint fails. The lint runs as part of `devtools verify --lab`, not the fast default path: diff --git a/docs/topology-status.md b/docs/topology-status.md index edf28fce99..59dc23283b 100644 --- a/docs/topology-status.md +++ b/docs/topology-status.md @@ -28,12 +28,12 @@ Generated by `devtools render topology-status`. Reads `docs/plans/topology-targe ### Summary -- **Stable** (no move scoped): 778 +- **Stable** (no move scoped): 779 - **Kernel** (polylogue/ root): 10 - **Primitives** (storage-root): 18 - **TBD** (cell needs explicit assignment): 6 -- **Total declared**: 932 -- **Realized polylogue/**/*.py**: 932 files declared +- **Total declared**: 933 +- **Realized polylogue/**/*.py**: 933 files declared ### TBD cells (require explicit routing) diff --git a/polylogue/storage/sqlite/lifecycle.py b/polylogue/storage/sqlite/lifecycle.py new file mode 100644 index 0000000000..0a792f9de1 --- /dev/null +++ b/polylogue/storage/sqlite/lifecycle.py @@ -0,0 +1,235 @@ +"""Declared, disposable fast-forward plans for rebuildable SQLite tiers. + +Index-tier schema versions remain rebuildable derived state. A declaration in +this module is not a migration chain: it is a narrowly scoped, version-pair +proof that a clone may be brought forward without raw replay. Any transition +whose result depends on parser semantics is explicitly routed to reprocess or +full rebuild instead. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from typing import TypedDict + + +class DerivedDeltaClass(StrEnum): + """Meaning of one derived-tier schema delta.""" + + CONSTRAINT_ONLY = "constraint-only" + VIEW_ONLY = "view-only" + INDEX_ONLY = "index-only" + FTS_REINDEX = "fts-reindex" + SEMANTIC_REPARSE = "semantic-reparse" + + +class FastForwardOperationKind(StrEnum): + """Generated-SQL operation backed by canonical index DDL.""" + + REPLACE_TABLE = "replace-table" + REPLACE_VIEW = "replace-view" + CREATE_INDEX = "create-index" + REBUILD_FTS = "rebuild-fts" + + +@dataclass(frozen=True, slots=True) +class FastForwardOperation: + """One canonical object operation in a clone-only fast-forward plan.""" + + name: str + kind: FastForwardOperationKind + objects: tuple[tuple[str, str], ...] + + +@dataclass(frozen=True, slots=True) +class IndexDeltaDeclaration: + """The declared meaning and SQL surface of one index schema version.""" + + version: int + classes: tuple[DerivedDeltaClass, ...] + operations: tuple[FastForwardOperation, ...] = () + + @property + def requires_semantic_reparse(self) -> bool: + return DerivedDeltaClass.SEMANTIC_REPARSE in self.classes + + +@dataclass(frozen=True, slots=True) +class IndexFastForwardPlan: + """A disposable plan for one exact source/target version pair.""" + + source_version: int + target_version: int + declarations: tuple[IndexDeltaDeclaration, ...] + + @property + def requires_semantic_reparse(self) -> bool: + return any(declaration.requires_semantic_reparse for declaration in self.declarations) + + @property + def eligible_for_sql_fast_forward(self) -> bool: + return ( + bool(self.declarations) + and not self.requires_semantic_reparse + and all(declaration.classes for declaration in self.declarations) + and all(declaration.operations for declaration in self.declarations) + ) + + @property + def canonical_objects(self) -> tuple[tuple[str, str], ...]: + return tuple( + dict.fromkeys( + object_ref + for declaration in self.declarations + for operation in declaration.operations + for object_ref in operation.objects + ) + ) + + @property + def stage_names(self) -> tuple[str, ...]: + """Return the stable executor-stage order declared by this plan.""" + return tuple( + dict.fromkeys(operation.name for declaration in self.declarations for operation in declaration.operations) + ) + + +class IndexDeltaDeclarationReport(TypedDict): + """Static coverage result consumed by the schema-version policy lint.""" + + compatibility_floor: int + declared_versions: tuple[int, ...] + missing_versions: tuple[int, ...] + duplicate_versions: tuple[int, ...] + invalid_versions: tuple[int, ...] + ok: bool + + +# Deliberately bounded: v32 is the oldest observed live generation for which +# we retain a clone-proof plan. Earlier versions continue to rebuild from raw +# evidence rather than silently acquiring an unsupported upgrade path. +INDEX_FAST_FORWARD_COMPATIBILITY_FLOOR = 32 + +INDEX_DELTA_DECLARATIONS: tuple[IndexDeltaDeclaration, ...] = ( + IndexDeltaDeclaration( + version=33, + classes=(DerivedDeltaClass.CONSTRAINT_ONLY,), + operations=( + FastForwardOperation( + name="v33-insight-check", + kind=FastForwardOperationKind.REPLACE_TABLE, + objects=(("table", "insight_materialization"),), + ), + ), + ), + IndexDeltaDeclaration( + version=34, + classes=(DerivedDeltaClass.INDEX_ONLY, DerivedDeltaClass.VIEW_ONLY), + operations=( + FastForwardOperation( + name="v34-index-and-delegations", + kind=FastForwardOperationKind.CREATE_INDEX, + objects=(("index", "idx_web_constructs_message"),), + ), + FastForwardOperation( + name="v34-index-and-delegations", + kind=FastForwardOperationKind.REPLACE_VIEW, + objects=(("view", "delegations"),), + ), + ), + ), + IndexDeltaDeclaration( + version=35, + classes=(DerivedDeltaClass.FTS_REINDEX,), + operations=( + FastForwardOperation( + name="v35-messages-fts", + kind=FastForwardOperationKind.REBUILD_FTS, + objects=( + ("table", "messages_fts"), + ("trigger", "messages_fts_ai"), + ("trigger", "messages_fts_ad"), + ("trigger", "messages_fts_au"), + ), + ), + FastForwardOperation( + name="v35-insight-fts", + kind=FastForwardOperationKind.REBUILD_FTS, + objects=( + ("table", "session_work_events_fts"), + ("trigger", "session_work_events_fts_ai"), + ("trigger", "session_work_events_fts_ad"), + ("trigger", "session_work_events_fts_au"), + ("table", "threads_fts"), + ("trigger", "threads_fts_ai"), + ("trigger", "threads_fts_ad"), + ("trigger", "threads_fts_au"), + ), + ), + ), + ), + IndexDeltaDeclaration( + version=36, + classes=(DerivedDeltaClass.SEMANTIC_REPARSE,), + ), +) + + +def index_delta_declaration_report(current_version: int) -> IndexDeltaDeclarationReport: + """Return the static declaration coverage used by the schema-policy lint.""" + versions = tuple(declaration.version for declaration in INDEX_DELTA_DECLARATIONS) + expected = tuple(range(INDEX_FAST_FORWARD_COMPATIBILITY_FLOOR + 1, current_version + 1)) + missing = tuple(version for version in expected if version not in versions) + duplicates = tuple(sorted({version for version in versions if versions.count(version) > 1})) + invalid = tuple( + declaration.version + for declaration in INDEX_DELTA_DECLARATIONS + if declaration.version > current_version + or not declaration.classes + or (not declaration.requires_semantic_reparse and not declaration.operations) + ) + return { + "compatibility_floor": INDEX_FAST_FORWARD_COMPATIBILITY_FLOOR, + "declared_versions": versions, + "missing_versions": missing, + "duplicate_versions": duplicates, + "invalid_versions": invalid, + "ok": not missing and not duplicates and not invalid, + } + + +def index_fast_forward_plan(source_version: int, target_version: int) -> IndexFastForwardPlan | None: + """Build a contiguous SQL plan, or ``None`` when rebuild/reprocess is required.""" + if source_version < INDEX_FAST_FORWARD_COMPATIBILITY_FLOOR or source_version >= target_version: + return None + declarations = tuple( + sorted( + ( + declaration + for declaration in INDEX_DELTA_DECLARATIONS + if source_version < declaration.version <= target_version + ), + key=lambda declaration: declaration.version, + ) + ) + if tuple(declaration.version for declaration in declarations) != tuple( + range(source_version + 1, target_version + 1) + ): + return None + plan = IndexFastForwardPlan(source_version, target_version, declarations) + return plan if plan.eligible_for_sql_fast_forward else None + + +__all__ = [ + "DerivedDeltaClass", + "FastForwardOperation", + "FastForwardOperationKind", + "INDEX_DELTA_DECLARATIONS", + "INDEX_FAST_FORWARD_COMPATIBILITY_FLOOR", + "IndexDeltaDeclaration", + "IndexDeltaDeclarationReport", + "IndexFastForwardPlan", + "index_delta_declaration_report", + "index_fast_forward_plan", +] diff --git a/tests/unit/devtools/test_index_fast_forward.py b/tests/unit/devtools/test_index_fast_forward.py index 0b76450606..d3bace9a85 100644 --- a/tests/unit/devtools/test_index_fast_forward.py +++ b/tests/unit/devtools/test_index_fast_forward.py @@ -22,6 +22,7 @@ ) from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.lifecycle import index_fast_forward_plan _OLD_MESSAGES_FTS = """ CREATE VIRTUAL TABLE messages_fts USING fts5( @@ -177,6 +178,15 @@ def _seed_v32_fixture(path: Path) -> dict[str, int]: """, (message_id, session_id, digest), ) + block_id = str(conn.execute("SELECT block_id FROM blocks").fetchone()[0]) + conn.execute( + """ + INSERT INTO web_content_constructs ( + session_id, message_id, block_id, position, provider, construct_type + ) VALUES (?, ?, ?, 0, 'chatgpt', 'content_reference') + """, + (session_id, message_id, block_id), + ) conn.execute( """ INSERT INTO threads(thread_id, search_text) @@ -233,6 +243,12 @@ def test_fast_forward_v32_fixture_applies_exact_deltas_without_raw_reparse(tmp_p assert receipt["status"] == "clone_ready" assert receipt["raw_reparse"] is False + plan = index_fast_forward_plan(32, FAST_FORWARD_TO_VERSION) + assert plan is not None + assert receipt["declared_stages"] == list(plan.stage_names) + assert receipt["delta_classes"] == [ + delta_class.value for declaration in plan.declarations for delta_class in declaration.classes + ] validation = validate_clone( clone, expected_counts=expected_counts, @@ -253,6 +269,16 @@ def test_fast_forward_v32_fixture_applies_exact_deltas_without_raw_reparse(tmp_p assert conn.execute( "SELECT 1 FROM sqlite_master WHERE type='index' AND name='idx_web_constructs_message'" ).fetchone() + web_construct_message_id = str(conn.execute("SELECT message_id FROM web_content_constructs").fetchone()[0]) + query_plan = " ".join( + str(row[3]) + for row in conn.execute( + "EXPLAIN QUERY PLAN SELECT 1 FROM web_content_constructs WHERE message_id = ? LIMIT 1", + (web_construct_message_id,), + ) + ) + assert "INDEX idx_web_constructs_message" in query_plan + assert "SCAN web_content_constructs" not in query_plan delegation_columns = [row[1] for row in conn.execute("PRAGMA table_info(delegations)")] assert "instruction_tool_use_block_id" in delegation_columns for fts_table in ("messages_fts", "session_work_events_fts", "threads_fts"): @@ -438,3 +464,56 @@ def test_failed_restart_contract_automatically_restores_v32_symlink( assert active_link.resolve() == active_db.resolve() rolled_back = json.loads(receipt_path.read_text()) assert rolled_back["status"] == "rolled_back_after_failed_activation" + + +def test_interrupted_activation_leaves_rollbackable_receipt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A crash between symlink swap and receipt write must not strand the archive. + + Anti-vacuity: writing rollback metadata only after the swap means a + SIGKILL/power loss in that window leaves an active clone with a + clone_ready receipt that rollback refuses. The pre-swap "activating" + receipt with rollback_target keeps rollback executable. + """ + import devtools.index_fast_forward as iff + + archive_root = tmp_path / "archive" + active_dir = archive_root / ".index-generations" / "gen-v32" + active_dir.mkdir(parents=True) + active_db = active_dir / "index.db" + _seed_v32_fixture(active_db) + active_hash = _sha256(active_db) + active_link = archive_root / "index.db" + active_link.symlink_to(active_db) + receipt_path = archive_root / "recovery" / "v35.json" + + receipt = create_and_fast_forward_generation( + active_link, + receipt_path, + max_io_full_avg10=1000, + max_memory_full_avg10=1000, + batch_rows=1, + ) + clone = Path(str(receipt["clone_path"])) + + real_swap = iff._swap_active_symlink + + def _swap_then_crash(source_link: Path, target: Path, *, label: str) -> None: + real_swap(source_link, target, label=label) + raise RuntimeError("simulated crash after swap, before receipt write") + + monkeypatch.setattr(iff, "_swap_active_symlink", _swap_then_crash) + with pytest.raises(RuntimeError, match="simulated crash"): + activate_generation(receipt_path) + monkeypatch.setattr(iff, "_swap_active_symlink", real_swap) + + on_disk = json.loads(receipt_path.read_text()) + assert on_disk["status"] == "activating" + assert on_disk["rollback_target"] == str(active_db) + assert active_link.resolve() == clone.resolve() + + rollback_generation(receipt_path) + assert active_link.resolve() == active_db.resolve() + assert _sha256(active_db) == active_hash diff --git a/tests/unit/storage/test_index_fast_forward_lifecycle.py b/tests/unit/storage/test_index_fast_forward_lifecycle.py new file mode 100644 index 0000000000..20cf32bed0 --- /dev/null +++ b/tests/unit/storage/test_index_fast_forward_lifecycle.py @@ -0,0 +1,128 @@ +"""Production plan contracts for derived index fast-forwards.""" + +from __future__ import annotations + +import json + +import pytest + +from devtools import verify_schema_upgrade_lane as schema_policy +from polylogue.storage.sqlite import lifecycle +from polylogue.storage.sqlite.archive_tiers.index import INDEX_SCHEMA_VERSION +from polylogue.storage.sqlite.lifecycle import ( + DerivedDeltaClass, + FastForwardOperation, + FastForwardOperationKind, + IndexDeltaDeclaration, + IndexFastForwardPlan, + index_delta_declaration_report, + index_fast_forward_plan, +) + +_DEPLOYED_FAST_FORWARD_TARGET = 35 + + +def test_v32_to_current_plan_declares_ma2_as_an_index_only_delta() -> None: + """Exercise the production declaration used by clone fast-forward selection.""" + plan = index_fast_forward_plan(32, _DEPLOYED_FAST_FORWARD_TARGET) + + assert plan is not None + ma2 = next(declaration for declaration in plan.declarations if declaration.version == 34) + assert DerivedDeltaClass.INDEX_ONLY in ma2.classes + assert ("index", "idx_web_constructs_message") in plan.canonical_objects + assert any(operation.kind is FastForwardOperationKind.CREATE_INDEX for operation in ma2.operations) + + +def test_semantic_delta_routes_a_plan_away_from_sql_fast_forward(monkeypatch: pytest.MonkeyPatch) -> None: + """A parser-dependent delta cannot be mistaken for a clone-only SQL repair.""" + declaration = IndexDeltaDeclaration( + version=36, + classes=(DerivedDeltaClass.SEMANTIC_REPARSE,), + operations=( + FastForwardOperation( + name="v36-parser-shape", + kind=FastForwardOperationKind.REPLACE_TABLE, + objects=(("table", "sessions"),), + ), + ), + ) + plan = IndexFastForwardPlan(source_version=35, target_version=36, declarations=(declaration,)) + monkeypatch.setattr(lifecycle, "INDEX_DELTA_DECLARATIONS", (declaration,)) + + assert plan.requires_semantic_reparse is True + assert plan.eligible_for_sql_fast_forward is False + assert lifecycle.index_fast_forward_plan(35, 36) is None + + +def test_current_index_schema_has_a_complete_delta_declaration() -> None: + """Exercise the exact declaration report consumed by the schema policy lint.""" + report = index_delta_declaration_report(INDEX_SCHEMA_VERSION) + + assert report["ok"] is True + assert report["missing_versions"] == () + + +def test_plan_orders_declarations_before_validating_contiguity(monkeypatch: pytest.MonkeyPatch) -> None: + """Registry order cannot silently downgrade a complete clone-safe plan.""" + monkeypatch.setattr(lifecycle, "INDEX_DELTA_DECLARATIONS", tuple(reversed(lifecycle.INDEX_DELTA_DECLARATIONS))) + + plan = lifecycle.index_fast_forward_plan(32, _DEPLOYED_FAST_FORWARD_TARGET) + + assert plan is not None + assert tuple(declaration.version for declaration in plan.declarations) == (33, 34, 35) + + +def test_nonsemantic_delta_without_operations_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + """A declared class cannot advance an index version without executable SQL.""" + empty_declaration = IndexDeltaDeclaration( + version=37, + classes=(DerivedDeltaClass.INDEX_ONLY,), + ) + monkeypatch.setattr( + lifecycle, + "INDEX_DELTA_DECLARATIONS", + (*lifecycle.INDEX_DELTA_DECLARATIONS, empty_declaration), + ) + + report = lifecycle.index_delta_declaration_report(37) + + assert report["ok"] is False + assert report["invalid_versions"] == (37,) + assert lifecycle.index_fast_forward_plan(32, 37) is None + + +def test_delta_without_a_declared_class_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + """An operation cannot create a clone route without a delta classification.""" + unclassified_declaration = IndexDeltaDeclaration( + version=37, + classes=(), + operations=( + FastForwardOperation( + name="v36-unclassified-index", + kind=FastForwardOperationKind.CREATE_INDEX, + objects=(("index", "idx_web_constructs_message"),), + ), + ), + ) + monkeypatch.setattr( + lifecycle, + "INDEX_DELTA_DECLARATIONS", + (*lifecycle.INDEX_DELTA_DECLARATIONS, unclassified_declaration), + ) + + assert lifecycle.index_delta_declaration_report(37)["invalid_versions"] == (37,) + assert lifecycle.index_fast_forward_plan(32, 37) is None + + +def test_schema_policy_rejects_an_index_bump_without_a_delta_declaration( + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The real lab command must fail before an undeclared bump reaches CI.""" + monkeypatch.setattr(schema_policy, "INDEX_SCHEMA_VERSION", INDEX_SCHEMA_VERSION + 1) + + exit_code = schema_policy.main(["--json"]) + + payload = json.loads(capsys.readouterr().out) + assert exit_code == 1 + assert payload["index_delta_declarations"]["missing_versions"] == [INDEX_SCHEMA_VERSION + 1]