fix(storage): make membership-replay conflicts retry-eligible by type

polylogue/storage/repair.py

@@ -103,6 +103,16 @@
 RAW_MATERIALIZATION_COMMIT_BATCH_SIZE = 20
 RAW_MATERIALIZATION_OUTCOME_SAMPLE_LIMIT = 8
 _TRANSIENT_LOCK_PARSE_ERROR = "OperationalError: database is locked"
+#: polylogue-5iz4: ``MembershipReplayConflictError``'s ``parse_error`` text
+#: (``f"{type(exc).__name__}: {exc}"``, set by ``mark_raw_parse_failed`` in
+#: ``storage/sqlite/archive_tiers/revision_governance.py``) always starts
+#: with this exact type name regardless of the exception's own
+#: human-readable message wording, which is not itself stable (it has
+#: already drifted once since PR #2718 introduced this guard under
+#: different phrasing). Matching on the type name here is what keeps a raw
+#: that hit this guard retry-eligible even after the message wording
+#: changes again.
+_MEMBERSHIP_REPLAY_CONFLICT_ERROR_PREFIX = "MembershipReplayConflictError:"
 _QUARANTINED_ACCEPTED_RAW_REPAIR_DETAIL = "repair:accepted_quarantined_raw_exact_byte_and_semantic_proof"
 _QUARANTINED_ACCEPTED_RAW_REPAIR_LIMIT = 100
 _QUARANTINED_ACCEPTED_RAW_REPAIR_BLOB_LIMIT_BYTES = 256 * 1024 * 1024
