From 0483582a86b419ed5b43df8d7a6e505aa296d8a4 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 30 Jul 2026 15:16:30 +0200 Subject: [PATCH] fix(storage): refuse a precedence write for a raw recorded ambiguous membership Problem: live measurement (origin='aistudio-drive') found 28 cohorts with raw_session_memberships.decision='ambiguous' on BOTH members under the SAME logical_source_key -- genuinely arbitrated, correctly refused a winner by classify_membership_revisions -- yet each cohort's session IS present in index.db with zero acquired attachments (641 attachments total, every one unfetched, while the enriched sibling holds the bytes in the blob store). Traced to ArchiveStore.apply_raw_membership_classification (the classify_membership_revisions consumer): proven innocent by reading its finalization block -- for a fully-ambiguous cohort it explicitly clears raw_sessions.parsed_at_ms and never touches sessions. The actual writer is _write_parsed_precedence_result (same file), reached via write_parsed_for_retained_raw/write_parsed_for_retained_raw_result with revision_authoritative=False (the default, used by the one-shot importer pipeline/services/archive_ingest.py and other non-membership-governed callers). Its only revision-authority awareness was a check against raw_revision_heads, populated ONLY when a cohort has an ACCEPTED winner -- a genuinely ambiguous cohort never gets one, so the check stays silent and the function falls through to its own browser-capture-precedence/ freshness logic, writing the session unconditionally on the raw's next reparse. repair.py:1075/repair.py:4432-4462 (the two gates the investigation started from) are both innocent: neither sits on this write path -- repair.py:4432 is a read-only reporting classifier (_raw_replay_plan_outcome) and repair.py:1075 is a narrow inspector for a different (source-v7/quarantined-accepted-raw) repair scenario. Solution: _write_parsed_precedence_result also refuses when the raw's own raw_session_memberships.decision = 'ambiguous', alongside the existing raw_revision_heads check. Known sibling hole, not fixed here: pipeline/services/ingest_batch/_core.py (the daemon's default batch-ingest write path for most non-drive origins) has the same shape -- its own precedence logic, no raw_session_memberships consultation. Filed as follow-up in polylogue-c737; needs its own read-only census before fixing (different file, out of this PR's ownership scope). Verification: - devtools test tests/unit/storage/test_revision_replay.py tests/unit/sources/test_revision_backfill.py tests/unit/sources/test_live_batch_support.py tests/unit/pipeline/test_archive_ingest_commit_batching.py tests/unit/pipeline/test_ingest_batch.py -- all passed (147+66 = 213 passed across the two runs; the broader combined run with test_live_watcher.py showed only the pre-existing unrelated test_end_to_end_hidden_root_file_creation_triggers_ingest timing flake). - devtools verify --quick -- exit 0. - mypy polylogue/storage/sqlite/archive_tiers/archive.py -- Success, no issues found. - Anti-vacuity: temporarily short-circuiting the new guard (`if False and ambiguous_membership is not None`) makes test_precedence_write_refuses_a_raw_recorded_ambiguous fail (`assert (1,) == (0,)` -- a session row exists when it must not), restored afterward. Ref polylogue-c737, polylogue-eqnv, polylogue-bu1i, polylogue-9dxn Co-Authored-By: Claude --- .../storage/sqlite/archive_tiers/archive.py | 33 ++++++++++ tests/unit/storage/test_revision_replay.py | 66 +++++++++++++++++++ 2 files changed, 99 insertions(+) diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 6c78bfa23b..dfab03b8c0 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -1668,6 +1668,39 @@ def _write_parsed_precedence_result( content_changed=False, counts=self._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". + ambiguous_membership = ( + self._ensure_source_conn() + .execute( + "SELECT 1 FROM raw_session_memberships WHERE raw_id = ? AND decision = 'ambiguous' LIMIT 1", + (raw_id,), + ) + .fetchone() + ) + if ambiguous_membership is not None: + return ArchiveRawParsedWriteResult( + raw_id=raw_id, + session_id=session_id, + content_changed=False, + counts=self._skipped_counts(session), + ) if source_index >= 0 and existing_raw_id and raw_id and existing_raw_id != raw_id: existing_is_dom_fallback = session_has_parser_ingest_flag( diff --git a/tests/unit/storage/test_revision_replay.py b/tests/unit/storage/test_revision_replay.py index d50ab6534c..4991d61313 100644 --- a/tests/unit/storage/test_revision_replay.py +++ b/tests/unit/storage/test_revision_replay.py @@ -2,6 +2,7 @@ import hashlib import json +import sqlite3 from itertools import permutations from pathlib import Path @@ -617,6 +618,71 @@ def parsed_solo(native_id: str, *texts: str) -> ParsedSession: assert second_plan.accepted_raw_ids == () +def test_precedence_write_refuses_a_raw_recorded_ambiguous(tmp_path: Path) -> None: + """A raw whose OWN logical identity is durably recorded + ``raw_session_memberships.decision = 'ambiguous'`` must never reach + ``sessions`` through the ordinary (non-revision-authoritative) parsed- + write path. + + ``ArchiveStore._write_parsed_precedence_result``'s only revision- + authority awareness before this fix was a check against + ``raw_revision_heads`` -- populated ONLY when a cohort has an ACCEPTED + winner (``apply_raw_membership_classification``/ + ``apply_raw_revision_replay``). A cohort ``classify_membership_ + revisions`` genuinely refused to arbitrate never gets an accepted head, + so that check stays silent and the ordinary browser-capture-precedence/ + freshness fallback below it writes the session unconditionally on the + next reparse -- arbitrary last-writer-wins over the exact invariant this + subsystem exists to enforce. Live evidence: 28 aistudio-drive cohorts + recorded ambiguous nonetheless materialized a session with 641 + attachments reported unfetched despite the bytes existing in the blob + store, because ``write_parsed_for_retained_raw`` (called from the + one-shot importer, ``revision_authoritative=False`` by default) never + consulted ``raw_session_memberships`` at all. + """ + initialize_active_archive_root(tmp_path) + + session = ParsedSession( + source_name=Provider.CHATGPT, + provider_session_id="s1", + messages=[ParsedMessage(provider_message_id="s1-0", role=Role.USER, text="left")], + ) + + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + raw_id = archive.write_raw_payload( + provider=Provider.CHATGPT, payload=b"aaa-left", source_path="a.json", acquired_at_ms=1 + ) + # Durable evidence that this raw's identity was already judged + # ambiguous -- the shape ``replace_raw_membership_census`` / + # ``apply_raw_membership_classification`` leave behind for a + # genuinely divergent cohort (reproduced directly here so the test + # isolates the WRITE-PATH guard from the classifier that produces + # this state). + source_conn = archive._ensure_source_conn() + with source_conn: + source_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 (?, 'chatgpt:s1', 's1', ?, ?, 1, 'ambiguous', 1) + """, + (raw_id, raw_id, bytes.fromhex(raw_id)), + ) + + returned_raw_id, session_id = archive.write_parsed_for_retained_raw( + session, + raw_id=raw_id, + source_path="a.json", + acquired_at_ms=2, + ) + + assert returned_raw_id == raw_id + with sqlite3.connect(tmp_path / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions WHERE session_id = ?", (session_id,)).fetchone() == (0,) + + def test_isolated_later_raw_does_not_override_cohort_retired_under_legacy_detail_string( tmp_path: Path, ) -> None: