fix(maintenance): harden durable schema currency gate (#3856)

docs/maintenance.md

@@ -9,8 +9,8 @@ common operational incidents.
 ## Applying a durable schema change train
 
 Durable schema changes are an offline release operation. Before applying a
-`source.db` or `user.db` migration above its adoption floor, confirm that the
-release contains the matching `migrations/{source,user}/NNN.train.json`
+`source.db`, `user.db`, or `audit.db` migration above its adoption floor,
+confirm that the release contains the matching `migrations/{source,user,audit}/NNN.train.json`
 sidecar. The sidecar reserves the exact slot and SQL hash and records the
 runtime and restart evidence needed for the change.
 
@@ -38,6 +38,34 @@ Restart health and runtime-consumer convergence are the final lifecycle proof
 and are recorded by the durable train lifecycle API, not inferred from this
 command's migration result alone.
 
+### Rebuild deployment-currency preflight
+
+Before a managed `rebuild-index`, confirm that the package selected for the
+operation owns the live durable schemas. The read-only preflight checks every
+canonical durable migration tier: `source.db`, `user.db`, and `audit.db`.
+`index.db` may be behind because the rebuild is the supported way to replace
+that derived tier.
+
+```bash
+polylogue ops maintenance rebuild-index --preflight --output-format json
+```
+
+It emits `rebuild-schema-currency` JSON with each durable tier's observed and
+package-expected `user_version`, and exits nonzero when a durable tier differs.
+The execution route checks it before consuming the schema-inference receipt,
+repeats it after archive ownership acquisition, and rejects daemon bulk
+transaction creation before any bookkeeping or candidate generation.
+
+For a safe deployment recovery, first choose the exact target package commit.
+With the daemon stopped, create a fresh verified full-evidence backup, run
+`migrate-tier source`, `migrate-tier user`, and `migrate-tier audit` when the
+target package requires them, then deploy that exact package. Run the preflight
+above and require a
+ready result before invoking `polylogue ops maintenance rebuild-index`; use
+that blue-green command rather than `ops reset --index` for an active managed
+generation. Restart the daemon only after the rebuilt generation is promoted
+and the post-deploy status shows no durable-tier mismatch.
+
 For the conceptual model behind derived insights and the FTS / blob
 substrate, see [architecture.md](architecture.md) and
 [internals.md](internals.md). For daemon ownership of the inline

polylogue/cli/commands/maintenance/_rebuild_index.py

@@ -388,6 +388,11 @@ def _rebuild_index_selection_plan(
         "single-writer path; unsupported with --daemon."
     ),
 )
