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: