diff --git a/polylogue/pipeline/services/ingest_batch/_core.py b/polylogue/pipeline/services/ingest_batch/_core.py index c7992d845e..a9c2f8f438 100644 --- a/polylogue/pipeline/services/ingest_batch/_core.py +++ b/polylogue/pipeline/services/ingest_batch/_core.py @@ -59,6 +59,7 @@ browser_capture_precedence, record_capture_gap_event, record_source_outage_events, + revision_authority_refuses_write, session_has_parser_ingest_flag, should_skip_stale_replace, stored_message_count, @@ -460,62 +461,18 @@ def _write_session( merge_append = False browser_precedence: BrowserCapturePrecedence = "default" - has_revision_heads = conn.execute( - "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'raw_revision_heads'" - ).fetchone() - if has_revision_heads is not None: - governed = conn.execute( - "SELECT 1 FROM raw_revision_heads WHERE session_id = ? LIMIT 1", - (payload.session_id,), - ).fetchone() - if governed 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 - - # 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 revision_authority_refuses_write( + conn, + source_conn, + session_id=payload.session_id, + raw_id=payload.raw_id or "", + provider_session_id=payload.parsed_session.provider_session_id, + ): + 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 diff --git a/polylogue/storage/sqlite/archive_tiers/ingest_precedence.py b/polylogue/storage/sqlite/archive_tiers/ingest_precedence.py index e9c0b5b156..93c17e2d04 100644 --- a/polylogue/storage/sqlite/archive_tiers/ingest_precedence.py +++ b/polylogue/storage/sqlite/archive_tiers/ingest_precedence.py @@ -142,6 +142,73 @@ def browser_capture_precedence( return "replace" if incoming_owns_browser_merge else "default" +def revision_authority_refuses_write( + conn: sqlite3.Connection, + source_conn: sqlite3.Connection | None, + *, + session_id: str, + raw_id: str, + provider_session_id: str, +) -> bool: + """The ONE revision-authority refusal gate for a retained-raw session write. + + Consolidated from two independently hand-maintained copies of the exact + same two checks (polylogue-c737: PR #3397 fixed + ``ArchiveStore._write_parsed_precedence_result`` + (``revision_governance.py``), then PR #3398 had to separately re-apply + the identical fix to the daemon batch-ingest path's ``_write_session`` + (``pipeline/services/ingest_batch/_core.py``) -- "the signature of + duplicated semantics rather than a missing check". polylogue-aggz + Invariant 2 makes this the only implementation; both write paths call it + before falling through to their own freshness/browser-capture precedence + logic, and neither may reimplement it locally. + + Two independent refusals, checked in order: + + - an ACCEPTED revision-authority head already exists for this + ``session_id`` (``raw_revision_heads``) -- some other raw in this + cohort already won, so this write is redundant, never authoritative. + - this raw's own recorded membership decision for this + ``provider_session_id`` is ``'ambiguous'`` -- ``classify_membership_ + revisions`` genuinely refused to arbitrate a winner for this cohort, + and falling through to ordinary freshness comparison would silently + pick one anyway by last-writer-wins -- exactly the "never silently + choose between branches" invariant this gate exists to enforce. + + ``source_conn`` may be ``None`` only when the caller genuinely has no + source.db handle (a synthetic index-only harness); a real batch ingest + or governed raw-parsed write always opens one, so the membership- + ambiguity leg runs there. ``raw_id`` empty/falsy also skips that leg + (there is no raw to look up membership for). + """ + has_revision_heads = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'raw_revision_heads'" + ).fetchone() + if has_revision_heads is not None: + governed = conn.execute( + "SELECT 1 FROM raw_revision_heads WHERE session_id = ? LIMIT 1", + (session_id,), + ).fetchone() + if governed is not None: + return True + if source_conn is None or not raw_id: + return False + has_memberships = source_conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'raw_session_memberships'" + ).fetchone() + if has_memberships is None: + return False + ambiguous_membership = source_conn.execute( + """ + SELECT 1 FROM raw_session_memberships + WHERE raw_id = ? AND provider_session_id = ? AND decision = 'ambiguous' + LIMIT 1 + """, + (raw_id, provider_session_id), + ).fetchone() + return ambiguous_membership is not None + + def stored_message_count(conn: sqlite3.Connection, session_id: str) -> int: row = conn.execute( "SELECT COUNT(*) FROM messages WHERE session_id = ?", @@ -285,6 +352,7 @@ def record_source_outage_events( "browser_capture_precedence", "record_capture_gap_event", "record_source_outage_events", + "revision_authority_refuses_write", "session_has_parser_ingest_flag", "stored_message_count", ] diff --git a/polylogue/storage/sqlite/archive_tiers/revision_governance.py b/polylogue/storage/sqlite/archive_tiers/revision_governance.py index bda2114397..41418a1e93 100644 --- a/polylogue/storage/sqlite/archive_tiers/revision_governance.py +++ b/polylogue/storage/sqlite/archive_tiers/revision_governance.py @@ -130,6 +130,7 @@ browser_capture_precedence, record_capture_gap_event, record_source_outage_events, + revision_authority_refuses_write, session_has_parser_ingest_flag, should_skip_stale_replace, stored_message_count, @@ -261,59 +262,13 @@ def _write_parsed_precedence_result( content_changed=True, counts=store._write_counts(session), ) - governed = store._conn.execute( - "SELECT 1 FROM raw_revision_heads WHERE session_id = ? LIMIT 1", - (session_id,), - ).fetchone() - if governed is not None: - return ArchiveRawParsedWriteResult( - raw_id=raw_id, - session_id=session_id, - content_changed=False, - counts=store._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". - # - # Scoped to the membership being written, not to the raw. One retained - # raw routinely lowers to many sessions -- a Claude Code transcript and - # its subagent sidechains, a bundle member set -- and those sessions are - # arbitrated independently. Measured on the live archive: 295 raws carry - # a mix of decisions, together holding 489 sessions whose own membership - # is NOT ambiguous, and one raw carries 106 memberships. A raw-scoped - # predicate suppresses every one of those sessions as soon as a single - # sibling membership is ambiguous, which trades a fidelity downgrade for - # outright absence -- a worse failure, and one that would have landed at - # the next full rebuild. - ambiguous_membership = ( - store._ensure_source_conn() - .execute( - """ - SELECT 1 FROM raw_session_memberships - WHERE raw_id = ? AND provider_session_id = ? AND decision = 'ambiguous' - LIMIT 1 - """, - (raw_id, session.provider_session_id), - ) - .fetchone() - ) - if ambiguous_membership is not None: + if revision_authority_refuses_write( + store._conn, + store._ensure_source_conn(), + session_id=session_id, + raw_id=raw_id, + provider_session_id=session.provider_session_id, + ): return ArchiveRawParsedWriteResult( raw_id=raw_id, session_id=session_id,