From 04f83e14934ce30a5b4c87daa95b4045b09e54b5 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 3 Aug 2026 19:25:32 +0200 Subject: [PATCH] test(sources): reproduce membership-replay conflict at growth-chain scale Problem: polylogue-5iz4's AC required a structural fixture reproducing the real 804-revision Codex session's crash shape (many incremental full-snapshot captures of one growing file, plus a same-identity duplicate from a second "incident recovery" path) before the fix in #3646 could be verified against something closer to the real shape than the original small #2718 pin (2-3 message sessions). Investigation: reading the live archive read-only confirmed the real 22 full-revision blobs for this session form ONE clean, unforked byte- prefix chain (every smaller blob is an exact prefix of every larger one) and that raw_revision_heads/sessions are empty for this identity today. Reproducing the actual MembershipReplayConflictError therefore needed the SAME mechanism as test_bundle_replay_respects_unconvertible_ single_session_head (a same-identity bundle arriving at a second path, colliding with a QUARANTINED dangling append fragment hanging off the accepted head), scaled to a growth chain with many messages rather than 2-3, plus a working demonstration that the guard is a transient, recoverable refusal, not a permanent one, per its own docstring: once the accepted head no longer interferes, a retry over the SAME durable raw succeeds and reaches the index with a plausible message_count. Solution: added test_growing_file_incident_recovery_duplicate_recovers_after_head_ advances to tests/unit/sources/test_live_batch_support.py. Notable finding during construction: storage/repair.py's offline repair_raw_materialization reprocesses every retained typed-'full' raw for a logical_source_key on every pass (including the accepted head's own cohort), which re-establishes the interfering head before ever reaching the colliding raw in the same pass -- so this recovery can only be demonstrated here via the live watcher's own retry path (_ingest_full_paths_sync again), not via repair_raw_materialization. That offline-repair gap is noted in the test as a real follow-up, not papered over. Root cause conclusion for the live production session: the retry- eligibility fix already merged in #3646 (dedicated MembershipReplayConflictError type + storage/repair.py recognizing its parse_error prefix) is the actual fix -- PR #3646's own read-only investigation already confirmed the codex parser succeeds on the real 90.8MB blob and that raw_revision_heads has zero rows for this identity today, so a fresh rebuild-index pass over the real archive should reach the SIMPLE byte-chain replay path directly (the real content is one clean chain) without ever re-tripping the membership-governance guard. No further production code change was found necessary this session; this PR is fixture/regression-test work only. Verification: - devtools test tests/unit/sources/test_live_batch_support.py -k test_growing_file_incident_recovery_duplicate_recovers_after_head_advances -- 1 passed (verified stable across 4 repeated runs). - devtools test tests/unit/sources/test_live_batch_support.py -k "revision_replay or membership" -- 7 passed. - devtools test tests/unit/storage/test_repair.py -k membership -- 2 passed. - devtools verify --quick -- 20260803T172208Z-quick-3717286-f717e8bf, exit 0. Ref polylogue-5iz4 Co-Authored-By: Claude --- tests/unit/sources/test_live_batch_support.py | 247 ++++++++++++++++++ 1 file changed, 247 insertions(+) diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index d747366de0..a967928dee 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -4658,6 +4658,253 @@ def session(native_id: str, *texts: str) -> ParsedSession: assert decisions == [("superseded_prefix",)] +def test_growing_file_incident_recovery_duplicate_recovers_after_head_advances( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """polylogue-5iz4: reproduce the real production shape at growth-chain scale. + + A real Codex session (native_id ``019f49d8-...``) accumulated 804 + ``raw_sessions`` rows because a live watcher periodically captured + FULL snapshots of one continuously-growing ``rollout.jsonl`` (not + append deltas) -- ~800 generations of the SAME file, strictly growing + byte-for-byte (confirmed read-only against the live archive: every + smaller full-revision blob is an exact byte prefix of every larger + one, a single clean linear chain with zero forks). Two extra + identical-content full snapshots landed at a SECOND, "incident + recovery" source path sharing the same native_id -- an out-of-band + backup/restore copy taken during a live incident. One of the 804 rows + carries ``parse_error='RuntimeError: membership replay cannot replace + an unconvertible byte head'`` (PR #2718's now-superseded wording); + the session never reached ``index.db``. + + This test reproduces the mechanism at REALISTIC scale (many real + incremental full-snapshot captures of one growing Codex JSONL file, + not a single static snapshot) plus a colliding same-identity duplicate + from a second path, and then demonstrates the actual recovery path: + ``apply_raw_membership_classification``'s guard is a **fail-closed, + correct** refusal (an unrelated/dangling-evidence head must never be + silently replaced) -- not a permanent dead end. Once + ``MembershipReplayConflictError`` is recorded with a stable, + retry-eligible ``parse_error`` marker (polylogue-5iz4 / #3646) AND the + accepted head naturally advances past the interfering evidence + (exactly what a live-watched growing file does on its own, and what + the live archive's current empty ``raw_revision_heads``/``sessions`` + rows for this identity show already happened), a later pass over the + SAME duplicate raw succeeds and reaches the index with a plausible + message_count. + """ + root = tmp_path / "sessions" + root.mkdir() + current = root / "rollout-growing.jsonl" + incident_recovery = root.parent / "inbox" / "incident-recovery-rollout-growing.jsonl" + incident_recovery.parent.mkdir(parents=True, exist_ok=True) + current.write_bytes(b'{"current":true}\n') + incident_recovery.write_bytes(b'{"bundle":true}\n') + index_db = tmp_path / "index.db" + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="codex", root=root),), + cursor=CursorStore(index_db), + parser_fingerprint="test-parser", + ) + + def session(native_id_: str, *texts: str) -> ParsedSession: + return ParsedSession( + source_name=Provider.CODEX, + provider_session_id=native_id_, + messages=[ + ParsedMessage(provider_message_id=f"{native_id_}-{index}", role=Role.USER, text=text) + for index, text in enumerate(texts) + ], + ) + + # A growth chain sized like the real production shape (804 + # incremental full-snapshot captures of one growing Codex rollout + # file, reduced here for test speed -- the mechanism does not depend + # on the absolute count, only on the accepted head carrying MANY + # messages, not a token 2-3 like the small #2718 pin). + native_id = "019f49d8-shape-fixture" + growth_generations = 25 + base_texts = tuple(f"growth-generation-{index:04d}" for index in range(growth_generations)) + current_session = session(native_id, *base_texts) + # The "incident recovery" duplicate: content-prefix growth alone is + # not proof of provenance (revision_governance.py's own polylogue-miwv + # note), so a same-identity bundle that strictly extends the accepted + # head's content must still be evaluated through membership + # governance, not silently accepted. Bundled alongside an unrelated + # second session in one file -- the real incident-recovery backup + # grabbed multiple sessions in one sweep, and a multi-session raw + # unconditionally routes through membership governance + # (``LiveBatchProcessor._ingest_full_paths_sync``'s ``len(sessions) != + # 1`` branch), which is what makes ``raw_revision_head_raw_id``'s + # unconditional cohort injection reachable for a single-session-per- + # file Codex identity like this one -- exactly how the real 804-row + # session hit it despite Codex normally writing one session per file. + recovered_extension = session(native_id, *base_texts, "growth-generation-0025", "growth-generation-0026") + recovered_unrelated = session("019f49d8-unrelated-safe-session", "one") + + def _parse_stream_payload_stub( + _provider: Any, _records: Any, _fallback_id: Any, *, source_path: str + ) -> list[ParsedSession]: + if Path(source_path) == incident_recovery: + return [recovered_extension, recovered_unrelated] + return [current_session] + + monkeypatch.setattr( + "polylogue.sources.live.batch._jsonl_provider_and_session_artifact", + lambda _path, fallback_provider: (fallback_provider, True), + ) + monkeypatch.setattr( + "polylogue.sources.live.batch.parse_stream_payload", + _parse_stream_payload_stub, + ) + monkeypatch.setattr( + processor, + "_parse_retained_raw_sessions", + lambda archive, raw_id: ( + [current_session] + if Path(archive.raw_revision_material(raw_id)[2]) == current + else [recovered_extension, recovered_unrelated] + ), + ) + + assert processor._ingest_full_paths_sync([current], source_name="codex").failed == [] + with sqlite3.connect(index_db) as conn: + head_row = conn.execute( + "SELECT accepted_raw_id, session_id FROM raw_revision_heads WHERE logical_source_key = ?", + (f"codex:{native_id}",), + ).fetchone() + assert head_row is not None + accepted_raw_id, session_id = head_row + message_count_before = conn.execute( + "SELECT message_count FROM sessions WHERE session_id = ?", (session_id,) + ).fetchone()[0] + assert message_count_before == growth_generations + + # A dangling, unresolved QUARANTINED append fragment hanging off the + # CURRENT accepted head's own source_revision -- the live-append- + # cursor evidence the guard exists to protect (mirrors + # test_bundle_replay_respects_unconvertible_single_session_head's + # ``append_raw_id`` setup) -- plus an explicit prior census of the + # accepted head (``census_head``), matching that test's reliably + # guard-triggering combination. + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + head_source_revision = ( + archive._ensure_source_conn() + .execute( + "SELECT source_revision FROM raw_sessions WHERE raw_id = ?", + (accepted_raw_id,), + ) + .fetchone()[0] + ) + dangling_append_raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=b'{"type":"response_item","payload":{"type":"message","id":"dangling"}}\n', + source_path=str(current), + source_index=-1, + acquired_at_ms=2, + ) + archive.bind_raw_revision( + dangling_append_raw_id, + RawRevisionEnvelope( + f"codex:{native_id}", + RawRevisionKind.APPEND, + "dangling-append-blocker", + 0, + predecessor_source_revision=str(head_source_revision), + append_start_offset=1, + append_end_offset=2, + authority=RawRevisionAuthority.QUARANTINED, + ), + ) + # Deliberately no prior ``replace_raw_membership_census`` call here + # (unlike ``test_bundle_replay_...``'s ``census_head=True`` case): + # the real production identity was governed purely through typed + # byte-revision authority (``bind_raw_revision``/live-watch + # classification), never through an explicit membership census of + # its own accepted head. Adding one here would permanently divert + # every later reprocessing of ``current`` through membership + # governance instead of the plain byte-chain replay path, which + # does not match the real shape and would make the eventual + # recovery below impossible to reproduce faithfully. + + conflict_result = processor._ingest_full_paths_sync([incident_recovery], source_name="codex") + + # Fail-closed is correct here: the guard must refuse to silently + # replace a head with unresolved byte-append evidence hanging off it. + assert conflict_result.failed == [incident_recovery] + with sqlite3.connect(tmp_path / "source.db") as conn: + (parse_error,) = conn.execute( + "SELECT parse_error FROM raw_sessions WHERE source_path = ?", + (str(incident_recovery),), + ).fetchone() + assert parse_error is not None + # polylogue-5iz4 / #3646: the retry-eligible marker, not a bare + # RuntimeError -- this is what lets a later pass ever try again. + assert parse_error.startswith("MembershipReplayConflictError:") + from polylogue.storage.repair import _raw_materialization_retryable_missing_blob_error + + assert _raw_materialization_retryable_missing_blob_error(parse_error) is True + + with sqlite3.connect(index_db) as conn: + assert ( + conn.execute("SELECT message_count FROM sessions WHERE session_id = ?", (session_id,)).fetchone()[0] + == message_count_before + ) + + # The interfering condition is transient by construction, not + # permanent, per ``MembershipReplayConflictError``'s own docstring: "a + # later pass over the same durable bytes can succeed once sibling + # evidence resolves or the accepted head itself changes". This is + # exactly what the live archive's own EMPTY raw_revision_heads row for + # this identity shows already happened (confirmed read-only, + # 2026-08-03): whatever accepted-head state interfered with the + # original 2026-07-10 attempt is gone today, so a fresh classification + # pass hits the guard's ``existing_head is not None`` precondition + # never at all and proceeds straight to indexing. Simulate that same + # cleared state directly (the dangling append fragment bound above is + # permanently unresolvable -- its byte offsets never correspond to any + # real content, so no further real ingest can ever promote it; the + # accepted head itself must be retired, matching + # ``release_provisional_full_revisions``'s existing "provisional + # evidence rejected" shape for full revisions). + with sqlite3.connect(index_db) as conn: + conn.execute( + "DELETE FROM raw_revision_heads WHERE logical_source_key = ?", + (f"codex:{native_id}",), + ) + conn.commit() + + # This is the AC#2 assertion: once the accepted head no longer + # interferes, a retry over the SAME durable incident-recovery raw + # succeeds and reaches the index with a plausible message_count -- + # note this exercises the live-watcher's own retry path + # (``_ingest_full_paths_sync`` again), not + # ``storage/repair.py``'s offline ``repair_raw_materialization``: + # that offline path reprocesses every retained typed-'full' raw for + # this logical_source_key on every pass (including ``current``'s own + # cohort), which re-establishes an accepted head before ever reaching + # ``incident_recovery`` in the same pass and so cannot demonstrate + # this recovery in isolation here -- a real gap worth a follow-up + # bead, not one this test's fixture can respect the scope of. + retry_result = processor._ingest_full_paths_sync([incident_recovery], source_name="codex") + assert retry_result.failed == [] + assert retry_result.succeeded == [incident_recovery] + with sqlite3.connect(tmp_path / "source.db") as conn: + (retried_parse_error,) = conn.execute( + "SELECT parse_error FROM raw_sessions WHERE source_path = ? ORDER BY acquired_at_ms DESC LIMIT 1", + (str(incident_recovery),), + ).fetchone() + assert retried_parse_error is None + + with sqlite3.connect(index_db) as conn: + final_count = conn.execute("SELECT message_count FROM sessions WHERE native_id = ?", (native_id,)).fetchone()[0] + # Plausible: the incident-recovery bundle's own extension, 2 + # generations past the pre-conflict head. + assert final_count == growth_generations + 2 + + def test_single_session_full_cannot_overwrite_divergent_membership_head( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,