From 7d1c2b24de846d823adadd14bac7fb150216e230 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 21 Jul 2026 14:43:58 +0200 Subject: [PATCH 1/2] fix(sources): widen 52l2 guard to recognize legacy retirement detail Problem Durable raw_membership_census rows written by sources/live/batch.py before PR #3234 carry detail="cross-route full revision governance" (the pre-fix literal at that call site) instead of the shared HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL marker the #3234 guard query matches on. Identities retired under that legacy literal are invisible to raw_membership_retired_full_revision_siblings, so the polylogue-52l2 isolated-singleton guard still fails to protect them -- the pre-existing discovery-order bug is preserved for every such identity. Solution Add RETIRED_FULL_REVISION_GOVERNANCE_DETAILS, a frozen tuple naming every detail literal that means "retired from full-revision byte governance to membership governance" -- the current shared constant plus the legacy pre-#3234 literal (documented read-only, never write). raw_membership_retired_full_revision_siblings now matches `c.detail IN (...)` against the whole tuple. Verification - devtools test tests/unit/storage/test_revision_replay.py tests/unit/sources/test_revision_backfill.py -> 61 passed, 1 pre-existing unrelated failure (test_backfill_resumes_after_replay_ batch_crash_discards_whole_batch_cleanly, messages_fts_identity UNIQUE constraint -- reproduces identically on unmodified master, confirmed by stash/rerun). - New regression test_isolated_later_raw_does_not_override_cohort_retired_under_legacy_detail_string mirrors the existing 52l2 test but retires siblings under the legacy literal directly; asserts the tuple names the legacy literal (anti-vacuity: removing it fails the test). - devtools test tests/property/test_sql_injection_boundary.py -> 40 passed (archive.py touched a SQL WHERE clause; uses the existing trusted `where_clause` interpolation pattern with an IN (...) placeholder list, values fully parameterized). - mypy --strict polylogue/archive/revision_authority.py polylogue/storage/sqlite/archive_tiers/archive.py -> clean. Ref polylogue-hm2f Co-Authored-By: Claude --- polylogue/archive/revision_authority.py | 22 +++++ .../storage/sqlite/archive_tiers/archive.py | 17 +++- tests/unit/storage/test_revision_replay.py | 91 +++++++++++++++++++ 3 files changed, 126 insertions(+), 4 deletions(-) diff --git a/polylogue/archive/revision_authority.py b/polylogue/archive/revision_authority.py index 5d1f17c5f8..56c956ef6a 100644 --- a/polylogue/archive/revision_authority.py +++ b/polylogue/archive/revision_authority.py @@ -33,6 +33,28 @@ class RawRevisionAuthority(StrEnum): #: later-discovered raw as an unconditional singleton byte-proven baseline. HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL = "historical non-prefix full revision governance" +#: All ``raw_membership_census.detail`` marker literals that mean "this raw +#: was retired from full-revision byte governance to membership governance" +#: (polylogue-hm2f, residual of polylogue-52l2). Durable ``raw_membership_ +#: census`` rows written before the #3234 fix used the literal +#: ``"cross-route full revision governance"`` at the live-watcher call site +#: (``sources/live/batch.py``, pre-fix) -- a DIFFERENT string from +#: ``HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL``, which only the offline +#: backfill call site used at the time. Those pre-fix rows are durable +#: (``source.db``) and were never rewritten in place (durable-tier changes +#: need an explicit additive migration, not a silent detail-string rewrite), +#: so identities retired under the legacy literal remain invisible to a guard +#: query keyed on the new marker alone. ``ArchiveStore.raw_membership_ +#: retired_full_revision_siblings`` matches against every literal in this +#: tuple so old and new retirements are both recognized. The legacy literal +#: is frozen here for read-compatibility only -- it must never be written by +#: new code (both current call sites use ``HISTORICAL_NON_PREFIX_GOVERNANCE_ +#: DETAIL`` exclusively). +RETIRED_FULL_REVISION_GOVERNANCE_DETAILS: tuple[str, ...] = ( + HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, + "cross-route full revision governance", # legacy pre-#3234 literal -- read-only, never write. +) + @dataclass(frozen=True) class RawRevisionEnvelope: diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 7151a4224e..ba89624592 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -38,7 +38,7 @@ QueryTextPredicate, ) from polylogue.archive.revision_authority import ( - HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, + RETIRED_FULL_REVISION_GOVERNANCE_DETAILS, HistoricalRawRevisionStream, RawRevisionAuthority, RawRevisionEnvelope, @@ -2003,18 +2003,27 @@ def raw_membership_retired_full_revision_siblings(self, logical_source_key: str) transition, so a later-arriving sibling for the same identity can still be told this identity has known, unresolved ambiguous evidence (polylogue-52l2) instead of being evaluated alone. + + Matches every literal in ``RETIRED_FULL_REVISION_GOVERNANCE_DETAILS`` + (polylogue-hm2f), not only the current + ``HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL`` marker: durable + ``raw_membership_census`` rows written before #3234 used a different, + now-legacy literal at the live-watcher call site, and durable-tier + detail strings are never silently rewritten in place. """ + detail_placeholders = ", ".join("?" for _ in RETIRED_FULL_REVISION_GOVERNANCE_DETAILS) + where_clause = f"c.detail IN ({detail_placeholders})" rows = ( self._ensure_source_conn() .execute( - """ + f""" SELECT m.raw_id FROM raw_session_memberships AS m JOIN raw_membership_census AS c ON c.raw_id = m.raw_id - WHERE m.logical_source_key = ? AND c.detail = ? + WHERE m.logical_source_key = ? AND {where_clause} ORDER BY m.raw_id """, - (logical_source_key, HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL), + (logical_source_key, *RETIRED_FULL_REVISION_GOVERNANCE_DETAILS), ) .fetchall() ) diff --git a/tests/unit/storage/test_revision_replay.py b/tests/unit/storage/test_revision_replay.py index 852659718a..54b342f015 100644 --- a/tests/unit/storage/test_revision_replay.py +++ b/tests/unit/storage/test_revision_replay.py @@ -536,6 +536,97 @@ def parsed_solo(native_id: str, *texts: str) -> ParsedSession: assert second_plan.accepted_raw_ids == () +def test_isolated_later_raw_does_not_override_cohort_retired_under_legacy_detail_string( + tmp_path: Path, +) -> None: + """polylogue-hm2f: the 52l2 guard must also recognize legacy-detail retirements. + + Durable ``raw_membership_census`` rows written by ``sources/live/batch.py`` + BEFORE #3234 carry ``detail="cross-route full revision governance"`` (the + pre-fix literal at that call site) instead of the shared + ``HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL`` marker the #3234 guard query + matches. This is a byte-for-byte mirror of + ``test_isolated_later_raw_does_not_override_known_ambiguous_cohort`` + except the two siblings are retired under that legacy literal directly + (as durable pre-#3234 rows would already be on disk) instead of the + current shared constant -- proving the guard's widened ``detail IN (...)`` + match, not just its original single-literal match. + """ + initialize_active_archive_root(tmp_path) + + def parsed_solo(native_id: str, *texts: str) -> ParsedSession: + return ParsedSession( + source_name=Provider.CHATGPT, + provider_session_id=native_id, + messages=[ + ParsedMessage(provider_message_id=f"{native_id}-{index}", role=Role.USER, text=text) + for index, text in enumerate(texts) + ], + ) + + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + raw_a = archive.write_raw_payload( + provider=Provider.CHATGPT, payload=b"aaa-left", source_path="a.json", acquired_at_ms=1 + ) + archive.bind_raw_revision( + raw_a, + RawRevisionEnvelope( + "chatgpt:s1", RawRevisionKind.FULL, raw_a, 0, authority=RawRevisionAuthority.QUARANTINED + ), + ) + raw_b = archive.write_raw_payload( + provider=Provider.CHATGPT, payload=b"bbb-right", source_path="b.json", acquired_at_ms=2 + ) + archive.bind_raw_revision( + raw_b, + RawRevisionEnvelope( + "chatgpt:s1", RawRevisionKind.FULL, raw_b, 0, authority=RawRevisionAuthority.QUARANTINED + ), + ) + + first_plan = archive.classify_raw_revision_cohort("chatgpt:s1") + assert first_plan.accepted_raw_ids == () + + # Retire both siblings under the LEGACY pre-#3234 literal, not the + # current shared constant -- this is what a durable row written + # before #3234 actually contains on disk. + for raw_id, session in ( + (raw_a, parsed_solo("s1", "base", "left")), + (raw_b, parsed_solo("s1", "base", "right")), + ): + archive.replace_raw_membership_census( + raw_id, + [session], + parser_fingerprint="revision-membership-v1", + censused_at_ms=0, + detail="cross-route full revision governance", + retire_full_revision_governance=True, + ) + + # A THIRD raw for the same logical identity, discovered afterward. + raw_c = archive.write_raw_payload( + provider=Provider.CHATGPT, payload=b"ccc-solo", source_path="c.json", acquired_at_ms=3 + ) + archive.bind_raw_revision( + raw_c, + RawRevisionEnvelope( + "chatgpt:s1", RawRevisionKind.FULL, raw_c, 0, authority=RawRevisionAuthority.QUARANTINED + ), + ) + second_plan = archive.classify_raw_revision_cohort("chatgpt:s1") + + # Same assertion as the shared-constant test: the isolated raw must not + # be promoted alone against siblings retired under the legacy literal. + assert second_plan.accepted_raw_ids == () + + # Anti-vacuity: removing the legacy literal from the tuple this guard + # matches against must make this exact test fail. Assert the tuple still + # names it explicitly, so a future edit that drops it is caught here too. + from polylogue.archive.revision_authority import RETIRED_FULL_REVISION_GOVERNANCE_DETAILS + + assert "cross-route full revision governance" in RETIRED_FULL_REVISION_GOVERNANCE_DETAILS + + def test_real_single_append_chain_folds_segmentation_distinct_full_snapshot(tmp_path: Path) -> None: initialize_active_archive_root(tmp_path) From 4c516ec8a284e845b6fe1692f67adb44030b93cc Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 21 Jul 2026 14:44:27 +0200 Subject: [PATCH 2/2] fix(sources): reunify retired revision siblings in the live watcher Problem The polylogue-52l2 guard (PR #3234) stops an isolated later-discovered raw from being silently accepted as the permanent content for a logical identity that already has retired, ambiguous siblings -- but it fails CLOSED: once classify_raw_revision_cohort's guard trips, the live incremental watcher (sources/live/batch.py) only logs a warning and marks the raw failed forever. Only the OFFLINE backfill path (sources/revision_backfill.py's census + convertible_full_revision_raw_ids retirement dance) can move a full-only cohort into membership governance where the real content-prefix classifier (classify_membership_revisions) can actually arbitrate it. There was no live-path route back to that resolution -- a real answer sat in convergence_debt-shaped limbo indefinitely. Solution When classify_raw_revision_cohort returns an empty cohort in the live watcher's single-session non-membership branch, query raw_membership_retired_full_revision_siblings(logical_source_key) before giving up: - No retired siblings: unchanged fail-closed behavior (a genuine, first-encounter divergence with nothing this tick can resolve). - Retired siblings exist (exactly what the 52l2 guard detects): fold this raw into membership governance via the SAME _apply_membership_sessions helper already used for the browser-capture and pre-existing-membership branches in this file, with a new `extra_member_raw_ids` parameter carrying the retired siblings. archive.raw_membership_raw_ids() alone cannot see them -- retirement leaves their raw_session_memberships row at the default revision_authority='quarantined', so only the durable detail marker (not that query) still names them. The real classifier then weighs the new raw against every known sibling instead of evaluating it alone; a genuine three-way conflict is still correctly quarantined as ambiguous (this recovers a real resolution when one exists, it never fabricates one). This is full within-tick reunification, not retire+defer: unlike the backfill's cross-page candidate bookkeeping (a local dict spanning its whole census run), the live path re-derives the sibling set from the same durable retirement marker the 52l2 guard itself reads, so no additional convergence_debt plumbing was needed to make the retry correct. Verification - New test test_live_third_raw_reunifies_with_backfill_retired_siblings (tests/unit/sources/test_live_batch_support.py) drives the exact live call sequence (bind_raw_revision -> classify_raw_revision_cohort) via the real LiveBatchProcessor._ingest_full_paths_sync entry point (not a hand-simulated call), after retiring two genuinely divergent siblings the same way backfill_historical_revision_evidence does. Asserts the third raw gets a real raw_session_memberships row and a decided outcome ("ambiguous", since it genuinely diverges from one sibling) alongside both retired siblings, rather than being silently dropped. Anti-vacuity confirmed by reverting only batch.py (git stash) and rerunning: fails with the exact pre-fix "no unique byte-revision candidate accepted ... surfacing as failed" warning and a missing raw_session_memberships row for the third raw. - devtools test tests/unit/sources/test_live_batch_support.py -> 126 passed, 5 pre-existing unrelated failures (messages_fts_identity UNIQUE constraint from #3235's rowid-reuse ledger under xdist; reproduces identically on unmodified master, confirmed by stash/rerun). - mypy --strict polylogue/sources/live/batch.py -> clean. Ref polylogue-hm2f Co-Authored-By: Claude --- polylogue/sources/live/batch.py | 128 +++++++++++---- tests/unit/sources/test_live_batch_support.py | 153 ++++++++++++++++++ 2 files changed, 254 insertions(+), 27 deletions(-) diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index dafdb51f16..648b0a8455 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -1823,36 +1823,92 @@ def _ingest_full_records_archive( ), ) plan = archive.classify_raw_revision_cohort(logical_source_key) - if not plan.accepted_raw_ids: - # classify_raw_revision_cohort is a synchronous, - # purely byte-level classification (no async - # conveyor revisits it) -- an empty cohort here - # is a genuine, terminal rewrite/divergence - # conflict, not a pending hand-off. Leave this - # raw out of both raw_ids and deferred_raw_ids so - # the caller's aggregation counts it as a real - # failure (fail-closed), with a log line so the - # cause is diagnosable. - logger.warning( - "live.watcher: no unique byte-revision candidate accepted for %s " - "(logical_source_key=%s) -- surfacing as failed", + if plan.accepted_raw_ids: + parsed_by_raw_id = self._parse_raw_revision_chain(archive, plan) + session_id, applied_raw_ids = archive.apply_raw_revision_replay( + plan, + parsed_by_raw_id, + acquired_at_ms=acquired_at_ms, + stage_timings_s=record_timings, + stage_timing_prefix="full", + ) + record_session_ids.append(session_id) + record_session_count = 1 + record_message_count = sum( + len(parsed_by_raw_id[raw_id].messages) for raw_id in applied_raw_ids + ) + else: + # polylogue-hm2f (live-path half of the + # polylogue-52l2 guard): an empty cohort here + # can mean two different things. + # + # (1) This identity has NO retired sibling + # evidence -- a genuine, terminal + # rewrite/divergence conflict between raws + # that are all still live 'full' rows. There + # is nothing this tick can safely resolve; + # leave this raw out of both raw_ids and + # deferred_raw_ids so the caller's aggregation + # counts it as a real failure (fail-closed), + # with a log line so the cause is diagnosable. + # + # (2) This identity DOES have retired sibling + # evidence (raw_membership_retired_full_ + # revision_siblings is non-empty): the 52l2 + # guard is exactly what emptied this cohort. + # Offline backfill reunites retired siblings + # with a newly discovered raw via its own + # connected-component/membership_candidates + # bookkeeping across the whole census pass; + # the live incremental path processes one + # record per tick with no such durable + # bookkeeping, so it re-derives the sibling + # set from the same durable marker the guard + # itself reads. Fold this raw into membership + # governance -- the same call sequence used + # above for pre-existing-membership and + # multi-session identities -- and pass the + # retired siblings in explicitly so the real + # content-prefix classifier + # (classify_membership_revisions) weighs this + # raw against every known sibling instead of + # evaluating it alone. If the siblings turn + # out to genuinely disagree, that classifier + # still refuses to pick a winner (ambiguous + # quarantine), so this can only ever recover a + # real resolution, never fabricate one. + retired_siblings = archive.raw_membership_retired_full_revision_siblings( + logical_source_key + ) + if not retired_siblings: + logger.warning( + "live.watcher: no unique byte-revision candidate accepted for %s " + "(logical_source_key=%s) -- surfacing as failed", + record.source_path, + logical_source_key, + ) + continue + logger.info( + "live.watcher: reunifying %s with %d retired sibling(s) under membership " + "governance (logical_source_key=%s)", record.source_path, + len(retired_siblings), logical_source_key, ) - continue - parsed_by_raw_id = self._parse_raw_revision_chain(archive, plan) - session_id, applied_raw_ids = archive.apply_raw_revision_replay( - plan, - parsed_by_raw_id, - acquired_at_ms=acquired_at_ms, - stage_timings_s=record_timings, - stage_timing_prefix="full", - ) - record_session_ids.append(session_id) - record_session_count = 1 - record_message_count = sum( - len(parsed_by_raw_id[raw_id].messages) for raw_id in applied_raw_ids - ) + ( + record_session_ids, + record_session_count, + record_message_count, + raw_authority_complete, + ) = self._apply_membership_sessions( + archive, + source_raw_id, + sessions, + acquired_at_ms=acquired_at_ms, + stage_timings_s=record_timings, + allow_current_complete_raw=True, + extra_member_raw_ids=retired_siblings, + ) else: archive.replace_raw_membership_census( source_raw_id, @@ -1958,7 +2014,22 @@ def _apply_membership_sessions( acquired_at_ms: int, stage_timings_s: dict[str, float] | None = None, allow_current_complete_raw: bool = False, + extra_member_raw_ids: tuple[str, ...] = (), ) -> tuple[list[str], int, int, bool]: + """Apply membership-governed classification for one logical identity. + + ``extra_member_raw_ids`` (polylogue-hm2f) names raws that carry known + membership evidence for this identity but are not currently + discoverable through ``archive.raw_membership_raw_ids`` -- concretely, + raws previously retired from full-revision byte governance + (``raw_membership_retired_full_revision_siblings``): their + ``raw_session_memberships`` row survives retirement with + ``revision_authority`` left at the default ``'quarantined'``, so the + ordinary byte-proven-or-caller-owned query never surfaces them. Forced + inclusion here is what lets a later-discovered raw be weighed by the + real content-prefix classifier against siblings the 52l2 guard is + aware of instead of being evaluated alone. + """ session_ids: list[str] = [] session_count = 0 message_count = 0 @@ -1996,6 +2067,9 @@ def _apply_membership_sessions( include_complete_raw_id=source_raw_id if allow_current_complete_raw else None, ) ) + for extra_raw_id in extra_member_raw_ids: + if extra_raw_id not in member_raw_ids: + member_raw_ids.append(extra_raw_id) accepted_head_raw_id = archive.raw_revision_head_raw_id(logical_source_key) if accepted_head_raw_id is not None and accepted_head_raw_id not in member_raw_ids: member_raw_ids.append(accepted_head_raw_id) diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index 5e20b327e2..9cf662eb21 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -15,6 +15,7 @@ import polylogue.sources.live.watcher as live_watcher from polylogue.archive.message.roles import Role from polylogue.archive.revision_authority import ( + HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, RawRevisionAuthority, RawRevisionEnvelope, RawRevisionKind, @@ -3393,6 +3394,158 @@ def conversation(native_id: str, *texts: str) -> dict[str, object]: ).fetchone() == (accepted_raw_id,) +def test_live_third_raw_reunifies_with_backfill_retired_siblings(tmp_path: Path) -> None: + """polylogue-hm2f: the live incremental path must reunite retired siblings, not drop new raws forever. + + Mirrors the exact live call sequence the polylogue-52l2 guard protects + (``bind_raw_revision`` -> ``classify_raw_revision_cohort``), then proves + the new routing this fix adds: when that cohort comes back empty AND + ``raw_membership_retired_full_revision_siblings`` shows this identity has + known siblings already retired to membership governance -- exactly the + durable state offline backfill (``sources/revision_backfill.py``, + ``convertible_full_revision_raw_ids`` + ``replace_raw_membership_census``) + leaves behind for a decided-ambiguous full-only cohort -- a newly + discovered THIRD raw for the same identity must be folded into that same + membership governance and weighed by the real content-prefix classifier + (``classify_membership_revisions``) alongside every known sibling, + instead of being silently dropped with only a warning log line (the + pre-fix behavior: ``bind_raw_revision`` succeeds, but no + ``raw_session_memberships`` row is ever written for the raw and the file + surfaces as failed with zero evidence trail). + + raw_a=["base","left"], raw_b=["base","right"] are byte-divergent (not a + prefix of one another) -- a genuine, decided ambiguous cohort, retired + here exactly the way ``backfill_historical_revision_evidence`` retires + one once ``classify_raw_revision_cohort`` returns no accepted chain. + raw_c=["base","left","extra"] then arrives through the live incremental + path (``LiveBatchProcessor._ingest_full_paths_sync``, the production + entry point, not a hand-simulated call). Content-wise raw_c does not + strictly dominate raw_b (they diverge at message index 1) so the real + classifier correctly still refuses to pick a winner -- but critically + that decision is reached by weighing raw_c against BOTH retired + siblings, and raw_c ends up in ``raw_session_memberships`` with a real, + decided outcome alongside raw_a and raw_b, proving reunification + happened rather than raw_c being evaluated alone or dropped. + """ + + def conversation(native_id: str, *texts: str) -> dict[str, object]: + mapping: dict[str, object] = { + "root": {"id": "root", "message": None, "parent": None, "children": [f"{native_id}-node-0"]} + } + for index, text in enumerate(texts): + node_id = f"{native_id}-node-{index}" + next_node = f"{native_id}-node-{index + 1}" if index + 1 < len(texts) else None + mapping[node_id] = { + "id": node_id, + "parent": "root" if index == 0 else f"{native_id}-node-{index - 1}", + "children": [] if next_node is None else [next_node], + "message": { + "id": f"{native_id}-message-{index}", + "author": {"role": "user"}, + "create_time": 1_780_000_000.0 + index, + "content": {"content_type": "text", "parts": [text]}, + "metadata": {}, + }, + } + return { + "id": native_id, + "title": native_id, + "current_node": f"{native_id}-node-{len(texts) - 1}", + "mapping": mapping, + } + + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as store: + raw_a = store.write_raw_payload( + provider=Provider.CHATGPT, + payload=json.dumps([conversation("shared", "base", "left")]).encode(), + source_path="a.json", + acquired_at_ms=1, + ) + store.bind_raw_revision( + raw_a, + RawRevisionEnvelope( + "chatgpt:shared", RawRevisionKind.FULL, raw_a, 0, authority=RawRevisionAuthority.QUARANTINED + ), + ) + raw_b = store.write_raw_payload( + provider=Provider.CHATGPT, + payload=json.dumps([conversation("shared", "base", "right")]).encode(), + source_path="b.json", + acquired_at_ms=2, + ) + store.bind_raw_revision( + raw_b, + RawRevisionEnvelope( + "chatgpt:shared", RawRevisionKind.FULL, raw_b, 0, authority=RawRevisionAuthority.QUARANTINED + ), + ) + + # Exactly the polylogue-52l2 guard-tripping sequence: no unique + # byte-prefix chain across a and b. + plan = store.classify_raw_revision_cohort("chatgpt:shared") + assert plan.accepted_raw_ids == () + + # Mirror backfill_historical_revision_evidence's own retirement step + # once a full-only cohort is decided ambiguous: move every + # convertible full raw to membership governance. + for raw_id in store.convertible_full_revision_raw_ids("chatgpt:shared"): + sessions = LiveBatchProcessor._parse_retained_raw_sessions(store, raw_id) + store.replace_raw_membership_census( + raw_id, + sessions, + parser_fingerprint="revision-membership-v1", + censused_at_ms=0, + detail=HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, + retire_full_revision_governance=True, + ) + store.commit() + retired_siblings = store.raw_membership_retired_full_revision_siblings("chatgpt:shared") + assert set(retired_siblings) == {raw_a, raw_b} + + # A THIRD raw for the same logical identity, discovered afterward + # through the actual live incremental entry point. + root = tmp_path / "inbox" + root.mkdir() + third = root / "third.json" + third.write_text(json.dumps([conversation("shared", "base", "left", "extra")]), encoding="utf-8") + index_db = tmp_path / "index.db" + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="inbox", root=root, suffixes=(".json",)),), + cursor=CursorStore(index_db), + parser_fingerprint="test-parser", + ) + third_result = processor._ingest_full_paths_sync([third], source_name="inbox") + + with sqlite3.connect(tmp_path / "source.db") as conn: + rows = conn.execute( + """ + SELECT r.source_path, m.decision + FROM raw_session_memberships AS m + JOIN raw_sessions AS r USING (raw_id) + WHERE m.logical_source_key = 'chatgpt:shared' + ORDER BY r.source_path + """ + ).fetchall() + + # Reunification proof: raw_c (third.json) has a raw_session_memberships + # row -- it was folded into the SAME membership cohort as raw_a/raw_b, + # not evaluated alone and not silently dropped. Every member of the + # cohort has a real DECIDED outcome (not NULL/pending, not simply + # absent). + by_path = dict(rows) + assert set(by_path) == {"a.json", "b.json", str(third)} + assert all(decision is not None for decision in by_path.values()) + + # The genuine three-way content divergence still correctly refuses to + # pick a winner -- reunification recovers a real resolution when one + # exists, it does not fabricate one. + assert by_path[str(third)] == "ambiguous" + assert third_result.failed == [third] + assert third_result.succeeded == [] + + def test_raw_membership_decision_pending_distinguishes_null_from_ambiguous(tmp_path: Path) -> None: """Pins the exact narrow scoping of the polylogue-emx2 fix (de0b2df7a regression, polylogue-lvz6 triage).