From 6df3972c673f219af44a6e316dc597957b989f3e Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 31 Jul 2026 23:30:24 +0200 Subject: [PATCH 1/4] fix(storage): gate ambiguous-verdict terminality by classifier fingerprint Problem: RAW_AUTHORITY_PARSER_FINGERPRINT existed as a proper constant in storage/raw_authority.py but sources/revision_backfill.py hardcoded the literal "revision-membership-v1" eight times instead of importing it, so bumping the constant would half-apply. Separately, storage/repair.py's terminal-decision check (~4432-4448) treated any persisted decision = 'ambiguous' row as durable terminal debt with no way to distinguish "ambiguous under the current classifier" from "ambiguous under a classifier we have since corrected" -- every improvement to classify_membership_revisions was therefore inert on existing data (polylogue-bu1i fixed 157/157 live aistudio-drive cohorts going forward, but the persisted ambiguous verdicts for those cohorts stayed terminal forever). Solution: - revision_backfill.py now imports RAW_AUTHORITY_PARSER_FINGERPRINT instead of repeating the literal at all 8 call sites (fingerprint writes to raw_authority_parser_census/raw_membership_census, and the resource-blocked-envelope fingerprint helper). - The quiescence gate (uncensused_historical_revision_raw_ids) now accepts ANY known fingerprint (current or superseded), not only the current one -- a bump answers "is this verdict still authoritative?", not "was this raw ever observed by a real parser?", so it must not force a full archive re-census. - repair.py's terminal-ambiguous query (covering both index_tier.raw_revision_applications and raw_session_memberships) now LEFT JOINs raw_authority_parser_census and excludes a raw from the terminal gate when its census fingerprint is listed in the new SUPERSEDED_MEMBERSHIP_FINGERPRINTS set. A raw with no census row (or a current-fingerprint row) stays conservative and remains terminal. - Bumped RAW_AUTHORITY_PARSER_FINGERPRINT to "revision-membership-v2" in the same commit as the gating (a bare bump alone would force a ~4h20m full reparse; the gating is what makes a bump targeted). Verification: devtools test tests/unit/storage/test_raw_authority_ledger.py tests/unit/storage/test_archive_readiness.py tests/unit/storage/test_revision_replay.py tests/unit/storage/test_quarantined_accepted_raw_repair.py -> 99 passed devtools test tests/unit/sources/test_revision_backfill.py -> 55 passed, 1 pre-existing failure unrelated to this change (test_parse_one_still_replays_real_claude_code_sessions_with_no_path_rule, a content-classification gate assertion untouched by this diff; reproduces identically on unmodified HEAD) devtools test tests/unit/storage/test_incremental_rebuild_equivalence.py -> 1 passed Anti-vacuity: test_ambiguous_verdict_under_superseded_fingerprint_is_replayable exercises repair._raw_replay_plan_outcome via the public build_raw_replay_plans/_raw_replay_plan_outcomes pair used by repair_raw_materialization (the daemon's live raw-materialization repair entrypoint). Reverting the LEFT JOIN + json_each NOT COALESCE guard in repair.py's terminal query makes this test fail by reclassifying the plan TERMINAL. Ref polylogue-9dxn --- polylogue/sources/revision_backfill.py | 53 ++++++++--- polylogue/storage/raw_authority.py | 18 +++- polylogue/storage/repair.py | 40 +++++++- tests/unit/sources/test_live_batch_support.py | 3 +- tests/unit/sources/test_revision_backfill.py | 10 +- tests/unit/storage/test_archive_readiness.py | 3 +- .../test_incremental_rebuild_equivalence.py | 3 +- .../test_quarantined_accepted_raw_repair.py | 9 +- .../unit/storage/test_raw_authority_ledger.py | 94 ++++++++++++++++++- tests/unit/storage/test_revision_replay.py | 5 +- 10 files changed, 203 insertions(+), 35 deletions(-) diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index fd65f31832..8c10a3ea3d 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -45,6 +45,10 @@ from polylogue.sources.parsers import hermes_state, hermes_verification from polylogue.sources.parsers.base import ParsedSession from polylogue.sources.sqlite_snapshot import looks_like_sqlite_bytes +from polylogue.storage.raw_authority import ( + RAW_AUTHORITY_PARSER_FINGERPRINT, + SUPERSEDED_MEMBERSHIP_FINGERPRINTS, +) from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.write import PreparedSessionRows, prepare_session_rows @@ -320,7 +324,7 @@ def __init__(self, raw_ids: list[str], limit_bytes: int, total_bytes: int) -> No def _resource_blocked_parser_fingerprint(max_payload_bytes: int) -> str: """Return the durable admission identity for one bounded census envelope.""" - return f"revision-membership-v1:resource-blocked:{max_payload_bytes}" + return f"{RAW_AUTHORITY_PARSER_FINGERPRINT}:resource-blocked:{max_payload_bytes}" def uncensused_historical_revision_raw_ids( @@ -331,10 +335,21 @@ def uncensused_historical_revision_raw_ids( ) -> tuple[str, ...]: """Return inputs whose current parser identity has not been persisted. - The dedicated receipt proves that the current parser actually observed - every relevant raw. Durable revision or membership rows alone may have - been produced by an older parser and therefore cannot establish current - quiescence. + The dedicated receipt proves that *some* parser version whose semantics + are still known to this codebase actually observed every relevant raw. + Durable revision or membership rows alone may have been produced by an + older parser and therefore cannot establish current quiescence. + + This deliberately accepts any *known* fingerprint (the current one, or + one listed in ``SUPERSEDED_MEMBERSHIP_FINGERPRINTS``), not only the + current one (polylogue-9dxn): the census answers "was this raw ever + observed by a real parser?", which a fingerprint bump alone does not + change -- only ``classify_membership_revisions`` semantics changing (a + superseded fingerprint) can make a *verdict* stale, which is a separate + question the terminal-decision check in ``storage/repair.py`` answers. + Treating a bump as forcing full re-census here would mean every + fingerprint bump re-parses the entire archive just to re-confirm facts + that did not change. """ if not raw_ids: return () @@ -342,6 +357,8 @@ def uncensused_historical_revision_raw_ids( resource_blocked_fingerprint = ( _resource_blocked_parser_fingerprint(max_payload_bytes) if max_payload_bytes is not None else None ) + known_fingerprints = [RAW_AUTHORITY_PARSER_FINGERPRINT, *sorted(SUPERSEDED_MEMBERSHIP_FINGERPRINTS)] + known_placeholders = ",".join("?" for _ in known_fingerprints) with sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True) as conn: rows = conn.execute( f""" @@ -350,7 +367,7 @@ def uncensused_historical_revision_raw_ids( LEFT JOIN raw_authority_parser_census AS c ON c.raw_id = r.raw_id WHERE r.raw_id IN ({placeholders}) AND NOT COALESCE( - c.parser_fingerprint = 'revision-membership-v1' + c.parser_fingerprint IN ({known_placeholders}) AND c.status = 'complete', 0 ) @@ -361,7 +378,7 @@ def uncensused_historical_revision_raw_ids( ) ORDER BY r.raw_id """, - [*raw_ids, resource_blocked_fingerprint], + [*raw_ids, *known_fingerprints, resource_blocked_fingerprint], ).fetchall() return tuple(str(row[0]) for row in rows) @@ -440,9 +457,9 @@ def _record_raw_authority_parser_census(archive_root: Path, raw_ids: tuple[str, membership_census = conn.execute( """ SELECT status, detail FROM raw_membership_census - WHERE raw_id = ? AND parser_fingerprint = 'revision-membership-v1' + WHERE raw_id = ? AND parser_fingerprint = ? """, - (raw_id,), + (raw_id, RAW_AUTHORITY_PARSER_FINGERPRINT), ).fetchone() membership_keys = [ str(row[0]) @@ -477,7 +494,7 @@ def _record_raw_authority_parser_census(archive_root: Path, raw_ids: tuple[str, INSERT INTO raw_authority_parser_census ( raw_id, parser_fingerprint, status, logical_keys_json, detail, censused_at_ms - ) VALUES (?, 'revision-membership-v1', ?, ?, ?, 0) + ) VALUES (?, ?, ?, ?, ?, 0) ON CONFLICT(raw_id) DO UPDATE SET parser_fingerprint = excluded.parser_fingerprint, status = excluded.status, @@ -485,7 +502,13 @@ def _record_raw_authority_parser_census(archive_root: Path, raw_ids: tuple[str, detail = excluded.detail, censused_at_ms = excluded.censused_at_ms """, - (raw_id, "complete" if complete else "failed", json.dumps(logical_keys), detail), + ( + raw_id, + RAW_AUTHORITY_PARSER_FINGERPRINT, + "complete" if complete else "failed", + json.dumps(logical_keys), + detail, + ), ) @@ -554,7 +577,7 @@ def apply_outcome( archive.replace_raw_membership_census( raw_id, None, - parser_fingerprint="revision-membership-v1", + parser_fingerprint=RAW_AUTHORITY_PARSER_FINGERPRINT, censused_at_ms=0, detail=BYTE_AUTHORITY_CENSUS_DETAIL, manage_transaction=not batched, @@ -567,7 +590,7 @@ def apply_outcome( archive.replace_raw_membership_census( raw_id, None, - parser_fingerprint="revision-membership-v1", + parser_fingerprint=RAW_AUTHORITY_PARSER_FINGERPRINT, censused_at_ms=0, detail=str(outcome), manage_transaction=not batched, @@ -598,7 +621,7 @@ def apply_outcome( archive.replace_raw_membership_census( raw_id, sessions, - parser_fingerprint="revision-membership-v1", + parser_fingerprint=RAW_AUTHORITY_PARSER_FINGERPRINT, censused_at_ms=0, manage_transaction=not batched, ) @@ -995,7 +1018,7 @@ def commit_replay_unit() -> None: archive.replace_raw_membership_census( raw_id, sessions, - parser_fingerprint="revision-membership-v1", + parser_fingerprint=RAW_AUTHORITY_PARSER_FINGERPRINT, censused_at_ms=0, detail=HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, retire_full_revision_governance=True, diff --git a/polylogue/storage/raw_authority.py b/polylogue/storage/raw_authority.py index f864710497..e6dc0507e6 100644 --- a/polylogue/storage/raw_authority.py +++ b/polylogue/storage/raw_authority.py @@ -24,7 +24,22 @@ from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.migration_runner import validate_migration_backup_manifest -RAW_AUTHORITY_PARSER_FINGERPRINT = "revision-membership-v1" +RAW_AUTHORITY_PARSER_FINGERPRINT = "revision-membership-v2" + +#: Fingerprints previously stamped by ``RAW_AUTHORITY_PARSER_FINGERPRINT`` +#: whose classification semantics are known to have been superseded by a +#: later, deliberately-corrected version of ``classify_membership_revisions`` +#: (polylogue-9dxn). A persisted ``ambiguous`` verdict recorded under one of +#: these fingerprints is stale, not authoritative -- the terminal-decision +#: check in ``storage/repair.py`` treats it as replayable instead of durable +#: debt. A verdict recorded under the CURRENT fingerprint, or with no census +#: row at all (never independently confirmed which parser produced it), +#: stays terminal -- absent evidence must default to conservative, not to +#: "assume it's fixed". This set only affects the *terminal* gate; the +#: *quiescence* gate (``uncensused_historical_revision_raw_ids``) accepts any +#: known fingerprint (current or superseded) so a bump does not force a full +#: archive re-census -- see that function's docstring. +SUPERSEDED_MEMBERSHIP_FINGERPRINTS = frozenset({"revision-membership-v1"}) RAW_AUTHORITY_CENSUS_QUERY_PREFIX = "polylogue://raw-authority-census/" RAW_AUTHORITY_DETAIL_QUERY_PREFIX = "polylogue://raw-authority-detail/" RAW_AUTHORITY_DETAIL_CHUNK_CHARS = 16_384 @@ -2229,6 +2244,7 @@ def prune_orphaned_index_revision_seeds( "RAW_AUTHORITY_DETAIL_CHUNK_CHARS", "RAW_AUTHORITY_DETAIL_QUERY_PREFIX", "RAW_AUTHORITY_PARSER_FINGERPRINT", + "SUPERSEDED_MEMBERSHIP_FINGERPRINTS", "RawAuthorityCensusReceipt", "RawAuthorityCensusResetCounts", "OrphanedIndexRevisionSeedCounts", diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py index ca82bd0f97..cd431aa969 100644 --- a/polylogue/storage/repair.py +++ b/polylogue/storage/repair.py @@ -56,6 +56,7 @@ ) from polylogue.storage.raw_authority import ( RAW_REPLAY_NO_PROGRESS_REASON, + SUPERSEDED_MEMBERSHIP_FINGERPRINTS, RawAuthorityCensusReceipt, RawReplayPlan, RawReplayPlanOutcome, @@ -4429,15 +4430,36 @@ def _raw_replay_plan_outcome( "authority comparison produced a durable deferred decision", "resolve the recorded authority conflict before retry", ) + # polylogue-9dxn: an 'ambiguous' decision recorded under a classifier + # fingerprint since superseded by a correction to + # ``classify_membership_revisions`` is stale, not authoritative -- it + # must be excluded from the terminal gate so the corrected classifier + # gets a chance to re-derive the verdict on replay. A raw with no + # ``raw_authority_parser_census`` row (fingerprint unknown) stays + # conservative and remains terminal, matching "absent evidence defaults + # to current". This gate applies ONLY to the ambiguous legs -- parse + # errors and failed membership census rows are unrelated to classifier + # semantics and stay unconditionally terminal. + superseded_json = json.dumps(sorted(SUPERSEDED_MEMBERSHIP_FINGERPRINTS)) terminal = conn.execute( f""" SELECT 1 - FROM index_tier.raw_revision_applications - WHERE raw_id IN ({placeholders}) AND decision = 'ambiguous' + FROM index_tier.raw_revision_applications AS a + LEFT JOIN raw_authority_parser_census AS c ON c.raw_id = a.raw_id + WHERE a.raw_id IN ({placeholders}) AND a.decision = 'ambiguous' + AND NOT COALESCE( + c.parser_fingerprint IN (SELECT value FROM json_each(?)), + 0 + ) UNION ALL SELECT 1 - FROM raw_session_memberships - WHERE raw_id IN ({placeholders}) AND decision = 'ambiguous' + FROM raw_session_memberships AS m + LEFT JOIN raw_authority_parser_census AS c ON c.raw_id = m.raw_id + WHERE m.raw_id IN ({placeholders}) AND m.decision = 'ambiguous' + AND NOT COALESCE( + c.parser_fingerprint IN (SELECT value FROM json_each(?)), + 0 + ) UNION ALL SELECT 1 FROM raw_sessions @@ -4450,7 +4472,15 @@ def _raw_replay_plan_outcome( WHERE raw_id IN ({placeholders}) AND status = 'failed' LIMIT 1 """, - (*component, *component, *component, _TRANSIENT_LOCK_PARSE_ERROR, *component), + ( + *component, + superseded_json, + *component, + superseded_json, + *component, + _TRANSIENT_LOCK_PARSE_ERROR, + *component, + ), ).fetchone() if terminal is not None: return RawReplayPlanOutcome( diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index d09d22067c..651d76c647 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -40,6 +40,7 @@ ) from polylogue.sources.live.cursor import CursorStore from polylogue.sources.parsers.base import ParsedMessage, ParsedSession +from polylogue.storage.raw_authority import RAW_AUTHORITY_PARSER_FINGERPRINT from polylogue.storage.sqlite.archive_tiers import archive as archive_tier_module from polylogue.storage.sqlite.archive_tiers import revision_governance as archive_revision_governance from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore @@ -3616,7 +3617,7 @@ def conversation(native_id: str, *texts: str) -> dict[str, object]: store.replace_raw_membership_census( raw_id, sessions, - parser_fingerprint="revision-membership-v1", + parser_fingerprint=RAW_AUTHORITY_PARSER_FINGERPRINT, censused_at_ms=0, detail=HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, retire_full_revision_governance=True, diff --git a/tests/unit/sources/test_revision_backfill.py b/tests/unit/sources/test_revision_backfill.py index 29c665ffce..3ab65d6fea 100644 --- a/tests/unit/sources/test_revision_backfill.py +++ b/tests/unit/sources/test_revision_backfill.py @@ -28,6 +28,7 @@ census_historical_revision_evidence, ) from polylogue.storage.blob_publication import ArchiveBlobPublisher +from polylogue.storage.raw_authority import RAW_AUTHORITY_PARSER_FINGERPRINT from polylogue.storage.sqlite.archive_tiers import revision_governance as archive_revision_governance from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root @@ -355,9 +356,10 @@ def test_historical_backfill_selects_prefix_newest_independent_of_acquisition_or """ SELECT status, COUNT(*) FROM raw_authority_parser_census - WHERE parser_fingerprint = 'revision-membership-v1' + WHERE parser_fingerprint = ? GROUP BY status ORDER BY status - """ + """, + (RAW_AUTHORITY_PARSER_FINGERPRINT,), ).fetchall() assert parser_census == [("complete", 2), ("failed", 1)] with sqlite3.connect(tmp_path / "index.db") as conn: @@ -694,9 +696,9 @@ def test_stale_pre_fix_identity_split_folds_into_one_ambiguous_cohort(tmp_path: """ INSERT INTO raw_authority_parser_census (raw_id, parser_fingerprint, status, logical_keys_json, detail, censused_at_ms) - VALUES (?, 'revision-membership-v1', 'complete', ?, 'pre-seeded for test', 0) + VALUES (?, ?, 'complete', ?, 'pre-seeded for test', 0) """, - (raw_id, json.dumps([key])), + (raw_id, RAW_AUTHORITY_PARSER_FINGERPRINT, json.dumps([key])), ) conn.commit() diff --git a/tests/unit/storage/test_archive_readiness.py b/tests/unit/storage/test_archive_readiness.py index 6714cc2701..811ad6b70f 100644 --- a/tests/unit/storage/test_archive_readiness.py +++ b/tests/unit/storage/test_archive_readiness.py @@ -15,6 +15,7 @@ raw_materialization_ready, ) from polylogue.storage.raw_authority import ( + RAW_AUTHORITY_PARSER_FINGERPRINT, RawReplayPlan, RawReplayPlanOutcome, RawReplayPlanStatus, @@ -265,7 +266,7 @@ def test_raw_materialization_snapshot_reads_append_census_writer_contract(tmp_pa archive.replace_raw_membership_census( raw_id, None, - parser_fingerprint="revision-membership-v1", + parser_fingerprint=RAW_AUTHORITY_PARSER_FINGERPRINT, censused_at_ms=0, detail=BYTE_AUTHORITY_CENSUS_DETAIL, ) diff --git a/tests/unit/storage/test_incremental_rebuild_equivalence.py b/tests/unit/storage/test_incremental_rebuild_equivalence.py index b95ba50d53..2afff5c859 100644 --- a/tests/unit/storage/test_incremental_rebuild_equivalence.py +++ b/tests/unit/storage/test_incremental_rebuild_equivalence.py @@ -33,6 +33,7 @@ from polylogue.maintenance.replay import rebuild_index_from_source from polylogue.sources.revision_backfill import backfill_historical_revision_evidence from polylogue.storage.index_generation import IndexGenerationStore, source_revision_snapshot +from polylogue.storage.raw_authority import RAW_AUTHORITY_PARSER_FINGERPRINT from polylogue.storage.repair import repair_session_insights from polylogue.storage.runtime import SESSION_INSIGHT_MATERIALIZER_VERSION from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore @@ -52,7 +53,7 @@ CANONICAL_TOKEN = "quartzneedle" STALE_TOKEN = "staleonlytoken" OVERLAY_TAG = "operator-canary" -PARSER_RECIPE = "revision-membership-v1" +PARSER_RECIPE = RAW_AUTHORITY_PARSER_FINGERPRINT # Volatile attempt timestamps are not canonical derivation output. The exact # semantic stamps beside them remain compared. diff --git a/tests/unit/storage/test_quarantined_accepted_raw_repair.py b/tests/unit/storage/test_quarantined_accepted_raw_repair.py index f5eb4b1cd5..acd3609f88 100644 --- a/tests/unit/storage/test_quarantined_accepted_raw_repair.py +++ b/tests/unit/storage/test_quarantined_accepted_raw_repair.py @@ -13,6 +13,7 @@ from polylogue.pipeline.ids import session_content_hash from polylogue.sources.revision_backfill import _parse_one from polylogue.storage.blob_store import BlobStore +from polylogue.storage.raw_authority import RAW_AUTHORITY_PARSER_FINGERPRINT from polylogue.storage.raw_reconciler import ( RawAuthorityActuator, RawAuthorityFrontierState, @@ -136,9 +137,9 @@ def _seed_invalid_head( """ INSERT INTO raw_membership_census ( raw_id, parser_fingerprint, status, member_count, censused_at_ms - ) VALUES (?, 'revision-membership-v1', 'complete', 1, 0) + ) VALUES (?, ?, 'complete', 1, 0) """, - (raw_id,), + (raw_id, RAW_AUTHORITY_PARSER_FINGERPRINT), ) source.commit() return raw_id @@ -382,9 +383,9 @@ def _seed_quarantined_raw_fanout(root: Path) -> tuple[str, tuple[tuple[str, str] """ INSERT INTO raw_membership_census ( raw_id, parser_fingerprint, status, member_count, censused_at_ms - ) VALUES (?, 'revision-membership-v1', 'complete', 1, 0) + ) VALUES (?, ?, 'complete', 1, 0) """, - (raw_id,), + (raw_id, RAW_AUTHORITY_PARSER_FINGERPRINT), ) source.commit() with sqlite3.connect(root / "source.db") as source: diff --git a/tests/unit/storage/test_raw_authority_ledger.py b/tests/unit/storage/test_raw_authority_ledger.py index 3f9c4148a5..57c5444bae 100644 --- a/tests/unit/storage/test_raw_authority_ledger.py +++ b/tests/unit/storage/test_raw_authority_ledger.py @@ -23,6 +23,7 @@ from polylogue.storage.archive_readiness import raw_materialization_readiness_snapshot, raw_materialization_ready from polylogue.storage.raw_authority import ( AUTO_STALE_PLAN_RESOLUTION, + RAW_AUTHORITY_PARSER_FINGERPRINT, RawReplayPlan, RawReplayPlanOutcome, RawReplayPlanStatus, @@ -1260,10 +1261,101 @@ def test_stale_per_raw_parser_fingerprint_is_recensused_before_planning(tmp_path "SELECT parser_fingerprint FROM raw_authority_parser_census WHERE raw_id = ?", (raw_id,), ).fetchone()[0] - == "revision-membership-v1" + == RAW_AUTHORITY_PARSER_FINGERPRINT ) +def _seed_ambiguous_membership_component( + tmp_path: Path, + *, + native_id: str, + parser_fingerprint: str | None, +) -> tuple[str, object]: + """Seed one raw whose membership decision is durably 'ambiguous'. + + ``parser_fingerprint`` controls what (if anything) the per-raw + ``raw_authority_parser_census`` row records: the CURRENT fingerprint (the + ambiguous verdict should still be terminal), a fingerprint listed in + ``SUPERSEDED_MEMBERSHIP_FINGERPRINTS`` (the verdict is stale and must be + replayable), or ``None`` (no census row at all -- absent evidence must + stay conservative and remain terminal). + """ + raw_id = _write_codex_raw(tmp_path, native_id=native_id, source_path=f"{native_id}.jsonl", acquired_at_ms=1) + logical_source_key = f"codex-session:{native_id}" + with sqlite3.connect(tmp_path / "source.db") as conn: + conn.execute( + """ + INSERT INTO raw_session_memberships ( + raw_id, logical_source_key, provider_session_id, source_revision, + normalized_content_hash, message_count, decision, decided_at_ms + ) VALUES (?, ?, ?, ?, ?, 1, 'ambiguous', 1) + """, + (raw_id, logical_source_key, native_id, "rev-1", bytes(32)), + ) + if parser_fingerprint is not None: + conn.execute( + """ + INSERT INTO raw_authority_parser_census ( + raw_id, parser_fingerprint, status, logical_keys_json, detail, censused_at_ms + ) VALUES (?, ?, 'complete', ?, 'test-seeded', 0) + """, + (raw_id, parser_fingerprint, json.dumps([logical_source_key])), + ) + conn.commit() + (plan,) = build_raw_replay_plans(tmp_path, [(raw_id,)]) + empty_remaining = repair_mod.RawMaterializationCandidates([], 0, 0) + (outcome,) = repair_mod._raw_replay_plan_outcomes(tmp_path, [plan], remaining=empty_remaining) + return raw_id, outcome + + +def test_ambiguous_verdict_under_current_fingerprint_stays_terminal(tmp_path: Path) -> None: + """polylogue-9dxn: an 'ambiguous' decision recorded under the CURRENT + classifier fingerprint is still authoritative -- it must not be + replayed without new evidence. + """ + initialize_active_archive_root(tmp_path) + _raw_id, outcome = _seed_ambiguous_membership_component( + tmp_path, native_id="current-ambiguous", parser_fingerprint=RAW_AUTHORITY_PARSER_FINGERPRINT + ) + assert outcome.status is RawReplayPlanStatus.TERMINAL + assert "ambiguous" in outcome.reason.lower() + + +def test_ambiguous_verdict_under_superseded_fingerprint_is_replayable(tmp_path: Path) -> None: + """polylogue-9dxn: an 'ambiguous' decision recorded under a fingerprint + listed in SUPERSEDED_MEMBERSHIP_FINGERPRINTS is stale -- a corrected + classifier deserves a chance to re-derive it, so it must not be + terminal. + + Anti-vacuity: this exercises the real production route + ``repair._raw_replay_plan_outcome`` (via the public + ``build_raw_replay_plans``/``_raw_replay_plan_outcomes`` pair used by + ``repair_raw_materialization``, the daemon's live raw-materialization + repair entrypoint). Reverting the fingerprint-gating clause added to the + terminal query in ``storage/repair.py`` (the ``LEFT JOIN + raw_authority_parser_census`` + ``NOT COALESCE(... IN (SELECT value FROM + json_each(?)) ...)`` guard) makes this test fail by re-classifying the + plan as TERMINAL. + """ + initialize_active_archive_root(tmp_path) + superseded_fingerprint = next(iter(raw_authority_mod.SUPERSEDED_MEMBERSHIP_FINGERPRINTS)) + _raw_id, outcome = _seed_ambiguous_membership_component( + tmp_path, native_id="superseded-ambiguous", parser_fingerprint=superseded_fingerprint + ) + assert outcome.status is not RawReplayPlanStatus.TERMINAL + + +def test_ambiguous_verdict_with_no_census_row_stays_terminal(tmp_path: Path) -> None: + """polylogue-9dxn: absent census evidence must default to conservative + (terminal), not to "assume the classifier fix already applies". + """ + initialize_active_archive_root(tmp_path) + _raw_id, outcome = _seed_ambiguous_membership_component( + tmp_path, native_id="uncensused-ambiguous", parser_fingerprint=None + ) + assert outcome.status is RawReplayPlanStatus.TERMINAL + + def test_repair_result_bounds_public_plan_outcomes() -> None: outcomes = tuple( RawReplayPlanOutcome( diff --git a/tests/unit/storage/test_revision_replay.py b/tests/unit/storage/test_revision_replay.py index dfa84dbca1..7f2f056dc3 100644 --- a/tests/unit/storage/test_revision_replay.py +++ b/tests/unit/storage/test_revision_replay.py @@ -30,6 +30,7 @@ from polylogue.pipeline.ids import session_content_hash, session_revision_projection from polylogue.sources.dispatch import merge_parsed_session_chunks, parse_stream_payload from polylogue.sources.parsers.base import ParsedAttachment, ParsedMessage, ParsedSession +from polylogue.storage.raw_authority import RAW_AUTHORITY_PARSER_FINGERPRINT from polylogue.storage.sqlite.archive_tiers import revision_governance as archive_revision_governance from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root @@ -595,7 +596,7 @@ def parsed_solo(native_id: str, *texts: str) -> ParsedSession: archive.replace_raw_membership_census( raw_id, [session], - parser_fingerprint="revision-membership-v1", + parser_fingerprint=RAW_AUTHORITY_PARSER_FINGERPRINT, censused_at_ms=0, detail="historical non-prefix full revision governance", retire_full_revision_governance=True, @@ -818,7 +819,7 @@ def parsed_solo(native_id: str, *texts: str) -> ParsedSession: archive.replace_raw_membership_census( raw_id, [session], - parser_fingerprint="revision-membership-v1", + parser_fingerprint=RAW_AUTHORITY_PARSER_FINGERPRINT, censused_at_ms=0, detail="cross-route full revision governance", retire_full_revision_governance=True, From bdffadc2dd2e8b52ff38eefde2f357331b454034 Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 31 Jul 2026 23:44:25 +0200 Subject: [PATCH 2/4] perf(sources): replay rebuild logical keys in lineage order, not lexicographic Problem: revision_backfill.py's rebuild replay loop visited sorted(logical_keys) -- a lexicographic string sort with zero relationship to parent/child lineage. During a cold/full rebuild this means a child (resume/fork) replays before its parent roughly as often as not. A child replayed before its parent is stored WHOLE (a full duplicate of the eventual shared prefix); when the parent finally arrives, _resolve_session_graph must walk every such orphaned child and normalize it (delete the duplicate prefix rows, remap session_events refs, delete prefix-scoped dependents) -- the #2467 deferred-tail path, confirmed linear but real O(orphaned_children * shared_prefix_size) row-mutation work (tests/benchmarks/test_graph_resolve_deferred_tail.py, 260s for one codex-session parent in a live 2026-07-03 rebuild batch). Solution: _lineage_aware_replay_order (new) computes roots-first, children-after-parent ordering for one rebuild's logical_keys, using the ParsedSession.parent_session_provider_id the census phase already parsed and spilled -- no extra reparsing on the happy path (spill.for_raw falls back to a bounded reparse only if a key's representative raw fell out of the spill cache). Falls back to the previous lexicographic order for any key whose parent is unresolvable (missing/external/cross-batch parent, or a lineage cycle) -- nothing is ever skipped. Scheduling-only: it changes the ORDER backfill_historical_revision_evidence's replay loop (and its pipeline-decode prefetcher, which now shares the same order) visits logical keys in, never what gets replayed or adopted. Verification: devtools test tests/unit/sources/test_revision_backfill.py -k lineage_aware -> 4 passed: - test_lineage_aware_replay_order_visits_parent_before_children - test_lineage_aware_replay_order_falls_back_for_unresolvable_parent - test_lineage_aware_replay_order_reduces_deferred_tail_hits (AC1): measures _reextract_prefix_tail_db call count on a 1-parent/5-child codex resume fixture -- lexicographic order hits it 5 times (once per child), lineage order hits it 0 times - test_lineage_aware_replay_order_preserves_outcome_parity (AC2): same fixture, lineage vs. forced-lexicographic order reach byte-identical index.db content (_index_content_manifest: sessions/messages/blocks/session_links) and identical RevisionBackfillResult counts devtools test tests/unit/sources/test_revision_backfill.py -> 59 passed, 1 pre-existing failure unrelated to this change (same as prior commit: test_parse_one_still_replays_real_claude_code_sessions_with_no_path_rule) devtools test tests/benchmarks/test_graph_resolve_deferred_tail.py tests/unit/storage/test_incremental_rebuild_equivalence.py -> 2 passed Anti-vacuity: test_lineage_aware_replay_order_reduces_deferred_tail_hits exercises the real production route (backfill_historical_revision_evidence -> its replay loop's ordered_logical_keys, computed by _lineage_aware_replay_order at its actual call site) with a real Codex resume-shaped fixture. Reverting the call site back to sorted(logical_keys) (verified live during implementation via monkeypatch.setattr(revision_backfill, "_lineage_aware_replay_order", lambda *a: sorted(a[0]))) makes lineage_hits jump from 0 to 5, equal to lexicographic_hits -- the exact failure this test is built to catch. Ref polylogue-5q2u --- polylogue/sources/revision_backfill.py | 112 ++++++++++- tests/unit/sources/test_revision_backfill.py | 199 +++++++++++++++++++ 2 files changed, 309 insertions(+), 2 deletions(-) diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index 8c10a3ea3d..123a15ca04 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -777,6 +777,104 @@ def census_historical_revision_evidence( ) +def _lineage_aware_replay_order( + logical_keys: set[str], + archive: ArchiveStore, + spill: _ParsedSessionSpill, + archive_root: Path, +) -> list[str]: + """Order one rebuild's byte-typed logical keys so a parent's cohort + replays before any of its children's (polylogue-5q2u). + + Replaying a child before its parent forces ``_resolve_session_graph`` to + store the child's shared prefix WHOLE, then re-walk and normalize it + (delete the duplicate prefix rows, remap ``session_events`` refs, delete + prefix-scoped dependents) once the parent finally arrives -- the + #2467 deferred-tail path, O(orphaned_children * shared_prefix_size) real + row-mutation work. The previous ``sorted(logical_keys)`` lexicographic + order has zero relationship to parent/child lineage, so it triggers this + expensive path roughly as often as not during a cold/full rebuild. + Visiting roots first (and each child only after its parent) minimizes + how often it triggers. + + This is deliberately scheduling-only: it must never change WHAT gets + replayed or adopted, only the order this module's own replay loop visits + logical keys in. A key whose parent cannot be resolved here -- no + ``parent_session_provider_id``, a parent outside this rebuild's + ``logical_keys`` (missing/external/cross-batch parent), or a lineage + cycle -- degrades to the original lexicographic position among the + unresolved remainder. Nothing is ever skipped. + """ + sorted_keys = sorted(logical_keys) + if len(sorted_keys) <= 1: + return sorted_keys + + placeholders = ",".join("?" for _ in sorted_keys) + with sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True) as conn: + rows = conn.execute( + f""" + SELECT logical_source_key, raw_id + FROM raw_sessions + WHERE logical_source_key IN ({placeholders}) + ORDER BY logical_source_key, acquired_at_ms DESC + """, + sorted_keys, + ).fetchall() + representative_raw_id: dict[str, str] = {} + for logical_source_key, raw_id in rows: + representative_raw_id.setdefault(str(logical_source_key), str(raw_id)) + + parent_of: dict[str, str | None] = {} + for key in sorted_keys: + parent_key: str | None = None + raw_id = representative_raw_id.get(key) + if raw_id is not None: + try: + sessions, _payload_bytes = spill.for_raw(archive, raw_id) + except Exception: + # Lineage ordering is a scheduling optimization only -- any + # failure here degrades to "treat as unresolved", never to a + # replay/adoption failure. + sessions = [] + if sessions: + session = sessions[0] + parent_provider_id = session.parent_session_provider_id + if parent_provider_id: + parent_key = f"{session.source_name.value}:{parent_provider_id}" + parent_of[key] = parent_key + + children: dict[str, list[str]] = {} + roots: list[str] = [] + for key in sorted_keys: + parent_key = parent_of[key] + if parent_key is not None and parent_key in logical_keys and parent_key != key: + children.setdefault(parent_key, []).append(key) + else: + roots.append(key) + + ordered: list[str] = [] + seen: set[str] = set() + + def visit(key: str) -> None: + if key in seen: + return + seen.add(key) + ordered.append(key) + for child in children.get(key, ()): + visit(child) + + for key in roots: + visit(key) + # Cycles: every remaining member has a not-yet-visited parent inside the + # set. Fall back to lexicographic order for the unresolved remainder -- + # ``visit`` still walks each one's children once reached, so nothing is + # skipped or duplicated. + for key in sorted_keys: + if key not in seen: + visit(key) + return ordered + + def backfill_historical_revision_evidence( archive_root: Path, *, @@ -972,13 +1070,23 @@ def commit_replay_unit() -> None: and len(logical_keys) + len(membership_keys) >= _PIPELINE_DECODE_MIN_COHORTS ) ) + # polylogue-5q2u: replay in lineage order (roots, then children after + # their parent) instead of lexicographic order -- see + # ``_lineage_aware_replay_order``'s docstring. Scheduling-only: the + # SET of keys replayed and the plan/adoption outcome for each is + # unaffected, only wall-clock and how often the deferred-tail path + # (#2467) triggers. Both the pipeline-decode prefetcher and the + # writer's own replay loop consume this SAME order so the + # prefetcher's lookahead actually matches what the writer visits + # next. + ordered_logical_keys = _lineage_aware_replay_order(logical_keys, archive, spill, archive_root) decode_prefetcher: _ReplaySpillPrefetcher | None = None if effective_pipeline_decode: decode_prefetcher = _ReplaySpillPrefetcher(spill, archive_root=archive_root) spill.attach_prefetcher(decode_prefetcher) - decode_prefetcher.start_phase(sorted(logical_keys), provisional_full_raw_ids) + decode_prefetcher.start_phase(ordered_logical_keys, provisional_full_raw_ids) try: - for logical_key in sorted(logical_keys): + for logical_key in ordered_logical_keys: if decode_prefetcher is not None: decode_prefetcher.enter_key(logical_key) # polylogue-eqnv: the offline backfill/rebuild path is the one diff --git a/tests/unit/sources/test_revision_backfill.py b/tests/unit/sources/test_revision_backfill.py index 3ab65d6fea..e164a166f9 100644 --- a/tests/unit/sources/test_revision_backfill.py +++ b/tests/unit/sources/test_revision_backfill.py @@ -23,6 +23,7 @@ from polylogue.sources.revision_backfill import ( RawParsePrefetchCache, _browser_snapshot_fidelity, + _lineage_aware_replay_order, _parse_one, backfill_historical_revision_evidence, census_historical_revision_evidence, @@ -30,6 +31,7 @@ from polylogue.storage.blob_publication import ArchiveBlobPublisher from polylogue.storage.raw_authority import RAW_AUTHORITY_PARSER_FINGERPRINT from polylogue.storage.sqlite.archive_tiers import revision_governance as archive_revision_governance +from polylogue.storage.sqlite.archive_tiers import write as archive_tier_write from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root from tests.infra.revision_backfill_benchmark import ( @@ -2416,3 +2418,200 @@ def test_pipelined_decode_respects_batched_replay_commits(tmp_path: Path, monkey assert serial_result == pipelined_result assert _index_content_manifest(serial_root) == _index_content_manifest(pipelined_root) + + +def _codex_session_payload( + session_id: str, + message_texts: list[str], + *, + forked_from_id: str | None = None, +) -> bytes: + """Build a codex JSONL raw with one message per ``message_texts`` entry. + + Mirrors how a real Codex resume payload looks: a ``session_meta`` record + carrying ``forked_from_id`` when this session is a resume/fork, followed + by ``response_item`` message records. Passing the SAME leading + ``message_texts`` for a parent and one of its children (plus extra tail + entries on the child) reproduces the on-disk shape #2467's deferred-tail + extraction exists for: the child's JSONL physically re-contains the + parent's entire prefix. + """ + meta_payload: dict[str, object] = {"id": session_id, "timestamp": "2026-06-01T00:00:00Z"} + if forked_from_id is not None: + meta_payload["forked_from_id"] = forked_from_id + lines = [json.dumps({"type": "session_meta", "payload": meta_payload}, separators=(",", ":"))] + for position, text in enumerate(message_texts): + lines.append( + json.dumps( + { + "type": "response_item", + "payload": { + "type": "message", + "id": f"m{position}", + "role": "user" if position % 2 == 0 else "assistant", + "content": [{"type": "input_text", "text": text}], + }, + }, + separators=(",", ":"), + ) + ) + return ("\n".join(lines) + "\n").encode() + + +def _seed_lineage_fixture(root: Path, *, n_children: int) -> None: + """One parent (native_id sorts LAST lexicographically) plus N children + (native_ids sort BEFORE the parent) that each replay the parent's full + message prefix plus one new tail message -- a real Codex resume shape. + """ + initialize_active_archive_root(root) + parent_native_id = "zparent" + parent_texts = [f"parent-{i}" for i in range(4)] + with ArchiveStore.open_existing(root, read_only=False) as archive: + archive.write_raw_payload( + provider=Provider.CODEX, + payload=_codex_session_payload(parent_native_id, parent_texts), + source_path=f"{parent_native_id}.jsonl", + acquired_at_ms=1, + ) + for index in range(n_children): + child_native_id = f"achild{index}" + child_texts = [*parent_texts, f"child-{index}-tail"] + archive.write_raw_payload( + provider=Provider.CODEX, + payload=_codex_session_payload(child_native_id, child_texts, forked_from_id=parent_native_id), + source_path=f"{child_native_id}.jsonl", + acquired_at_ms=2 + index, + ) + + +def test_lineage_aware_replay_order_visits_parent_before_children(tmp_path: Path) -> None: + """polylogue-5q2u: roots first, then each child only after its parent -- + NOT the lexicographic order a plain ``sorted()`` would produce (the + parent's native id, "zparent", sorts LAST here).""" + root = tmp_path / "archive" + _seed_lineage_fixture(root, n_children=5) + with ArchiveStore.open_existing(root, read_only=False) as archive: + revision_backfill._census_historical_revision_evidence( + archive, + revision_backfill._ParsedSessionSpill(root, max_cached_payload_bytes=None), + selected_raw_ids=None, + max_payload_bytes=None, + ) + archive.commit() + _expanded, logical_keys = archive.expand_raw_membership_selection(None) + with revision_backfill._ParsedSessionSpill(root, max_cached_payload_bytes=None) as spill: + order = _lineage_aware_replay_order(set(logical_keys), archive, spill, root) + + assert order[0] == "codex:zparent" + parent_position = order.index("codex:zparent") + for index in range(5): + child_key = f"codex:achild{index}" + assert child_key in order + assert order.index(child_key) > parent_position + # Lexicographic order would have put every child before the parent. + assert sorted(logical_keys)[0] != "codex:zparent" + + +def test_lineage_aware_replay_order_falls_back_for_unresolvable_parent(tmp_path: Path) -> None: + """A parent outside this call's ``logical_keys`` set (missing/external/ + cross-batch) must not crash or drop the child -- it degrades to the + lexicographic position among the unresolved remainder. Two orphans (not + one) so the real DB lookup + ``spill.for_raw`` path is exercised instead + of the single-key short-circuit.""" + root = tmp_path / "archive" + initialize_active_archive_root(root) + with ArchiveStore.open_existing(root, read_only=False) as archive: + for native_id in ("zorphan", "aorphan"): + archive.write_raw_payload( + provider=Provider.CODEX, + payload=_codex_session_payload(native_id, ["only-message"], forked_from_id="never-ingested-parent"), + source_path=f"{native_id}.jsonl", + acquired_at_ms=1, + ) + with revision_backfill._ParsedSessionSpill(root, max_cached_payload_bytes=None) as spill: + order = _lineage_aware_replay_order({"codex:zorphan", "codex:aorphan"}, archive, spill, root) + assert sorted(order) == ["codex:aorphan", "codex:zorphan"] + # Neither key's parent is in the set, so both are roots -- fallback + # degrades to lexicographic order among them. + assert order == ["codex:aorphan", "codex:zorphan"] + + +def test_lineage_aware_replay_order_reduces_deferred_tail_hits(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """polylogue-5q2u AC1: lineage-aware replay must trigger the #2467 + deferred-tail/orphaned-child normalization path (``_reextract_prefix_tail_db``) + strictly less often than the previous lexicographic order for a + representative parent-with-many-children fixture, where the parent's + native id sorts lexicographically AFTER its children's. + + Anti-vacuity: reverting the ``_lineage_aware_replay_order`` call at the + ``for logical_key in ...:`` call site back to ``sorted(logical_keys)`` + makes this test fail (both counts become equal and >0, since every + child would then replay before the parent it depends on). + """ + lineage_root = tmp_path / "lineage" + lexicographic_root = tmp_path / "lexicographic" + n_children = 5 + _seed_lineage_fixture(lineage_root, n_children=n_children) + _seed_lineage_fixture(lexicographic_root, n_children=n_children) + + def _count_deferred_tail_hits(root: Path, *, force_lexicographic: bool) -> int: + calls = 0 + original = archive_tier_write._reextract_prefix_tail_db + + def counting_wrapper(*args: object, **kwargs: object) -> object: + nonlocal calls + calls += 1 + return original(*args, **kwargs) + + monkeypatch.setattr(archive_tier_write, "_reextract_prefix_tail_db", counting_wrapper) + if force_lexicographic: + monkeypatch.setattr( + revision_backfill, + "_lineage_aware_replay_order", + lambda logical_keys, archive, spill, archive_root: sorted(logical_keys), + ) + backfill_historical_revision_evidence(root) + monkeypatch.undo() + return calls + + lexicographic_hits = _count_deferred_tail_hits(lexicographic_root, force_lexicographic=True) + lineage_hits = _count_deferred_tail_hits(lineage_root, force_lexicographic=False) + + assert lexicographic_hits == n_children, ( + f"expected every one of the {n_children} children (native ids sorting before " + f"the parent's) to hit the deferred-tail path under lexicographic order, got {lexicographic_hits}" + ) + assert lineage_hits == 0, ( + f"lineage-aware order should replay the parent before any child, avoiding the " + f"deferred-tail path entirely; got {lineage_hits} hits" + ) + + +def test_lineage_aware_replay_order_preserves_outcome_parity(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """polylogue-5q2u AC2: lineage-aware scheduling must not change WHAT gets + replayed/adopted -- only the order. Two archives seeded identically, + replayed once under lineage order and once forced to the previous + lexicographic order, must reach byte-identical index.db content + (sessions/messages/blocks/session_links), matching the equivalence + currency ``_index_content_manifest`` already uses for other replay-order + equivalence proofs in this file (e.g. + ``test_pipelined_decode_respects_batched_replay_commits``). + """ + lineage_root = tmp_path / "lineage" + lexicographic_root = tmp_path / "lexicographic" + _seed_lineage_fixture(lineage_root, n_children=5) + _seed_lineage_fixture(lexicographic_root, n_children=5) + + lineage_result = backfill_historical_revision_evidence(lineage_root) + + monkeypatch.setattr( + revision_backfill, + "_lineage_aware_replay_order", + lambda logical_keys, archive, spill, archive_root: sorted(logical_keys), + ) + lexicographic_result = backfill_historical_revision_evidence(lexicographic_root) + + assert lineage_result.replayed_logical_sources == lexicographic_result.replayed_logical_sources + assert lineage_result.quarantined == lexicographic_result.quarantined + assert lineage_result.adoption_deferred == lexicographic_result.adoption_deferred + assert _index_content_manifest(lineage_root) == _index_content_manifest(lexicographic_root) From 49b443b86ec4e164140be5425633e43115f0a028 Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 31 Jul 2026 23:58:19 +0200 Subject: [PATCH 3/4] fix(storage): report raw-materialization dry-run preview as success Problem: repair_raw_materialization's dry-run branch (the path reached once a preview has validly identified executable authority components) unconditionally returned success=False, conflating "nothing was mutated" (which repaired_count=0 already encodes correctly) with "the preview failed". Repro: seed one raw_sessions row with a real blob and call repair_raw_materialization(config, dry_run=True) -- candidate_count=1, repaired_count=0 (correct), success=False (wrong: the preview completed validly). devtools/scale_regression_probe.py's own raw_materialization_debt_detected check had been adapted to assert `dry_run.success is False` as the "ok" condition, silently codifying the bug as expected behavior rather than fixing it. Solution: dry_run success now means "the requested preview phase completed validly", matching the convention the adjacent "no candidate_raw_ids" and census-pending branches in the same function already use. repaired_count (unconditionally 0 for every dry-run outcome) remains the sole signal for "nothing was mutated" -- preview success and preview mutation are now two independent, honestly-named facts instead of one field trying to carry both. Migrated 6 existing test_repair.py assertions plus scale_regression_probe.py's own check to the corrected semantics; no production caller depends on dry-run success being False (grepped every `repair_raw_materialization`/`repair_materialization` call site -- the daemon's two callers, daemon/cli.py:1169/1231, always pass dry_run=False and are unaffected). Non-goal (per lane scope): this fixes the specific dry-run outcome defect and the scale-probe assertions it broke, not the broader class-level typed MaintenanceOutcome/receipt vocabulary (phase + candidate/eligible/blocked/planned/applied/already-satisfied/failed/ remaining counts across every repair/cleanup handler) polylogue-f57q's design section describes. That census remains open scope; a follow-up tracking item should own it explicitly rather than this fix silently claiming it. Verification: devtools test tests/unit/storage/test_repair.py -> 65 passed devtools test tests/unit/devtools/test_scale_regression_probe.py tests/unit/storage/test_raw_authority_ledger.py -> 40 passed devtools test tests/unit/daemon/test_daemon_cli.py tests/unit/daemon/test_raw_materialization_parse_stage_equivalence.py -> 112 passed, 2 pre-existing failures unrelated to this change (test_maybe_run_raw_materialization_whale_pass_* fail with `TypeError: () got an unexpected keyword argument '_bootstrap'` inside polylogue/config.py:2173 -- a mock-signature mismatch with no relation to repair.py/dry_run/success, reproduces on unmodified files) Anti-vacuity: test_scale_regression_probe_runs_seeded_bug_class_checks and test_scale_regression_probe_main_emits_json exercise the real production route repair_raw_materialization(dry_run=True) through run_scale_regression_probe's raw_materialization_debt_detected check. Reverting the `success=True` mutation in repair.py's dry-run branch back to `success=False` makes both tests fail again with the exact `assert False is True` / `assert 1 == 0` shape observed before this fix. Ref polylogue-f57q --- devtools/scale_regression_probe.py | 6 +++++- polylogue/storage/repair.py | 13 ++++++++++++- tests/unit/storage/test_repair.py | 12 ++++++------ 3 files changed, 23 insertions(+), 8 deletions(-) diff --git a/devtools/scale_regression_probe.py b/devtools/scale_regression_probe.py index 3e92f8ac45..5753127e05 100644 --- a/devtools/scale_regression_probe.py +++ b/devtools/scale_regression_probe.py @@ -405,10 +405,14 @@ def _check_raw_materialization_backlog(root: Path) -> ScaleRegressionCheck: preview = repair_mod.raw_materialization_replay_backlog(config, limit=5) dry_run = repair_mod.repair_raw_materialization(config, dry_run=True) selected_plan_count = len(dry_run.plan_outcomes) + # polylogue-f57q: a dry-run preview that identifies exactly one + # candidate/eligible/planned raw and mutates nothing is a phase-honest + # SUCCESS -- repaired_count staying 0 is what proves the preview never + # mutated, not dry_run.success being False. ok = ( preview["candidate_count"] == 1 and dry_run.repaired_count == 0 - and dry_run.success is False + and dry_run.success is True and selected_plan_count == 1 ) return ScaleRegressionCheck( diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py index cd431aa969..88f43cf496 100644 --- a/polylogue/storage/repair.py +++ b/polylogue/storage/repair.py @@ -6591,7 +6591,18 @@ def repair_raw_materialization( return _internal_derived_repair_result( "raw_materialization", repaired_count=0, - success=False, + # polylogue-f57q: a dry-run PREVIEW that reaches this branch has + # validly identified an executable plan -- it is a phase-honest + # SUCCESS (the preview completed), never a failure. `repaired_count` + # (always 0 here, correctly) already carries the "nothing was + # mutated" fact; conflating "no mutation" with "failed" collided + # with devtools/scale_regression_probe.py's contract, which the + # previous dry-run + `success=False` shape made impossible to + # satisfy honestly. This mirrors the same convention the + # "no candidate_raw_ids" branch above and the census-pending + # short-circuit below already use: success answers "did the + # requested phase complete validly", not "was anything applied". + success=True, detail=detail, metrics=metrics, plan_outcomes=plan_outcomes, diff --git a/tests/unit/storage/test_repair.py b/tests/unit/storage/test_repair.py index d939729456..421f28c5b3 100644 --- a/tests/unit/storage/test_repair.py +++ b/tests/unit/storage/test_repair.py @@ -255,7 +255,7 @@ def test_raw_materialization_preview_counts_replayable_rows_without_erasing_miss result = repair_mod.repair_raw_materialization(config, dry_run=True) assert result.repaired_count == 0 - assert result.success is False + assert result.success is True assert result.metrics == { "raw_materialization_candidate_count": 1.0, "raw_materialization_selected_count": 1.0, @@ -323,7 +323,7 @@ def test_raw_materialization_replays_same_native_when_index_raw_link_is_dangling result = repair_mod.repair_raw_materialization(config, dry_run=True) - assert result.success is False + assert result.success is True assert result.repaired_count == 0 assert result.metrics["raw_materialization_candidate_count"] == 1.0 @@ -686,7 +686,7 @@ def test_raw_materialization_replays_parsed_rows_when_index_is_empty(tmp_path: P result = repair_mod.repair_raw_materialization(config, dry_run=True) assert result.repaired_count == 0 - assert result.success is False + assert result.success is True assert result.metrics["raw_materialization_candidate_count"] == 1.0 assert result.metrics["raw_materialization_already_parsed_count"] == 1.0 assert "already parsed but not materialized" in result.detail @@ -748,7 +748,7 @@ def test_raw_materialization_replays_parsed_rows_after_interrupted_index_rebuild result = repair_mod.repair_raw_materialization(config, dry_run=True) assert result.repaired_count == 0 - assert result.success is False + assert result.success is True assert result.metrics["raw_materialization_candidate_count"] == 1.0 assert result.metrics["raw_materialization_already_parsed_count"] == 1.0 assert "already parsed but not materialized" in result.detail @@ -1429,7 +1429,7 @@ def test_raw_materialization_dry_run_reports_limited_selection( result, incomplete_censuses = _complete_bounded_raw_census(config, limit=2) assert len(incomplete_censuses) == 1 - assert result.success is False + assert result.success is True assert result.repaired_count == 0 assert "Would: classify and replay" in result.detail assert result.metrics["raw_materialization_candidate_count"] == 4.0 @@ -2095,7 +2095,7 @@ def test_raw_materialization_reports_the_active_custom_payload_envelope(tmp_path result = repair_mod.repair_raw_materialization(_config(tmp_path), dry_run=True, max_payload_bytes=max_payload_bytes) - assert result.success is False + assert result.success is True assert result.metrics["raw_materialization_execute_blob_limit_bytes"] == float(max_payload_bytes) assert "100 B" in result.detail assert "1.0 GiB" not in result.detail From 56551862ebfea5d99e087e5d093be4301955342f Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 1 Aug 2026 00:03:48 +0200 Subject: [PATCH 4/4] fix(test): satisfy mypy --strict for the 9dxn/5q2u/f57q test additions devtools verify --quick's mypy step failed on two additions from this branch's earlier commits: - test_raw_authority_ledger.py: _seed_ambiguous_membership_component (polylogue-9dxn) was typed to return tuple[str, object]; narrow it to tuple[str, RawReplayPlanOutcome] so .status/.reason access type-checks. - test_revision_backfill.py: the deferred-tail-hit counting_wrapper (polylogue-5q2u) took *args: object/**kwargs: object, which mypy rejects against _reextract_prefix_tail_db's concrete signature; switch to *args: Any/**kwargs: Any (the standard shape for an untyped passthrough wrapper around a function whose exact signature the test does not care about). Verification: python -m mypy -> Success: no issues found in 2426 source files. devtools test tests/unit/storage/test_raw_authority_ledger.py tests/unit/sources/test_revision_backfill.py -> 97 passed, 1 pre-existing unrelated failure (test_parse_one_still_replays_real_claude_code_sessions_with_no_path_rule). --- tests/unit/sources/test_revision_backfill.py | 3 ++- tests/unit/storage/test_raw_authority_ledger.py | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/unit/sources/test_revision_backfill.py b/tests/unit/sources/test_revision_backfill.py index e164a166f9..34547d8542 100644 --- a/tests/unit/sources/test_revision_backfill.py +++ b/tests/unit/sources/test_revision_backfill.py @@ -5,6 +5,7 @@ import time from io import BytesIO from pathlib import Path +from typing import Any import pytest @@ -2558,7 +2559,7 @@ def _count_deferred_tail_hits(root: Path, *, force_lexicographic: bool) -> int: calls = 0 original = archive_tier_write._reextract_prefix_tail_db - def counting_wrapper(*args: object, **kwargs: object) -> object: + def counting_wrapper(*args: Any, **kwargs: Any) -> Any: nonlocal calls calls += 1 return original(*args, **kwargs) diff --git a/tests/unit/storage/test_raw_authority_ledger.py b/tests/unit/storage/test_raw_authority_ledger.py index 57c5444bae..ca79c02b1e 100644 --- a/tests/unit/storage/test_raw_authority_ledger.py +++ b/tests/unit/storage/test_raw_authority_ledger.py @@ -1270,7 +1270,7 @@ def _seed_ambiguous_membership_component( *, native_id: str, parser_fingerprint: str | None, -) -> tuple[str, object]: +) -> tuple[str, RawReplayPlanOutcome]: """Seed one raw whose membership decision is durably 'ambiguous'. ``parser_fingerprint`` controls what (if anything) the per-raw