From 84a5fae7b522519e86fb2ae76338ca624abc0df6 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 22:17:12 +0200 Subject: [PATCH 01/13] fix: reconcile offline rebuild provenance recovery Problem: successful external-ground-truth rehashes refreshed the validator result but not the in-pass token, and offline operation-id admission could resume a transaction whose generation was already active after interrupted promotion attestation. What changed: retain the refreshed inventory token in RebuildProvenanceContext, extract the existing active-generation reconciliation into the maintenance rebuild engine, and use it from both offline and daemon transaction resolution. Add real-route coverage for inventory scan reuse and terminal offline recovery. Compatibility/migration: receipt schemas, typed reason codes, operation identifiers, and production write behavior are unchanged. --- polylogue/daemon/bulk_rebuild.py | 41 ++--------- polylogue/maintenance/rebuild_index.py | 42 ++++++++++++ .../test_rebuild_index_provenance_gate.py | 68 +++++++++++++++++++ 3 files changed, 115 insertions(+), 36 deletions(-) diff --git a/polylogue/daemon/bulk_rebuild.py b/polylogue/daemon/bulk_rebuild.py index 85b1aa740d..276b51f900 100644 --- a/polylogue/daemon/bulk_rebuild.py +++ b/polylogue/daemon/bulk_rebuild.py @@ -46,6 +46,10 @@ from polylogue.config import Config from polylogue.logging import get_logger from polylogue.maintenance.archive_verification import read_raw_failure_lifecycle +from polylogue.maintenance.rebuild_index import ( + _REBUILD_TERMINAL_NOT_RESUMABLE, + _reconcile_active_generation_transaction, +) from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation, assert_owns_archive_location from polylogue.storage.index_generation import ( IndexGenerationStore, @@ -149,7 +153,7 @@ def _raise_daemon_cleanup_failures(cleanup_errors: list[BaseException], *, label #: ``stale`` (source evidence changed mid-build), and ``failed`` (a pass #: raised; automagic doctrine retries rather than waiting on an operator to #: intervene). -_TERMINAL_NOT_RESUMABLE = frozenset({"promoted", "promoted-attestation-failed", "stale", "failed"}) +_TERMINAL_NOT_RESUMABLE = _REBUILD_TERMINAL_NOT_RESUMABLE def _preflight_raw_failure_lifecycle(root: Path) -> None: @@ -164,41 +168,6 @@ def _preflight_raw_failure_lifecycle(root: Path) -> None: raise RuntimeError(f"daemon bulk-rebuild raw failure lifecycle preflight failed: {reason}") -def _reconcile_active_generation_transaction( - store: IndexGenerationStore, transaction: IndexRebuildTransaction -) -> IndexRebuildTransaction: - """Turn a post-pointer-write transaction into terminal state on restart. - - Promotion changes the active pointer before the transaction attestation is - durable. If both the normal and recovery checkpoints fail, the persisted - transaction can still look resumable even though its generation is active. - The next resolver pass must record that observed fact before returning it - to the rebuild loop, otherwise the daemon retries an already-active - generation forever. - """ - - if transaction.status in _TERMINAL_NOT_RESUMABLE: - return transaction - try: - generation = store.load(transaction.generation_id) - active_path = store.active_pointer.resolve(strict=True) - generation_path = Path(generation.index_path).resolve(strict=True) - except (FileNotFoundError, OSError, ValueError): - return transaction - if generation.state != "active" or active_path != generation_path: - return transaction - return store.checkpoint_transaction( - transaction, - status="promoted-attestation-failed", - error="reconciled active generation after interrupted promotion attestation", - post_promotion_attestation={ - "status": "reconciled-after-restart", - "generation_id": generation.generation_id, - "generation_state": generation.state, - }, - ) - - def resolve_or_start_daemon_bulk_rebuild_transaction( root: Path, *, schema_inference_receipt_path: Path | None = None ) -> IndexRebuildTransaction: diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index f82673d9f5..26eebd4221 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -159,6 +159,9 @@ def validate(self, *, verify_blob_integrity: bool = False, refresh_blob_integrit verify_blob_integrity=verify_blob_integrity, verified_blob_integrity_snapshot=cached_snapshot, ) + refreshed_inventory_token = validated.get("external_ground_truth_inventory_token") + if isinstance(refreshed_inventory_token, dict): + self.external_inventory_token = refreshed_inventory_token if verify_blob_integrity: refreshed_snapshot = validated.pop("_verified_blob_integrity_snapshot", None) if isinstance(refreshed_snapshot, dict): @@ -204,6 +207,44 @@ def _validate_rebuild_provenance_receipt( raise RebuildProvenanceError(f"rebuild schema-inference preflight gate failed: {exc}") from exc +_REBUILD_TERMINAL_NOT_RESUMABLE = frozenset({"promoted", "promoted-attestation-failed", "stale", "failed"}) + + +def _reconcile_active_generation_transaction( + store: IndexGenerationStore, transaction: IndexRebuildTransaction +) -> IndexRebuildTransaction: + """Turn a post-pointer-write transaction into terminal state on restart. + + Promotion changes the active pointer before the transaction attestation is + durable. If both the normal and recovery checkpoints fail, the persisted + transaction can still look resumable even though its generation is active. + The next resolver pass must record that observed fact before returning it + to the rebuild loop, otherwise the caller retries an already-active + generation forever. + """ + + if transaction.status in _REBUILD_TERMINAL_NOT_RESUMABLE: + return transaction + try: + generation = store.load(transaction.generation_id) + active_path = store.active_pointer.resolve(strict=True) + generation_path = Path(generation.index_path).resolve(strict=True) + except (FileNotFoundError, OSError, ValueError): + return transaction + if generation.state != "active" or active_path != generation_path: + return transaction + return store.checkpoint_transaction( + transaction, + status="promoted-attestation-failed", + error="reconciled active generation after interrupted promotion attestation", + post_promotion_attestation={ + "status": "reconciled-after-restart", + "generation_id": generation.generation_id, + "generation_state": generation.state, + }, + ) + + def _mark_rebuild_transaction_stale_after_provenance_failure( root: Path, operation_id: str | None, error: RebuildProvenanceError ) -> None: @@ -1410,6 +1451,7 @@ async def _rebuild_index_from_source_owned( if resumable_full_source: if request.operation_id is not None: transaction = generation_store.load_transaction(request.operation_id) + transaction = _reconcile_active_generation_transaction(generation_store, transaction) else: transaction = _create_rebuild_transaction_after_receipt_validation( generation_store, diff --git a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py index 1a1ead2f7e..7a76ae13a5 100644 --- a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py +++ b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py @@ -201,6 +201,60 @@ def test_valid_receipt_allows_real_candidate_acceptance_and_promotion(tmp_path: assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().exists() +def test_rebuild_context_reuses_refreshed_inventory_token_after_detector_change( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A successful metadata-only rehash updates the pass token for later checkpoints. + + Anti-vacuity: this runs the real offline rebuild entry point and validator. + Touching the external corpus after the first context validation forces one + successful full inventory refresh. Every later context validation must use + the refreshed token instead of scanning that corpus again. + """ + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + origin = receipt["ground_truth_inputs"]["origins"]["codex-session"] + external_path = Path(origin["declared_roots"][0]) / origin["external_inventory"][0]["relative_path"] + + full_inventory_calls = 0 + original_inventory = schema_gate_module._external_inventory + + def counted_inventory(roots: object) -> object: + nonlocal full_inventory_calls + full_inventory_calls += 1 + return original_inventory(roots) # type: ignore[arg-type] + + monkeypatch.setattr(schema_gate_module, "_external_inventory", counted_inventory) + result = rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path, promote=False) + ) + + assert result.status == "replayed" + consumed_evidence = result.consumed_evidence + inventory_token = cast(dict[str, object], consumed_evidence["external_ground_truth_inventory_token"]) + provenance = rebuild_index_module.RebuildProvenanceContext( + root=root, + receipt_path=receipt_path, + source_snapshot=str(consumed_evidence["source_snapshot"]), + consumed_evidence=consumed_evidence, + external_inventory_token=inventory_token, + ) + stat = external_path.stat() + os.utime(external_path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000)) + inventory_calls_before_refresh = full_inventory_calls + + rebuild_index_module._validate_before_derived_state(provenance) + inventory_calls_after_refresh = full_inventory_calls + rebuild_index_module._validate_before_derived_state(provenance) + + assert inventory_calls_after_refresh == inventory_calls_before_refresh + 1 + assert full_inventory_calls == inventory_calls_after_refresh, ( + "the refreshed pass token must prevent a second full inventory scan" + ) + + def test_resume_revalidates_external_mapping_before_more_replay(tmp_path: Path) -> None: root = tmp_path / "archive" _seed(root, count=2) @@ -853,6 +907,20 @@ def fail_attestation_checkpoint(self: IndexGenerationStore, transaction: object, assert store.load(transaction.generation_id).state == "active" monkeypatch.undo() + with pytest.raises(RuntimeError, match="promoted-attestation-failed; start a new operation"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID, + promote=True, + ) + ) + + offline_terminal = store.load_transaction(bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID) + assert offline_terminal.status == "promoted-attestation-failed" + assert store.load(offline_terminal.generation_id).state == "active" + reconciled = bulk_rebuild_module.resolve_or_start_daemon_bulk_rebuild_transaction( root, schema_inference_receipt_path=receipt_path, From ff058ca368223b679db3bc3462a0b072bfb7ce80 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 22:35:41 +0200 Subject: [PATCH 02/13] test(maintenance): bind ownership fixtures to rebuild receipts Problem: the mandatory schema-inference receipt gate made six existing ownership tests stop before exercising their intended preflight or lock behavior.\n\nWhat changed: file-backed fixtures now create identity-matching blobs, initialize the tiers required by the raw-frontier gate, and pass a valid receipt into the rebuild route.\n\nVerification: devtools test tests/unit/maintenance/test_rebuild_index_provenance_gate.py tests/unit/maintenance/test_rebuild_index_ownership.py\n\nRef polylogue-q4qpl. --- .../test_rebuild_index_ownership.py | 51 +++++++++++++++---- 1 file changed, 40 insertions(+), 11 deletions(-) diff --git a/tests/unit/maintenance/test_rebuild_index_ownership.py b/tests/unit/maintenance/test_rebuild_index_ownership.py index afa31a2bd0..a7950307d1 100644 --- a/tests/unit/maintenance/test_rebuild_index_ownership.py +++ b/tests/unit/maintenance/test_rebuild_index_ownership.py @@ -24,6 +24,7 @@ ) from polylogue.storage.archive_identity import ArchiveLocation, ArchiveOwnershipError, OwnedArchiveLocation from polylogue.storage.archive_readiness import probe_archive_tier +from polylogue.storage.blob_store import BlobStore 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 @@ -162,11 +163,14 @@ def test_rebuild_refuses_when_archive_location_already_owned(tmp_path: Path) -> """ root = tmp_path / "archive" _init_empty_source(root) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") location = ArchiveLocation.resolve(root) owned = OwnedArchiveLocation.acquire(location, owner_id="concurrent-campaign") try: with pytest.raises(ArchiveOwnershipError): - rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) # Failure happened before any generation bookkeeping was created. assert not (root / ".index-generations").exists() # The rebuild lease is now deliberately acquired before the general @@ -176,7 +180,9 @@ def test_rebuild_refuses_when_archive_location_already_owned(tmp_path: Path) -> owned.release() # Releasing the concurrent holder's ownership lets the rebuild proceed. - receipt = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + receipt = rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) assert receipt.status == "empty-source" @@ -186,6 +192,8 @@ def test_rebuild_blocks_unsafe_cursor_authority_before_generation_creation( ) -> None: root = tmp_path / "archive" _init_empty_source(root) + cursor_payload = b"cursor-authority-fixture" + cursor_blob_hash, _ = BlobStore(root / "blob").write_from_bytes(cursor_payload) with sqlite3.connect(root / "source.db") as conn: conn.execute( """ @@ -194,18 +202,21 @@ def test_rebuild_blocks_unsafe_cursor_authority_before_generation_creation( blob_size, acquired_at_ms, logical_source_key, revision_kind, source_revision, acquisition_generation, revision_authority ) VALUES ('raw-1', 'codex-session', 'session-1', 'source.jsonl', 0, ?, - 1, 1, 'codex:session-1', 'full', 'revision-0', 0, 'byte_proven') + ?, 1, 'codex:session-1', 'full', 'revision-0', 0, 'byte_proven') """, - (bytes(32),), + (bytes.fromhex(cursor_blob_hash), len(cursor_payload)), ) conn.commit() monkeypatch.setattr( "polylogue.readiness.capability.raw_frontier_source_selection_block_reason", lambda _root: "1 ingest cursor row committed past accepted raw material", ) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") with pytest.raises(RuntimeError, match="raw frontier integrity"): - rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) assert not (root / ".index-generations").exists() @@ -221,9 +232,13 @@ def test_rebuild_source_preflight_rejects_orphaned_blob_refs_before_generation_c """, (b"o" * 32,), ) + conn.commit() + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") with pytest.raises(RuntimeError, match="reindex source preflight gate failed: blob-refs-liveness"): - rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) assert not (root / ".index-generations").exists() @@ -231,20 +246,27 @@ def test_rebuild_source_preflight_rejects_orphaned_blob_refs_before_generation_c def test_rebuild_source_preflight_rejects_unexplained_raw_failure(tmp_path: Path) -> None: root = tmp_path / "archive" _init_empty_source(root) + initialize_archive_database(root / "index.db", ArchiveTier.INDEX) + initialize_archive_database(root / "ops.db", ArchiveTier.OPS) + failed_payload = b"raw-failure-fixture" + failed_blob_hash, _ = BlobStore(root / "blob").write_from_bytes(failed_payload) with sqlite3.connect(root / "source.db") as conn: conn.execute( """ INSERT INTO raw_sessions( raw_id, origin, native_id, source_path, blob_hash, blob_size, acquired_at_ms, parse_error - ) VALUES ('raw-failed', 'codex-session', 'failed', '/x', ?, 10, 100, 'unexpected parser failure') + ) VALUES ('raw-failed', 'codex-session', 'failed', '/x', ?, ?, 100, 'unexpected parser failure') """, - (b"u" * 32,), + (bytes.fromhex(failed_blob_hash), len(failed_payload)), ) conn.commit() + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") with pytest.raises(RuntimeError, match="raw-failure-lifecycle"): - rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) assert not (root / ".index-generations").exists() @@ -264,9 +286,13 @@ def test_rebuild_preflight_exposes_unreconciled_source_ref_types(tmp_path: Path) (b"h" * 32, "hook-gone", "hook_payload"), ), ) + conn.commit() + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") with pytest.raises(RuntimeError) as exc_info: - rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) message = str(exc_info.value) assert "reindex source preflight gate failed: blob-refs-liveness" in message @@ -286,8 +312,11 @@ def test_rebuild_releases_ownership_lock_after_completion(tmp_path: Path) -> Non """ root = tmp_path / "archive" _init_empty_source(root) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") - receipt = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + receipt = rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) assert receipt.status == "empty-source" location = ArchiveLocation.resolve(root) From 6e0d391f7245286b663c659fb272c967df02d8f6 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 22:44:30 +0200 Subject: [PATCH 03/13] test(maintenance): document rebuild preflight ordering Problem: the raw-failure fixture now satisfies schema-inference and raw-frontier prerequisites before asserting lifecycle classification, but that ordering was implicit.\n\nWhat changed: document the fixture contract at the test boundary.\n\nVerification: devtools test tests/unit/maintenance/test_rebuild_index_ownership.py::test_rebuild_source_preflight_rejects_unexplained_raw_failure\n\nRef polylogue-q4qpl. --- tests/unit/maintenance/test_rebuild_index_ownership.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/unit/maintenance/test_rebuild_index_ownership.py b/tests/unit/maintenance/test_rebuild_index_ownership.py index a7950307d1..834ebb6b95 100644 --- a/tests/unit/maintenance/test_rebuild_index_ownership.py +++ b/tests/unit/maintenance/test_rebuild_index_ownership.py @@ -244,6 +244,7 @@ def test_rebuild_source_preflight_rejects_orphaned_blob_refs_before_generation_c def test_rebuild_source_preflight_rejects_unexplained_raw_failure(tmp_path: Path) -> None: + """Reach raw-failure classification after satisfying earlier readiness gates.""" root = tmp_path / "archive" _init_empty_source(root) initialize_archive_database(root / "index.db", ArchiveTier.INDEX) From af625c41abf61680274cd527cc8fe3214e46fed2 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 22:56:11 +0200 Subject: [PATCH 04/13] fix(rebuild): verify generation ownership during recovery Problem: an interrupted transaction could checkpoint an active generation whose owner no longer matched the transaction record.\n\nWhat changed: require the generation owner and transaction owner to match before recording promoted-attestation-failed, with a real-route regression test.\n\nVerification: .venv/bin/python -m devtools test tests/unit/maintenance/test_rebuild_index_provenance_gate.py tests/unit/maintenance/test_rebuild_index_ownership.py (50 passed).\n\nRef polylogue-q4qpl. --- polylogue/maintenance/rebuild_index.py | 6 +++- .../test_rebuild_index_provenance_gate.py | 36 +++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 26eebd4221..8ea969c960 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -231,7 +231,11 @@ def _reconcile_active_generation_transaction( generation_path = Path(generation.index_path).resolve(strict=True) except (FileNotFoundError, OSError, ValueError): return transaction - if generation.state != "active" or active_path != generation_path: + if ( + generation.owner_id != transaction.generation_owner_id + or generation.state != "active" + or active_path != generation_path + ): return transaction return store.checkpoint_transaction( transaction, diff --git a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py index 7a76ae13a5..c31aadfd79 100644 --- a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py +++ b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py @@ -934,6 +934,42 @@ def fail_attestation_checkpoint(self: IndexGenerationStore, transaction: object, } +def test_active_generation_reconciliation_requires_transaction_owner_match( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A stale transaction cannot checkpoint a generation owned by another pass.""" + root = tmp_path / "archive" + _seed(root, count=1) + store = IndexGenerationStore.for_archive_root(root) + generation = IndexGeneration( + generation_id="gen-active", + owner_id="current-owner", + archive_root=str(root), + index_path=str(root / "index.db"), + state="active", + created_at_ms=1, + ) + transaction = IndexRebuildTransaction( + operation_id="rebuild-stale-owner", + generation_id=generation.generation_id, + generation_owner_id="stale-owner", + source_snapshot="source-snapshot", + status="ready", + created_at_ms=1, + updated_at_ms=1, + ) + monkeypatch.setattr(store, "load", lambda _generation_id: generation) + + def fail_checkpoint(*args: object, **kwargs: object) -> object: + raise AssertionError("owner-mismatched transaction must not checkpoint") + + monkeypatch.setattr(store, "checkpoint_transaction", fail_checkpoint) + + reconciled = rebuild_index_module._reconcile_active_generation_transaction(store, transaction) + + assert reconciled == transaction + + def test_daemon_does_not_route_promoted_attestation_failure_back_to_rebuild( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: From c19bea64e5c2d9de8a7416f88b33562be97b54b7 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 23:06:51 +0200 Subject: [PATCH 05/13] docs(rebuild): state recovery ownership invariant Document that interrupted-promotion reconciliation requires the transaction and active generation owners to match.\n\nRef polylogue-q4qpl. --- polylogue/maintenance/rebuild_index.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 8ea969c960..a1cbf658ad 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -220,7 +220,9 @@ def _reconcile_active_generation_transaction( transaction can still look resumable even though its generation is active. The next resolver pass must record that observed fact before returning it to the rebuild loop, otherwise the caller retries an already-active - generation forever. + generation forever. The transaction's generation owner is also required + to match the active generation owner, so a stale transaction cannot + attest a generation owned by another rebuild. """ if transaction.status in _REBUILD_TERMINAL_NOT_RESUMABLE: From d1d664b9481434c805391925054c920662e35a0b Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 00:17:01 +0200 Subject: [PATCH 06/13] fix(rebuild): preserve refreshed receipt evidence Problem: resumed rebuild validation could retain a refreshed external inventory token only in an ephemeral context, and an invalid receipt could stale a transaction whose owned generation was already active. The daemon recovery test also passed through the offline terminal path instead of exercising resolver recovery.\n\nWhat changed: propagate refreshed inventory evidence through the pass context and active validator, reconcile owned active generations before stale classification, and reset the fixture to the interrupted ready state before invoking daemon recovery.\n\nVerification: 50 focused maintenance tests and devtools verify --quick. --- polylogue/maintenance/rebuild_index.py | 10 +++++++++- .../maintenance/test_rebuild_index_provenance_gate.py | 5 ++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index a1cbf658ad..e575a52d7a 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -162,6 +162,8 @@ def validate(self, *, verify_blob_integrity: bool = False, refresh_blob_integrit refreshed_inventory_token = validated.get("external_ground_truth_inventory_token") if isinstance(refreshed_inventory_token, dict): self.external_inventory_token = refreshed_inventory_token + self.consumed_evidence["external_ground_truth_inventory_token"] = refreshed_inventory_token + _ACTIVE_EXTERNAL_INVENTORY_TOKEN.set(refreshed_inventory_token) if verify_blob_integrity: refreshed_snapshot = validated.pop("_verified_blob_integrity_snapshot", None) if isinstance(refreshed_snapshot, dict): @@ -1299,7 +1301,7 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde the *location* it resolved, catching e.g. a concurrent devtools campaign or a foreign/rotated root before this rebuild can act on stale identity). """ - from polylogue.storage.index_generation import RebuildLease + from polylogue.storage.index_generation import IndexGenerationStore, RebuildLease from polylogue.storage.sqlite.connection_profile import ( check_mapped_bytes_budget_against_cgroup_limit, log_mapped_bytes_budget_check, @@ -1379,6 +1381,12 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde # can create or mutate a candidate/transaction. with RebuildLease(root): if initial_provenance_error is not None: + if request.operation_id is not None: + recovery_store = IndexGenerationStore(owned.location) + existing = recovery_store.load_transaction(request.operation_id) + existing = _reconcile_active_generation_transaction(recovery_store, existing) + if existing.status in _REBUILD_TERMINAL_NOT_RESUMABLE: + raise initial_provenance_error _mark_rebuild_transaction_stale_after_provenance_failure( root, request.operation_id, initial_provenance_error ) diff --git a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py index c31aadfd79..a8a0631a17 100644 --- a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py +++ b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py @@ -921,6 +921,9 @@ def fail_attestation_checkpoint(self: IndexGenerationStore, transaction: object, assert offline_terminal.status == "promoted-attestation-failed" assert store.load(offline_terminal.generation_id).state == "active" + # Recreate the interrupted pre-attestation state so this assertion reaches + # the daemon resolver's recovery branch rather than its terminal fast path. + ready_transaction = store.checkpoint_transaction(offline_terminal, status="ready") reconciled = bulk_rebuild_module.resolve_or_start_daemon_bulk_rebuild_transaction( root, schema_inference_receipt_path=receipt_path, @@ -929,7 +932,7 @@ def fail_attestation_checkpoint(self: IndexGenerationStore, transaction: object, assert reconciled.status == "promoted-attestation-failed" assert reconciled.post_promotion_attestation == { "status": "reconciled-after-restart", - "generation_id": transaction.generation_id, + "generation_id": ready_transaction.generation_id, "generation_state": "active", } From 8bedb085383a9f152511c5f08970364ca301dcea Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 03:35:50 +0200 Subject: [PATCH 07/13] fix(rebuild): preserve recovery provenance guards Problem: automated review found recovery and inventory-refresh paths that could lose ownership or refreshed evidence when provenance admission failed.\n\nWhat changed: carry refreshed inventory tokens into non-resumable rebuild validation, reconcile active generations without losing the original provenance error when transaction state is unreadable, and test both paths.\n\nVerification: focused provenance tests passed 4; devtools verify --quick passed all 24 steps. --- polylogue/maintenance/rebuild_index.py | 15 +++- .../test_rebuild_index_provenance_gate.py | 76 +++++++++++++------ 2 files changed, 63 insertions(+), 28 deletions(-) diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index e575a52d7a..35ab16792f 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -1383,10 +1383,16 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde if initial_provenance_error is not None: if request.operation_id is not None: recovery_store = IndexGenerationStore(owned.location) - existing = recovery_store.load_transaction(request.operation_id) - existing = _reconcile_active_generation_transaction(recovery_store, existing) - if existing.status in _REBUILD_TERMINAL_NOT_RESUMABLE: - raise initial_provenance_error + try: + existing = recovery_store.load_transaction(request.operation_id) + except Exception: + # The stale-marker path below owns transaction-load + # failures and preserves the provenance rejection. + pass + else: + existing = _reconcile_active_generation_transaction(recovery_store, existing) + if existing.status in _REBUILD_TERMINAL_NOT_RESUMABLE: + raise initial_provenance_error _mark_rebuild_transaction_stale_after_provenance_failure( root, request.operation_id, initial_provenance_error ) @@ -1567,6 +1573,7 @@ async def _rebuild_index_from_source_owned( external_inventory_token=inventory_token, ) precreate_provenance.validate() + inventory_token = precreate_provenance.external_inventory_token generation = generation_store.create(source_snapshot=rebuild_source_evidence_snapshot(root)) try: selection_evidence = rebuild_selection_evidence( diff --git a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py index a8a0631a17..556c8b99a8 100644 --- a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py +++ b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py @@ -201,15 +201,16 @@ def test_valid_receipt_allows_real_candidate_acceptance_and_promotion(tmp_path: assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().exists() -def test_rebuild_context_reuses_refreshed_inventory_token_after_detector_change( +def test_nonresumable_rebuild_reuses_refreshed_inventory_token_after_detector_change( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - """A successful metadata-only rehash updates the pass token for later checkpoints. + """A nonresumable rebuild carries a metadata refresh to later validations. Anti-vacuity: this runs the real offline rebuild entry point and validator. - Touching the external corpus after the first context validation forces one - successful full inventory refresh. Every later context validation must use - the refreshed token instead of scanning that corpus again. + The real raw-id selection boundary touches the external corpus after the + initial receipt validation. Pre-creation validation then refreshes its + detector token once; every subsequent validation must use that token + instead of scanning the corpus again. """ root = tmp_path / "archive" _seed(root, count=1) @@ -227,32 +228,59 @@ def counted_inventory(roots: object) -> object: return original_inventory(roots) # type: ignore[arg-type] monkeypatch.setattr(schema_gate_module, "_external_inventory", counted_inventory) + original_select = rebuild_index_module.select_rebuild_raw_ids + inventory_calls_before_refresh: int | None = None + + def select_then_touch(request: RebuildIndexRequest) -> tuple[int, list[str], int]: + nonlocal inventory_calls_before_refresh + selected = original_select(request) + stat = external_path.stat() + os.utime(external_path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000)) + inventory_calls_before_refresh = full_inventory_calls + return selected + + monkeypatch.setattr(rebuild_index_module, "select_rebuild_raw_ids", select_then_touch) result = rebuild_index_from_source_sync( - RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path, promote=False) + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + raw_ids=tuple(_raw_ids(root)), + promote=False, + ) ) assert result.status == "replayed" - consumed_evidence = result.consumed_evidence - inventory_token = cast(dict[str, object], consumed_evidence["external_ground_truth_inventory_token"]) - provenance = rebuild_index_module.RebuildProvenanceContext( - root=root, - receipt_path=receipt_path, - source_snapshot=str(consumed_evidence["source_snapshot"]), - consumed_evidence=consumed_evidence, - external_inventory_token=inventory_token, + assert inventory_calls_before_refresh is not None + assert full_inventory_calls == inventory_calls_before_refresh + 1, ( + "the refreshed pass token must prevent a second full inventory scan" ) - stat = external_path.stat() - os.utime(external_path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000)) - inventory_calls_before_refresh = full_inventory_calls - rebuild_index_module._validate_before_derived_state(provenance) - inventory_calls_after_refresh = full_inventory_calls - rebuild_index_module._validate_before_derived_state(provenance) - assert inventory_calls_after_refresh == inventory_calls_before_refresh + 1 - assert full_inventory_calls == inventory_calls_after_refresh, ( - "the refreshed pass token must prevent a second full inventory scan" - ) +@pytest.mark.parametrize("transaction_payload", [None, "{"], ids=["missing", "malformed"]) +def test_invalid_receipt_preserves_provenance_error_when_recovery_state_is_unreadable( + tmp_path: Path, transaction_payload: str | None +) -> None: + """Recovery load failures cannot replace the admission-gate rejection.""" + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + receipt["generated_at"] = "2000-01-01T00:00:00Z" + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + operation_id = "unreadable-recovery" + if transaction_payload is not None: + transaction_path = root / ".index-rebuild-transactions" / f"{operation_id}.json" + transaction_path.parent.mkdir() + transaction_path.write_text(transaction_payload, encoding="utf-8") + + with pytest.raises(rebuild_index_module.RebuildProvenanceError, match="schema-inference preflight gate failed"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=operation_id, + ) + ) def test_resume_revalidates_external_mapping_before_more_replay(tmp_path: Path) -> None: From b0968823a46d7d8dc292b8921fae3da9362ee0de Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 03:53:08 +0200 Subject: [PATCH 08/13] fix(rebuild): carry refreshed provenance through checkpoints Problem: rebuild recovery could retire an active owned generation after a receipt failure, while checkpoint and pass-receipt validation depended on ambient inventory-token state. What changed: make the pass provenance context the explicit validation carrier for transaction creation, checkpoints, and persisted receipts. Reconcile active owned generations before either stale-retirement path. Separate daemon recovery from offline recovery in real-route tests. Verification: devtools test tests/unit/maintenance/test_rebuild_index_provenance_gate.py (43 passed); devtools verify --quick (passed). Co-Authored-By: Codex --- polylogue/maintenance/rebuild_index.py | 146 +++++++----------- .../test_rebuild_index_provenance_gate.py | 109 ++++++++++--- 2 files changed, 148 insertions(+), 107 deletions(-) diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 35ab16792f..2ec14b696b 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -253,6 +253,18 @@ def _reconcile_active_generation_transaction( ) +def _reconcile_active_generation_before_stale_retirement(store: IndexGenerationStore, operation_id: str | None) -> bool: + """Record an already-active owned generation before retiring its transaction.""" + if operation_id is None: + return False + try: + transaction = store.load_transaction(operation_id) + except Exception: + return False + transaction = _reconcile_active_generation_transaction(store, transaction) + return transaction.status in _REBUILD_TERMINAL_NOT_RESUMABLE + + def _mark_rebuild_transaction_stale_after_provenance_failure( root: Path, operation_id: str | None, error: RebuildProvenanceError ) -> None: @@ -400,17 +412,13 @@ def _cleanup_nonresumable_generation_failure( def _create_rebuild_transaction_after_receipt_validation( generation_store: IndexGenerationStore, request: RebuildIndexRequest, - root: Path, - *, - inventory_token: dict[str, object] | None = None, + provenance: RebuildProvenanceContext, ) -> IndexRebuildTransaction: """Create the first candidate only after an ownership-bound validation.""" from polylogue.storage.index_generation import rebuild_source_evidence_snapshot - consumed_evidence = _validate_rebuild_provenance_receipt( - root, request.schema_inference_receipt_path, inventory_token=inventory_token - ) - source_snapshot = rebuild_source_evidence_snapshot(root) + provenance.validate() + source_snapshot = rebuild_source_evidence_snapshot(provenance.root) transaction = generation_store.create_transaction( source_snapshot=source_snapshot, pass_byte_budget=( @@ -421,10 +429,8 @@ def _create_rebuild_transaction_after_receipt_validation( ), ) try: - _validate_rebuild_provenance_receipt( - root, request.schema_inference_receipt_path, inventory_token=inventory_token - ) - if rebuild_source_evidence_snapshot(root) != source_snapshot: + provenance.validate() + if rebuild_source_evidence_snapshot(provenance.root) != source_snapshot: raise RebuildProvenanceError( "rebuild schema-inference preflight gate failed: source evidence changed during transaction creation" ) @@ -432,9 +438,9 @@ def _create_rebuild_transaction_after_receipt_validation( _cleanup_transaction_after_provenance_failure( generation_store, transaction, - root, - request.schema_inference_receipt_path, - consumed_evidence, + provenance.root, + provenance.receipt_path, + provenance.consumed_evidence, exc, ) raise @@ -444,8 +450,7 @@ def _create_rebuild_transaction_after_receipt_validation( def _checkpoint_rebuild_transaction_after_receipt_validation( generation_store: IndexGenerationStore, transaction: IndexRebuildTransaction, - root: Path, - receipt_path: Path | None, + provenance: RebuildProvenanceContext, *, status: str, last_blob_hash_hex: str | None = None, @@ -455,16 +460,10 @@ def _checkpoint_rebuild_transaction_after_receipt_validation( error: str | None = None, derived_stores_cleared: bool | None = None, post_promotion_attestation: dict[str, object] | None = None, - inventory_token: dict[str, object] | None = None, verify_blob_integrity: bool = False, ) -> IndexRebuildTransaction: """Validate immediately before every persisted rebuild state transition.""" - _validate_rebuild_provenance_receipt( - root, - receipt_path, - inventory_token=inventory_token, - verify_blob_integrity=verify_blob_integrity, - ) + provenance.validate(verify_blob_integrity=verify_blob_integrity) return generation_store.checkpoint_transaction( transaction, status=status, @@ -482,12 +481,10 @@ def _save_rebuild_pass_receipt_after_receipt_validation( generation_store: IndexGenerationStore, operation_id: str, pass_receipt: RebuildIndexReceipt, - root: Path, - receipt_path: Path | None, - inventory_token: dict[str, object] | None = None, + provenance: RebuildProvenanceContext, ) -> None: """Validate before publishing a pass receipt tied to rebuild state.""" - _validate_rebuild_provenance_receipt(root, receipt_path, inventory_token=inventory_token) + provenance.validate() generation_store.save_pass_receipt(operation_id, pass_receipt.to_dict()) @@ -1381,18 +1378,9 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde # can create or mutate a candidate/transaction. with RebuildLease(root): if initial_provenance_error is not None: - if request.operation_id is not None: - recovery_store = IndexGenerationStore(owned.location) - try: - existing = recovery_store.load_transaction(request.operation_id) - except Exception: - # The stale-marker path below owns transaction-load - # failures and preserves the provenance rejection. - pass - else: - existing = _reconcile_active_generation_transaction(recovery_store, existing) - if existing.status in _REBUILD_TERMINAL_NOT_RESUMABLE: - raise initial_provenance_error + recovery_store = IndexGenerationStore(owned.location) + if _reconcile_active_generation_before_stale_retirement(recovery_store, request.operation_id): + raise initial_provenance_error _mark_rebuild_transaction_stale_after_provenance_failure( root, request.operation_id, initial_provenance_error ) @@ -1406,6 +1394,9 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde ), ) except RebuildProvenanceError as exc: + recovery_store = IndexGenerationStore(owned.location) + if _reconcile_active_generation_before_stale_retirement(recovery_store, request.operation_id): + raise _mark_rebuild_transaction_stale_after_provenance_failure(root, request.operation_id, exc) raise return await _rebuild_index_from_source_owned( @@ -1440,8 +1431,15 @@ async def _rebuild_index_from_source_owned( from polylogue.storage.repair import repair_session_insights generation_store = IndexGenerationStore(owned.location) - inventory_token = cast(dict[str, object], consumed_evidence.get("external_ground_truth_inventory_token", {})) - _ACTIVE_EXTERNAL_INVENTORY_TOKEN.set(inventory_token) + provenance = RebuildProvenanceContext( + root=root, + receipt_path=request.schema_inference_receipt_path, + source_snapshot=str(consumed_evidence.get("source_snapshot", "")), + consumed_evidence=consumed_evidence, + external_inventory_token=cast( + dict[str, object], consumed_evidence.get("external_ground_truth_inventory_token", {}) + ), + ) # ``rebuild_index_from_source`` already acquired this root's # ``RebuildLease`` before any operation mutation. Retain this scope only # to preserve the body's indentation and make the outer ownership boundary @@ -1474,10 +1472,7 @@ async def _rebuild_index_from_source_owned( transaction = _reconcile_active_generation_transaction(generation_store, transaction) else: transaction = _create_rebuild_transaction_after_receipt_validation( - generation_store, - request, - root, - inventory_token=inventory_token, + generation_store, request, provenance ) transaction_created_here = True if transaction.status in {"promoted", "promoted-attestation-failed", "stale"}: @@ -1503,8 +1498,7 @@ async def _rebuild_index_from_source_owned( _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, transaction, - root, - request.schema_inference_receipt_path, + provenance, status="stale", error="source evidence changed since this rebuild was planned", ) @@ -1527,13 +1521,12 @@ async def _rebuild_index_from_source_owned( raise RuntimeError(f"rebuild operation {transaction.operation_id} lost its inactive candidate") if not transaction.derived_stores_cleared: try: - _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + provenance.validate() _clear_bulk_build_derived_stores(Path(generation.index_path)) transaction = _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, transaction, - root, - request.schema_inference_receipt_path, + provenance, status=transaction.status, derived_stores_cleared=True, ) @@ -1565,15 +1558,7 @@ async def _rebuild_index_from_source_owned( raw_count, selected_raw_ids, skipped_by_blob_limit_count = select_rebuild_raw_ids(request) selection_elapsed_s = time.perf_counter() - selection_started_at selected_raw_count = len(selected_raw_ids) - precreate_provenance = RebuildProvenanceContext( - root=root, - receipt_path=request.schema_inference_receipt_path, - source_snapshot=str(consumed_evidence.get("source_snapshot", "")), - consumed_evidence=consumed_evidence, - external_inventory_token=inventory_token, - ) - precreate_provenance.validate() - inventory_token = precreate_provenance.external_inventory_token + provenance.validate() generation = generation_store.create(source_snapshot=rebuild_source_evidence_snapshot(root)) try: selection_evidence = rebuild_selection_evidence( @@ -1595,13 +1580,6 @@ async def _rebuild_index_from_source_owned( primary=exc, ) raise - provenance = RebuildProvenanceContext( - root=root, - receipt_path=request.schema_inference_receipt_path, - source_snapshot=str(consumed_evidence.get("source_snapshot", "")), - consumed_evidence=consumed_evidence, - external_inventory_token=inventory_token, - ) sharded_replay = request.shard_count > 1 and len(selected_raw_ids) >= request.shard_count source_drifted = False try: @@ -1742,8 +1720,7 @@ def _check_pass_deadline() -> None: transaction = _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, transaction, - root, - request.schema_inference_receipt_path, + provenance, status="stale", error="source evidence changed during deadline-interrupted rebuild pass", ) @@ -1754,8 +1731,7 @@ def _check_pass_deadline() -> None: transaction = _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, transaction, - root, - request.schema_inference_receipt_path, + provenance, status="deferred", error=str(exc), ) @@ -1830,8 +1806,7 @@ def _check_pass_deadline() -> None: generation_store, transaction.operation_id, pass_receipt, - root, - request.schema_inference_receipt_path, + provenance, ) return pass_receipt pass_elapsed_s = time.perf_counter() - pass_started_at_s @@ -1849,12 +1824,11 @@ def _check_pass_deadline() -> None: "rebuild schema-inference preflight gate failed: " "source evidence changed during this bounded rebuild pass" ) - _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + provenance.validate() transaction = _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, transaction, - root, - request.schema_inference_receipt_path, + provenance, status="stale", error="source evidence changed during this bounded rebuild pass", ) @@ -1872,8 +1846,7 @@ def _check_pass_deadline() -> None: transaction = _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, transaction, - root, - request.schema_inference_receipt_path, + provenance, status=status, last_blob_hash_hex=last_blob_hash_hex, last_raw_id=last_raw_id, @@ -1931,8 +1904,7 @@ def _check_pass_deadline() -> None: generation_store, transaction.operation_id, pass_receipt, - root, - request.schema_inference_receipt_path, + provenance, ) return pass_receipt # polylogue-o56w: terminal-stage costs used to survive only as log @@ -1965,12 +1937,11 @@ def _check_pass_deadline() -> None: "source evidence changed before terminal readiness" ) if transaction is not None: - _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + provenance.validate() transaction = _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, transaction, - root, - request.schema_inference_receipt_path, + provenance, status="stale", error="source evidence changed before terminal readiness", ) @@ -2099,8 +2070,7 @@ def _check_pass_deadline() -> None: transaction = _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, transaction, - root, - request.schema_inference_receipt_path, + provenance, status="ready", ) if request.promote: @@ -2110,7 +2080,7 @@ def _check_pass_deadline() -> None: # caught before clobbering someone else's activation rather # than after (polylogue-ovme.2 AC3). assert_owns_archive_location(owned, ArchiveLocation.resolve(root)) - _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + provenance.validate() terminal_started_at = time.perf_counter() generation = generation_store.promote(generation) terminal_timings_s["terminal.promote"] = time.perf_counter() - terminal_started_at @@ -2186,8 +2156,7 @@ def _check_pass_deadline() -> None: _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, transaction, - root, - request.schema_inference_receipt_path, + provenance, status="failed", error="bounded rebuild pass failed; candidate retained for diagnosis or explicit recovery", ) @@ -2292,8 +2261,7 @@ def _check_pass_deadline() -> None: generation_store, transaction.operation_id, final_receipt, - root, - request.schema_inference_receipt_path, + provenance, ) except BaseException as exc: if transaction is None: diff --git a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py index 556c8b99a8..b17ed5da24 100644 --- a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py +++ b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py @@ -256,6 +256,66 @@ def select_then_touch(request: RebuildIndexRequest) -> tuple[int, list[str], int ) +def test_resumable_checkpoint_and_pass_receipt_reuse_refreshed_inventory_token( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """One refreshed token remains authoritative through a resumable pass receipt. + + Anti-vacuity: this uses the transaction page selection, checkpoint, and + pass-receipt paths. Replacing the shared provenance context in either + helper makes a second full inventory scan observable after the refresh. + """ + root = tmp_path / "archive" + _seed(root, count=2) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + origin = receipt["ground_truth_inputs"]["origins"]["codex-session"] + external_path = Path(origin["declared_roots"][0]) / origin["external_inventory"][0]["relative_path"] + + full_inventory_calls = 0 + original_inventory = schema_gate_module._external_inventory + + def counted_inventory(roots: object) -> object: + nonlocal full_inventory_calls + full_inventory_calls += 1 + return original_inventory(roots) # type: ignore[arg-type] + + monkeypatch.setattr(schema_gate_module, "_external_inventory", counted_inventory) + original_next_raw_page = IndexGenerationStore.next_raw_page + inventory_calls_before_refresh: int | None = None + + def select_then_touch(self: IndexGenerationStore, *args: object, **kwargs: object) -> object: + nonlocal inventory_calls_before_refresh + page = original_next_raw_page(self, *args, **kwargs) # type: ignore[arg-type] + if inventory_calls_before_refresh is None: + stat = external_path.stat() + os.utime(external_path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000)) + inventory_calls_before_refresh = full_inventory_calls + return page + + monkeypatch.setattr(IndexGenerationStore, "next_raw_page", select_then_touch) + result = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + raw_batch_size=1, + promote=False, + ) + ) + + assert result.status == "paused" + assert result.transaction is not None + assert inventory_calls_before_refresh is not None + assert full_inventory_calls == inventory_calls_before_refresh + 1 + operation_id = str(result.transaction["operation_id"]) + pass_receipt_path = next((root / ".index-rebuild-transactions" / f"{operation_id}.receipts").glob("pass-*.json")) + persisted_receipt = json.loads(pass_receipt_path.read_text(encoding="utf-8")) + assert ( + persisted_receipt["consumed_evidence"]["external_ground_truth_inventory_token"] + == result.consumed_evidence["external_ground_truth_inventory_token"] + ) + + @pytest.mark.parametrize("transaction_payload", [None, "{"], ids=["missing", "malformed"]) def test_invalid_receipt_preserves_provenance_error_when_recovery_state_is_unreadable( tmp_path: Path, transaction_payload: str | None @@ -935,23 +995,6 @@ def fail_attestation_checkpoint(self: IndexGenerationStore, transaction: object, assert store.load(transaction.generation_id).state == "active" monkeypatch.undo() - with pytest.raises(RuntimeError, match="promoted-attestation-failed; start a new operation"): - rebuild_index_from_source_sync( - RebuildIndexRequest( - archive_root=root, - schema_inference_receipt_path=receipt_path, - operation_id=bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID, - promote=True, - ) - ) - - offline_terminal = store.load_transaction(bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID) - assert offline_terminal.status == "promoted-attestation-failed" - assert store.load(offline_terminal.generation_id).state == "active" - - # Recreate the interrupted pre-attestation state so this assertion reaches - # the daemon resolver's recovery branch rather than its terminal fast path. - ready_transaction = store.checkpoint_transaction(offline_terminal, status="ready") reconciled = bulk_rebuild_module.resolve_or_start_daemon_bulk_rebuild_transaction( root, schema_inference_receipt_path=receipt_path, @@ -960,11 +1003,41 @@ def fail_attestation_checkpoint(self: IndexGenerationStore, transaction: object, assert reconciled.status == "promoted-attestation-failed" assert reconciled.post_promotion_attestation == { "status": "reconciled-after-restart", - "generation_id": ready_transaction.generation_id, + "generation_id": transaction.generation_id, "generation_state": "active", } +def test_provenance_failure_reconciles_active_generation_before_stale_retirement(tmp_path: Path) -> None: + """Receipt rejection preserves an already-promoted owned generation's lifecycle fact.""" + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + store = IndexGenerationStore.for_archive_root(root) + transaction = store.create_transaction( + source_snapshot=rebuild_source_evidence_snapshot(root), operation_id="active-before-stale-retirement" + ) + active_generation = store.promote(store.load(transaction.generation_id)) + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + receipt["generated_at"] = "2000-01-01T00:00:00Z" + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + + with pytest.raises(rebuild_index_module.RebuildProvenanceError, match="schema-inference preflight gate failed"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=transaction.operation_id, + promote=True, + ) + ) + + reconciled = store.load_transaction(transaction.operation_id) + assert reconciled.status == "promoted-attestation-failed" + assert reconciled.generation_id == active_generation.generation_id + assert store.load(reconciled.generation_id).state == "active" + + def test_active_generation_reconciliation_requires_transaction_owner_match( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: From e6138c6ce7e1cdad51f1f6b4b98ebe029ec3806f Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 04:18:49 +0200 Subject: [PATCH 09/13] fix(rebuild): persist refreshed provenance evidence Carry the latest receipt-bound inventory evidence through transaction checkpoints and pass receipts, including non-resumable rebuilds. Preserve provenance failures when recovery metadata is malformed and reconcile active generations before stale retirement. --- polylogue/maintenance/rebuild_index.py | 50 ++++++------- polylogue/storage/index_generation.py | 12 ++- .../test_rebuild_index_provenance_gate.py | 73 +++++++++++++++++-- 3 files changed, 101 insertions(+), 34 deletions(-) diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 2ec14b696b..1fbb89beec 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -16,7 +16,7 @@ import sqlite3 import time from collections.abc import Iterable, Sequence -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass, field, replace from hashlib import sha256 from http import HTTPStatus from pathlib import Path @@ -180,6 +180,10 @@ def validate_cleanup(self) -> None: if not self.consumed_evidence or not self.source_snapshot: raise RuntimeError("rebuild cleanup has no validated provenance context") + def receipt_evidence(self) -> dict[str, object]: + """Snapshot the latest receipt-bound evidence for a durable record.""" + return dict(self.consumed_evidence) + def _validate_rebuild_provenance_receipt( root: Path, @@ -233,7 +237,7 @@ def _reconcile_active_generation_transaction( generation = store.load(transaction.generation_id) active_path = store.active_pointer.resolve(strict=True) generation_path = Path(generation.index_path).resolve(strict=True) - except (FileNotFoundError, OSError, ValueError): + except (FileNotFoundError, OSError, TypeError, ValueError): return transaction if ( generation.owner_id != transaction.generation_owner_id @@ -253,18 +257,6 @@ def _reconcile_active_generation_transaction( ) -def _reconcile_active_generation_before_stale_retirement(store: IndexGenerationStore, operation_id: str | None) -> bool: - """Record an already-active owned generation before retiring its transaction.""" - if operation_id is None: - return False - try: - transaction = store.load_transaction(operation_id) - except Exception: - return False - transaction = _reconcile_active_generation_transaction(store, transaction) - return transaction.status in _REBUILD_TERMINAL_NOT_RESUMABLE - - def _mark_rebuild_transaction_stale_after_provenance_failure( root: Path, operation_id: str | None, error: RebuildProvenanceError ) -> None: @@ -284,6 +276,7 @@ def _mark_rebuild_transaction_stale_after_provenance_failure( try: store = IndexGenerationStore.for_archive_root(root) transaction = store.load_transaction(operation_id) + transaction = _reconcile_active_generation_transaction(store, transaction) except Exception as load_error: error.add_note(f"could not load rebuild transaction to mark it stale: {load_error}") return @@ -427,6 +420,7 @@ def _create_rebuild_transaction_after_receipt_validation( pass_deadline_ms=( int(request.pass_deadline_seconds * 1000) if request.pass_deadline_seconds is not None else None ), + consumed_evidence=provenance.receipt_evidence(), ) try: provenance.validate() @@ -474,6 +468,7 @@ def _checkpoint_rebuild_transaction_after_receipt_validation( error=error, derived_stores_cleared=derived_stores_cleared, post_promotion_attestation=post_promotion_attestation, + consumed_evidence=provenance.receipt_evidence(), ) @@ -482,10 +477,12 @@ def _save_rebuild_pass_receipt_after_receipt_validation( operation_id: str, pass_receipt: RebuildIndexReceipt, provenance: RebuildProvenanceContext, -) -> None: +) -> RebuildIndexReceipt: """Validate before publishing a pass receipt tied to rebuild state.""" provenance.validate() + pass_receipt = replace(pass_receipt, consumed_evidence=provenance.receipt_evidence()) generation_store.save_pass_receipt(operation_id, pass_receipt.to_dict()) + return pass_receipt #: Passed through to ``CensusParseStage.warm_raw_ids``'s ``max_payload_bytes`` @@ -1298,7 +1295,7 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde the *location* it resolved, catching e.g. a concurrent devtools campaign or a foreign/rotated root before this rebuild can act on stale identity). """ - from polylogue.storage.index_generation import IndexGenerationStore, RebuildLease + from polylogue.storage.index_generation import RebuildLease from polylogue.storage.sqlite.connection_profile import ( check_mapped_bytes_budget_against_cgroup_limit, log_mapped_bytes_budget_check, @@ -1378,9 +1375,6 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde # can create or mutate a candidate/transaction. with RebuildLease(root): if initial_provenance_error is not None: - recovery_store = IndexGenerationStore(owned.location) - if _reconcile_active_generation_before_stale_retirement(recovery_store, request.operation_id): - raise initial_provenance_error _mark_rebuild_transaction_stale_after_provenance_failure( root, request.operation_id, initial_provenance_error ) @@ -1394,9 +1388,6 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde ), ) except RebuildProvenanceError as exc: - recovery_store = IndexGenerationStore(owned.location) - if _reconcile_active_generation_before_stale_retirement(recovery_store, request.operation_id): - raise _mark_rebuild_transaction_stale_after_provenance_failure(root, request.operation_id, exc) raise return await _rebuild_index_from_source_owned( @@ -1800,9 +1791,9 @@ def _check_pass_deadline() -> None: ), selection_evidence=selection_evidence, timings_s=cast(dict[str, float], pass_cost.to_dict()), - consumed_evidence=consumed_evidence, + consumed_evidence=provenance.receipt_evidence(), ) - _save_rebuild_pass_receipt_after_receipt_validation( + pass_receipt = _save_rebuild_pass_receipt_after_receipt_validation( generation_store, transaction.operation_id, pass_receipt, @@ -1898,9 +1889,9 @@ def _check_pass_deadline() -> None: ), selection_evidence=selection_evidence, timings_s=cast(dict[str, float], pass_cost.to_dict()), - consumed_evidence=consumed_evidence, + consumed_evidence=provenance.receipt_evidence(), ) - _save_rebuild_pass_receipt_after_receipt_validation( + pass_receipt = _save_rebuild_pass_receipt_after_receipt_validation( generation_store, transaction.operation_id, pass_receipt, @@ -2108,6 +2099,7 @@ def _check_pass_deadline() -> None: transaction, status="promoted", post_promotion_attestation=attestation, + consumed_evidence=provenance.receipt_evidence(), ) except Exception as attestation_error: source_drifted = True @@ -2201,7 +2193,7 @@ def _check_pass_deadline() -> None: replay=replay, terminal_timings_s=terminal_timings_s, ), - consumed_evidence=consumed_evidence, + consumed_evidence=provenance.receipt_evidence(), ) try: _persist_candidate_receipt(generation, final_receipt.to_dict()) @@ -2220,6 +2212,7 @@ def _check_pass_deadline() -> None: "generation_state": "active", "error": str(attestation_error), }, + consumed_evidence=provenance.receipt_evidence(), ) except BaseException as recovery_error: attestation_error.add_note( @@ -2249,6 +2242,7 @@ def _check_pass_deadline() -> None: "generation_state": "active", "error": str(attestation_error), }, + consumed_evidence=provenance.receipt_evidence(), ) except BaseException as recovery_error: attestation_error.add_note( @@ -2257,7 +2251,7 @@ def _check_pass_deadline() -> None: ) raise else: - _save_rebuild_pass_receipt_after_receipt_validation( + final_receipt = _save_rebuild_pass_receipt_after_receipt_validation( generation_store, transaction.operation_id, final_receipt, diff --git a/polylogue/storage/index_generation.py b/polylogue/storage/index_generation.py index 121ce6decd..d7e27c4692 100644 --- a/polylogue/storage/index_generation.py +++ b/polylogue/storage/index_generation.py @@ -13,7 +13,7 @@ import time import uuid from contextlib import closing -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, field from enum import StrEnum from pathlib import Path from types import TracebackType @@ -191,6 +191,10 @@ class IndexRebuildTransaction: # checks, so a post-flip observation failure cannot leave a resumable # transaction claiming that its candidate is merely ready. post_promotion_attestation: dict[str, object] | None = None + # The schema-inference evidence admitted for this checkpoint. This is + # distinct from ``source_snapshot``: it records the receipt-bound external + # inventory token that authorized the current replay state. + consumed_evidence: dict[str, object] = field(default_factory=dict) @property def cursor(self) -> str | None: @@ -507,6 +511,7 @@ def create_transaction( operation_id: str | None = None, pass_byte_budget: int | None = None, pass_deadline_ms: int | None = None, + consumed_evidence: dict[str, object] | None = None, ) -> IndexRebuildTransaction: """Create an inactive candidate and its resumable transaction record.""" op_id = operation_id or str(uuid.uuid4()) @@ -528,6 +533,7 @@ def create_transaction( owner_pid=os.getpid(), owner_host=socket.gethostname(), heartbeat_at_ms=now, + consumed_evidence=dict(consumed_evidence or {}), ) self.save_transaction(transaction) return transaction @@ -591,6 +597,7 @@ def checkpoint_transaction( error: str | None = None, derived_stores_cleared: bool | None = None, post_promotion_attestation: dict[str, object] | None = None, + consumed_evidence: dict[str, object] | None = None, ) -> IndexRebuildTransaction: """Persist one state transition without changing candidate ownership.""" return self.save_transaction( @@ -615,6 +622,9 @@ def checkpoint_transaction( "post_promotion_attestation": post_promotion_attestation if post_promotion_attestation is not None else transaction.post_promotion_attestation, + "consumed_evidence": ( + dict(consumed_evidence) if consumed_evidence is not None else transaction.consumed_evidence + ), } ) ) diff --git a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py index b17ed5da24..2eab46f66f 100644 --- a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py +++ b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py @@ -201,8 +201,9 @@ def test_valid_receipt_allows_real_candidate_acceptance_and_promotion(tmp_path: assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().exists() -def test_nonresumable_rebuild_reuses_refreshed_inventory_token_after_detector_change( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch +@pytest.mark.parametrize("selection", ["raw-ids", "only-missing", "max-blob-mb"]) +def test_nonresumable_rebuild_persists_refreshed_inventory_evidence_after_detector_change( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, selection: str ) -> None: """A nonresumable rebuild carries a metadata refresh to later validations. @@ -240,20 +241,42 @@ def select_then_touch(request: RebuildIndexRequest) -> tuple[int, list[str], int return selected monkeypatch.setattr(rebuild_index_module, "select_rebuild_raw_ids", select_then_touch) - result = rebuild_index_from_source_sync( - RebuildIndexRequest( + if selection == "only-missing": + request = RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + only_missing=True, + promote=False, + ) + elif selection == "max-blob-mb": + request = RebuildIndexRequest( archive_root=root, schema_inference_receipt_path=receipt_path, raw_ids=tuple(_raw_ids(root)), + max_blob_mb=1.0, promote=False, ) - ) + else: + request = RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + raw_ids=tuple(_raw_ids(root)), + promote=False, + ) + result = rebuild_index_from_source_sync(request) assert result.status == "replayed" assert inventory_calls_before_refresh is not None assert full_inventory_calls == inventory_calls_before_refresh + 1, ( "the refreshed pass token must prevent a second full inventory scan" ) + candidate_receipt = Path(cast(str, result.generation["index_path"])).parent / "rebuild-receipt.json" + persisted_receipt = json.loads(candidate_receipt.read_text(encoding="utf-8")) + assert persisted_receipt["consumed_evidence"] == result.consumed_evidence + assert ( + persisted_receipt["consumed_evidence"]["external_ground_truth_inventory_token"] + == result.consumed_evidence["external_ground_truth_inventory_token"] + ) def test_resumable_checkpoint_and_pass_receipt_reuse_refreshed_inventory_token( @@ -308,6 +331,8 @@ def select_then_touch(self: IndexGenerationStore, *args: object, **kwargs: objec assert inventory_calls_before_refresh is not None assert full_inventory_calls == inventory_calls_before_refresh + 1 operation_id = str(result.transaction["operation_id"]) + checkpoint = IndexGenerationStore.for_archive_root(root).load_transaction(operation_id) + assert checkpoint.consumed_evidence == result.consumed_evidence pass_receipt_path = next((root / ".index-rebuild-transactions" / f"{operation_id}.receipts").glob("pass-*.json")) persisted_receipt = json.loads(pass_receipt_path.read_text(encoding="utf-8")) assert ( @@ -343,6 +368,42 @@ def test_invalid_receipt_preserves_provenance_error_when_recovery_state_is_unrea ) +@pytest.mark.parametrize("metadata", ["transaction", "generation"]) +def test_invalid_receipt_preserves_provenance_error_when_recovery_metadata_is_readable_but_malformed( + tmp_path: Path, metadata: str +) -> None: + """Metadata-shape failures during stale retirement cannot replace the gate error.""" + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + store = IndexGenerationStore.for_archive_root(root) + transaction = store.create_transaction( + source_snapshot=rebuild_source_evidence_snapshot(root), operation_id=f"malformed-{metadata}" + ) + if metadata == "transaction": + metadata_path = root / ".index-rebuild-transactions" / f"{transaction.operation_id}.json" + payload = json.loads(metadata_path.read_text(encoding="utf-8")) + payload["generation_id"] = None + metadata_path.write_text(json.dumps(payload), encoding="utf-8") + else: + metadata_path = Path(store.load(transaction.generation_id).index_path).parent / "generation.json" + payload = json.loads(metadata_path.read_text(encoding="utf-8")) + payload["index_path"] = None + metadata_path.write_text(json.dumps(payload), encoding="utf-8") + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + receipt["generated_at"] = "2000-01-01T00:00:00Z" + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + + with pytest.raises(rebuild_index_module.RebuildProvenanceError, match="schema-inference preflight gate failed"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=transaction.operation_id, + ) + ) + + def test_resume_revalidates_external_mapping_before_more_replay(tmp_path: Path) -> None: root = tmp_path / "archive" _seed(root, count=2) @@ -551,6 +612,7 @@ def expire_after_transaction( operation_id: str | None = None, pass_byte_budget: int | None = None, pass_deadline_ms: int | None = None, + consumed_evidence: dict[str, object] | None = None, ) -> IndexRebuildTransaction: transaction = original_create_transaction( store, @@ -558,6 +620,7 @@ def expire_after_transaction( operation_id=operation_id, pass_byte_budget=pass_byte_budget, pass_deadline_ms=pass_deadline_ms, + consumed_evidence=consumed_evidence, ) assert store.load(transaction.generation_id).state == "inactive" assert store.load_transaction(transaction.operation_id).operation_id == transaction.operation_id From 143deb9ad9f9a9c460064ad5b29ced6abc81329b Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 05:38:10 +0200 Subject: [PATCH 10/13] chore: refresh rebuild provenance gate metadata Refresh the commit-bound CI receipt after correcting the ready PR scope carrier. The empty trigger commit carries no product diff and will disappear in the squash merge. From 83e77aed6533dfaa8bce874d297df107e1bb2b1d Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 05:53:43 +0200 Subject: [PATCH 11/13] chore: finalize rebuild provenance gate metadata Trigger a fresh CI evaluation after recomputing the ready PR scope carrier for the final branch head. This metadata-only trigger has no product diff and disappears in the squash merge. From 91d56f4523d8cb1b271baf10b6926cc0a3d9a644 Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 06:02:14 +0200 Subject: [PATCH 12/13] chore: align rebuild provenance CI carrier Trigger CI after the ready PR carrier was regenerated for the branch head. This metadata-only trigger has no product diff and disappears in the squash merge. From a7b960595e18e63f8b1d8216e80d3e7afe867a41 Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 06:19:01 +0200 Subject: [PATCH 13/13] test(maintenance): pin active generation reconciliation preconditions Problem: the owner-mismatch regression test also passed when unrelated path or state guards returned early.\n\nWhat changed: assert the active generation state and pointer target before exercising reconciliation, so the owner comparison is the only early-return condition under test.\n\nVerification: focused provenance test passed. --- tests/unit/maintenance/test_rebuild_index_provenance_gate.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py index 2eab46f66f..4f624e2011 100644 --- a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py +++ b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py @@ -1127,6 +1127,11 @@ def test_active_generation_reconciliation_requires_transaction_owner_match( ) monkeypatch.setattr(store, "load", lambda _generation_id: generation) + # Pin the non-owner short circuits so the owner mismatch is the only + # reason reconciliation can return the unchanged transaction. + assert generation.state == "active" + assert store.active_pointer.resolve(strict=True) == Path(generation.index_path).resolve(strict=True) + def fail_checkpoint(*args: object, **kwargs: object) -> object: raise AssertionError("owner-mismatched transaction must not checkpoint")