@@ -3910,6 +3920,19 @@ def _raw_materialization_candidate_ids(
                 OR (
                   r.parse_error LIKE 'decode:%No such file or directory:%'
                 )
+                OR (
+                  -- polylogue-5iz4: MembershipReplayConflictError
+                  -- (storage/sqlite/archive_tiers/revision_governance.py) is a
+                  -- transient, retry-eligible refusal by construction -- a
+                  -- later pass over the SAME durable raw bytes can succeed
+                  -- once sibling evidence resolves or the accepted head
+                  -- itself changes. Matching by exception TYPE name (stable)
+                  -- rather than the human-readable message text (which has
+                  -- already drifted twice since #2718 introduced this guard)
+                  -- is what keeps this retry-eligible even after the guard's
+                  -- own wording changes again.
+                  r.parse_error LIKE '{_MEMBERSHIP_REPLAY_CONFLICT_ERROR_PREFIX}%'
+                )
               )
               AND NOT (
                 COALESCE(r.validation_status, '') = 'skipped'
@@ -4143,8 +4166,19 @@ def _raw_materialization_component_stream_safe(
 def _raw_materialization_retryable_missing_blob_error(parse_error: object) -> bool:
     if not isinstance(parse_error, str):
         return False
-    return parse_error == _TRANSIENT_LOCK_PARSE_ERROR or (
-        parse_error.startswith("decode:") and "No such file or directory" in parse_error
+    return (
+        parse_error == _TRANSIENT_LOCK_PARSE_ERROR
+        or (parse_error.startswith("decode:") and "No such file or directory" in parse_error)
+        # polylogue-5iz4: MembershipReplayConflictError
+        # (storage/sqlite/archive_tiers/revision_governance.py) is a
+        # transient, retry-eligible refusal by construction -- see the SQL
+        # candidate query's matching clause above for the full rationale.
+        # This Python-side check re-validates every row the SQL WHERE
+        # clause already passed and is the true single source of truth for
+        # retry eligibility (the SQL clause is a pre-filter, not merely an
+        # optimization the SQL and this function must independently agree,
+        # or a row that clears the SQL gate is silently re-excluded here).
+        or parse_error.startswith(_MEMBERSHIP_REPLAY_CONFLICT_ERROR_PREFIX)
     )
 
 

polylogue/storage/sqlite/archive_tiers/archive.py

@@ -159,6 +159,7 @@
 from polylogue.storage.sqlite.archive_tiers.revision_governance import (
     ActiveByteRevisionChainError,
     ArchiveRawParsedWriteResult,
+    MembershipReplayConflictError,
     _authorize_full_snapshot_fold,
     _flush_pending_raw_parse_states,
     _index_parsed_for_retained_raw,
@@ -11590,4 +11591,5 @@ def _month_bucket_end_ms(bucket: str) -> int:
     "ArchiveStore",
     "ArchiveSessionSearchHit",
     "ArchiveSessionSummary",
+    "MembershipReplayConflictError",
 ]

polylogue/storage/sqlite/archive_tiers/revision_governance.py

@@ -163,6 +163,31 @@ class ActiveByteRevisionChainError(RuntimeError):
     """A byte-identical revision chain cannot admit a conflicting sibling."""
 
 
+class MembershipReplayConflictError(RuntimeError):
+    """Membership replay refused to move an accepted head this pass.
+
+    Raised by ``apply_raw_membership_classification`` when it cannot safely
+    retire or replace the currently accepted ``raw_revision_heads`` row for a
+    logical identity (an unrelated accepted head, or a head with unresolved
+    byte-append evidence still hanging off it). This is a **transient,
+    retry-eligible** refusal, not proof the raw is unparseable: a later pass
+    over the same durable raw bytes can succeed once sibling evidence
+    resolves or the accepted head itself changes (polylogue-5iz4).
+
+    A dedicated subclass exists so ``mark_raw_parse_failed``'s ``parse_error``
+    text (``f"{type(exc).__name__}: {exc}"``) carries a stable, matchable
+    marker for retry-eligibility checks (``storage/repair.py``'s raw
+    materialization candidate query) independent of this class's own
+    human-readable message wording, which has already drifted twice (#2718's
+    original "unconvertible byte head" phrasing no longer appears anywhere in
+    this module) and will keep drifting as the guard is refined. A plain
+    ``RuntimeError`` gives the retry-candidate query nothing durable to match
+    on beyond the exact message text, which is how a raw that hit this guard
+    under old wording got silently excluded from every future rebuild even
+    after the guard's conditions no longer applied to it.
+    """
+
+
 class RawRevisionGovernanceHost(Protocol):
     """The narrow slice of ``ArchiveStore`` this module is allowed to touch.
 
@@ -2444,7 +2469,7 @@ def apply_raw_membership_classification(
                         or persisted_session is None
                         or (persisted_raw != existing_raw_id and persisted_raw not in classified_raw_ids)
                     ):
-                        raise RuntimeError(
+                        raise MembershipReplayConflictError(
                             "membership replay cannot retire an unrelated accepted head: "
                             f"logical_source_key={logical_source_key!r} "
                             f"existing_head(raw_id={existing_raw_id!r}, session_id={str(existing_head[3])!r}, "
@@ -2504,7 +2529,7 @@ def apply_raw_membership_classification(
                             (logical_source_key, existing_raw_id, *classified_raw_ids, existing_raw_id),
                         ).fetchone()
                         if dangling_append is not None:
-                            raise RuntimeError(
+                            raise MembershipReplayConflictError(
                                 "membership replay cannot replace a head with unresolved byte-append "
                                 f"evidence: logical_source_key={logical_source_key!r} "
                                 f"existing_head(raw_id={existing_raw_id!r})"

tests/unit/sources/test_live_batch_support.py

@@ -4570,6 +4570,15 @@ def session(native_id: str, *texts: str) -> ParsedSession:
         assert parse_error is None
     else:
         assert parse_error is not None
+        # polylogue-5iz4: this guard's refusal is transient/retry-eligible by
+        # construction (a later pass over the same durable bytes can succeed
+        # once sibling evidence resolves), but a plain RuntimeError leaves the
+        # retry-candidate query (storage/repair.py) nothing stable to match
+        # once the message text drifts -- exactly what happened to a real
+        # production session that hit this guard under #2718's original
+        # wording and was never retried again. MembershipReplayConflictError
+        # gives that query a message-text-independent marker to key on.
+        assert parse_error.startswith("MembershipReplayConflictError:")
     with sqlite3.connect(index_db) as conn:
         assert conn.execute("SELECT message_count FROM sessions WHERE native_id = 'shared'").fetchone() == (2,)
         head_after = conn.execute(

tests/unit/storage/test_repair.py

@@ -416,6 +416,76 @@ def test_raw_materialization_retries_typed_transient_lock_failure(tmp_path: Path
         ).fetchone() == (1, None)
 
 
+def test_raw_materialization_retries_membership_replay_conflict_failure(tmp_path: Path) -> None:
+    """polylogue-5iz4: a MembershipReplayConflictError parse_error is retryable.
+
+    ``apply_raw_membership_classification``'s two head-conflict guards
+    (storage/sqlite/archive_tiers/revision_governance.py) are explicitly
+    transient/retry-eligible refusals, not proof a raw is unparseable -- a
+    later pass over the same durable bytes can succeed once sibling evidence
+    resolves or the accepted head changes. But the raw materialization
+    candidate query only recognizes retry-eligible parse_error text through a
+    literal allowlist; a plain ``RuntimeError`` message (the type both guards
+    raised before this fix, and the type a real production Codex session's
+    parse_error was frozen as under PR #2718's now-superseded wording) is
+    indistinguishable from a genuinely terminal parse failure and is silently
+    excluded from every future rebuild attempt forever. Anti-vacuity: a
+    sibling row carrying the OLD plain-``RuntimeError`` text must remain
+    excluded -- this test would pass vacuously (0 candidates either way) if
+    the candidate query's parse_error filter were simply deleted instead of
+    extended.
+    """
+    config = _config(tmp_path)
+    initialize_archive_database(tmp_path / "source.db", ArchiveTier.SOURCE)
+    initialize_archive_database(tmp_path / "index.db", ArchiveTier.INDEX)
+    blob_store = BlobStore(tmp_path / "blob")
+    retryable_raw_id, retryable_size = blob_store.write_from_bytes(b'{"mapping":{}}')
+    stale_raw_id, stale_size = blob_store.write_from_bytes(b'{"mapping":{"stale":{}}}')
+
+    with sqlite3.connect(tmp_path / "source.db") as source_conn:
+        source_conn.executemany(
+            """
+            INSERT INTO raw_sessions (
+                raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size,
+                acquired_at_ms, parsed_at_ms, parse_error
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            [
+                (
+                    retryable_raw_id,
+                    "codex-session",
+                    "native-membership-conflict",
+                    "whale.jsonl",
+                    0,
+                    bytes.fromhex(retryable_raw_id),
+                    retryable_size,
+                    2,
+                    3,
+                    "MembershipReplayConflictError: membership replay cannot replace a head "
+                    "with unresolved byte-append evidence: logical_source_key='codex:whale'",
+                ),
+                (
+                    stale_raw_id,
+                    "codex-session",
+                    "native-stale-plain-runtime-error",
+                    "stale.jsonl",
+                    0,
+                    bytes.fromhex(stale_raw_id),
+                    stale_size,
+                    1,
+                    4,
+                    "RuntimeError: membership replay cannot replace an unconvertible byte head",
+                ),
+            ],
+        )
+        source_conn.commit()
+
+    result = repair_mod.repair_raw_materialization(config, dry_run=True)
+
+    assert result.metrics["raw_materialization_candidate_count"] == 1.0
+    assert result.metrics["raw_materialization_total_blob_bytes"] == float(retryable_size)
+
+
 def test_raw_materialization_split_root_classifies_parsed_sidecar_from_routed_blob(tmp_path: Path) -> None:
     configured_root = tmp_path / "configured"
     routed_root = tmp_path / "routed"