fix(storage): refuse a precedence write for a raw recorded ambiguous membership

polylogue/storage/sqlite/archive_tiers/archive.py

@@ -1668,6 +1668,39 @@ def _write_parsed_precedence_result(
                 content_changed=False,
                 counts=self._skipped_counts(session),
             )
+        # polylogue-c737: ``governed`` above only catches a logical
+        # identity with an ACCEPTED revision-authority head
+        # (``raw_revision_heads``, populated by
+        # ``apply_raw_membership_classification``/``apply_raw_revision_replay``
+        # only when a cohort has a winner). A cohort that ``classify_
+        # membership_revisions`` refused to arbitrate -- genuinely
+        # ``raw_session_memberships.decision = 'ambiguous'`` -- never gets an
+        # accepted head, so ``governed`` stays ``None`` here even though this
+        # raw's own identity is recorded authority debt. Falling through to
+        # the ordinary browser-capture-precedence/freshness logic below then
+        # writes this raw's session unconditionally on its next parse --
+        # last-writer-wins, exactly the "never silently choose between
+        # branches" invariant this whole subsystem exists to enforce, and
+        # the fidelity-losing side of an aistudio-drive ambiguous pair reaches
+        # the index every time this reparses (measured live: 28 cohorts, 641
+        # attachments reported unfetched despite the bytes existing in the
+        # blob store). Refuse this raw explicitly instead of relying on an
+        # absent head to imply "unclaimed, free to write".
+        ambiguous_membership = (
+            self._ensure_source_conn()
+            .execute(
+                "SELECT 1 FROM raw_session_memberships WHERE raw_id = ? AND decision = 'ambiguous' LIMIT 1",
+                (raw_id,),
+            )
+            .fetchone()
+        )
+        if ambiguous_membership is not None:
+            return ArchiveRawParsedWriteResult(
+                raw_id=raw_id,
+                session_id=session_id,
+                content_changed=False,
+                counts=self._skipped_counts(session),
+            )
 
         if source_index >= 0 and existing_raw_id and raw_id and existing_raw_id != raw_id:
             existing_is_dom_fallback = session_has_parser_ingest_flag(

tests/unit/storage/test_revision_replay.py

@@ -2,6 +2,7 @@
 
 import hashlib
 import json
+import sqlite3
 from itertools import permutations
 from pathlib import Path
 
@@ -617,6 +618,71 @@ def parsed_solo(native_id: str, *texts: str) -> ParsedSession:
     assert second_plan.accepted_raw_ids == ()
 
 
+def test_precedence_write_refuses_a_raw_recorded_ambiguous(tmp_path: Path) -> None:
+    """A raw whose OWN logical identity is durably recorded
+    ``raw_session_memberships.decision = 'ambiguous'`` must never reach
+    ``sessions`` through the ordinary (non-revision-authoritative) parsed-
+    write path.
+
+    ``ArchiveStore._write_parsed_precedence_result``'s only revision-
+    authority awareness before this fix was a check against
+    ``raw_revision_heads`` -- populated ONLY when a cohort has an ACCEPTED
+    winner (``apply_raw_membership_classification``/
+    ``apply_raw_revision_replay``). A cohort ``classify_membership_
+    revisions`` genuinely refused to arbitrate never gets an accepted head,
+    so that check stays silent and the ordinary browser-capture-precedence/
+    freshness fallback below it writes the session unconditionally on the
+    next reparse -- arbitrary last-writer-wins over the exact invariant this
+    subsystem exists to enforce. Live evidence: 28 aistudio-drive cohorts
+    recorded ambiguous nonetheless materialized a session with 641
+    attachments reported unfetched despite the bytes existing in the blob
+    store, because ``write_parsed_for_retained_raw`` (called from the
+    one-shot importer, ``revision_authoritative=False`` by default) never
+    consulted ``raw_session_memberships`` at all.
+    """
+    initialize_active_archive_root(tmp_path)
+
+    session = ParsedSession(
+        source_name=Provider.CHATGPT,
+        provider_session_id="s1",
+        messages=[ParsedMessage(provider_message_id="s1-0", role=Role.USER, text="left")],
+    )
+
+    with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
+        raw_id = archive.write_raw_payload(
+            provider=Provider.CHATGPT, payload=b"aaa-left", source_path="a.json", acquired_at_ms=1
+        )
+        # Durable evidence that this raw's identity was already judged
+        # ambiguous -- the shape ``replace_raw_membership_census`` /
+        # ``apply_raw_membership_classification`` leave behind for a
+        # genuinely divergent cohort (reproduced directly here so the test
+        # isolates the WRITE-PATH guard from the classifier that produces
+        # this state).
+        source_conn = archive._ensure_source_conn()
+        with source_conn:
+            source_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 (?, 'chatgpt:s1', 's1', ?, ?, 1, 'ambiguous', 1)
+                """,
+                (raw_id, raw_id, bytes.fromhex(raw_id)),
+            )
+
+        returned_raw_id, session_id = archive.write_parsed_for_retained_raw(
+            session,
+            raw_id=raw_id,
+            source_path="a.json",
+            acquired_at_ms=2,
+        )
+
+    assert returned_raw_id == raw_id
+    with sqlite3.connect(tmp_path / "index.db") as conn:
+        assert conn.execute("SELECT COUNT(*) FROM sessions WHERE session_id = ?", (session_id,)).fetchone() == (0,)
+
+
 def test_isolated_later_raw_does_not_override_cohort_retired_under_legacy_detail_string(
     tmp_path: Path,
 ) -> None: