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..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 @@ -159,6 +159,11 @@ 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 + 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): @@ -175,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, @@ -204,6 +213,50 @@ 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. 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: + 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, TypeError, ValueError): + return transaction + if ( + generation.owner_id != transaction.generation_owner_id + or 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: @@ -223,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 @@ -351,17 +405,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=( @@ -370,12 +420,11 @@ 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: - _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" ) @@ -383,9 +432,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 @@ -395,8 +444,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, @@ -406,16 +454,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, @@ -426,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(), ) @@ -433,13 +476,13 @@ 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, -) -> None: + provenance: RebuildProvenanceContext, +) -> RebuildIndexReceipt: """Validate before publishing a pass receipt tied to rebuild state.""" - _validate_rebuild_provenance_receipt(root, receipt_path, inventory_token=inventory_token) + 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`` @@ -1379,8 +1422,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 @@ -1410,12 +1460,10 @@ 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, - request, - root, - inventory_token=inventory_token, + generation_store, request, provenance ) transaction_created_here = True if transaction.status in {"promoted", "promoted-attestation-failed", "stale"}: @@ -1441,8 +1489,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", ) @@ -1465,13 +1512,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, ) @@ -1503,14 +1549,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() + provenance.validate() generation = generation_store.create(source_snapshot=rebuild_source_evidence_snapshot(root)) try: selection_evidence = rebuild_selection_evidence( @@ -1532,13 +1571,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: @@ -1679,8 +1711,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", ) @@ -1691,8 +1722,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), ) @@ -1761,14 +1791,13 @@ 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, - root, - request.schema_inference_receipt_path, + provenance, ) return pass_receipt pass_elapsed_s = time.perf_counter() - pass_started_at_s @@ -1786,12 +1815,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", ) @@ -1809,8 +1837,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, @@ -1862,14 +1889,13 @@ 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, - root, - request.schema_inference_receipt_path, + provenance, ) return pass_receipt # polylogue-o56w: terminal-stage costs used to survive only as log @@ -1902,12 +1928,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", ) @@ -2036,8 +2061,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: @@ -2047,7 +2071,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 @@ -2075,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 @@ -2123,8 +2148,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", ) @@ -2169,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()) @@ -2188,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( @@ -2217,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( @@ -2225,12 +2251,11 @@ 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, - root, - request.schema_inference_receipt_path, + provenance, ) except BaseException as exc: if transaction is None: 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_ownership.py b/tests/unit/maintenance/test_rebuild_index_ownership.py index afa31a2bd0..834ebb6b95 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,30 +232,42 @@ 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() 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) + 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 +287,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 +313,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) diff --git a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py index 1a1ead2f7e..4f624e2011 100644 --- a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py +++ b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py @@ -201,6 +201,209 @@ def test_valid_receipt_allows_real_candidate_acceptance_and_promotion(tmp_path: assert IndexGenerationStore.for_archive_root(root).active_pointer.resolve().exists() +@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. + + Anti-vacuity: this runs the real offline rebuild entry point and validator. + 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) + 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_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) + 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( + 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"]) + 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 ( + 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 +) -> 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, + ) + ) + + +@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) @@ -409,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, @@ -416,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 @@ -866,6 +1071,77 @@ def fail_attestation_checkpoint(self: IndexGenerationStore, transaction: object, } +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: + """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) + + # 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") + + 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: