From f2ae88e2d918510cf1fcfff4afb5c69680d9eb27 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 10:38:48 +0200 Subject: [PATCH 1/3] fix(maintenance): gate reindex on durable schema currency Problem A rebuild accepted a source.db whose user_version lagged the package that would parse and rebuild from it. The live archive reached that state at source v28 while the installed package expected v24 and master expects v29. What changed The rebuild route now checks source.db and user.db before provenance, ownership, or candidate creation. A read-only --preflight option exposes the same structured diagnostic. index.db remains exempt because rebuilding it is the operation's purpose. Compatibility/migration Operators must migrate durable tiers and deploy the matching package before rebuilding index.db. The maintenance runbook records the ordered recovery sequence. Co-Authored-By: Codex --- docs/maintenance.md | 25 +++++++++ .../commands/maintenance/_rebuild_index.py | 22 ++++++++ polylogue/maintenance/rebuild_index.py | 56 +++++++++++++++++++ .../unit/cli/test_archive_maintenance_cli.py | 25 +++++++++ .../test_rebuild_index_ownership.py | 44 ++++++++++++++- 5 files changed, 170 insertions(+), 2 deletions(-) diff --git a/docs/maintenance.md b/docs/maintenance.md index 62f1375a44..c3b94e9bb7 100644 --- a/docs/maintenance.md +++ b/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 diff --git a/polylogue/cli/commands/maintenance/_rebuild_index.py b/polylogue/cli/commands/maintenance/_rebuild_index.py index ed2e128aae..8f7eedf4d3 100644 --- a/polylogue/cli/commands/maintenance/_rebuild_index.py +++ b/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, diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 2805347939..8bea29f817 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/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 diff --git a/tests/unit/cli/test_archive_maintenance_cli.py b/tests/unit/cli/test_archive_maintenance_cli.py index fd9222171d..4f4bff0761 100644 --- a/tests/unit/cli/test_archive_maintenance_cli.py +++ b/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: diff --git a/tests/unit/maintenance/test_rebuild_index_ownership.py b/tests/unit/maintenance/test_rebuild_index_ownership.py index ab5b43fcf8..35f3605fce 100644 --- a/tests/unit/maintenance/test_rebuild_index_ownership.py +++ b/tests/unit/maintenance/test_rebuild_index_ownership.py @@ -16,10 +16,15 @@ 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: @@ -27,6 +32,41 @@ def _init_empty_source(root: Path) -> None: 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 From 38d02ade5c6a0cace858eee09fb3fd7b6e526e70 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 11:26:54 +0200 Subject: [PATCH 2/3] fix(maintenance): harden durable schema currency gate Problem The initial currency gate checked only source and user tiers, leaving audit, daemon bulk transaction setup, an ownership-acquisition race, and the CLI empty-source path outside the durable schema boundary. The daemon surfaced a currency mismatch as an unstructured 500 response. What changed The gate now derives every durable migration tier from the canonical set, rechecks after archive ownership is acquired, and runs before daemon bulk bookkeeping. The CLI delegates empty sources to the guarded operation and rejects daemon preflight. Currency errors carry the original diagnostic to the daemon HTTP route with conflict semantics. Compatibility/migration Operators must bring source, user, and audit durable tiers to the deployed package versions before a rebuild. No migration or live archive mutation is performed by this change. Co-Authored-By: Codex --- docs/maintenance.md | 14 +-- .../commands/maintenance/_rebuild_index.py | 27 ++---- polylogue/daemon/bulk_rebuild.py | 3 + polylogue/daemon/http.py | 4 + polylogue/maintenance/rebuild_index.py | 13 ++- .../unit/cli/test_archive_maintenance_cli.py | 32 +++++++ .../daemon/test_bulk_rebuild_ownership.py | 25 +++++- .../unit/daemon/test_daemon_http_contracts.py | 43 +++++++++ .../test_rebuild_index_ownership.py | 89 ++++++++++++++++++- 9 files changed, 220 insertions(+), 30 deletions(-) diff --git a/docs/maintenance.md b/docs/maintenance.md index c3b94e9bb7..850d71b30f 100644 --- a/docs/maintenance.md +++ b/docs/maintenance.md @@ -41,18 +41,20 @@ 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. +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 either differs. The -execution route repeats this check before it consumes the schema-inference -receipt, acquires archive ownership, or creates a candidate generation. +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 diff --git a/polylogue/cli/commands/maintenance/_rebuild_index.py b/polylogue/cli/commands/maintenance/_rebuild_index.py index 8f7eedf4d3..1eed173200 100644 --- a/polylogue/cli/commands/maintenance/_rebuild_index.py +++ b/polylogue/cli/commands/maintenance/_rebuild_index.py @@ -429,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: @@ -484,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 @@ -555,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( @@ -573,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 diff --git a/polylogue/daemon/bulk_rebuild.py b/polylogue/daemon/bulk_rebuild.py index a5f0d7c5fc..91b9dfdc7a 100644 --- a/polylogue/daemon/bulk_rebuild.py +++ b/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. diff --git a/polylogue/daemon/http.py b/polylogue/daemon/http.py index b7c9b499bb..449f6cfd2f 100644 --- a/polylogue/daemon/http.py +++ b/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, diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 8bea29f817..503a8f84ba 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/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,9 +68,11 @@ class RebuildDerivedStateProvenanceError(RebuildProvenanceError): """A derived-state stage was blocked by a failed provenance recheck.""" -class RebuildSchemaCurrencyError(RuntimeError): +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"] @@ -85,14 +89,14 @@ 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 + 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.archive_tiers.types import ArchiveTier + from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS checks: list[dict[str, object]] = [] - for tier in (ArchiveTier.SOURCE, ArchiveTier.USER): + for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value): probe = probe_archive_tier(tier, root / f"{tier.value}.db") checks.append( { @@ -1140,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 diff --git a/tests/unit/cli/test_archive_maintenance_cli.py b/tests/unit/cli/test_archive_maintenance_cli.py index 4f4bff0761..8e9c210f9a 100644 --- a/tests/unit/cli/test_archive_maintenance_cli.py +++ b/tests/unit/cli/test_archive_maintenance_cli.py @@ -2015,6 +2015,38 @@ def test_rebuild_index_preflight_reports_durable_schema_currency( 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: diff --git a/tests/unit/daemon/test_bulk_rebuild_ownership.py b/tests/unit/daemon/test_bulk_rebuild_ownership.py index 1868b7eb16..792084701a 100644 --- a/tests/unit/daemon/test_bulk_rebuild_ownership.py +++ b/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: diff --git a/tests/unit/daemon/test_daemon_http_contracts.py b/tests/unit/daemon/test_daemon_http_contracts.py index ea6e098c10..84a14eae43 100644 --- a/tests/unit/daemon/test_daemon_http_contracts.py +++ b/tests/unit/daemon/test_daemon_http_contracts.py @@ -40,6 +40,7 @@ from http import HTTPStatus from io import BytesIO from pathlib import Path +from types import SimpleNamespace from typing import TYPE_CHECKING, cast from unittest.mock import MagicMock @@ -193,6 +194,48 @@ def _archive_state_hash(archive_root: Path) -> str: return h.hexdigest() +def test_rebuild_index_schema_currency_conflict_preserves_preflight_diagnostic( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The actual maintenance route returns the shared diagnostic, not a 500.""" + 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 + + root = tmp_path / "archive" + root.mkdir() + for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value): + initialize_archive_database(root / f"{tier.value}.db", tier) + 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}") + monkeypatch.setattr("polylogue.paths.archive_root", lambda: root) + + handler = _make_handler("POST", "/api/maintenance/rebuild-index", body=b"{}") + handler.server.write_bridge = SimpleNamespace( # type: ignore[assignment] + run_sync_with_timeout=lambda _actor, _timeout, operation, request: operation(request) + ) + send_error, send_json = _capture_responses(handler) + + handler._handle_rebuild_index() + + send_error.assert_not_called() + status, payload = send_json.call_args.args + assert status == HTTPStatus.CONFLICT + assert payload["kind"] == "rebuild-schema-currency" + assert payload["status"] == "blocked" + assert payload["blocking_tiers"] == [ + { + "tier": "source", + "path": str(root / "source.db"), + "actual_user_version": source_probe.expected_user_version + 1, + "expected_user_version": source_probe.expected_user_version, + "status": "mismatch", + } + ] + + def test_cli_query_post_forwards_root_request_to_daemon_compiler() -> None: """The UDS-only envelope carries raw root flags, not a client-built SQL query.""" diff --git a/tests/unit/maintenance/test_rebuild_index_ownership.py b/tests/unit/maintenance/test_rebuild_index_ownership.py index 35f3605fce..afa31a2bd0 100644 --- a/tests/unit/maintenance/test_rebuild_index_ownership.py +++ b/tests/unit/maintenance/test_rebuild_index_ownership.py @@ -13,6 +13,7 @@ import sqlite3 from pathlib import Path +from typing import cast import pytest @@ -22,14 +23,17 @@ rebuild_index_from_source_sync, ) 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_active_archive_root, initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS 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) + for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value): + initialize_archive_database(root / f"{tier.value}.db", tier) def test_rebuild_rejects_source_schema_behind_runtime_before_candidate_creation(tmp_path: Path) -> None: @@ -67,6 +71,89 @@ def test_rebuild_rejects_source_schema_behind_runtime_before_candidate_creation( assert not (root / ".index-rebuild-transactions").exists() +def test_rebuild_rejects_source_schema_ahead_of_runtime_before_candidate_creation(tmp_path: Path) -> None: + """A newer source tier is as unsafe to rebuild as an older one.""" + 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: + rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + + assert exc_info.value.diagnostic["blocking_tiers"] == [ + { + "tier": "source", + "path": str(root / "source.db"), + "actual_user_version": source_probe.expected_user_version + 1, + "expected_user_version": source_probe.expected_user_version, + "status": "mismatch", + } + ] + assert not (root / ".index-generations").exists() + + +@pytest.mark.parametrize("mode", ["missing", "mismatched"]) +def test_rebuild_rejects_missing_or_mismatched_audit_tier_before_candidate_creation(tmp_path: Path, mode: str) -> None: + """Every canonical durable tier, including audit, must be package-current.""" + root = tmp_path / "archive" + _init_empty_source(root) + audit_path = root / "audit.db" + expected = probe_archive_tier(ArchiveTier.AUDIT, audit_path).expected_user_version + if mode == "missing": + audit_path.unlink() + actual: int | None = None + status = "missing" + else: + with sqlite3.connect(audit_path) as conn: + conn.execute(f"PRAGMA user_version = {expected + 1}") + actual = expected + 1 + status = "mismatch" + + with pytest.raises(RebuildSchemaCurrencyError) as exc_info: + rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + + assert exc_info.value.diagnostic["blocking_tiers"] == [ + { + "tier": "audit", + "path": str(audit_path), + "actual_user_version": actual, + "expected_user_version": expected, + "status": status, + } + ] + assert not (root / ".index-generations").exists() + + +def test_rebuild_rechecks_schema_currency_after_acquiring_archive_ownership( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Schema drift after the early guard cannot reach the candidate path.""" + root = tmp_path / "archive" + _init_empty_source(root) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") + source_probe = probe_archive_tier(ArchiveTier.SOURCE, root / "source.db") + original_acquire = OwnedArchiveLocation.acquire + + def acquire_then_advance_schema(location: ArchiveLocation) -> OwnedArchiveLocation: + owned = original_acquire(location) + with sqlite3.connect(root / "source.db") as conn: + conn.execute(f"PRAGMA user_version = {source_probe.expected_user_version + 1}") + return owned + + monkeypatch.setattr("polylogue.maintenance.rebuild_index.OwnedArchiveLocation.acquire", acquire_then_advance_schema) + + with pytest.raises(RebuildSchemaCurrencyError, match="schema currency") as exc_info: + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) + + 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() + + 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 From 1ac4749772bb9207c356ab9a32e6fa14c9db194a Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 11:44:53 +0200 Subject: [PATCH 3/3] docs(maintenance): name all durable rebuild tiers Problem The rebuild schema-currency gate checks source, user, and audit tiers, while operator text still described only source and user. What changed Align the maintenance recovery instructions and rebuild preflight help with the canonical durable-tier set. Compatibility Documentation and CLI help only. The maintenance behavior is unchanged. Co-Authored-By: Codex --- docs/maintenance.md | 9 +++++---- polylogue/cli/commands/maintenance/_rebuild_index.py | 2 +- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/docs/maintenance.md b/docs/maintenance.md index 850d71b30f..c91fa8789a 100644 --- a/docs/maintenance.md +++ b/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. @@ -58,8 +58,9 @@ 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` and `migrate-tier user` when the target package requires -them, then deploy that exact package. Run the preflight above and require a +`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 diff --git a/polylogue/cli/commands/maintenance/_rebuild_index.py b/polylogue/cli/commands/maintenance/_rebuild_index.py index 1eed173200..2bf6d0e428 100644 --- a/polylogue/cli/commands/maintenance/_rebuild_index.py +++ b/polylogue/cli/commands/maintenance/_rebuild_index.py @@ -391,7 +391,7 @@ def _rebuild_index_selection_plan( @click.option( "--preflight", is_flag=True, - help="Read-only: report whether durable source/user tiers match this package before rebuilding index.db.", + help="Read-only: report whether durable tiers match this package before rebuilding index.db.", ) def rebuild_index_command( only_missing: bool,