Skip to content

refactor(storage): collapse duplicated session-write authority gate to one function #3505

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Sinity merged 1 commit into from
Aug 1, 2026
Merged

refactor(storage): collapse duplicated session-write authority gate to one function #3505

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 13 additions & 56 deletions polylogue/pipeline/services/ingest_batch/_core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down
68 changes: 68 additions & 0 deletions polylogue/storage/sqlite/archive_tiers/ingest_precedence.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = ?",
Expand Down Expand Up @@ -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",
]
61 changes: 8 additions & 53 deletions polylogue/storage/sqlite/archive_tiers/revision_governance.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down