+@click.option(
+    "--preflight",
+    is_flag=True,
+    help="Read-only: report whether durable tiers match this package before rebuilding index.db.",
+)
 def rebuild_index_command(
     only_missing: bool,
     raw_ids: tuple[str, ...],
@@ -404,6 +409,7 @@ def rebuild_index_command(
     use_daemon: bool,
     daemon_url: str,
     shard_count: int,
+    preflight: bool,
 ) -> None:
     """Inspect or execute an authority-safe source-to-index rebuild.
 
@@ -423,6 +429,8 @@ def rebuild_index_command(
         raise click.BadParameter("plan limit must be positive", param_hint="--plan-limit")
     if use_daemon and plan_only:
         raise click.UsageError("--daemon executes a rebuild; --plan is always a local read-only preview")
+    if use_daemon and preflight:
+        raise click.UsageError("--preflight cannot be combined with --daemon")
     if shard_count <= 0:
         raise click.BadParameter("shard count must be positive", param_hint="--shard-count")
     if use_daemon and shard_count > 1:
@@ -441,6 +449,22 @@ def rebuild_index_command(
         raise click.UsageError("resumed rebuild budgets are durable; omit pass budget options with --operation-id")
 
     root = archive_root()
+    if preflight:
+        from polylogue.maintenance.rebuild_index import rebuild_schema_currency_preflight
+
+        payload = rebuild_schema_currency_preflight(root)
+        if output_format == "json":
+            click.echo(json.dumps(payload, indent=2, sort_keys=True))
+        else:
+            click.echo(f"Archive root: {root}")
+            for tier in cast(list[dict[str, object]], payload["tiers"]):
+                click.echo(
+                    f"{tier['tier']}.db: {tier['actual_user_version']} (package expects "
+                    f"{tier['expected_user_version']}; {tier['status']})"
+                )
+        if payload["status"] != "ready":
+            raise click.ClickException("rebuild schema currency preflight failed; migrate or deploy before rebuilding")
+        return
     if use_daemon:
         payload = _run_daemon_rebuild(
             daemon_url,
@@ -462,23 +486,8 @@ def rebuild_index_command(
         click.echo(f"Replayed:     {int(cast(Any, payload['replayed_logical_source_count'])):,} logical source(s)")
         click.echo(f"Quarantined:  {int(cast(Any, payload['quarantined_raw_count'])):,} raw row(s)")
         return
-    raw_count = _count_source_raw_sessions(root)
-    if raw_count == 0:
-        payload = {
-            "archive_root": str(root),
-            "raw_session_count": 0,
-            "selected_raw_count": 0,
-            "skipped_by_blob_limit_count": 0,
-            "status": "empty-source",
-            "materialized": False,
-        }
-        if output_format == "json":
-            click.echo(json.dumps(payload, indent=2, sort_keys=True))
-        else:
-            click.echo(f"Archive root: {root}")
-            click.echo("No source.db raw_sessions rows found.")
-        return
     if plan_only:
+        raw_count = _count_source_raw_sessions(root)
         selected_raw_ids = (
             list(dict.fromkeys(raw_ids))
             if raw_ids
@@ -533,7 +542,11 @@ def rebuild_index_command(
                         f"blob={int(group['blob_bytes']):,} source={group['source_path']}"
                     )
         return
-    from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync
+    from polylogue.maintenance.rebuild_index import (
+        RebuildIndexRequest,
+        RebuildSchemaCurrencyError,
+        rebuild_index_from_source_sync,
+    )
 
     try:
         receipt = rebuild_index_from_source_sync(
@@ -551,7 +564,7 @@ def rebuild_index_command(
                 shard_count=shard_count,
             )
         )
-    except (RuntimeError, ValueError) as exc:
+    except (RebuildSchemaCurrencyError, RuntimeError, ValueError) as exc:
         raise click.ClickException(str(exc)) from exc
     payload = receipt.to_dict()
     result = payload

polylogue/daemon/bulk_rebuild.py

@@ -182,6 +182,9 @@ def resolve_or_start_daemon_bulk_rebuild_transaction(
     generation directory, so both must fail closed against a foreign/rotated
     archive location before touching disk, not just the eventual write pass.
     """
+    from polylogue.maintenance.rebuild_index import require_rebuild_schema_currency
+
+    require_rebuild_schema_currency(root)
     _validate_rebuild_provenance_receipt(root, schema_inference_receipt_path)
     # This must precede transaction resolution, because retiring a terminal
     # transaction and creating its replacement also creates generation state.

polylogue/daemon/http.py

@@ -1085,6 +1085,10 @@ def wrapper(self: DaemonAPIHandler, *args: object, **kwargs: object) -> None:
                 if 100 <= exc.http_status_code <= 599
                 else HTTPStatus.INTERNAL_SERVER_ERROR
             )
+            diagnostic = getattr(exc, "diagnostic", None)
+            if isinstance(diagnostic, dict):
+                self._send_json(status, diagnostic)
+                return
             field = getattr(exc, "field", None)
             self._send_json(
                 status,

polylogue/maintenance/rebuild_index.py

@@ -16,10 +16,12 @@
 import time
 from dataclasses import asdict, dataclass, field
 from hashlib import sha256
+from http import HTTPStatus
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 
 from polylogue.config import Config
+from polylogue.core.errors import PolylogueError
 from polylogue.logging import get_logger
 from polylogue.maintenance.offline_guard import offline_maintenance_block_reason
 from polylogue.paths import render_root
@@ -66,6 +68,63 @@ class RebuildDerivedStateProvenanceError(RebuildProvenanceError):
     """A derived-state stage was blocked by a failed provenance recheck."""
 
 
+class RebuildSchemaCurrencyError(PolylogueError):
+    """The durable tiers do not match the package that would rebuild them."""
+
+    http_status_code = HTTPStatus.CONFLICT
+
+    def __init__(self, diagnostic: dict[str, object]) -> None:
+        self.diagnostic = diagnostic
+        blocked = diagnostic["blocking_tiers"]
+        assert isinstance(blocked, list)
+        detail = ", ".join(
+            f"{item['tier']}.db:{item['actual_user_version']}!={item['expected_user_version']}"
+            for item in blocked
+            if isinstance(item, dict)
+        )
+        super().__init__(f"rebuild schema currency preflight failed: {detail}")
+
+
+def rebuild_schema_currency_preflight(root: Path) -> dict[str, object]:
+    """Report whether durable source evidence matches this runtime package.
+
+    ``index.db`` is intentionally absent: rebuilding it is the operation's
+    purpose, while a durable-tier mismatch means this package can interpret or
+    write durable evidence using a schema it does not own.
+    """
+    from polylogue.storage.archive_readiness import probe_archive_tier
+    from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS
+
+    checks: list[dict[str, object]] = []
+    for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value):
+        probe = probe_archive_tier(tier, root / f"{tier.value}.db")
+        checks.append(
+            {
+                "tier": tier.value,
+                "path": probe.path,
+                "actual_user_version": probe.user_version,
+                "expected_user_version": probe.expected_user_version,
+                "status": probe.version_status,
+            }
+        )
+    blocking = [check for check in checks if check["status"] != "ok"]
+    return {
+        "kind": "rebuild-schema-currency",
+        "archive_root": str(root),
+        "status": "ready" if not blocking else "blocked",
+        "tiers": checks,
+        "blocking_tiers": blocking,
+    }
+
+
+def require_rebuild_schema_currency(root: Path) -> dict[str, object]:
+    """Reject a rebuild before it consumes evidence or creates a generation."""
+    diagnostic = rebuild_schema_currency_preflight(root)
+    if diagnostic["status"] != "ready":
+        raise RebuildSchemaCurrencyError(diagnostic)
+    return diagnostic
+
+
 @dataclass(frozen=True, slots=True)
 class RebuildProvenanceContext:
     """Validated evidence shared by every mutation in one rebuild pass.
@@ -1047,6 +1106,7 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde
     log_mapped_bytes_budget_check(logger, check_mapped_bytes_budget_against_cgroup_limit())
     validate_rebuild_index_request(request)
     root = request.archive_root
+    require_rebuild_schema_currency(root)
     consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path)
     location = ArchiveLocation.resolve(root)
     # The joined raw-frontier projection is rooted at the co-located active
@@ -1084,6 +1144,7 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde
     owned = OwnedArchiveLocation.acquire(location)
     try:
         assert_owns_archive_location(owned, location)
+        require_rebuild_schema_currency(root)
         consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path)
         # The lease is itself lifecycle state guarded by the provenance gate.
         # Revalidate again under the lease immediately before the owned body

tests/unit/cli/test_archive_maintenance_cli.py

@@ -1990,6 +1990,63 @@ def test_rebuild_index_force_write_option_is_retired(cli_runner: CliRunner) -> N
     assert "--force-write" in result.output
 
 
+def test_rebuild_index_preflight_reports_durable_schema_currency(
+    cli_workspace: dict[str, Path], cli_runner: CliRunner
+) -> None:
+    root = cli_workspace["archive_root"]
+    with sqlite3.connect(root / "source.db") as conn:
+        conn.execute("DROP INDEX idx_raw_failure_disposition_receipts_disposed_at")
+        conn.execute("DROP TABLE raw_failure_disposition_receipts")
+        conn.execute("PRAGMA user_version = 28")
+
+    result = cli_runner.invoke(
+        cli,
+        ["--plain", "ops", "maintenance", "rebuild-index", "--preflight", "--output-format", "json"],
+        catch_exceptions=False,
+    )
+
+    assert result.exit_code == 1
+    payload = json.loads(result.stdout)
+    assert payload["kind"] == "rebuild-schema-currency"
+    assert payload["status"] == "blocked"
+    assert payload["blocking_tiers"][0]["tier"] == "source"
+    assert payload["blocking_tiers"][0]["actual_user_version"] == 28
+    assert payload["blocking_tiers"][0]["expected_user_version"] == 29
+    assert "migrate or deploy before rebuilding" in result.stderr
+
+
+def test_rebuild_index_empty_source_still_runs_the_schema_currency_guard(
+    cli_workspace: dict[str, Path], cli_runner: CliRunner
+) -> None:
+    root = cli_workspace["archive_root"]
+    with sqlite3.connect(root / "audit.db") as conn:
+        expected = int(conn.execute("PRAGMA user_version").fetchone()[0])
+        conn.execute(f"PRAGMA user_version = {expected + 1}")
+
+    result = cli_runner.invoke(
+        cli,
+        ["--plain", "ops", "maintenance", "rebuild-index", "--output-format", "json"],
+        catch_exceptions=False,
+    )
+
+    assert result.exit_code == 1
+    assert "audit.db" in result.stderr
+    assert not (root / ".index-generations").exists()
+
+
+def test_rebuild_index_rejects_daemon_schema_preflight_combination(
+    cli_workspace: dict[str, Path], cli_runner: CliRunner
+) -> None:
+    result = cli_runner.invoke(
+        cli,
+        ["--plain", "ops", "maintenance", "rebuild-index", "--preflight", "--daemon"],
+        catch_exceptions=False,
+    )
+
+    assert result.exit_code == 2
+    assert "--preflight cannot be combined with --daemon" in result.output
+
+
 def test_rebuild_index_daemon_path_posts_the_real_selection_request(
     cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/unit/daemon/test_bulk_rebuild_ownership.py

@@ -15,19 +15,42 @@
 
 from __future__ import annotations
 
+import sqlite3
 from pathlib import Path
+from typing import cast
 
 import pytest
 
 from polylogue.daemon.bulk_rebuild import resolve_or_start_daemon_bulk_rebuild_transaction
+from polylogue.maintenance.rebuild_index import RebuildSchemaCurrencyError
 from polylogue.storage.archive_identity import ArchiveLocation, ArchiveOwnershipError, OwnedArchiveLocation
+from polylogue.storage.archive_readiness import probe_archive_tier
 from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
+from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS
 
 
 def _init_empty_source(root: Path) -> None:
     root.mkdir(parents=True, exist_ok=True)
-    initialize_archive_database(root / "source.db", ArchiveTier.SOURCE)
+    for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value):
+        initialize_archive_database(root / f"{tier.value}.db", tier)
+
+
+def test_daemon_bulk_rebuild_rejects_schema_mismatch_before_transaction_bookkeeping(tmp_path: Path) -> None:
+    """The daemon's direct transaction entry cannot bypass the shared gate."""
+    root = tmp_path / "archive"
+    _init_empty_source(root)
+    source_probe = probe_archive_tier(ArchiveTier.SOURCE, root / "source.db")
+    with sqlite3.connect(root / "source.db") as conn:
+        conn.execute(f"PRAGMA user_version = {source_probe.expected_user_version + 1}")
+
+    with pytest.raises(RebuildSchemaCurrencyError) as exc_info:
+        resolve_or_start_daemon_bulk_rebuild_transaction(root)
+
+    blocking_tiers = cast(list[dict[str, object]], exc_info.value.diagnostic["blocking_tiers"])
+    assert blocking_tiers[0]["tier"] == "source"
+    assert not (root / ".index-generations").exists()
+    assert not (root / ".index-rebuild-transactions").exists()
 
 
 def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned(tmp_path: Path) -> None: