From a613a38b2c8dc992eb6186a6eb8ec6b025b23ca3 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 21 Jul 2026 00:06:30 +0200 Subject: [PATCH 1/2] fix(sources): refuse isolated singleton byte-chain acceptance over known ambiguous cohort Problem Ref polylogue-52l2. classify_raw_revision_cohort classifies a byte-prefix chain from whichever raw_sessions rows currently carry revision_kind='full' for a logical_source_key -- not against the complete sibling population for that identity. Retiring an ambiguous sibling to membership governance (replace_raw_membership_census(..., retire_full_revision_governance=True), what the backfill/live-watcher callers do once a cohort is decided ambiguous) nulls that raw's raw_sessions.logical_source_key, so it disappears from the query. A THIRD raw for the same identity, discovered afterward (e.g. a re-acquired browser-capture snapshot), is then evaluated completely alone: classify_historical_full_revision_streams unconditionally accepts a one-member stream as a byte-proven baseline (no sibling to prove a byte prefix against), so the isolated raw permanently becomes the accepted session content -- an outcome that depends on incremental discovery order, not on which content is actually correct. Reproduced directly against ArchiveStore (test_revision_replay.py), mirroring the live incremental watcher's own call sequence (sources/live/batch.py: bind_raw_revision then classify_raw_revision_cohort, no census-phase re-derivation in between). The equivalent two-call scenario through backfill_historical_revision_evidence does NOT reproduce: its census phase unconditionally re-parses every still-unindexed raw and its connected-component selection expansion reunites retired siblings by shared membership evidence before classification runs -- protections the live incremental path lacks. Solution - New ArchiveStore.raw_membership_retired_full_revision_siblings(): finds raw_session_memberships rows for a logical_source_key whose raw_membership_census.detail matches the (now shared) HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL marker written at retirement -- survives the raw_sessions.logical_source_key NULL-out. - classify_raw_revision_cohort refuses the byte-chain path entirely (empty full_rows) whenever this identity has retired sibling evidence, so the caller's existing "no accepted chain" fallback (convertible_full_revision_raw_ids) folds the newly-discovered raw into membership governance instead, where the real prefix-based classifier weighs every known sibling together, rather than silently establishing a wrong permanent head. - Unified the two retire_full_revision_governance call sites' previously-divergent detail strings ("historical non-prefix full revision governance" in revision_backfill.py, "cross-route full revision governance" in live/batch.py) onto the one shared constant so the guard recognizes retirement from either path. Deferred (not in scope for this fix): once an identity's retired siblings are recognized, nothing in the LIVE incremental watcher path (unlike offline backfill) currently re-unites them with the new raw for a real membership decision -- it fails closed instead (logs "no unique byte-revision candidate accepted ... surfacing as failed"), correctly never establishing wrong content, but requiring a later offline backfill/repair pass to resolve the identity. Also out of scope: PR #3204 already tracks a known, separate limitation (a content-ahead capture still yielding to a byte-kind chain head) as polylogue-nfl5. Verification devtools test tests/unit/storage/test_revision_replay.py tests/unit/sources/test_revision_backfill.py tests/unit/storage/test_raw_retention.py -> 178 passed (60 + 59 + 59). New regression test_isolated_later_raw_does_not_override_known_ambiguous_cohort fails on unmodified master (asserts accepted_raw_ids == () but gets the isolated raw's id) and passes after this change. mypy --strict clean on the four touched modules. Ref polylogue-52l2 Co-Authored-By: Claude --- polylogue/archive/revision_authority.py | 10 ++ polylogue/sources/live/batch.py | 3 +- polylogue/sources/revision_backfill.py | 3 +- .../storage/sqlite/archive_tiers/archive.py | 54 +++++++++++ tests/unit/storage/test_revision_replay.py | 97 +++++++++++++++++++ 5 files changed, 165 insertions(+), 2 deletions(-) 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..bc45e8db53 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) 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/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) From 5b1708515e3c1b94af8714070a057cd9b0ec6e19 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 21 Jul 2026 00:17:01 +0200 Subject: [PATCH 2/2] perf(sources): dedup census parse by blob_hash across source paths Problem Ref polylogue-869u. Live evidence (2026-07-19, source.db): the newest-revision-per-logical_source_key population is 87,177 rows / 52.1 GiB, but only 85,066 DISTINCT blob_hash values / 43.4 GiB -- the same bytes (e.g. one 442MB codex computer-use rollout) recur under up to 8 different logical_source_keys / source paths, from re-acquisition or re-export. The existing dedup (#3151, _parse_retained_raws) only collapses rows sharing BOTH blob_hash AND source_path, so this whole cross-path duplicate population -- 8.7GB / 17% of newest-only bytes -- still paid a full parse per row. Solution _parse_retained_raws' grouping key now drops source_path for a new _PATH_INDEPENDENT_PARSE_PROVIDERS allowlist (ChatGPT, Claude web/Code, Codex, Gemini/Gemini CLI, Grok, Drive): those parsers derive session identity purely from payload bytes, so identical bytes decode identically regardless of acquired path. Providers whose parse DOES depend on source_path keep the original (blob_hash, source_path) key unchanged: Beads derives workspace-scoped native ids from source_path (sources/parsers/beads.py:_repository_root), Antigravity's brain-metadata mode and Hermes' ATOF/ATIF/verification-evidence modes derive profile_root from source_path (sources/dispatch.py). Provider.UNKNOWN stays path-scoped too, out of caution. Updated the existing test_parse_retained_raws_dedupes_identical_blob_and_path (renamed test_..._across_paths_for_safe_providers) to assert the new cross-path Codex dedup; added test_parse_retained_raws_preserves_path_scoped_dedup_for_path_dependent_providers to prove Beads rows at different paths still parse separately (the safety property this change must not regress). Verification (dedup receipt) Synthetic corpus: 40 distinct ~300KB Codex payloads, each duplicated across 5 different source paths (200 raw rows total, mirroring the live re-acquisition-stampede shape) -- measured via tests/infra/revision_backfill_benchmark.py-style corpus construction, counting real _parse_retained_raw invocations: before (blob_hash+source_path key): 200 parse calls, 0.516s wall clock after (blob_hash key, safe providers): 40 parse calls, 0.104s wall clock 160 avoided parse calls (80% reduction) on this corpus; proportional to duplicate ratio on the live archive's 17% figure. devtools test tests/unit/sources/test_revision_backfill.py -> 40 passed. devtools test tests/unit/storage/test_revision_replay.py tests/unit/storage/test_raw_retention.py tests/property/test_sql_injection_boundary.py -> 120 passed. mypy --strict clean. Ref polylogue-869u Co-Authored-By: Claude --- polylogue/sources/revision_backfill.py | 59 +++++++++++++++---- tests/unit/sources/test_revision_backfill.py | 61 +++++++++++++++++--- 2 files changed, 100 insertions(+), 20 deletions(-) diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index bc45e8db53..00c43f2ded 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -1045,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], @@ -1056,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 @@ -1084,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/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: