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: