fix(sources): fix cross-material isolation guard + dedup census parse by blob_hash

polylogue/archive/revision_authority.py

@@ -23,6 +23,16 @@ class RawRevisionAuthority(StrEnum):
 
 BYTE_AUTHORITY_CENSUS_DETAIL = "append fragments are governed by byte revision authority"
 
+#: ``raw_membership_census.detail`` marker written when a full-only,
+#: non-prefix-chain cohort is retired from byte-revision governance to
+#: membership governance (``ArchiveStore.replace_raw_membership_census``,
+#: ``retire_full_revision_governance=True``). Shared between the writer and
+#: ``ArchiveStore.classify_raw_revision_cohort``'s polylogue-52l2 guard,
+#: which uses it to detect that a logical identity already has retired,
+#: previously-ambiguous sibling evidence before ever accepting a
+#: later-discovered raw as an unconditional singleton byte-proven baseline.
+HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL = "historical non-prefix full revision governance"
+
 
 @dataclass(frozen=True)
 class RawRevisionEnvelope:

polylogue/sources/live/batch.py

@@ -24,6 +24,7 @@
     NATIVE_BROWSER_CAPTURE_INGEST_FLAG,
 )
 from polylogue.archive.revision_authority import (
+    HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL,
     RawRevisionAuthority,
     RawRevisionEnvelope,
     RawRevisionKind,
@@ -1983,7 +1984,7 @@ def _apply_membership_sessions(
                     retained_sessions,
                     parser_fingerprint=self._current_parser_fingerprint(),
                     censused_at_ms=acquired_at_ms,
-                    detail="cross-route full revision governance",
+                    detail=HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL,
                     retire_full_revision_governance=True,
                 )
             member_sessions: dict[str, Any] = {}

polylogue/sources/revision_backfill.py

@@ -25,6 +25,7 @@
 )
 from polylogue.archive.revision_authority import (
     BYTE_AUTHORITY_CENSUS_DETAIL,
+    HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL,
     RawRevisionAuthority,
     RawRevisionEnvelope,
     RawRevisionKind,
@@ -760,7 +761,7 @@ def commit_replay_unit() -> None:
                         sessions,
                         parser_fingerprint="revision-membership-v1",
                         censused_at_ms=0,
-                        detail="historical non-prefix full revision governance",
+                        detail=HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL,
                         retire_full_revision_governance=True,
                     )
                     membership_candidates.setdefault(logical_key, set()).add(raw_id)
@@ -1044,6 +1045,33 @@ def _pool_dispatch_amortizes(pool_raw_ids: list[str], payload_sizes: dict[str, i
     return total >= _parse_pool_min_aggregate_bytes()
 
 
+#: Providers whose parsed session identity is derived purely from payload
+#: bytes, never from ``source_path`` -- safe to dedup census parse ACROSS
+#: source paths sharing a ``blob_hash`` (polylogue-869u). Excluded
+#: deliberately: ``Provider.BEADS`` derives workspace-scoped native ids from
+#: ``source_path`` (``sources/parsers/beads.py:_repository_root``);
+#: ``Provider.ANTIGRAVITY``'s brain-metadata mode derives its
+#: ``profile_root``/artifact path from ``source_path``
+#: (``sources/dispatch.py``'s ``antigravity.parse_brain_metadata`` call);
+#: ``Provider.HERMES``'s ATOF/ATIF/verification-evidence modes likewise
+#: derive ``profile_root`` from ``source_path``. Those three keep the
+#: conservative same-path-only dedup below. ``Provider.UNKNOWN`` (browser
+#: capture / unclassified) is also excluded out of caution -- its identity
+#: derivation is not centrally audited here.
+_PATH_INDEPENDENT_PARSE_PROVIDERS: Final[frozenset[Provider]] = frozenset(
+    {
+        Provider.CHATGPT,
+        Provider.CLAUDE_AI,
+        Provider.CLAUDE_CODE,
+        Provider.CODEX,
+        Provider.GEMINI,
+        Provider.GEMINI_CLI,
+        Provider.GROK,
+        Provider.DRIVE,
+    }
+)
+
+
 def _parse_retained_raws(
     archive: ArchiveStore,
     raw_ids: list[str],
@@ -1055,15 +1083,22 @@ def _parse_retained_raws(
 
     Returns each outcome keyed by raw_id: either the parsed
     ``(sessions, payload_bytes, revision_kind)`` tuple or the caught
-    exception. Rows sharing the same ``(blob_hash, source_path)`` are parsed
-    exactly once and the outcome fanned out: identical bytes at an identical
-    path decode deterministically identically, so re-parsing them is pure
-    waste (measured live 2026-07-19: 17% of newest-only bytes — e.g. one
-    442MB codex rollout stored under 8 raw rows — were byte-identical
-    duplicates each paying a full parse). ``source_path`` stays in the key
-    because some parsers derive identity from the path (e.g. beads workspace
-    ids), so cross-path duplicates are deliberately NOT deduplicated.
-    Per-row ``revision_kind`` is re-attached from each row's own descriptor.
+    exception. Rows sharing the same ``blob_hash`` for a
+    ``_PATH_INDEPENDENT_PARSE_PROVIDERS`` provider are parsed exactly once
+    and the outcome fanned out, regardless of ``source_path``: identical
+    bytes decode deterministically identically for those providers, so
+    re-parsing them per row (even under a DIFFERENT acquired path -- a
+    common re-acquisition/re-export shape) is pure waste. For every other
+    provider the dedup key still includes ``source_path`` (some parsers
+    derive identity from the path, e.g. beads workspace ids -- see
+    ``_PATH_INDEPENDENT_PARSE_PROVIDERS``'s docstring for the excluded
+    providers), so cross-path duplicates for those stay deliberately NOT
+    deduplicated. Live evidence (polylogue-869u, 2026-07-19): 87,177
+    newest-revision raws / 52.1 GiB but only 85,066 distinct blob hashes /
+    43.4 GiB -- the same bytes (e.g. one 442 MB codex rollout) recur under
+    up to 8 different ``logical_source_key``s / source paths and, before
+    this cross-path widening, paid a full parse each time. Per-row
+    ``revision_kind`` is re-attached from each row's own descriptor.
 
     ``prefetch_cache`` (polylogue-m6tp phase (a)) is consulted BEFORE any of
     the above: a raw_id already popped from the cache is used directly and
@@ -1083,10 +1118,11 @@ def _parse_retained_raws(
                 remaining_raw_ids.append(raw_id)
             else:
                 results[raw_id] = cached
-    grouped: dict[tuple[str, str], list[str]] = {}
+    grouped: dict[tuple[Provider, str, str], list[str]] = {}
     for raw_id in remaining_raw_ids:
-        _provider, blob_hash, source_path, _kind, _size = descriptors[raw_id]
-        grouped.setdefault((blob_hash, source_path), []).append(raw_id)
+        provider, blob_hash, source_path, _kind, _size = descriptors[raw_id]
+        dedup_path = "" if provider in _PATH_INDEPENDENT_PARSE_PROVIDERS else source_path
+        grouped.setdefault((provider, blob_hash, dedup_path), []).append(raw_id)
     representatives = [members[0] for members in grouped.values()]
     unique = _parse_unique_retained_raws(
         archive, representatives, descriptors=descriptors, ingest_workers=ingest_workers

polylogue/storage/sqlite/archive_tiers/archive.py

@@ -38,6 +38,7 @@
     QueryTextPredicate,
 )
 from polylogue.archive.revision_authority import (
+    HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL,
     HistoricalRawRevisionStream,
     RawRevisionAuthority,
     RawRevisionEnvelope,
@@ -1986,6 +1987,39 @@ def raw_append_revision_parent(
         row = rows[0]
         return str(row[0]), str(row[1]), int(row[2]) + 1
 
+    def raw_membership_retired_full_revision_siblings(self, logical_source_key: str) -> tuple[str, ...]:
+        """Return raws previously retired from full-revision byte governance for this key.
+
+        ``replace_raw_membership_census(..., retire_full_revision_governance=True)``
+        nulls the retired raw's ``raw_sessions.logical_source_key`` and sets
+        ``revision_authority='quarantined'`` -- it becomes invisible both to
+        ``classify_raw_revision_cohort``'s own byte-row query and to
+        ``raw_membership_rebuild_raw_ids``'s deliberately byte-proven-only
+        filter (polylogue-lkrc/#2822 guards a different hazard: reopening a
+        quarantined member against an already-established head). Its
+        ``raw_session_memberships`` row, keyed by the raw's own parsed
+        logical identity, survives retirement together with a
+        ``raw_membership_census.detail`` marker naming this specific
+        transition, so a later-arriving sibling for the same identity can
+        still be told this identity has known, unresolved ambiguous
+        evidence (polylogue-52l2) instead of being evaluated alone.
+        """
+        rows = (
+            self._ensure_source_conn()
+            .execute(
+                """
+                SELECT m.raw_id
+                FROM raw_session_memberships AS m
+                JOIN raw_membership_census AS c ON c.raw_id = m.raw_id
+                WHERE m.logical_source_key = ? AND c.detail = ?
+                ORDER BY m.raw_id
+                """,
+                (logical_source_key, HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL),
+            )
+            .fetchall()
+        )
+        return tuple(str(row[0]) for row in rows)
+
     def classify_raw_revision_cohort(self, logical_source_key: str) -> RevisionReplayPlan:
         """Promote only a unique byte-prefix full chain and contiguous appends."""
         if self._blob_publisher is None:
@@ -1999,6 +2033,26 @@ def classify_raw_revision_cohort(self, logical_source_key: str) -> RevisionRepla
             """,
             (logical_source_key,),
         ).fetchall()
+        # polylogue-52l2: a byte chain is classified against whichever full
+        # rows the caller happens to have discovered/censused so far, not
+        # against the complete sibling population for this logical identity
+        # -- an earlier pass can have already retired ambiguous siblings to
+        # membership governance (nulling their logical_source_key, see
+        # raw_membership_retired_full_revision_siblings). If that leaves a
+        # later-discovered raw as the ONLY remaining 'full' row here, it
+        # would be evaluated as a trivial one-member "chain" and
+        # unconditionally accepted as a byte-proven baseline by
+        # classify_historical_full_revision_streams (no sibling to prove a
+        # byte prefix against) -- permanently establishing session content
+        # from whichever raw happened to be discovered last, independent of
+        # which content is actually correct. Refuse the byte-chain path
+        # entirely whenever this identity has retired sibling evidence: the
+        # caller's existing "no accepted chain" fallback
+        # (convertible_full_revision_raw_ids) folds these full rows into
+        # membership governance instead, where the real prefix-based
+        # classifier weighs every known sibling together.
+        if full_rows and self.raw_membership_retired_full_revision_siblings(logical_source_key):
+            full_rows = []
         historical: list[HistoricalRawRevisionStream] = []
         for row in full_rows:
 

tests/unit/sources/test_revision_backfill.py

@@ -1122,12 +1122,15 @@ def test_partition_raws_by_dispatch_size_routes_small_to_pool_large_sequential()
     assert sequential_ids == ["large-1", "large-2"]
 
 
-def test_parse_retained_raws_dedupes_identical_blob_and_path(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Byte-identical rows at the same source_path parse once and the outcome
-    fans out with each row's own revision_kind; the same bytes at a DIFFERENT
-    path still parse separately (path participates in some parsers' identity).
-    Live shape (polylogue-869u): one 442MB codex rollout acquired 8x within
-    2.3s during a stampede — 8 raw rows, one blob, 8 full parses."""
+def test_parse_retained_raws_dedupes_identical_blob_across_paths_for_safe_providers(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """polylogue-869u: for a path-independent provider (Codex here), rows
+    sharing a ``blob_hash`` parse once and fan out -- INCLUDING across
+    different ``source_path``s, since those parsers' session identity comes
+    entirely from the payload bytes. Live shape: one 442MB codex rollout
+    acquired 8x within 2.3s during a stampede, at up to 8 different acquired
+    paths -- 8 raw rows, one blob, formerly 8 full parses."""
     descriptors = {
         "dup-1": (Provider.CODEX, "hash-A", "same.jsonl", RawRevisionKind.FULL, 10),
         "dup-2": (Provider.CODEX, "hash-A", "same.jsonl", RawRevisionKind.UNKNOWN, 10),
@@ -1155,13 +1158,55 @@ def fake_parse(archive: object, raw_id: str) -> tuple[list[ParsedSession], int,
         ingest_workers=1,
     )
 
-    # one parse per distinct (blob_hash, source_path): dup-2/dup-3 reuse dup-1's
-    assert parsed == ["dup-1", "other-path", "other-bytes"]
+    # one parse per distinct blob_hash: dup-2/dup-3/other-path all reuse
+    # dup-1's outcome despite other-path having a different source_path.
+    assert parsed == ["dup-1", "other-bytes"]
     assert set(results) == set(descriptors)
     sessions, size, kind = results["dup-2"]  # type: ignore[misc]
     assert (sessions, size, kind) == ([], 10, RawRevisionKind.UNKNOWN)
     _sessions, _size, dup3_kind = results["dup-3"]  # type: ignore[misc]
     assert dup3_kind == RawRevisionKind.FULL
+    _sessions, other_path_size, other_path_kind = results["other-path"]  # type: ignore[misc]
+    assert (other_path_size, other_path_kind) == (10, RawRevisionKind.FULL)
+
+
+def test_parse_retained_raws_preserves_path_scoped_dedup_for_path_dependent_providers(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """polylogue-869u: Beads derives workspace-scoped native ids from
+    ``source_path`` (``sources/parsers/beads.py:_repository_root``), so
+    byte-identical Beads rows at DIFFERENT paths must keep parsing
+    separately -- unlike the path-independent providers, cross-path
+    identity is not provably safe here."""
+    descriptors = {
+        "same-path-a": (Provider.BEADS, "hash-A", "workspace-one/issues.jsonl", RawRevisionKind.FULL, 10),
+        "same-path-b": (Provider.BEADS, "hash-A", "workspace-one/issues.jsonl", RawRevisionKind.FULL, 10),
+        "other-path": (Provider.BEADS, "hash-A", "workspace-two/issues.jsonl", RawRevisionKind.FULL, 10),
+    }
+
+    class FakeArchive:
+        def raw_revision_descriptor(self, raw_id: str) -> tuple[Provider, str, str, RawRevisionKind, int]:
+            return descriptors[raw_id]
+
+    parsed: list[str] = []
+
+    def fake_parse(archive: object, raw_id: str) -> tuple[list[ParsedSession], int, RawRevisionKind]:
+        parsed.append(raw_id)
+        descriptor = descriptors[raw_id]
+        return [], descriptor[4], descriptor[3]
+
+    monkeypatch.setattr(revision_backfill, "_parse_retained_raw", fake_parse)
+
+    results = revision_backfill._parse_retained_raws(
+        FakeArchive(),  # type: ignore[arg-type]
+        list(descriptors),
+        ingest_workers=1,
+    )
+
+    # same-path-b reuses same-path-a's outcome; other-path (different
+    # source_path, same bytes) still pays its own parse.
+    assert parsed == ["same-path-a", "other-path"]
+    assert set(results) == set(descriptors)
 
 
 def test_parse_retained_raws_fans_out_exceptions_to_duplicate_rows(monkeypatch: pytest.MonkeyPatch) -> None:

tests/unit/storage/test_revision_replay.py

@@ -439,6 +439,103 @@ def parsed(*messages: tuple[str, str]) -> ParsedSession:
         assert tuple(head) == (folded, 20)
 
 
+def test_isolated_later_raw_does_not_override_known_ambiguous_cohort(tmp_path: Path) -> None:
+    """polylogue-52l2: a raw discovered for a logical identity that already
+    has quarantined/ambiguous siblings must not be accepted as an
+    unambiguous singleton byte-proven baseline.
+
+    This mirrors the LIVE incremental watcher's own call sequence
+    (``sources/live/batch.py``): ``bind_raw_revision`` then
+    ``classify_raw_revision_cohort`` directly, with no census-phase
+    re-derivation or connected-component re-expansion in between (those only
+    happen in the offline ``backfill_historical_revision_evidence`` path,
+    which is why this bug does not reproduce through that entry point).
+
+    ``classify_raw_revision_cohort`` only ever queries
+    ``raw_sessions WHERE logical_source_key = ? AND revision_kind = 'full'``.
+    Retiring an ambiguous sibling to membership governance
+    (``replace_raw_membership_census(..., retire_full_revision_governance=True)``,
+    exactly what the backfill caller does with
+    ``convertible_full_revision_raw_ids`` once a cohort is decided
+    ambiguous) nulls its ``raw_sessions.logical_source_key`` -- it becomes
+    invisible to that query. A THIRD raw for the same identity, discovered
+    afterward, is then evaluated completely alone:
+    ``classify_historical_full_revision_streams`` unconditionally accepts a
+    singleton stream as a "byte-proven baseline" (there is no sibling to
+    compare a byte-prefix against), so the isolated raw would permanently
+    become the accepted session content -- an outcome that depends on
+    incremental discovery order, not on which content is actually correct.
+    """
+    initialize_active_archive_root(tmp_path)
+
+    def parsed_solo(native_id: str, *texts: str) -> ParsedSession:
+        return ParsedSession(
+            source_name=Provider.CHATGPT,
+            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)
+            ],
+        )
+
+    with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
+        raw_a = archive.write_raw_payload(
+            provider=Provider.CHATGPT, payload=b"aaa-left", source_path="a.json", acquired_at_ms=1
+        )
+        archive.bind_raw_revision(
+            raw_a,
+            RawRevisionEnvelope(
+                "chatgpt:s1", RawRevisionKind.FULL, raw_a, 0, authority=RawRevisionAuthority.QUARANTINED
+            ),
+        )
+        raw_b = archive.write_raw_payload(
+            provider=Provider.CHATGPT, payload=b"bbb-right", source_path="b.json", acquired_at_ms=2
+        )
+        archive.bind_raw_revision(
+            raw_b,
+            RawRevisionEnvelope(
+                "chatgpt:s1", RawRevisionKind.FULL, raw_b, 0, authority=RawRevisionAuthority.QUARANTINED
+            ),
+        )
+
+        first_plan = archive.classify_raw_revision_cohort("chatgpt:s1")
+        assert first_plan.accepted_raw_ids == ()
+
+        # Both siblings genuinely disagree (no byte-prefix relation) --
+        # exactly what the backfill caller does when a cohort is decided
+        # ambiguous: move it to membership governance so parsed-content
+        # prefix rules can still arbitrate it later.
+        for raw_id, session in (
+            (raw_a, parsed_solo("s1", "base", "left")),
+            (raw_b, parsed_solo("s1", "base", "right")),
+        ):
+            archive.replace_raw_membership_census(
+                raw_id,
+                [session],
+                parser_fingerprint="revision-membership-v1",
+                censused_at_ms=0,
+                detail="historical non-prefix full revision governance",
+                retire_full_revision_governance=True,
+            )
+
+        # A THIRD raw for the same logical identity, discovered afterward.
+        raw_c = archive.write_raw_payload(
+            provider=Provider.CHATGPT, payload=b"ccc-solo", source_path="c.json", acquired_at_ms=3
+        )
+        archive.bind_raw_revision(
+            raw_c,
+            RawRevisionEnvelope(
+                "chatgpt:s1", RawRevisionKind.FULL, raw_c, 0, authority=RawRevisionAuthority.QUARANTINED
+            ),
+        )
+        second_plan = archive.classify_raw_revision_cohort("chatgpt:s1")
+
+    # The isolated raw must not be promoted alone: this identity has known,
+    # unresolved ambiguous siblings that a real classifier must weigh it
+    # against, not silently outrank by discovery order.
+    assert second_plan.accepted_raw_ids == ()
+
+
 def test_real_single_append_chain_folds_segmentation_distinct_full_snapshot(tmp_path: Path) -> None:
     initialize_active_archive_root(tmp_path)