From 9d81e774fb62619ae1c9be8224a41ed4b6471a89 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 18:00:51 +0200 Subject: [PATCH 1/8] test(reindex): harden Codex 804 recovery census Problem: The Codex 804 proof did not preserve baseline message timestamps explicitly, its interruption wait observed only post-checkpoint progress, and recovery accepted permissive authority and membership outcomes. The focused resume regression also lacked the current schema-inference receipt gate. What changed: Preserve and compare baseline timestamps across four revisions. Add a pre-checkpoint subprocess boundary with transaction and receipt assertions, retain the committed-page boundary, and trace production restart selections to prove exact suffix replay. Require byte-authority conservation for all 804 raws, complete application coverage, and one terminal head. Supply valid receipts to the maintenance parity fixture. Verification: direnv exec . devtools test tests/unit/scenarios/test_codex_804_live_proof.py (1 passed in 276.97s); direnv exec . devtools test tests/unit/maintenance/test_rebuild_index_resume_correctness.py (1 passed in 7.41s); direnv exec . devtools verify --quick (exit 0). Disposition: implementation-complete; live-proof-pending. Successor: polylogue-codex-804-live-proof. --- .../test_rebuild_index_resume_correctness.py | 38 +++- .../scenarios/test_codex_804_live_proof.py | 163 +++++++++++++++++- 2 files changed, 190 insertions(+), 11 deletions(-) diff --git a/tests/unit/maintenance/test_rebuild_index_resume_correctness.py b/tests/unit/maintenance/test_rebuild_index_resume_correctness.py index 2bec4fa1fd..f5c4ef67f7 100644 --- a/tests/unit/maintenance/test_rebuild_index_resume_correctness.py +++ b/tests/unit/maintenance/test_rebuild_index_resume_correctness.py @@ -122,6 +122,7 @@ def test_committed_page_interrupt_resumes_only_suffix_and_matches_clean_rebuild( clean_root = tmp_path / "clean" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) _seed(root, monkeypatch=monkeypatch) + resumed_receipt_path = write_valid_rebuild_receipt(root, tmp_path / "resumed-receipt.json") original_checkpoint = IndexGenerationStore.checkpoint_transaction interrupted = False @@ -137,7 +138,13 @@ def interrupt_after_committed_page(self: IndexGenerationStore, transaction: obje with monkeypatch.context() as scoped: scoped.setattr(IndexGenerationStore, "checkpoint_transaction", interrupt_after_committed_page) with pytest.raises(InjectedInterruptError, match="after committed page"): - rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root, raw_batch_size=1)) + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + raw_batch_size=1, + schema_inference_receipt_path=resumed_receipt_path, + ) + ) store = IndexGenerationStore.for_archive_root(root) operation_id = next(path.stem for path in store.transactions_root.glob("*.json")) @@ -162,7 +169,12 @@ async def recording_replay(*args: object, **kwargs: object) -> dict[str, object] monkeypatch.setattr(replay_module, "rebuild_index_from_source", recording_replay) while True: receipt = rebuild_index_from_source_sync( - RebuildIndexRequest(archive_root=root, operation_id=operation_id, raw_batch_size=1) + RebuildIndexRequest( + archive_root=root, + operation_id=operation_id, + raw_batch_size=1, + schema_inference_receipt_path=resumed_receipt_path, + ) ) if receipt.status == "replayed": break @@ -170,20 +182,38 @@ async def recording_replay(*args: object, **kwargs: object) -> dict[str, object] assert [len(page) for page in replayed_raw_pages] == [1, 1] assert len({raw_id for page in replayed_raw_pages for raw_id in page}) == 2 + with sqlite3.connect(root / "source.db") as conn: + first_committed_raw_id = str( + conn.execute("SELECT raw_id FROM raw_sessions ORDER BY blob_hash, raw_id LIMIT 1").fetchone()[0] + ) + all_raw_ids = {str(row[0]) for row in conn.execute("SELECT raw_id FROM raw_sessions")} + resumed_raw_ids = {raw_id for page in replayed_raw_pages for raw_id in page} + # Mutation that resets the persisted cursor would replay the committed raw + # again and fail this exact suffix conservation check. + assert first_committed_raw_id not in resumed_raw_ids + assert resumed_raw_ids == all_raw_ids - {first_committed_raw_id} assert receipt.operation["cursor"] is not None assert receipt.operation["heartbeat"]["at_ms"] is not None # type: ignore[index] assert receipt.operation["recovery_state"] == "promoted" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(clean_root)) _seed(clean_root, monkeypatch=monkeypatch) - clean = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=clean_root, raw_batch_size=1)) + clean_receipt_path = write_valid_rebuild_receipt(clean_root, tmp_path / "clean-receipt.json") + clean = rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=clean_root, raw_batch_size=1, schema_inference_receipt_path=clean_receipt_path) + ) assert clean.status == "paused" assert clean.transaction is not None clean_operation = clean.transaction["operation_id"] assert isinstance(clean_operation, str) while clean.status != "replayed": clean = rebuild_index_from_source_sync( - RebuildIndexRequest(archive_root=clean_root, operation_id=clean_operation, raw_batch_size=1) + RebuildIndexRequest( + archive_root=clean_root, + operation_id=clean_operation, + raw_batch_size=1, + schema_inference_receipt_path=clean_receipt_path, + ) ) resumed_snapshot = _semantic_snapshot(root) diff --git a/tests/unit/scenarios/test_codex_804_live_proof.py b/tests/unit/scenarios/test_codex_804_live_proof.py index 873755c5b7..a3bdaf8648 100644 --- a/tests/unit/scenarios/test_codex_804_live_proof.py +++ b/tests/unit/scenarios/test_codex_804_live_proof.py @@ -55,6 +55,8 @@ SESSION_NATIVE_ID = "codex-sanitized-804-session" SOURCE_PATH = "codex/incident-804-sanitized.jsonl" NEAR_TERMINAL_PREDECESSOR_BYTES = 32 * 1024 * 1024 +_BASELINE_MESSAGE_IDS = tuple(f"{SESSION_NATIVE_ID}-message-{index}" for index in range(2)) +_BASELINE_MESSAGE_TIMESTAMPS = ("2026-07-31T04:25:20Z", "2026-07-31T04:25:20Z") def _wire_target_bytes(revision: int) -> int: @@ -89,7 +91,7 @@ def _codex_payload(revision: int, *, terminal: bool) -> bytes: records.append( { "type": "response_item", - "timestamp": revision_timestamp, + "timestamp": _BASELINE_MESSAGE_TIMESTAMPS[message_index], "payload": { "type": "message", "id": f"{SESSION_NATIVE_ID}-message-{message_index}", @@ -162,6 +164,24 @@ def _codex_payload(revision: int, *, terminal: bool) -> bytes: return payload +def _baseline_message_timestamps(payload: bytes) -> tuple[str, ...]: + timestamps: dict[str, str] = {} + for line in payload.splitlines(): + record = json.loads(line) + if not isinstance(record, dict): + continue + record_payload = record.get("payload") + if not isinstance(record_payload, dict) or record_payload.get("type") != "message": + continue + message_id = record_payload.get("id") + timestamp = record.get("timestamp") + if isinstance(message_id, str) and message_id in _BASELINE_MESSAGE_IDS: + if not isinstance(timestamp, str): + raise AssertionError(f"baseline message {message_id} has no string timestamp") + timestamps[message_id] = timestamp + return tuple(timestamps[message_id] for message_id in _BASELINE_MESSAGE_IDS) + + def _incident_program() -> CorpusProgram: operations = tuple( Acquire( @@ -335,6 +355,12 @@ def test_sanitized_codex_804_revision_recovery_proof( assert current_payload.startswith(previous_payload) assert b"incident witness user baseline" in first_revision_payload assert b"parsed milestone revision 1" in second_revision_payload + # Mutation that assigns every recursively extended baseline message the + # current revision timestamp fails this exact multi-revision comparison. + for revision, terminal in ((0, False), (1, False), (800, False), (803, True)): + assert _baseline_message_timestamps(_codex_payload(revision, terminal=terminal)) == ( + _BASELINE_MESSAGE_TIMESTAMPS + ) program_digest = hashlib.sha256(program_json.encode("utf-8")).hexdigest() @@ -377,6 +403,62 @@ def test_sanitized_codex_804_revision_recovery_proof( assert transaction.status == "running" transaction_path = store.transactions_root / f"{operation_id}.json" assert transaction_path.is_file() + precheckpoint_script = """ +import os +import sys +from pathlib import Path + +from polylogue.storage.index_generation import IndexGenerationStore + +original_checkpoint = IndexGenerationStore.checkpoint_transaction + +def terminate_before_page_checkpoint(self, transaction, **kwargs): + if kwargs.get("status") == "paused" and kwargs.get("processed_raw_count", 0) > 0: + os._exit(97) + return original_checkpoint(self, transaction, **kwargs) + +IndexGenerationStore.checkpoint_transaction = terminate_before_page_checkpoint +from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync + +root = Path(sys.argv[1]) +operation_id = sys.argv[2] +receipt = Path(sys.argv[3]) +rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + operation_id=operation_id, + promote=False, + schema_inference_receipt_path=receipt, + raw_batch_size=8, + ) +) +raise SystemExit("checkpoint seam was not reached") +""" + precheckpoint_process = subprocess.run( + [sys.executable, "-c", precheckpoint_script, str(root), operation_id, str(schema_inference_receipt_path)], + cwd=Path.cwd(), + capture_output=True, + text=True, + check=False, + timeout=120, + ) + assert precheckpoint_process.returncode == 97, ( + f"pre-checkpoint boundary did not terminate at the checkpoint seam: " + f"returncode={precheckpoint_process.returncode}; " + f"stdout={precheckpoint_process.stdout}; stderr={precheckpoint_process.stderr}" + ) + persisted_before_page = store.load_transaction(operation_id) + # Mutation that advances the cursor before checkpoint_transaction returns + # fails these pre-checkpoint invariants and the receipt census below. + assert persisted_before_page.status == "running" + assert persisted_before_page.processed_raw_count == 0 + assert persisted_before_page.last_raw_id is None + assert persisted_before_page.last_blob_hash_hex is None + receipt_directory = store.transactions_root / f"{operation_id}.receipts" + assert not tuple(receipt_directory.glob("pass-*.json")), "a pre-checkpoint kill emitted a false paused receipt" + committed_page = store.next_raw_page(persisted_before_page, limit=8) + committed_page_raw_ids = tuple(row[0] for row in committed_page.rows) + assert len(committed_page_raw_ids) == 8 replay_script = """ import sys from pathlib import Path @@ -427,7 +509,7 @@ def test_sanitized_codex_804_revision_recovery_proof( assert replay_returncode == -9 persisted = store.load_transaction(operation_id) assert persisted.status == "paused" - assert persisted.processed_raw_count > 0 + assert persisted.processed_raw_count == len(committed_page_raw_ids) interrupted_generation = store.load(persisted.generation_id) with sqlite3.connect(interrupted_generation.index_path) as conn: assert int(conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]) > 0 @@ -447,9 +529,21 @@ def test_sanitized_codex_804_revision_recovery_proof( resume_before = _resource_sample(root) resume_started = time.perf_counter() resume_script = """ +import json import sys from pathlib import Path +import polylogue.maintenance.rebuild_index as rebuild_module + +trace = Path(sys.argv[4]) +real_selection = rebuild_module.rebuild_selection_evidence + +def recording_selection(raw_ids, **kwargs): + with trace.open("a", encoding="utf-8") as stream: + stream.write(json.dumps(list(raw_ids), sort_keys=True) + "\\n") + return real_selection(raw_ids, **kwargs) + +rebuild_module.rebuild_selection_evidence = recording_selection from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync root = Path(sys.argv[1]) @@ -470,13 +564,26 @@ def test_sanitized_codex_804_revision_recovery_proof( else: raise SystemExit("persisted rebuild did not reach replayed") """ + selection_trace_path = tmp_path / "rebuild-selection-trace.jsonl" resumed_process = subprocess.run( - [sys.executable, "-c", resume_script, str(root), operation_id, str(schema_inference_receipt_path)], + [ + sys.executable, + "-c", + resume_script, + str(root), + operation_id, + str(schema_inference_receipt_path), + str(selection_trace_path), + ], cwd=Path.cwd(), - check=True, + check=False, capture_output=True, text=True, ) + assert resumed_process.returncode == 0, ( + f"restart subprocess failed: returncode={resumed_process.returncode}; " + f"stdout={resumed_process.stdout}; stderr={resumed_process.stderr}" + ) generation_id = resumed_process.stdout.strip().splitlines()[-1] assert generation_id.startswith("gen-") persisted_after_restart = store.load_transaction(operation_id) @@ -485,6 +592,18 @@ def test_sanitized_codex_804_revision_recovery_proof( assert candidate.state == "inactive" assert Path(candidate.index_path).is_file() assert store.active_pointer.resolve(strict=True) == active_before + resumed_raw_pages = tuple( + tuple(json.loads(line)) for line in selection_trace_path.read_text(encoding="utf-8").splitlines() + ) + assert resumed_raw_pages, "restart observed no production replay selections for the suffix" + resumed_raw_ids = {raw_id for page in resumed_raw_pages for raw_id in page} + with sqlite3.connect(root / "source.db") as conn: + all_raw_ids = { + str(row[0]) for row in conn.execute("SELECT raw_id FROM raw_sessions ORDER BY blob_hash, raw_id") + } + assert all(isinstance(raw_id, str) for page in resumed_raw_pages for raw_id in page) + assert set(committed_page_raw_ids).isdisjoint(resumed_raw_ids) + assert resumed_raw_ids == all_raw_ids - set(committed_page_raw_ids) phases.append( _phase( "postflight", @@ -502,8 +621,7 @@ def test_sanitized_codex_804_revision_recovery_proof( assert max_blob_size == TERMINAL_WIRE_BYTES assert source_path_count == 1 assert parse_error_count == 0 - assert authorities - assert set(authorities) <= {"asserted", "byte_proven", "quarantined"} + assert authorities == ("byte_proven",) assert tuple(row[1] for row in raw_rows) == tuple( _wire_target_bytes(revision) for revision in range(REVISION_COUNT) ) @@ -529,7 +647,31 @@ def test_sanitized_codex_804_revision_recovery_proof( "WHERE logical_source_key IS NOT NULL ORDER BY logical_source_key" ) ) - assert post_recovery_membership_count in {0, REVISION_COUNT} + raw_ids = {str(row[0]) for row in conn.execute("SELECT raw_id FROM raw_sessions")} + authority_counts = tuple( + (str(row[0]), int(row[1])) + for row in conn.execute("SELECT revision_authority, COUNT(*) FROM raw_sessions GROUP BY revision_authority") + ) + membership_rows = tuple( + (str(row[0]), str(row[1]), None if row[2] is None else str(row[2])) + for row in conn.execute( + "SELECT raw_id, revision_authority, decision FROM raw_session_memberships ORDER BY raw_id" + ) + ) + census_rows = tuple( + (str(row[0]), str(row[1])) + for row in conn.execute("SELECT raw_id, status FROM raw_membership_census ORDER BY raw_id") + ) + # This fixture is the byte-revision authority route, so semantic + # membership census remains exactly empty. The application ledger below + # is the production coverage relation for all 804 byte revisions. + assert post_recovery_membership_count == 0 + # Mutation that omits one authority row from the final census fails the + # exact raw, membership, and complete-census conservation below. + assert len(raw_ids) == REVISION_COUNT + assert authority_counts == (("byte_proven", REVISION_COUNT),) + assert membership_rows == () + assert census_rows == () assert len(source_keys) == 1 terminal_logical_source_key = membership_keys[0] if membership_keys else source_keys[0] @@ -565,6 +707,13 @@ def test_sanitized_codex_804_revision_recovery_proof( "SELECT accepted_raw_id FROM raw_revision_heads WHERE logical_source_key = ?", (terminal_logical_source_key,), ).fetchall() + application_rows = conn.execute( + "SELECT raw_id, decision, accepted_raw_id FROM raw_revision_applications ORDER BY raw_id" + ).fetchall() + assert len(application_rows) == REVISION_COUNT + assert {str(row[0]) for row in application_rows} == raw_ids + assert {str(row[1]) for row in application_rows} <= {"selected_baseline", "applied_append", "superseded"} + assert all(row[2] is not None for row in application_rows) assert len(indexed) == 1 assert len(selected_heads) == 1 selected_raw_id = str(selected_heads[0][0]) From 7d34ef10d77a97362865684816af60bf6a97c646 Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 17:44:02 +0200 Subject: [PATCH 2/8] chore: refresh PR scope authority Reissue the exact carrier head after rebasing onto current master. From d4b9bbf318fb7f51eb58486aa51f9e221caac241 Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 18:33:58 +0200 Subject: [PATCH 3/8] test(scenarios): bind Codex recovery proof to frozen source Problem: the Codex 804 proof observed a helper-level selection calculation instead of the production replay path, and its fixture let source remediation occur after the candidate receipt was frozen. What changed: run the production source remediation route in an isolated archive before freezing the candidate receipt, instrument the actual replay seam across resumed pages, assert terminal raw-head ownership, and verify persisted baseline timestamps in the candidate index. Compatibility/migration: this changes only the incident-scale proof fixture; no production archive or migration behavior is changed. Co-Authored-By: Claude --- .../scenarios/test_codex_804_live_proof.py | 52 +++++++++++++++---- 1 file changed, 42 insertions(+), 10 deletions(-) diff --git a/tests/unit/scenarios/test_codex_804_live_proof.py b/tests/unit/scenarios/test_codex_804_live_proof.py index a3bdaf8648..5396974ebf 100644 --- a/tests/unit/scenarios/test_codex_804_live_proof.py +++ b/tests/unit/scenarios/test_codex_804_live_proof.py @@ -19,11 +19,13 @@ import json import os import resource +import shutil import sqlite3 import subprocess import sys import time from dataclasses import replace +from datetime import datetime from functools import cache from pathlib import Path @@ -38,6 +40,7 @@ from polylogue.scenarios.workload import raw_authority_fixed_point_spec from polylogue.schemas.operator.receipt import package_hashes_for_registry from polylogue.schemas.registry import SCHEMA_DIR, SchemaRegistry +from polylogue.sources.revision_backfill import backfill_historical_revision_evidence from polylogue.storage.archive_readiness import raw_materialization_readiness_snapshot from polylogue.storage.index_generation import IndexGenerationStore, rebuild_source_evidence_snapshot from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root @@ -293,7 +296,7 @@ def _readiness_count(readiness: dict[str, object], key: str) -> int: return int(value) -@pytest.mark.timeout(300) +@pytest.mark.timeout(420) @pytest.mark.uses_real_clock("waits for a real subprocess replay checkpoint and kill/resume boundary") def test_sanitized_codex_804_revision_recovery_proof( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest @@ -390,6 +393,18 @@ def test_sanitized_codex_804_revision_recovery_proof( ) ) + # Source remediation is a phase-2 input to candidate construction. Run + # the production remediation route in an isolated archive, then carry its + # finalized durable source tier into this fresh-index candidate fixture. + # This keeps the rebuild receipt frozen after remediation, so a candidate + # replay cannot hide a source mutation behind its own provenance gate. + source_ready_root = tmp_path / "codex-804-source-ready" + initialize_active_archive_root(source_ready_root) + shutil.copy2(root / "source.db", source_ready_root / "source.db") + shutil.copytree(root / "blob", source_ready_root / "blob") + backfill_historical_revision_evidence(source_ready_root, ingest_workers=1, bulk_fts=True) + shutil.copy2(source_ready_root / "source.db", root / "source.db") + schema_inference_receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-gate-receipt.json") store = IndexGenerationStore.for_archive_root(root) active_before = store.active_pointer.resolve(strict=True) @@ -533,17 +548,17 @@ def terminate_before_page_checkpoint(self, transaction, **kwargs): import sys from pathlib import Path -import polylogue.maintenance.rebuild_index as rebuild_module +import polylogue.maintenance.replay as replay_module trace = Path(sys.argv[4]) -real_selection = rebuild_module.rebuild_selection_evidence +real_replay = replay_module.rebuild_index_from_source -def recording_selection(raw_ids, **kwargs): +async def recording_replay(*args, **kwargs): with trace.open("a", encoding="utf-8") as stream: - stream.write(json.dumps(list(raw_ids), sort_keys=True) + "\\n") - return real_selection(raw_ids, **kwargs) + stream.write(json.dumps(list(kwargs["raw_ids"]), sort_keys=True) + "\\n") + return await real_replay(*args, **kwargs) -rebuild_module.rebuild_selection_evidence = recording_selection +replay_module.rebuild_index_from_source = recording_replay from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync root = Path(sys.argv[1]) @@ -564,7 +579,7 @@ def recording_selection(raw_ids, **kwargs): else: raise SystemExit("persisted rebuild did not reach replayed") """ - selection_trace_path = tmp_path / "rebuild-selection-trace.jsonl" + replay_trace_path = tmp_path / "rebuild-replay-trace.jsonl" resumed_process = subprocess.run( [ sys.executable, @@ -573,7 +588,7 @@ def recording_selection(raw_ids, **kwargs): str(root), operation_id, str(schema_inference_receipt_path), - str(selection_trace_path), + str(replay_trace_path), ], cwd=Path.cwd(), check=False, @@ -593,7 +608,7 @@ def recording_selection(raw_ids, **kwargs): assert Path(candidate.index_path).is_file() assert store.active_pointer.resolve(strict=True) == active_before resumed_raw_pages = tuple( - tuple(json.loads(line)) for line in selection_trace_path.read_text(encoding="utf-8").splitlines() + tuple(json.loads(line)) for line in replay_trace_path.read_text(encoding="utf-8").splitlines() ) assert resumed_raw_pages, "restart observed no production replay selections for the suffix" resumed_raw_ids = {raw_id for page in resumed_raw_pages for raw_id in page} @@ -713,6 +728,10 @@ def recording_selection(raw_ids, **kwargs): assert len(application_rows) == REVISION_COUNT assert {str(row[0]) for row in application_rows} == raw_ids assert {str(row[1]) for row in application_rows} <= {"selected_baseline", "applied_append", "superseded"} + accepted_application_ids = {str(row[2]) for row in application_rows if row[2] is not None} + assert len(accepted_application_ids) == 1 + assert accepted_application_ids <= raw_ids + assert accepted_application_ids == {terminal_raw_id} assert all(row[2] is not None for row in application_rows) assert len(indexed) == 1 assert len(selected_heads) == 1 @@ -726,6 +745,19 @@ def recording_selection(raw_ids, **kwargs): ).fetchone() assert selected_source_row == (TERMINAL_WIRE_BYTES, terminal_blob_hash) assert terminal_block_count > 0 + expected_baseline_timestamps = tuple( + int(datetime.fromisoformat(timestamp.replace("Z", "+00:00")).timestamp() * 1000) + for timestamp in _BASELINE_MESSAGE_TIMESTAMPS + ) + with sqlite3.connect(candidate.index_path) as conn: + persisted_baseline_timestamps = tuple( + int(row[0]) + for row in conn.execute( + "SELECT occurred_at_ms FROM messages WHERE message_id LIKE ? ORDER BY message_id", + (f"%{SESSION_NATIVE_ID}-message-%",), + ) + ) + assert persisted_baseline_timestamps == expected_baseline_timestamps quiescent = _resource_sample(root) phases.append( From 28c8b9c83b8e280258108eb87bb36970cdd367a8 Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 18:36:59 +0200 Subject: [PATCH 4/8] chore(test): record Codex recovery timeout budget Problem: the expanded Codex 804 proof exceeded the recorded timeout exception after adding the real source-remediation and replay paths. What changed: update the timeout manifest to the measured 420-second bound and record the source-remediation plus interrupted/resumed replay rationale. Compatibility/migration: test policy metadata only; production behavior is unchanged. Co-Authored-By: Claude --- devtools/pytest_timeout_overrides.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/devtools/pytest_timeout_overrides.toml b/devtools/pytest_timeout_overrides.toml index 8d6dc73875..952c941ef0 100644 --- a/devtools/pytest_timeout_overrides.toml +++ b/devtools/pytest_timeout_overrides.toml @@ -10,5 +10,5 @@ rationale = "The single-process coverage gate needs a bounded diagnostic budget [[exception]] path = "tests/unit/scenarios/test_codex_804_live_proof.py" -value = 300 -rationale = "The sanitized 804-revision production replay includes an approximately 90 MiB terminal wire artifact and needs a bounded incident-scale replay budget." +value = 420 +rationale = "The sanitized 804-revision proof runs source remediation plus an interrupted and resumed production replay; the measured replay takes about 311 seconds and needs a bounded incident-scale budget." From 48b44edf026b8c4a80f68e2f647f3a9c5b6aca2c Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 18:41:01 +0200 Subject: [PATCH 5/8] chore(ci): refresh Codex proof scope carrier Refresh the non-draft PR trigger after the verified Codex 804 proof and exact-head carrier update. No product or test behavior changes.\n\nCo-Authored-By: Claude From 8f87d63287a7c1d4cd079e25d8867dda747a1dc3 Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 19:00:22 +0200 Subject: [PATCH 6/8] test(scenarios): reject duplicate recovery replay Problem: the Codex recovery proof collapsed resumed replay selections into a set, allowing a duplicated suffix raw to pass if every expected raw eventually appeared. What changed: retain the production replay page order, reject duplicate raw IDs, and compare the resumed sequence with the exact source-order suffix after the committed page. Compatibility/migration: proof-only assertion strengthening; production replay behavior is unchanged. Ref polylogue-27522 Co-Authored-By: Claude --- tests/unit/scenarios/test_codex_804_live_proof.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/tests/unit/scenarios/test_codex_804_live_proof.py b/tests/unit/scenarios/test_codex_804_live_proof.py index 5396974ebf..6948757914 100644 --- a/tests/unit/scenarios/test_codex_804_live_proof.py +++ b/tests/unit/scenarios/test_codex_804_live_proof.py @@ -611,14 +611,16 @@ async def recording_replay(*args, **kwargs): tuple(json.loads(line)) for line in replay_trace_path.read_text(encoding="utf-8").splitlines() ) assert resumed_raw_pages, "restart observed no production replay selections for the suffix" - resumed_raw_ids = {raw_id for page in resumed_raw_pages for raw_id in page} + resumed_raw_sequence = tuple(raw_id for page in resumed_raw_pages for raw_id in page) with sqlite3.connect(root / "source.db") as conn: - all_raw_ids = { + all_raw_sequence = tuple( str(row[0]) for row in conn.execute("SELECT raw_id FROM raw_sessions ORDER BY blob_hash, raw_id") - } + ) assert all(isinstance(raw_id, str) for page in resumed_raw_pages for raw_id in page) - assert set(committed_page_raw_ids).isdisjoint(resumed_raw_ids) - assert resumed_raw_ids == all_raw_ids - set(committed_page_raw_ids) + assert len(resumed_raw_sequence) == len(set(resumed_raw_sequence)), "restart replayed a raw more than once" + assert len(all_raw_sequence) == len(set(all_raw_sequence)) + assert set(committed_page_raw_ids).isdisjoint(set(resumed_raw_sequence)) + assert resumed_raw_sequence == all_raw_sequence[len(committed_page_raw_ids) :] phases.append( _phase( "postflight", From f11f12d69e3edfbd5e6fc8d8722fdd9437afedef Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 19:49:17 +0200 Subject: [PATCH 7/8] test(reindex): measure Codex recovery envelope Problem: The recovery receipt started measuring after source remediation and the pre-checkpoint crash boundary, while the test did not prove that the killed process had already materialized candidate work.\n\nWhat changed: Extend the existing replay phase over source remediation and the crash boundary, and assert the inactive generation contains the first page immediately after the pre-checkpoint kill.\n\nCompatibility/migration: The workload phase contract remains unchanged.\n\nCo-Authored-By: Claude --- tests/unit/scenarios/test_codex_804_live_proof.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/unit/scenarios/test_codex_804_live_proof.py b/tests/unit/scenarios/test_codex_804_live_proof.py index 6948757914..613eb28c3e 100644 --- a/tests/unit/scenarios/test_codex_804_live_proof.py +++ b/tests/unit/scenarios/test_codex_804_live_proof.py @@ -398,6 +398,10 @@ def test_sanitized_codex_804_revision_recovery_proof( # finalized durable source tier into this fresh-index candidate fixture. # This keeps the rebuild receipt frozen after remediation, so a candidate # replay cannot hide a source mutation behind its own provenance gate. + # The replay phase intentionally begins before source remediation and the + # crash boundary so its receipt covers the complete recovery envelope. + replay_before = _resource_sample(root) + replay_started = time.perf_counter() source_ready_root = tmp_path / "codex-804-source-ready" initialize_active_archive_root(source_ready_root) shutil.copy2(root / "source.db", source_ready_root / "source.db") @@ -471,6 +475,9 @@ def terminate_before_page_checkpoint(self, transaction, **kwargs): assert persisted_before_page.last_blob_hash_hex is None receipt_directory = store.transactions_root / f"{operation_id}.receipts" assert not tuple(receipt_directory.glob("pass-*.json")), "a pre-checkpoint kill emitted a false paused receipt" + precheckpoint_generation = store.load(persisted_before_page.generation_id) + with sqlite3.connect(precheckpoint_generation.index_path) as conn: + assert int(conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]) > 0 committed_page = store.next_raw_page(persisted_before_page, limit=8) committed_page_raw_ids = tuple(row[0] for row in committed_page.rows) assert len(committed_page_raw_ids) == 8 @@ -494,8 +501,6 @@ def terminate_before_page_checkpoint(self, transaction, **kwargs): ) print(result.status) """ - replay_before = _resource_sample(root) - replay_started = time.perf_counter() replay_process = subprocess.Popen( [sys.executable, "-c", replay_script, str(root), operation_id, str(schema_inference_receipt_path)], cwd=Path.cwd(), From ad18a3858c9ef8263e50b15d0bb72e6e7214fb10 Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 7 Aug 2026 20:19:52 +0200 Subject: [PATCH 8/8] chore(ci): refresh Codex proof carrier Problem: The structured PR scope carrier was refreshed after the last proof commit, but the CI status still belongs to the preceding head. What changed: Create a new verification boundary so Circle and automated review evaluate the exact carrier-bound proof head. Compatibility/migration: No production or archive mutation.