fix(maintenance): gate reindex on durable schema currency (#3851)

docs/maintenance.md

@@ -38,6 +38,31 @@ 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 deliberately
+checks `source.db` and `user.db` only: `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 either differs. The
+execution route repeats this check before it consumes the schema-inference
+receipt, acquires archive ownership, or creates a 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` and `migrate-tier user` 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 source/user 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.
 
@@ -441,6 +447,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,

polylogue/maintenance/rebuild_index.py

@@ -66,6 +66,61 @@ class RebuildDerivedStateProvenanceError(RebuildProvenanceError):
     """A derived-state stage was blocked by a failed provenance recheck."""
 
 
+class RebuildSchemaCurrencyError(RuntimeError):
+    """The durable tiers do not match the package that would rebuild them."""
+
+    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 source/user 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.archive_tiers.types import ArchiveTier
+
+    checks: list[dict[str, object]] = []
+    for tier in (ArchiveTier.SOURCE, ArchiveTier.USER):
+        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 +1102,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

tests/unit/cli/test_archive_maintenance_cli.py

@@ -1990,6 +1990,31 @@ 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_daemon_path_posts_the_real_selection_request(
     cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/unit/maintenance/test_rebuild_index_ownership.py

@@ -16,17 +16,57 @@
 
 import pytest
 
-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,
+)
 from polylogue.storage.archive_identity import ArchiveLocation, ArchiveOwnershipError, OwnedArchiveLocation
-from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
+from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_database
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
+from tests.infra.rebuild_receipt import write_valid_rebuild_receipt
 
 
 def _init_empty_source(root: Path) -> None:
     root.mkdir(parents=True, exist_ok=True)
     initialize_archive_database(root / "source.db", ArchiveTier.SOURCE)
 
 
+def test_rebuild_rejects_source_schema_behind_runtime_before_candidate_creation(tmp_path: Path) -> None:
+    """A real v28 source tier must not reach the v29 rebuild package.
+
+    The test builds ordinary file-backed archive tiers, removes exactly v29's
+    additive objects, and supplies a valid rebuild receipt. The production
+    rebuild route used to accept this archive and return ``empty-source``.
+    """
+    root = tmp_path / "archive"
+    initialize_active_archive_root(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")
+    receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json")
+
+    with pytest.raises(RebuildSchemaCurrencyError) as exc_info:
+        rebuild_index_from_source_sync(
+            RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path)
+        )
+
+    diagnostic = exc_info.value.diagnostic
+    assert diagnostic["status"] == "blocked"
+    assert diagnostic["blocking_tiers"] == [
+        {
+            "tier": "source",
+            "path": str(root / "source.db"),
+            "actual_user_version": 28,
+            "expected_user_version": 29,
+            "status": "mismatch",
+        }
+    ]
+    assert not (root / ".index-generations").exists()
+    assert not (root / ".index-rebuild-transactions").exists()
+
+
 def test_rebuild_refuses_when_archive_location_already_owned(tmp_path: Path) -> None:
     """A concurrent holder of the archive-location ownership lock must block
     an offline rebuild before any generation directory or SQLite tier is