fix(pipeline): refuse ingest-batch writes for ambiguous raw memberships

polylogue/pipeline/services/ingest_batch/_core.py

@@ -393,6 +393,7 @@ def _write_session(
     stage_timings_s: dict[str, float] | None = None,
     blob_publisher: ArchiveBlobPublisher | None = None,
     pending_attachment_receipts: list[tuple[str, bytes]] | None = None,
+    source_conn: sqlite3.Connection | None = None,
 ) -> tuple[bool, dict[str, int]]:
     """Write one parsed session payload into the current archive index.
 
@@ -437,6 +438,48 @@ def _write_session(
             counts["skipped_session_events"] = len(payload.parsed_session.session_events)
             return False, counts
 
+    # polylogue-c737: mirrors ArchiveStore._write_parsed_precedence_result's
+    # ambiguous-membership refusal (#3397/#3398). ``governed`` above only
+    # catches a logical identity with an ACCEPTED revision-authority head
+    # (``raw_revision_heads``, populated only when a cohort has a winner). A
+    # cohort ``classify_membership_revisions`` genuinely refused to
+    # arbitrate never gets an accepted head, so ``governed`` stays ``None``
+    # here even though this raw's own membership is recorded authority
+    # debt -- and this batch write path, the daemon's default for most
+    # non-drive origins, never consulted ``raw_session_memberships`` at all
+    # before this fix. Falling through to the freshness/precedence logic
+    # below then writes the session unconditionally on the raw's next
+    # reparse -- last-writer-wins over the "never silently choose between
+    # branches" invariant.
+    #
+    # Scoped to the membership actually being written (raw_id AND
+    # provider_session_id), not to the raw alone: one retained raw routinely
+    # lowers to many independently-arbitrated sessions (a Claude Code
+    # transcript plus its subagent sidechains, a bundle member set), and
+    # #3398 measured 295 raws carrying a mix of decisions with 489 sessions
+    # whose own membership is NOT ambiguous -- a raw-scoped predicate would
+    # suppress all of those too, trading a fidelity downgrade for outright
+    # absence.
+    if source_conn is not None and payload.raw_id:
+        has_memberships = source_conn.execute(
+            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'raw_session_memberships'"
+        ).fetchone()
+        if has_memberships is not None:
+            ambiguous_membership = source_conn.execute(
+                """
+                SELECT 1 FROM raw_session_memberships
+                WHERE raw_id = ? AND provider_session_id = ? AND decision = 'ambiguous'
+                LIMIT 1
+                """,
+                (payload.raw_id, payload.parsed_session.provider_session_id),
+            ).fetchone()
+            if ambiguous_membership is not None:
+                counts["skipped_sessions"] = 1
+                counts["skipped_messages"] = payload.message_count
+                counts["skipped_attachments"] = payload.attachment_count
+                counts["skipped_session_events"] = len(payload.parsed_session.session_events)
+                return False, counts
+
     if (
         not force_write
         and not payload.append_only
@@ -671,6 +714,7 @@ def _write_session_entry(
     signature_cache: dict[str, list[tuple[str, str]]] | None = None,
     blob_publisher: ArchiveBlobPublisher | None = None,
     pending_attachment_receipts: list[tuple[str, bytes]] | None = None,
+    source_conn: sqlite3.Connection | None = None,
 ) -> bool:
     try:
         t_write = time.perf_counter()
@@ -683,6 +727,7 @@ def _write_session_entry(
             stage_timings_s=write_stage_timings,
             blob_publisher=blob_publisher,
             pending_attachment_receipts=pending_attachment_receipts,
+            source_conn=source_conn,
         )
         for stage, elapsed_s in write_stage_timings.items():
             summary.stage_timings_s[stage] = summary.stage_timings_s.get(stage, 0.0) + elapsed_s
@@ -794,6 +839,7 @@ def _drain_ready_session_entries(
     force_write: bool = False,
     blob_publisher: ArchiveBlobPublisher | None = None,
     pending_attachment_receipts: list[tuple[str, bytes]] | None = None,
+    source_conn: sqlite3.Connection | None = None,
 ) -> int:
     _delete_stale_sessions_for_raw_entries(conn, ready_entries)
     written_count = 0
@@ -812,6 +858,7 @@ def _drain_ready_session_entries(
             signature_cache=signature_cache,
             blob_publisher=blob_publisher,
             pending_attachment_receipts=pending_attachment_receipts,
+            source_conn=source_conn,
         )
         discard_session_data_payload(cdata)
         if not wrote:
@@ -1052,6 +1099,7 @@ def _drain_ingest_result(
     force_write: bool = False,
     blob_publisher: ArchiveBlobPublisher | None = None,
     pending_attachment_receipts: list[tuple[str, bytes]] | None = None,
+    source_conn: sqlite3.Connection | None = None,
 ) -> None:
     _record_outcome(summary, ir)
     _observe_current_rss(summary)
@@ -1108,6 +1156,7 @@ def _drain_ingest_result(
         force_write=force_write,
         blob_publisher=blob_publisher,
         pending_attachment_receipts=pending_attachment_receipts,
+        source_conn=source_conn,
     )
     if written_count == 0:
         summary.skipped_raw_ids.add(ir.raw_id)
@@ -1140,6 +1189,7 @@ def _consume_ingest_results(
     force_process_pool: bool = False,
     blob_publisher: ArchiveBlobPublisher | None = None,
     pending_attachment_receipts: list[tuple[str, bytes]] | None = None,
+    source_conn: sqlite3.Connection | None = None,
 ) -> bool:
     result_iterator = iter(
         _iter_ingest_results_sync(
@@ -1188,6 +1238,7 @@ def ensure_index_transaction() -> None:
                 force_write=force_write,
                 blob_publisher=blob_publisher,
                 pending_attachment_receipts=pending_attachment_receipts,
+                source_conn=source_conn,
             )
         finally:
             discard_ingest_result_payload(ir)
@@ -1347,6 +1398,15 @@ def _process_ingest_batch_sync(
     materialized_ids: set[str] = set()
     blob_publisher = ArchiveBlobPublisher(archive_root / "source.db", archive_root / "blob")
     pending_attachment_receipts: list[tuple[str, bytes]] = []
+    # polylogue-c737: read-only source.db handle used solely to consult
+    # ``raw_session_memberships`` decisions during the write-precedence
+    # check in ``_write_session`` (mirrors ArchiveStore's own
+    # ``_ensure_source_conn`` read path). Opened once per batch rather than
+    # per session write.
+    source_db_path = archive_root / "source.db"
+    source_conn: sqlite3.Connection | None = None
+    if source_db_path.exists():
+        source_conn = sqlite3.connect(str(source_db_path), timeout=DB_TIMEOUT)
     _observe_current_rss(summary)
     transaction_started = False
     try:
@@ -1367,6 +1427,7 @@ def _process_ingest_batch_sync(
             force_process_pool=force_process_pool,
             blob_publisher=blob_publisher,
             pending_attachment_receipts=pending_attachment_receipts,
+            source_conn=source_conn,
         )
         _flush_ingest_results(
             conn,
@@ -1463,6 +1524,8 @@ def _process_ingest_batch_sync(
             with contextlib.suppress(Exception):
                 conn.execute("PRAGMA foreign_keys = ON")
         conn.close()
+        if source_conn is not None:
+            source_conn.close()
     summary.worker_progress_in_flight = len(progress.in_flight_raw_ids)
     summary.worker_progress_completed = progress.completed_raw_count
     summary.worker_progress_total = progress.total_raw_count

tests/unit/pipeline/test_ingest_batch.py

@@ -1615,6 +1615,113 @@ def test_write_session_allows_existing_upsert_even_without_messages(tmp_path: Pa
         assert counts["skipped_sessions"] == 0
 
 
+def test_write_session_refuses_a_raw_recorded_ambiguous_membership(tmp_path: Path) -> None:
+    """``_write_session`` -- the daemon's default batch-ingest write path,
+    used for most non-drive origins -- must refuse a session whose OWN
+    ``raw_session_memberships.decision`` is recorded ``'ambiguous'``.
+
+    This mirrors ``ArchiveStore._write_parsed_precedence_result``'s guard
+    (#3397/#3398, polylogue-c737). Before this fix, this path had zero
+    ``raw_session_memberships`` awareness: its only revision-authority check
+    was against ``raw_revision_heads``, populated ONLY when a cohort has an
+    ACCEPTED winner. A cohort ``classify_membership_revisions`` genuinely
+    refused to arbitrate never gets an accepted head, so that check stayed
+    silent and the ordinary freshness/precedence fallback below it wrote the
+    session unconditionally on the raw's next reparse -- arbitrary
+    last-writer-wins over the recorded verdict.
+
+    Scoped per-membership (``raw_id`` AND ``provider_session_id``), not
+    per-raw: one retained raw routinely lowers to many independently-
+    arbitrated sessions (a Claude Code transcript plus its subagent
+    sidechains, a bundle member set). #3398 measured 295 raws carrying a mix
+    of decisions on the live archive, together holding 489 sessions whose own
+    membership is NOT ambiguous. This test builds that exact shape -- one
+    raw, two memberships, one ``ambiguous`` and one ``applied`` -- and
+    asserts BOTH halves, so a raw-scoped predicate (which would refuse both)
+    fails it just as it would have fixed nothing for the settled sibling.
+    """
+    archive_root = tmp_path / "archive"
+    initialize_active_archive_root(archive_root)
+    raw_id = "abcd1234abcd1234"
+    source_db_path = archive_root / "source.db"
+
+    with sqlite3.connect(str(source_db_path)) as source_setup_conn:
+        source_setup_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 (?, 'codex:s-ambiguous', 's-ambiguous', 'rev-1', ?, 1, 'ambiguous', 1)
+            """,
+            (raw_id, b"0" * 32),
+        )
+        source_setup_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 (?, 'codex:s-settled', 's-settled', 'rev-1', ?, 1, 'applied', 1)
+            """,
+            (raw_id, b"1" * 32),
+        )
+        source_setup_conn.commit()
+
+    with (
+        open_connection(archive_root / "index.db") as conn,
+        sqlite3.connect(str(source_db_path)) as source_conn,
+    ):
+        ambiguous_msg = _message_tuple(
+            "a-0",
+            "codex-session:s-ambiguous",
+            role="user",
+            text="left",
+            content_hash="hash-a",
+            sort_key=1.0,
+        )
+        settled_msg = _message_tuple(
+            "b-0",
+            "codex-session:s-settled",
+            role="user",
+            text="right",
+            content_hash="hash-b",
+            sort_key=1.0,
+        )
+        ambiguous_payload = _session_data(
+            "codex-session:s-ambiguous",
+            content_hash="hash-ambiguous",
+            raw_id=raw_id,
+            message_tuples=[ambiguous_msg],
+        )
+        settled_payload = _session_data(
+            "codex-session:s-settled",
+            content_hash="hash-settled",
+            raw_id=raw_id,
+            message_tuples=[settled_msg],
+        )
+
+        changed_ambiguous, counts_ambiguous = _write_session(conn, ambiguous_payload, source_conn=source_conn)
+        changed_settled, counts_settled = _write_session(conn, settled_payload, source_conn=source_conn)
+        conn.commit()
+
+    assert changed_ambiguous is False
+    assert counts_ambiguous["skipped_sessions"] == 1
+
+    assert changed_settled is True
+    assert counts_settled["skipped_sessions"] == 0
+
+    with sqlite3.connect(str(archive_root / "index.db")) as verify_conn:
+        # The ambiguous membership is still refused ...
+        assert verify_conn.execute(
+            "SELECT COUNT(*) FROM sessions WHERE session_id = ?", ("codex-session:s-ambiguous",)
+        ).fetchone() == (0,)
+        # ... and its settled sibling on the same raw is not collateral damage.
+        assert verify_conn.execute(
+            "SELECT COUNT(*) FROM sessions WHERE session_id = ?", ("codex-session:s-settled",)
+        ).fetchone() == (1,)
+
+
 def test_iter_ingest_results_sync_runs_inline_for_single_worker(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None: