From 5d477860f96292284873c599c16a8fa62dcfa43b Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 30 Jul 2026 16:15:49 +0200 Subject: [PATCH] fix(pipeline): refuse ingest-batch writes for ambiguous raw memberships ## Summary Extends the #3397/#3398 ambiguous-membership refusal to the daemon's default batch-ingest write path, `_write_session` in `pipeline/services/ingest_batch/_core.py`. ## Problem `_write_session` had the exact same shape as the pre-#3397 `ArchiveStore._write_parsed_precedence_result`: 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 stays silent and the ordinary freshness/browser-precedence fallback below it writes the session unconditionally on the raw's next reparse -- last-writer-wins over the recorded `ambiguous` verdict. This path is the daemon's default for most non-drive origins, which is why quarantined-but-parsed counts concentrate here (chatgpt-export 7,050, codex-session 3,633, claude-code-session 2,450, claude-ai-export 1,562 raws with `revision_authority='quarantined'` and `parsed_at_ms` set). Left unfixed, a planned full index rebuild would re-corrupt every one of those origins even after #3397/#3398 fixed the drive-specific write path. ## Solution Added the same membership check directly to `_write_session`, not routed through a shared helper with `archive.py`. The two write paths don't share a connection: `ArchiveStore` keeps a lazily-opened persistent `source.db` connection as instance state (`_ensure_source_conn`), while `_core.py`'s writer is a free function operating on a plain `sqlite3.Connection` for index.db with no existing source.db handle in scope. Extracting a shared predicate would require either threading a `source_conn` through both call graphs anyway (no real duplication saved beyond the six-line SQL predicate) or introducing a new cross-module coupling between `archive_tiers/archive.py` and `ingest_batch/_core.py` for a single SELECT. The duplication that mattered was the *authority semantics* (never write a session for a membership recorded ambiguous), not the six-line SQL string; both copies now encode identical semantics and are covered by mirrored tests. A read-only-use `source.db` connection is opened once per batch (alongside the existing `blob_publisher`, which already opens its own source.db handle for a different purpose) and threaded through `_consume_ingest_results` -> `_drain_ingest_result` -> `_drain_ready_session_entries` -> `_write_session_entry` -> `_write_session`, closed in the batch's existing `finally`. The parameter defaults to `None` everywhere, so the ~40 existing direct `_write_session(conn, payload)` calls in `tests/unit/pipeline/ test_ingest_batch.py` and friends are unaffected. The predicate matches `raw_id` AND `provider_session_id`, not raw_id alone -- #3398's exact scoping lesson. One retained raw routinely lowers to many independently-arbitrated sessions (a Claude Code transcript plus its subagent sidechains, a bundle member set); a raw-scoped predicate would suppress every session on the raw the moment one sibling membership is ambiguous, turning a fidelity downgrade into outright absence at the next full rebuild. ## Verification `test_write_session_refuses_a_raw_recorded_ambiguous_membership` builds one raw with two `raw_session_memberships` rows -- one `ambiguous`, one `applied` -- and asserts BOTH halves: the ambiguous membership is refused, its settled sibling on the same raw is written. Asserting only the refusal would pass against an over-broad raw-scoped predicate; this was checked directly (see anti-vacuity below). ``` devtools test tests/unit/pipeline/test_ingest_batch.py -k test_write_session_refuses_a_raw_recorded_ambiguous_membership 1 passed devtools test tests/unit/pipeline/test_ingest_batch.py 56 passed devtools test tests/unit/pipeline/test_ingest_batch_fts_repair.py tests/unit/pipeline/test_ingest_append_replay.py tests/unit/pipeline/test_ingest_batch_resource_bounds.py 8 passed devtools test tests/unit/pipeline/test_ingest_batch_wal_checkpoint.py tests/unit/pipeline/test_blob_publication_crash_matrix.py tests/unit/pipeline/test_parsing_service.py 41 passed mypy --strict polylogue/pipeline/services/ingest_batch/_core.py Success: no issues found in 1 source file ``` **Anti-vacuity, run and reverted twice:** 1. Reverting the predicate to the raw-scoped form (`WHERE raw_id = ? AND decision = 'ambiguous'`, dropping the `provider_session_id` match) fails the settled-sibling assertion (`assert False is True`) -- proves the scoping is load-bearing, not incidental. 2. Short-circuiting the guard entirely (`if False and source_conn is not None ...`) fails the ambiguous-refusal assertion (`assert True is False`) -- proves the guard itself, not an unrelated earlier clause, is what refuses the write. Ref polylogue-c737 Co-Authored-By: Claude --- .../pipeline/services/ingest_batch/_core.py | 63 +++++++++++ tests/unit/pipeline/test_ingest_batch.py | 107 ++++++++++++++++++ 2 files changed, 170 insertions(+) diff --git a/polylogue/pipeline/services/ingest_batch/_core.py b/polylogue/pipeline/services/ingest_batch/_core.py index 370250bce3..1f10dd712b 100644 --- a/polylogue/pipeline/services/ingest_batch/_core.py +++ b/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 diff --git a/tests/unit/pipeline/test_ingest_batch.py b/tests/unit/pipeline/test_ingest_batch.py index 6dbb711b33..44d2459672 100644 --- a/tests/unit/pipeline/test_ingest_batch.py +++ b/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: