diff --git a/polylogue/archive/revision_authority.py b/polylogue/archive/revision_authority.py index 0cdb9766fa..5d1f17c5f8 100644 --- a/polylogue/archive/revision_authority.py +++ b/polylogue/archive/revision_authority.py @@ -23,6 +23,16 @@ class RawRevisionAuthority(StrEnum): BYTE_AUTHORITY_CENSUS_DETAIL = "append fragments are governed by byte revision authority" +#: ``raw_membership_census.detail`` marker written when a full-only, +#: non-prefix-chain cohort is retired from byte-revision governance to +#: membership governance (``ArchiveStore.replace_raw_membership_census``, +#: ``retire_full_revision_governance=True``). Shared between the writer and +#: ``ArchiveStore.classify_raw_revision_cohort``'s polylogue-52l2 guard, +#: which uses it to detect that a logical identity already has retired, +#: previously-ambiguous sibling evidence before ever accepting a +#: later-discovered raw as an unconditional singleton byte-proven baseline. +HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL = "historical non-prefix full revision governance" + @dataclass(frozen=True) class RawRevisionEnvelope: diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index a227ebbe04..dafdb51f16 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -24,6 +24,7 @@ NATIVE_BROWSER_CAPTURE_INGEST_FLAG, ) from polylogue.archive.revision_authority import ( + HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, RawRevisionAuthority, RawRevisionEnvelope, RawRevisionKind, @@ -1983,7 +1984,7 @@ def _apply_membership_sessions( retained_sessions, parser_fingerprint=self._current_parser_fingerprint(), censused_at_ms=acquired_at_ms, - detail="cross-route full revision governance", + detail=HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, retire_full_revision_governance=True, ) member_sessions: dict[str, Any] = {} diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index 66e72bfdb6..00c43f2ded 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -25,6 +25,7 @@ ) from polylogue.archive.revision_authority import ( BYTE_AUTHORITY_CENSUS_DETAIL, + HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, RawRevisionAuthority, RawRevisionEnvelope, RawRevisionKind, @@ -760,7 +761,7 @@ def commit_replay_unit() -> None: sessions, parser_fingerprint="revision-membership-v1", censused_at_ms=0, - detail="historical non-prefix full revision governance", + detail=HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, retire_full_revision_governance=True, ) membership_candidates.setdefault(logical_key, set()).add(raw_id) @@ -1044,6 +1045,33 @@ def _pool_dispatch_amortizes(pool_raw_ids: list[str], payload_sizes: dict[str, i return total >= _parse_pool_min_aggregate_bytes() +#: Providers whose parsed session identity is derived purely from payload +#: bytes, never from ``source_path`` -- safe to dedup census parse ACROSS +#: source paths sharing a ``blob_hash`` (polylogue-869u). Excluded +#: deliberately: ``Provider.BEADS`` derives workspace-scoped native ids from +#: ``source_path`` (``sources/parsers/beads.py:_repository_root``); +#: ``Provider.ANTIGRAVITY``'s brain-metadata mode derives its +#: ``profile_root``/artifact path from ``source_path`` +#: (``sources/dispatch.py``'s ``antigravity.parse_brain_metadata`` call); +#: ``Provider.HERMES``'s ATOF/ATIF/verification-evidence modes likewise +#: derive ``profile_root`` from ``source_path``. Those three keep the +#: conservative same-path-only dedup below. ``Provider.UNKNOWN`` (browser +#: capture / unclassified) is also excluded out of caution -- its identity +#: derivation is not centrally audited here. +_PATH_INDEPENDENT_PARSE_PROVIDERS: Final[frozenset[Provider]] = frozenset( + { + Provider.CHATGPT, + Provider.CLAUDE_AI, + Provider.CLAUDE_CODE, + Provider.CODEX, + Provider.GEMINI, + Provider.GEMINI_CLI, + Provider.GROK, + Provider.DRIVE, + } +) + + def _parse_retained_raws( archive: ArchiveStore, raw_ids: list[str], @@ -1055,15 +1083,22 @@ def _parse_retained_raws( Returns each outcome keyed by raw_id: either the parsed ``(sessions, payload_bytes, revision_kind)`` tuple or the caught - exception. Rows sharing the same ``(blob_hash, source_path)`` are parsed - exactly once and the outcome fanned out: identical bytes at an identical - path decode deterministically identically, so re-parsing them is pure - waste (measured live 2026-07-19: 17% of newest-only bytes — e.g. one - 442MB codex rollout stored under 8 raw rows — were byte-identical - duplicates each paying a full parse). ``source_path`` stays in the key - because some parsers derive identity from the path (e.g. beads workspace - ids), so cross-path duplicates are deliberately NOT deduplicated. - Per-row ``revision_kind`` is re-attached from each row's own descriptor. + exception. Rows sharing the same ``blob_hash`` for a + ``_PATH_INDEPENDENT_PARSE_PROVIDERS`` provider are parsed exactly once + and the outcome fanned out, regardless of ``source_path``: identical + bytes decode deterministically identically for those providers, so + re-parsing them per row (even under a DIFFERENT acquired path -- a + common re-acquisition/re-export shape) is pure waste. For every other + provider the dedup key still includes ``source_path`` (some parsers + derive identity from the path, e.g. beads workspace ids -- see + ``_PATH_INDEPENDENT_PARSE_PROVIDERS``'s docstring for the excluded + providers), so cross-path duplicates for those stay deliberately NOT + deduplicated. Live evidence (polylogue-869u, 2026-07-19): 87,177 + newest-revision raws / 52.1 GiB but only 85,066 distinct blob hashes / + 43.4 GiB -- the same bytes (e.g. one 442 MB codex rollout) recur under + up to 8 different ``logical_source_key``s / source paths and, before + this cross-path widening, paid a full parse each time. Per-row + ``revision_kind`` is re-attached from each row's own descriptor. ``prefetch_cache`` (polylogue-m6tp phase (a)) is consulted BEFORE any of the above: a raw_id already popped from the cache is used directly and @@ -1083,10 +1118,11 @@ def _parse_retained_raws( remaining_raw_ids.append(raw_id) else: results[raw_id] = cached - grouped: dict[tuple[str, str], list[str]] = {} + grouped: dict[tuple[Provider, str, str], list[str]] = {} for raw_id in remaining_raw_ids: - _provider, blob_hash, source_path, _kind, _size = descriptors[raw_id] - grouped.setdefault((blob_hash, source_path), []).append(raw_id) + provider, blob_hash, source_path, _kind, _size = descriptors[raw_id] + dedup_path = "" if provider in _PATH_INDEPENDENT_PARSE_PROVIDERS else source_path + grouped.setdefault((provider, blob_hash, dedup_path), []).append(raw_id) representatives = [members[0] for members in grouped.values()] unique = _parse_unique_retained_raws( archive, representatives, descriptors=descriptors, ingest_workers=ingest_workers diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index ba6c568640..7151a4224e 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -38,6 +38,7 @@ QueryTextPredicate, ) from polylogue.archive.revision_authority import ( + HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, HistoricalRawRevisionStream, RawRevisionAuthority, RawRevisionEnvelope, @@ -1986,6 +1987,39 @@ def raw_append_revision_parent( row = rows[0] return str(row[0]), str(row[1]), int(row[2]) + 1 + def raw_membership_retired_full_revision_siblings(self, logical_source_key: str) -> tuple[str, ...]: + """Return raws previously retired from full-revision byte governance for this key. + + ``replace_raw_membership_census(..., retire_full_revision_governance=True)`` + nulls the retired raw's ``raw_sessions.logical_source_key`` and sets + ``revision_authority='quarantined'`` -- it becomes invisible both to + ``classify_raw_revision_cohort``'s own byte-row query and to + ``raw_membership_rebuild_raw_ids``'s deliberately byte-proven-only + filter (polylogue-lkrc/#2822 guards a different hazard: reopening a + quarantined member against an already-established head). Its + ``raw_session_memberships`` row, keyed by the raw's own parsed + logical identity, survives retirement together with a + ``raw_membership_census.detail`` marker naming this specific + 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. + """ + rows = ( + self._ensure_source_conn() + .execute( + """ + 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 = ? + ORDER BY m.raw_id + """, + (logical_source_key, HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL), + ) + .fetchall() + ) + return tuple(str(row[0]) for row in rows) + def classify_raw_revision_cohort(self, logical_source_key: str) -> RevisionReplayPlan: """Promote only a unique byte-prefix full chain and contiguous appends.""" if self._blob_publisher is None: @@ -1999,6 +2033,26 @@ def classify_raw_revision_cohort(self, logical_source_key: str) -> RevisionRepla """, (logical_source_key,), ).fetchall() + # polylogue-52l2: a byte chain is classified against whichever full + # rows the caller happens to have discovered/censused so far, not + # against the complete sibling population for this logical identity + # -- an earlier pass can have already retired ambiguous siblings to + # membership governance (nulling their logical_source_key, see + # raw_membership_retired_full_revision_siblings). If that leaves a + # later-discovered raw as the ONLY remaining 'full' row here, it + # would be evaluated as a trivial one-member "chain" and + # unconditionally accepted as a byte-proven baseline by + # classify_historical_full_revision_streams (no sibling to prove a + # byte prefix against) -- permanently establishing session content + # from whichever raw happened to be discovered last, independent of + # which content is actually correct. Refuse the byte-chain path + # entirely whenever this identity has retired sibling evidence: the + # caller's existing "no accepted chain" fallback + # (convertible_full_revision_raw_ids) folds these full rows into + # membership governance instead, where the real prefix-based + # classifier weighs every known sibling together. + if full_rows and self.raw_membership_retired_full_revision_siblings(logical_source_key): + full_rows = [] historical: list[HistoricalRawRevisionStream] = [] for row in full_rows: diff --git a/tests/unit/sources/test_revision_backfill.py b/tests/unit/sources/test_revision_backfill.py index dcb919fef8..d4400c2ac6 100644 --- a/tests/unit/sources/test_revision_backfill.py +++ b/tests/unit/sources/test_revision_backfill.py @@ -1122,12 +1122,15 @@ def test_partition_raws_by_dispatch_size_routes_small_to_pool_large_sequential() assert sequential_ids == ["large-1", "large-2"] -def test_parse_retained_raws_dedupes_identical_blob_and_path(monkeypatch: pytest.MonkeyPatch) -> None: - """Byte-identical rows at the same source_path parse once and the outcome - fans out with each row's own revision_kind; the same bytes at a DIFFERENT - path still parse separately (path participates in some parsers' identity). - Live shape (polylogue-869u): one 442MB codex rollout acquired 8x within - 2.3s during a stampede — 8 raw rows, one blob, 8 full parses.""" +def test_parse_retained_raws_dedupes_identical_blob_across_paths_for_safe_providers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """polylogue-869u: for a path-independent provider (Codex here), rows + sharing a ``blob_hash`` parse once and fan out -- INCLUDING across + different ``source_path``s, since those parsers' session identity comes + entirely from the payload bytes. Live shape: one 442MB codex rollout + acquired 8x within 2.3s during a stampede, at up to 8 different acquired + paths -- 8 raw rows, one blob, formerly 8 full parses.""" descriptors = { "dup-1": (Provider.CODEX, "hash-A", "same.jsonl", RawRevisionKind.FULL, 10), "dup-2": (Provider.CODEX, "hash-A", "same.jsonl", RawRevisionKind.UNKNOWN, 10), @@ -1155,13 +1158,55 @@ def fake_parse(archive: object, raw_id: str) -> tuple[list[ParsedSession], int, ingest_workers=1, ) - # one parse per distinct (blob_hash, source_path): dup-2/dup-3 reuse dup-1's - assert parsed == ["dup-1", "other-path", "other-bytes"] + # one parse per distinct blob_hash: dup-2/dup-3/other-path all reuse + # dup-1's outcome despite other-path having a different source_path. + assert parsed == ["dup-1", "other-bytes"] assert set(results) == set(descriptors) sessions, size, kind = results["dup-2"] # type: ignore[misc] assert (sessions, size, kind) == ([], 10, RawRevisionKind.UNKNOWN) _sessions, _size, dup3_kind = results["dup-3"] # type: ignore[misc] assert dup3_kind == RawRevisionKind.FULL + _sessions, other_path_size, other_path_kind = results["other-path"] # type: ignore[misc] + assert (other_path_size, other_path_kind) == (10, RawRevisionKind.FULL) + + +def test_parse_retained_raws_preserves_path_scoped_dedup_for_path_dependent_providers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """polylogue-869u: Beads derives workspace-scoped native ids from + ``source_path`` (``sources/parsers/beads.py:_repository_root``), so + byte-identical Beads rows at DIFFERENT paths must keep parsing + separately -- unlike the path-independent providers, cross-path + identity is not provably safe here.""" + descriptors = { + "same-path-a": (Provider.BEADS, "hash-A", "workspace-one/issues.jsonl", RawRevisionKind.FULL, 10), + "same-path-b": (Provider.BEADS, "hash-A", "workspace-one/issues.jsonl", RawRevisionKind.FULL, 10), + "other-path": (Provider.BEADS, "hash-A", "workspace-two/issues.jsonl", RawRevisionKind.FULL, 10), + } + + class FakeArchive: + def raw_revision_descriptor(self, raw_id: str) -> tuple[Provider, str, str, RawRevisionKind, int]: + return descriptors[raw_id] + + parsed: list[str] = [] + + def fake_parse(archive: object, raw_id: str) -> tuple[list[ParsedSession], int, RawRevisionKind]: + parsed.append(raw_id) + descriptor = descriptors[raw_id] + return [], descriptor[4], descriptor[3] + + monkeypatch.setattr(revision_backfill, "_parse_retained_raw", fake_parse) + + results = revision_backfill._parse_retained_raws( + FakeArchive(), # type: ignore[arg-type] + list(descriptors), + ingest_workers=1, + ) + + # same-path-b reuses same-path-a's outcome; other-path (different + # source_path, same bytes) still pays its own parse. + assert parsed == ["same-path-a", "other-path"] + assert set(results) == set(descriptors) def test_parse_retained_raws_fans_out_exceptions_to_duplicate_rows(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/unit/storage/test_revision_replay.py b/tests/unit/storage/test_revision_replay.py index 0050e94e6e..852659718a 100644 --- a/tests/unit/storage/test_revision_replay.py +++ b/tests/unit/storage/test_revision_replay.py @@ -439,6 +439,103 @@ def parsed(*messages: tuple[str, str]) -> ParsedSession: assert tuple(head) == (folded, 20) +def test_isolated_later_raw_does_not_override_known_ambiguous_cohort(tmp_path: Path) -> None: + """polylogue-52l2: a raw discovered for a logical identity that already + has quarantined/ambiguous siblings must not be accepted as an + unambiguous singleton byte-proven baseline. + + This mirrors the LIVE incremental watcher's own call sequence + (``sources/live/batch.py``): ``bind_raw_revision`` then + ``classify_raw_revision_cohort`` directly, with no census-phase + re-derivation or connected-component re-expansion in between (those only + happen in the offline ``backfill_historical_revision_evidence`` path, + which is why this bug does not reproduce through that entry point). + + ``classify_raw_revision_cohort`` only ever queries + ``raw_sessions WHERE logical_source_key = ? AND revision_kind = 'full'``. + Retiring an ambiguous sibling to membership governance + (``replace_raw_membership_census(..., retire_full_revision_governance=True)``, + exactly what the backfill caller does with + ``convertible_full_revision_raw_ids`` once a cohort is decided + ambiguous) nulls its ``raw_sessions.logical_source_key`` -- it becomes + invisible to that query. A THIRD raw for the same identity, discovered + afterward, is then evaluated completely alone: + ``classify_historical_full_revision_streams`` unconditionally accepts a + singleton stream as a "byte-proven baseline" (there is no sibling to + compare a byte-prefix against), so the isolated raw would permanently + become the accepted session content -- an outcome that depends on + incremental discovery order, not on which content is actually correct. + """ + 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 == () + + # Both siblings genuinely disagree (no byte-prefix relation) -- + # exactly what the backfill caller does when a cohort is decided + # ambiguous: move it to membership governance so parsed-content + # prefix rules can still arbitrate it later. + 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="historical non-prefix 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") + + # The isolated raw must not be promoted alone: this identity has known, + # unresolved ambiguous siblings that a real classifier must weigh it + # against, not silently outrank by discovery order. + assert second_plan.accepted_raw_ids == () + + def test_real_single_append_chain_folds_segmentation_distinct_full_snapshot(tmp_path: Path) -> None: initialize_active_archive_root(tmp_path)