diff --git a/devtools/command_catalog.py b/devtools/command_catalog.py index 68770fba08..6449868323 100644 --- a/devtools/command_catalog.py +++ b/devtools/command_catalog.py @@ -606,7 +606,7 @@ def to_dict(self) -> dict[str, object]: ), examples=( "devtools workspace raw-authority-scale-proof --json", - "devtools workspace raw-authority-scale-proof --components 10163 --raws 15264 --pass-limit 64 --keep --json", + "devtools workspace raw-authority-scale-proof --components 10163 --raws 15264 --expanded-raws 21398 --pass-limit 64 --keep --json", ), ), CommandSpec( diff --git a/devtools/raw_authority_scale_proof.py b/devtools/raw_authority_scale_proof.py index 607fe06257..aeb9754e15 100644 --- a/devtools/raw_authority_scale_proof.py +++ b/devtools/raw_authority_scale_proof.py @@ -19,6 +19,7 @@ import sqlite3 import tempfile import time +from collections import Counter from dataclasses import asdict, dataclass from pathlib import Path from typing import TextIO, cast @@ -258,12 +259,16 @@ def _write_payload(path: Path, *, native_id: str, revision: int, target_size: in remaining -= amount -def _independent_payload(*, native_id: str, target_size: int) -> bytes: - """Build one bounded standalone JSONL raw without a disk staging file.""" - header = f'{{"type":"session_meta","payload":{{"id":"{native_id}","timestamp":"2026-07-15T00:00:00Z"}}}}\n'.encode() - if target_size < len(header): - raise ValueError("scenario payload allocation cannot preserve valid JSONL evidence") - return header + (b" " * (target_size - len(header))) +def _explicit_component_cohorts( + *, components: int, direct_candidates: int, expanded_candidates: int +) -> tuple[tuple[int, int, int], ...]: + """Build an exact aggregate topology without inventing an unexpanded corpus.""" + direct = _component_counts(direct_candidates, components=components) + expanded = _component_counts(expanded_candidates, components=components) + return tuple( + (raw_count, direct_count, count) + for (raw_count, direct_count), count in sorted(Counter(zip(expanded, direct, strict=True)).items()) + ) def _component_counts(total: int, *, components: int) -> list[int]: @@ -507,6 +512,12 @@ def _record_repair_pass( wall_ms = int((time.perf_counter() - started) * 1000) after = _process_sample() metrics = result.metrics + hard_failure_metrics = ( + "raw_materialization_plan_conservation_error_count", + "raw_materialization_unresolved_blocker_count", + ) + if not result.success and any(float(metrics.get(key, 0)) > 0 for key in hard_failure_metrics): + raise RuntimeError(f"raw-authority scale proof repair pass failed: {result.detail}") candidate_value = metrics.get("raw_materialization_candidate_count") executable_candidate_value = metrics.get("raw_materialization_executable_candidate_count", candidate_value) if ( @@ -566,6 +577,7 @@ def run_raw_authority_scale_proof( *, components: int = 16, raws: int = 24, + expanded_raws: int | None = None, scenario: RawAuthorityScaleScenario | None = None, pass_limit: int = 4, keep: bool = False, @@ -582,11 +594,17 @@ def run_raw_authority_scale_proof( if scenario is None: if components < 1 or raws < components: raise ValueError("require components >= 1 and raws >= components") + expanded = raws if expanded_raws is None else expanded_raws scenario = RawAuthorityScaleScenario( components=components, direct_candidates=raws, - expanded_candidates=raws, - total_payload_bytes=raws * 1024, + expanded_candidates=expanded, + total_payload_bytes=expanded * 1024, + component_cohorts=( + _explicit_component_cohorts(components=components, direct_candidates=raws, expanded_candidates=expanded) + if expanded != raws + else None + ), ) if components != 16 and components != scenario.components: raise ValueError("components and scenario.components disagree") @@ -601,6 +619,7 @@ def run_raw_authority_scale_proof( max_memory_full_avg10=max_memory_full_avg10, ) generation_samples = [admission_sample] + replay_samples: list[ProcessSample] = [] def check_generation_pressure() -> None: sample = _process_sample() @@ -611,6 +630,15 @@ def check_generation_pressure() -> None: ) generation_samples.append(sample) + def check_replay_pressure() -> None: + sample = _process_sample() + _assert_admission( + sample, + max_io_full_avg10=max_io_full_avg10, + max_memory_full_avg10=max_memory_full_avg10, + ) + replay_samples.append(sample) + root = workdir.expanduser().resolve() / "raw-authority-scale-proof" if root.exists(): shutil.rmtree(root) @@ -647,21 +675,16 @@ def check_generation_pressure() -> None: if not _uses_independent_component_members(scenario) else f"{session_native_id}-member-{member:05d}" ) - if _uses_independent_component_members(scenario): - blob_hash, blob_size = publisher.write_from_bytes( - _independent_payload(native_id=row_native_id, target_size=row_size) - ) - else: - payload_path = temporary_root / f"{source_label}.jsonl" - _write_payload( - payload_path, - native_id=row_native_id, - revision=member, - target_size=row_size, - previous=previous, - ) - blob_hash, blob_size = publisher.write_from_path(payload_path) - payload_path.unlink() + payload_path = temporary_root / f"{source_label}.jsonl" + _write_payload( + payload_path, + native_id=row_native_id, + revision=member, + target_size=row_size, + previous=None if _uses_independent_component_members(scenario) else previous, + ) + blob_hash, blob_size = publisher.write_from_path(payload_path) + payload_path.unlink() previous = publisher.blob_path(blob_hash) source_path = component_source_path pending_rows.append((row_native_id, source_path, blob_hash, blob_size, terminalized, component)) @@ -793,6 +816,7 @@ def check_generation_pressure() -> None: ) pass_receipts: list[RawAuthorityScalePass] = [] for number in range(1, (scenario.components * 3) + 4): + check_replay_pressure() pass_receipt, _digest = _record_repair_pass( number=number, mode="apply", @@ -807,6 +831,7 @@ def check_generation_pressure() -> None: raise RuntimeError("raw-authority scale proof did not drain bounded apply passes") fixed_point_digests: list[str] = [] for _ in range(2): + check_replay_pressure() pass_receipt, digest = _record_repair_pass( number=len(pass_receipts) + 1, mode="dry_run", @@ -878,6 +903,7 @@ def check_generation_pressure() -> None: "achieved_shape": achieved_shape, "admission_sample": asdict(admission_sample), "generation_samples": [asdict(sample) for sample in generation_samples], + "replay_samples": [asdict(sample) for sample in replay_samples], "passes": [asdict(item) for item in pass_receipts], "fixed_point_digests": fixed_point_digests, "receipt": receipt.to_payload(), @@ -893,6 +919,12 @@ def main(argv: list[str] | None = None, *, stdout: TextIO | None = None) -> int: parser.add_argument("--workdir", type=Path, default=Path(".cache") / "raw-authority-scale-proof") parser.add_argument("--components", type=int, default=16) parser.add_argument("--raws", type=int, default=24) + parser.add_argument( + "--expanded-raws", + type=int, + default=None, + help="Exact expanded authority candidates; defaults to --raws when omitted.", + ) parser.add_argument( "--scenario-profile", type=Path, @@ -938,6 +970,7 @@ def main(argv: list[str] | None = None, *, stdout: TextIO | None = None) -> int: args.workdir, components=args.components, raws=args.raws, + expanded_raws=args.expanded_raws, scenario=scenario, pass_limit=args.pass_limit, keep=args.keep, diff --git a/polylogue/storage/archive_readiness.py b/polylogue/storage/archive_readiness.py index 0c6f86dcb8..0a0948ef64 100644 --- a/polylogue/storage/archive_readiness.py +++ b/polylogue/storage/archive_readiness.py @@ -107,6 +107,7 @@ def raw_materialization_ready(readiness: Mapping[str, Any] | object | None) -> b "unchecked", "affected_unchecked", "raw_authority_frontier_blocking_count", + "raw_authority_blocker_count", "raw_authority_pending_census_count", ) return all(_read_int(readiness, key) == 0 for key in blocking_keys) diff --git a/polylogue/storage/raw_authority.py b/polylogue/storage/raw_authority.py index f20ad75fe7..a9e2df841b 100644 --- a/polylogue/storage/raw_authority.py +++ b/polylogue/storage/raw_authority.py @@ -778,6 +778,32 @@ def record_raw_authority_census( scope_json = _canonical_json(scope) residual_json = _canonical_json(residual) with closing(sqlite3.connect(archive_root / "source.db")) as conn, conn: + # A frontier preview authorizes an immutable execution exactly once. + # This runs in the census INSERT transaction, so two offline callers + # cannot both turn one unchanged preview into execution receipts. + if mode == "apply" and scope.get("schema") == "polylogue.raw-authority-frontier-scope.v1": + preview_census_id = scope.get("preview_census_id") + if isinstance(preview_census_id, str) and preview_census_id and selected_plan_ids: + placeholders = ",".join("?" for _ in selected_plan_ids) + duplicate = conn.execute( + f""" + SELECT existing.census_id + FROM raw_authority_censuses AS existing + JOIN raw_authority_census_plans AS cp ON cp.census_id = existing.census_id + WHERE existing.mode = 'apply' + AND json_extract(existing.scope_json, '$.schema') = + 'polylogue.raw-authority-frontier-scope.v1' + AND json_extract(existing.scope_json, '$.preview_census_id') = ? + AND cp.selected = 1 + AND cp.plan_id IN ({placeholders}) + LIMIT 1 + """, + (preview_census_id, *sorted(selected_plan_ids)), + ).fetchone() + if duplicate is not None: + raise RuntimeError( + f"raw authority frontier preview plan is already claimed by census {duplicate[0]}" + ) previous = conn.execute( """ SELECT census_id, sequence_no, inventory_digest, residual_digest, @@ -1367,8 +1393,8 @@ def recover_interrupted_raw_authority_censuses( JOIN raw_authority_plans AS p ON p.plan_id = cp.plan_id WHERE c.lifecycle_status = 'planned' AND cp.selected = 1 AND cp.outcome_recorded = 0 - AND COALESCE(json_extract(p.authority_witness_json, '$.schema'), '') != - 'polylogue.raw-authority-frontier-plan.v1' + AND COALESCE(json_extract(c.scope_json, '$.schema'), '') != + 'polylogue.raw-authority-frontier-scope.v1' ORDER BY c.sequence_no, cp.ordinal """ ).fetchall() @@ -1379,15 +1405,8 @@ def recover_interrupted_raw_authority_censuses( SELECT census_id, scope_json FROM raw_authority_censuses WHERE lifecycle_status = 'planned' - AND EXISTS ( - SELECT 1 - FROM raw_authority_census_plans AS cp - JOIN raw_authority_plans AS p ON p.plan_id = cp.plan_id - WHERE cp.census_id = raw_authority_censuses.census_id - AND cp.selected = 1 AND cp.outcome_recorded = 0 - AND COALESCE(json_extract(p.authority_witness_json, '$.schema'), '') != - 'polylogue.raw-authority-frontier-plan.v1' - ) + AND COALESCE(json_extract(scope_json, '$.schema'), '') != + 'polylogue.raw-authority-frontier-scope.v1' ORDER BY sequence_no """ ) diff --git a/polylogue/storage/raw_reconciler.py b/polylogue/storage/raw_reconciler.py index 3b28135e53..ba2b891475 100644 --- a/polylogue/storage/raw_reconciler.py +++ b/polylogue/storage/raw_reconciler.py @@ -1381,22 +1381,13 @@ def recover_interrupted_raw_authority_frontier(config: Config) -> tuple[str, ... receipt, ) record_raw_replay_outcome(root, census_id, outcome) - elif related and all( - item.state in {RawAuthorityFrontierState.PROVEN_CURRENT, RawAuthorityFrontierState.SUPERSEDED} - for item in related - ): - outcome = RawReplayPlanOutcome( - plan.plan_id, - plan.input_raw_ids, - RawReplayPlanStatus.EXECUTED, - "interrupted application recovered from typed terminal frontier states", - "none", - receipt, - ) - record_raw_replay_outcome(root, census_id, outcome) else: from polylogue.storage.raw_authority import reject_stale_raw_replay_plan + # Terminal-looking frontier items do not prove this exact immutable + # strategy ran: ordinary ingest may have terminalized only part of + # a multi-input plan. Without its strategy receipt, never mint an + # EXECUTED outcome during crash recovery. reject_stale_raw_replay_plan(root, census_id, plan, receipt) recovered.append(plan.plan_id) post_state_counts = _state_counts(current_items) diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py index 0d2710e530..0644cffcf6 100644 --- a/polylogue/storage/repair.py +++ b/polylogue/storage/repair.py @@ -3768,8 +3768,11 @@ def _raw_materialization_candidate_ids( continue if bool(row["application_terminal"]): continue - if bool(row["membership_authority_complete"]): - continue + # Membership authority describes how to replay a shared raw; it is + # not evidence that the rebuildable index still contains the + # governed sessions. The candidate query has already proved this + # raw has no materialized index row, so retain complete censuses + # as replay inputs after an index reset. if bool(row["membership_authority_quarantined"]): authority_quarantined += 1 authority_quarantined_raw_ids.append(row_raw_id) @@ -3920,6 +3923,7 @@ def _raw_authority_scope( source_family: str | None, source_root: Path | None, raw_artifact_limit: int | None, + max_payload_bytes: int, ) -> dict[str, object]: return { "raw_artifact_id": raw_artifact_id, @@ -3927,6 +3931,7 @@ def _raw_authority_scope( "source_family": source_family, "source_root": str(source_root) if source_root is not None else None, "raw_artifact_limit": raw_artifact_limit, + "max_payload_bytes": max_payload_bytes, } @@ -3959,6 +3964,8 @@ def _raw_authority_residual( def _raw_authority_postflight_snapshot( archive_root: Path, candidates: RawMaterializationCandidates, + *, + max_payload_bytes: int, ) -> tuple[tuple[RawReplayPlan, ...], dict[str, object]]: """Build the complete post-pass plan inventory and typed residual debt.""" components = _raw_materialization_ordered_components(candidates, archive_root=archive_root) @@ -3968,7 +3975,7 @@ def _raw_authority_postflight_snapshot( plan.plan_id for component, plan in zip(components, plans, strict=True) if sum(_raw_materialization_component_blob_bytes(candidates, member) for member in component) - > RAW_MATERIALIZATION_EXECUTE_BLOB_LIMIT_BYTES + > max_payload_bytes ) ) return plans, _raw_authority_residual(candidates, resource_blocked_plan_ids=blocked_plan_ids) @@ -5571,10 +5578,15 @@ def repair_raw_materialization( archive_root = _raw_materialization_archive_root(config) recovered_censuses = recover_interrupted_raw_authority_censuses(archive_root) for recovered_census_id, recovered_scope in recovered_censuses: + recovered_envelope = recovered_scope.get("max_payload_bytes") + recovered_max_payload_bytes = ( + recovered_envelope if isinstance(recovered_envelope, int) and recovered_envelope > 0 else max_payload_bytes + ) recovered_candidates = _raw_authority_candidates_for_scope(config, recovered_scope) recovered_post_plans, recovered_post_residual = _raw_authority_postflight_snapshot( archive_root, recovered_candidates, + max_payload_bytes=recovered_max_payload_bytes, ) finalize_raw_authority_census( archive_root, @@ -5682,6 +5694,7 @@ def repair_raw_materialization( source_family=source_family, source_root=source_root, raw_artifact_limit=raw_artifact_limit, + max_payload_bytes=max_payload_bytes, ), residual=residual, ) @@ -5753,13 +5766,13 @@ def repair_raw_materialization( ] all_blocked_component_raw_ids = {raw_id for component in all_blocked_components for raw_id in component} resource_blocked_candidate_raw_ids = set(candidate_raw_ids).intersection(all_blocked_component_raw_ids) - selected_components = ( - ordered_components[:raw_artifact_limit] if raw_artifact_limit is not None else ordered_components - ) deferred_plan_ids = raw_replay_plan_deferred_for_envelope(archive_root, max_payload_bytes=max_payload_bytes) - selected_components = [ - component for component in selected_components if plan_by_component[component].plan_id not in deferred_plan_ids + admissible_components = [ + component for component in ordered_components if plan_by_component[component].plan_id not in deferred_plan_ids ] + selected_components = ( + admissible_components[:raw_artifact_limit] if raw_artifact_limit is not None else admissible_components + ) blocked_components = [ component for component in selected_components if all_blocked_component_raw_ids.intersection(component) ] @@ -5837,9 +5850,17 @@ def repair_raw_materialization( source_family=source_family, source_root=source_root, raw_artifact_limit=raw_artifact_limit, + max_payload_bytes=max_payload_bytes, ) if not dry_run and not selected_components and deferred_plan_ids: retained_census_receipt = latest_raw_authority_census_receipt(archive_root, scope=scope) + if retained_census_receipt is None: + # v1 receipts predate envelope identity. They are safe to reuse + # only in this all-deferred branch: the persisted deferred-plan + # reason was already matched against this active envelope above. + legacy_scope = dict(scope) + legacy_scope.pop("max_payload_bytes") + retained_census_receipt = latest_raw_authority_census_receipt(archive_root, scope=legacy_scope) if retained_census_receipt is None: raise RuntimeError("resource-deferred raw replay lacks a completed durable census receipt") return _internal_derived_repair_result( @@ -5994,6 +6015,7 @@ def repair_raw_materialization( stale_post_plans, stale_post_residual = _raw_authority_postflight_snapshot( archive_root, stale_candidates, + max_payload_bytes=max_payload_bytes, ) census_receipt = finalize_raw_authority_census( archive_root, @@ -6158,7 +6180,9 @@ def repair_raw_materialization( } ) plan_outcomes = tuple(execution_outcomes) + blocked_plan_outcomes - post_plans, post_residual = _raw_authority_postflight_snapshot(archive_root, remaining) + post_plans, post_residual = _raw_authority_postflight_snapshot( + archive_root, remaining, max_payload_bytes=max_payload_bytes + ) census_receipt = finalize_raw_authority_census( archive_root, census_receipt.census_id, diff --git a/tests/unit/devtools/test_raw_authority_scale_proof.py b/tests/unit/devtools/test_raw_authority_scale_proof.py index 83afb781de..575009129d 100644 --- a/tests/unit/devtools/test_raw_authority_scale_proof.py +++ b/tests/unit/devtools/test_raw_authority_scale_proof.py @@ -94,6 +94,46 @@ def test_raw_authority_scale_proof_rechecks_pressure_during_generation( ) +def test_raw_authority_scale_proof_rechecks_pressure_during_replay( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + samples = iter( + ( + ProcessSample(1, 1, 0, 0, 0, 0, 0.0, 0.0), + ProcessSample(1, 1, 0, 0, 0, 0, 0.0, 0.0), + ProcessSample(1, 1, 0, 0, 0, 0, 0.0, 0.0), + ProcessSample(1, 1, 0, 0, 0, 0, 2.1, 0.0), + ) + ) + monkeypatch.setattr("devtools.raw_authority_scale_proof._process_sample", lambda: next(samples)) + + with pytest.raises(RuntimeError, match="I/O pressure gate"): + run_raw_authority_scale_proof(tmp_path, components=1, raws=1, max_io_full_avg10=2.0, max_memory_full_avg10=None) + + +def test_raw_authority_scale_proof_rejects_unsuccessful_production_pass( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setattr( + repair, + "repair_raw_materialization", + lambda *_args, **_kwargs: type( + "Failure", + (), + { + "success": False, + "detail": "conservation mismatch", + "metrics": {"raw_materialization_plan_conservation_error_count": 1}, + }, + )(), + ) + + with pytest.raises(RuntimeError, match="repair pass failed: conservation mismatch"): + run_raw_authority_scale_proof( + tmp_path, components=1, raws=1, max_io_full_avg10=None, max_memory_full_avg10=None + ) + + def test_raw_authority_scale_proof_consumes_reservations_and_has_stable_corpus_identity(tmp_path: Path) -> None: first = run_raw_authority_scale_proof( tmp_path / "first", @@ -373,7 +413,7 @@ def test_raw_authority_scale_proof_preserves_private_free_joint_byte_cohorts(tmp assert int(largest_blob[0]) < 4_096 -def test_explicit_cohorts_publish_bounded_payloads_without_disk_staging( +def test_explicit_cohorts_stream_bounded_payloads_without_allocating_full_bytes( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: scenario = RawAuthorityScaleScenario( @@ -385,10 +425,10 @@ def test_explicit_cohorts_publish_bounded_payloads_without_disk_staging( component_byte_cohorts=((2, 2, 2048, 1),), ) - def reject_staged_publication(*_args: object, **_kwargs: object) -> tuple[str, int]: - raise AssertionError("explicit cohorts must publish bytes without a staged payload path") + def reject_full_payload(*_args: object, **_kwargs: object) -> tuple[str, int]: + raise AssertionError("explicit cohorts must not allocate a full payload bytes object") - monkeypatch.setattr(ArchiveBlobPublisher, "write_from_path", reject_staged_publication) + monkeypatch.setattr(ArchiveBlobPublisher, "write_from_bytes", reject_full_payload) payload = run_raw_authority_scale_proof( tmp_path, scenario=scenario, diff --git a/tests/unit/storage/test_archive_readiness.py b/tests/unit/storage/test_archive_readiness.py index b5b646c7e0..a01920eb6c 100644 --- a/tests/unit/storage/test_archive_readiness.py +++ b/tests/unit/storage/test_archive_readiness.py @@ -38,6 +38,16 @@ def test_raw_materialization_readiness_requires_completed_frontier_census() -> N ) +def test_raw_materialization_readiness_rejects_unresolved_authority_blockers() -> None: + readiness = { + "available": True, + "raw_authority_frontier": {"lifecycle_status": "completed"}, + "raw_authority_blocker_count": 1, + } + + assert raw_materialization_ready(readiness) is False + + def test_readiness_uses_frontier_postflight_not_preapply_scope(tmp_path: Path) -> None: """An applied repair must not remain blocked by its immutable preflight.""" initialize_active_archive_root(tmp_path) diff --git a/tests/unit/storage/test_raw_authority_ledger.py b/tests/unit/storage/test_raw_authority_ledger.py index 63e151eda2..d28b1adbdf 100644 --- a/tests/unit/storage/test_raw_authority_ledger.py +++ b/tests/unit/storage/test_raw_authority_ledger.py @@ -26,6 +26,7 @@ read_raw_authority_detail, record_raw_authority_census, record_raw_replay_outcome, + recover_interrupted_raw_authority_censuses, reject_stale_raw_replay_plan, resolve_raw_authority_blocker, validate_raw_replay_plan, @@ -731,6 +732,58 @@ def test_identical_stale_rejection_after_resolution_creates_new_open_blocker(tmp assert cast(dict[str, object], blockers[0])["blocker_id"] == second_blocker +def test_recovery_returns_planned_census_after_all_outcomes_are_recorded(tmp_path: Path) -> None: + """A crash after outcome commit still needs durable postflight finalization.""" + initialize_active_archive_root(tmp_path) + plan = RawReplayPlan( + plan_id="raw-replay:outcome-recorded", + input_digest="a" * 64, + input_raw_ids=("raw-outcome-recorded",), + logical_keys=("codex:outcome-recorded",), + authority_witness={}, + source_preconditions={}, + index_preconditions={}, + ) + census = record_raw_authority_census( + tmp_path, + (plan,), + selected_plan_ids={plan.plan_id}, + mode="apply", + quiescent=True, + scope={"test": "outcome-recorded"}, + residual={}, + ) + record_raw_replay_outcome( + tmp_path, + census.census_id, + RawReplayPlanOutcome(plan.plan_id, plan.input_raw_ids, RawReplayPlanStatus.EXECUTED, "done", "none"), + ) + + assert recover_interrupted_raw_authority_censuses(tmp_path) == ((census.census_id, {"test": "outcome-recorded"}),) + + +def test_frontier_preview_cannot_claim_one_plan_twice(tmp_path: Path) -> None: + initialize_active_archive_root(tmp_path) + plan = RawReplayPlan( + plan_id="raw-authority-frontier:" + "a" * 64, + input_digest="b" * 64, + input_raw_ids=("raw-once",), + logical_keys=("codex:once",), + authority_witness={"schema": "polylogue.raw-authority-frontier-plan.v1"}, + source_preconditions={}, + index_preconditions={}, + ) + scope = {"schema": "polylogue.raw-authority-frontier-scope.v1", "preview_census_id": "preview:once"} + record_raw_authority_census( + tmp_path, (plan,), selected_plan_ids={plan.plan_id}, mode="apply", quiescent=True, scope=scope, residual={} + ) + + with pytest.raises(RuntimeError, match="already claimed"): + record_raw_authority_census( + tmp_path, (plan,), selected_plan_ids={plan.plan_id}, mode="apply", quiescent=True, scope=scope, residual={} + ) + + def test_fixed_point_compares_residual_identity_and_parser_fingerprint(tmp_path: Path) -> None: initialize_active_archive_root(tmp_path) first = record_raw_authority_census( diff --git a/tests/unit/storage/test_repair.py b/tests/unit/storage/test_repair.py index 2d8c925ea6..7f89292126 100644 --- a/tests/unit/storage/test_repair.py +++ b/tests/unit/storage/test_repair.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import sqlite3 from collections.abc import Iterator, Sequence from contextlib import contextmanager @@ -834,7 +835,7 @@ def test_raw_materialization_receipts_partition_terminal_deferred_and_executable assert candidates.adoption_deferred == 1 -def test_raw_materialization_retires_only_complete_governed_bundle_membership(tmp_path: Path) -> None: +def test_raw_materialization_replays_complete_governed_bundle_membership_after_index_loss(tmp_path: Path) -> None: config = _config(tmp_path) initialize_archive_database(tmp_path / "source.db", ArchiveTier.SOURCE) initialize_archive_database(tmp_path / "index.db", ArchiveTier.INDEX) @@ -892,7 +893,7 @@ def test_raw_materialization_retires_only_complete_governed_bundle_membership(tm candidates = repair_mod._raw_materialization_candidate_ids(config) - assert set(candidates.raw_ids) == {raw_ids[0], raw_ids[2]} + assert set(candidates.raw_ids) == {raw_ids[0], raw_ids[2], raw_ids[3]} assert candidates.authority_quarantined == 1 @@ -1852,6 +1853,52 @@ def test_raw_materialization_blocks_aggregate_sub_limit_cohort_before_blob_open( assert "aggregate payload exceeds 1.0 GiB" in result.detail +def test_raw_materialization_reuses_pre_envelope_deferred_receipt(tmp_path: Path) -> None: + """Upgrading does not strand a completed deferred receipt without envelope identity.""" + from polylogue.core.enums import Provider + from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=b'{"type":"session_meta","payload":{"id":"legacy-deferred"}}\n', + source_path="legacy-deferred.jsonl", + acquired_at_ms=1, + ) + with sqlite3.connect(tmp_path / "source.db") as conn: + conn.execute( + "UPDATE raw_sessions SET blob_size = ? WHERE raw_id = ?", + (repair_mod.RAW_MATERIALIZATION_EXECUTE_BLOB_LIMIT_BYTES + 1, raw_id), + ) + conn.commit() + census_historical_revision_evidence(tmp_path, selected_raw_ids=[raw_id]) + first = repair_mod.repair_raw_materialization(_config(tmp_path), raw_artifact_limit=1) + assert first.census_receipt is not None + with sqlite3.connect(tmp_path / "source.db") as conn: + scope = json.loads( + str( + conn.execute( + "SELECT scope_json FROM raw_authority_censuses WHERE census_id = ?", + (first.census_receipt.census_id,), + ).fetchone()[0] + ) + ) + scope.pop("max_payload_bytes") + conn.execute( + "UPDATE raw_authority_censuses SET scope_json = ? WHERE census_id = ?", + (json.dumps(scope, sort_keys=True, separators=(",", ":")), first.census_receipt.census_id), + ) + conn.commit() + + repeated = repair_mod.repair_raw_materialization(_config(tmp_path), raw_artifact_limit=1) + + assert repeated.success is True + assert repeated.census_receipt is not None + assert repeated.census_receipt.census_id == first.census_receipt.census_id + + def test_raw_materialization_processes_independent_components_across_bounded_passes(tmp_path: Path) -> None: from polylogue.core.enums import Provider from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore