From 52027b4242c6a20b9ded4e77e66cee47b439b19f Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 2 Aug 2026 18:24:21 +0200 Subject: [PATCH 1/2] fix(storage): collapse byte-equal duplicates before revision-chain proof Problem: classify_historical_full_revision_streams (and its eager sibling) treated any two full-revision captures that tied on size as an unprovable fork, quarantining the *entire* cohort even when the tie was two byte-identical copies of the same capture -- the ordinary result of re-acquiring an unchanged file. A live audit (polylogue-lb39z) measured 13,671 quarantined raw_sessions rows / 50.2GB (79% of quarantined bytes) caused by exactly this: duplicate captures misread as ambiguity, with the verdict written cohort-wide instead of localized to any real divergence. A stuck baseline from this bug also let the live watcher pump unbounded quarantined rows for one growing file (one Codex rollout produced 799 quarantined snapshots / 6.28GB of overlapping tails). What changed: both classifiers now (1) collapse byte-identical revisions onto one representative before any chain-order proof runs (I4) -- content hashed directly for in-memory payloads, streamed via one read pass for the streaming variant, so no extra I/O beyond what already ran; every non-representative duplicate mirrors its representative's verdict with a new `relation="duplicate"`; and (2) localize any residual divergence to only the fork tip instead of the whole cohort (I5): a shared ancestor with two mutually-incomparable children now classifies the ancestor BYTE_PROVEN and only the fork's children QUARANTINED, rather than quarantining all three. Two disconnected/incomparable roots (no shared ancestor at all) still quarantine everything -- there is no anchor to localize from, matching prior behavior for that genuinely irreducible case. classify_untyped_full_revision_groups' "whole cohort must be one provable chain" contract is preserved by checking every decision instead of only the first. Alternatives rejected: promoting the fork's larger/newer-looking child as the accepted head was considered and rejected -- it would make an arbitrary sort-order or size tie-break silently pick a winner among genuinely competing evidence, which is exactly the kind of silent-wrong-guess this subsystem has zero tolerance for (see polylogue-yla8's write-gate history). Leaving the fork tip quarantined and letting the (separate, still-unwired) judgment-assertion flow adjudicate it is the safe default. Verification: devtools test tests/unit/storage/test_raw_revision_authority.py (24 passed, including new anti-vacuity tests proving the pre-fix classifier quarantined a byte-equal duplicate pair and a shared-root fork wholesale, and the post-fix classifier does not); devtools test -k "raw_revision or revision_governance or raw_authority" (115 passed, 6 failed -- all 6 reproduce identically with this change reverted via `git stash`, confirmed pre-existing/unrelated: 5 are devtools/raw_authority_scale_proof.py synthetic-corpus repair-pass assertions and 1 is an integration daemon-health probe timeout); mypy --strict on both touched modules (no issues); devtools lab policy schema-versioning (intact); ruff check + format (clean). Ref polylogue-lb39z (Phase 1, item 1 of 5). Co-Authored-By: Claude --- polylogue/archive/revision_authority.py | 254 ++++++++++++------ .../archive_tiers/revision_governance.py | 10 +- .../storage/test_raw_revision_authority.py | 139 +++++++++- 3 files changed, 313 insertions(+), 90 deletions(-) diff --git a/polylogue/archive/revision_authority.py b/polylogue/archive/revision_authority.py index 56c956ef6a..ba9822fb25 100644 --- a/polylogue/archive/revision_authority.py +++ b/polylogue/archive/revision_authority.py @@ -112,7 +112,7 @@ class HistoricalRawRevisionStream: class HistoricalRevisionDecision: raw_id: str authority: RawRevisionAuthority - relation: Literal["baseline", "predecessor", "ambiguous"] + relation: Literal["baseline", "predecessor", "ambiguous", "duplicate"] predecessor_raw_id: str | None = None @@ -121,80 +121,165 @@ def append_source_revision(predecessor_revision: str, payload_hash: str) -> str: return sha256(f"{predecessor_revision}\0{payload_hash}".encode()).hexdigest() -def classify_historical_full_revisions( - revisions: list[HistoricalRawRevision], -) -> list[HistoricalRevisionDecision]: - """Prove a unique byte-prefix chain; quarantine every ambiguous cohort. - - Acquisition time, source path, provider timestamps, and raw-id ordering are - intentionally absent. Equal or divergent payloads do not establish which - capture is newer. +def _classify_deduped_nodes( + node_ids: list[str], + sizes: dict[str, int], + is_prefix: Callable[[str, str], bool], +) -> dict[str, HistoricalRevisionDecision]: + """Prove a byte-prefix DAG over already-deduplicated (distinct-content) nodes. + + Implements I4/I5 (polylogue-lb39z): a divergence quarantines only the + genuinely-divergent suffix, never a proven prefix chain or an unrelated + sibling that happens to share a cohort with it. + + ``is_prefix(parent, child)`` must only be evaluated for pairs where + ``sizes[parent] < sizes[child]`` (the caller guarantees this so streamed + implementations can skip same-size/reverse-size comparisons entirely). + + A node is only ever classified ``BYTE_PROVEN`` if it sits on a single, + unbranched path back to the cohort's one true root: every ancestor, + including the node itself, has exactly one maximal parent candidate and + is the sole child of that parent. The moment a node's parent has more + than one child (a fork) or a node has more than one incomparable maximal + parent (an ambiguous parent set), that node -- and everything built on + top of it -- is quarantined, while everything already proven up to that + point (including the fork point itself) is left alone. If the cohort has + zero or more than one root (no shared ancestor at all, e.g. two + unrelated same-size captures), there is no anchor to localize from and + the whole cohort is quarantined, matching prior behavior for that case. """ - if not revisions: - return [] - by_id = {revision.raw_id: revision for revision in revisions} + ordered = sorted(node_ids, key=lambda raw_id: (sizes[raw_id], raw_id)) parents: dict[str, list[str]] = {} - children: dict[str, list[str]] = {raw_id: [] for raw_id in by_id} - for child in revisions: + children: dict[str, list[str]] = {raw_id: [] for raw_id in ordered} + for child in ordered: candidates = [ - parent.raw_id - for parent in revisions - if parent.raw_id != child.raw_id - and len(parent.payload) < len(child.payload) - and child.payload.startswith(parent.payload) + parent + for parent in ordered + if parent != child and sizes[parent] < sizes[child] and is_prefix(parent, child) ] maximal = [ candidate for candidate in candidates if not any( - candidate != other - and len(by_id[candidate].payload) < len(by_id[other].payload) - and by_id[other].payload.startswith(by_id[candidate].payload) + candidate != other and sizes[candidate] < sizes[other] and is_prefix(candidate, other) for other in candidates ) ] - parents[child.raw_id] = maximal + parents[child] = maximal for parent in maximal: - children[parent].append(child.raw_id) - roots = [raw_id for raw_id, parent_ids in parents.items() if not parent_ids] - leaves = [raw_id for raw_id, child_ids in children.items() if not child_ids] - unique_chain = ( - len(roots) == 1 - and len(leaves) == 1 - and all(len(parent_ids) <= 1 for parent_ids in parents.values()) - and all(len(child_ids) <= 1 for child_ids in children.values()) - ) - if not unique_chain: - return [ - HistoricalRevisionDecision( - raw_id=revision.raw_id, authority=RawRevisionAuthority.QUARANTINED, relation="ambiguous" + children[parent].append(child) + + roots = [raw_id for raw_id in ordered if not parents[raw_id]] + if len(roots) != 1: + return { + raw_id: HistoricalRevisionDecision( + raw_id=raw_id, authority=RawRevisionAuthority.QUARANTINED, relation="ambiguous" ) - for revision in revisions - ] + for raw_id in ordered + } root = roots[0] - decisions: list[HistoricalRevisionDecision] = [] - current: str | None = root - while current is not None: - predecessor: str | None = parents[current][0] if parents[current] else None - relation: Literal["baseline", "predecessor"] = "baseline" if not parents[current] else "predecessor" - decisions.append( - HistoricalRevisionDecision( - raw_id=current, + clean: dict[str, bool] = {root: True} + for raw_id in ordered: + if raw_id == root: + continue + parent_set = parents[raw_id] + if len(parent_set) != 1: + clean[raw_id] = False + continue + (parent,) = parent_set + clean[raw_id] = clean.get(parent, False) and len(children[parent]) == 1 + + decisions: dict[str, HistoricalRevisionDecision] = {} + for raw_id in ordered: + if clean.get(raw_id, False): + parent_set = parents[raw_id] + predecessor = parent_set[0] if parent_set else None + relation: Literal["baseline", "predecessor"] = "baseline" if predecessor is None else "predecessor" + decisions[raw_id] = HistoricalRevisionDecision( + raw_id=raw_id, authority=RawRevisionAuthority.BYTE_PROVEN, relation=relation, predecessor_raw_id=predecessor, ) - ) - current = children[current][0] if children[current] else None + else: + decisions[raw_id] = HistoricalRevisionDecision( + raw_id=raw_id, authority=RawRevisionAuthority.QUARANTINED, relation="ambiguous" + ) return decisions -def _stream_size(revision: HistoricalRawRevisionStream) -> int: +def classify_historical_full_revisions( + revisions: list[HistoricalRawRevision], +) -> list[HistoricalRevisionDecision]: + """Prove a unique byte-prefix chain; quarantine only the divergent suffix. + + Acquisition time, source path, provider timestamps, and raw-id ordering are + intentionally absent. Equal or divergent payloads do not establish which + capture is newer. + + I4 (polylogue-lb39z): byte-identical captures of one source are one + evidence node observed twice, not competing revisions -- they are + collapsed onto a single representative (the lexicographically smallest + ``raw_id``) before the chain proof runs, and every non-representative + duplicate mirrors its representative's verdict with ``relation= + "duplicate"``. Output order matches input size order (ascending, ties + broken by ``raw_id``), preserving the existing "first is oldest, last is + head" contract relied on by callers such as + ``classify_untyped_full_revision_groups``. + """ + if not revisions: + return [] + groups: dict[bytes, list[HistoricalRawRevision]] = {} + for revision in revisions: + groups.setdefault(revision.payload, []).append(revision) + representative_payload: dict[str, bytes] = {} + members_of: dict[str, list[str]] = {} + for payload, members in groups.items(): + representative = min(members, key=lambda revision: revision.raw_id) + representative_payload[representative.raw_id] = payload + members_of[representative.raw_id] = sorted(member.raw_id for member in members) + + def is_prefix(parent: str, child: str) -> bool: + return representative_payload[child].startswith(representative_payload[parent]) + + sizes = {raw_id: len(payload) for raw_id, payload in representative_payload.items()} + rep_decisions = _classify_deduped_nodes(list(representative_payload), sizes, is_prefix) + return _expand_duplicate_decisions(rep_decisions, members_of, sizes) + + +def _expand_duplicate_decisions( + rep_decisions: dict[str, HistoricalRevisionDecision], + members_of: dict[str, list[str]], + rep_sizes: dict[str, int], +) -> list[HistoricalRevisionDecision]: + size_by_raw_id: dict[str, int] = {} + decision_by_raw_id: dict[str, HistoricalRevisionDecision] = {} + for representative_id, member_ids in members_of.items(): + rep_decision = rep_decisions[representative_id] + decision_by_raw_id[representative_id] = rep_decision + size_by_raw_id[representative_id] = rep_sizes[representative_id] + for member_id in member_ids: + if member_id == representative_id: + continue + decision_by_raw_id[member_id] = HistoricalRevisionDecision( + raw_id=member_id, + authority=rep_decision.authority, + relation="duplicate", + predecessor_raw_id=None, + ) + size_by_raw_id[member_id] = rep_sizes[representative_id] + ordered_ids = sorted(decision_by_raw_id, key=lambda raw_id: (size_by_raw_id[raw_id], raw_id)) + return [decision_by_raw_id[raw_id] for raw_id in ordered_ids] + + +def _stream_size_and_hash(revision: HistoricalRawRevisionStream) -> tuple[int, str]: size = 0 + digest = sha256() with revision.open_payload() as handle: while chunk := handle.read(1024 * 1024): size += len(chunk) - return size + digest.update(chunk) + return size, digest.hexdigest() def _stream_is_prefix( @@ -220,45 +305,42 @@ def _stream_is_prefix( def classify_historical_full_revision_streams( revisions: list[HistoricalRawRevisionStream], ) -> list[HistoricalRevisionDecision]: - """Stream the same unique-prefix proof as the eager byte classifier.""" + """Stream the same dedup-first, localized-ambiguity proof, without eager payloads. + + Each stream is read exactly once to learn its true size and content hash + (never trusting the caller-supplied ``payload_size`` alone); streams whose + hash matches are byte-identical duplicates and are collapsed per I4 before + any prefix comparison runs. Prefix comparisons between distinct-content + representatives are themselves streamed (``_stream_is_prefix``), so no + full payload is ever held in memory at once. + """ if not revisions: return [] - actual_sizes = {revision.raw_id: _stream_size(revision) for revision in revisions} - ordered = sorted(revisions, key=lambda revision: (actual_sizes[revision.raw_id], revision.raw_id)) - if len({actual_sizes[revision.raw_id] for revision in ordered}) != len(ordered): - return [ - HistoricalRevisionDecision( - raw_id=revision.raw_id, authority=RawRevisionAuthority.QUARANTINED, relation="ambiguous" - ) - for revision in revisions - ] - decisions: list[HistoricalRevisionDecision] = [] - previous: HistoricalRawRevisionStream | None = None - for current in ordered: - predecessor = previous.raw_id if previous is not None else None - if previous is not None and not _stream_is_prefix( - previous, - current, - parent_size=actual_sizes[previous.raw_id], - child_size=actual_sizes[current.raw_id], - ): - return [ - HistoricalRevisionDecision( - raw_id=revision.raw_id, authority=RawRevisionAuthority.QUARANTINED, relation="ambiguous" - ) - for revision in revisions - ] - relation: Literal["baseline", "predecessor"] = "baseline" if predecessor is None else "predecessor" - decisions.append( - HistoricalRevisionDecision( - raw_id=current.raw_id, - authority=RawRevisionAuthority.BYTE_PROVEN, - relation=relation, - predecessor_raw_id=predecessor, - ) + size_and_hash = {revision.raw_id: _stream_size_and_hash(revision) for revision in revisions} + by_hash: dict[str, list[HistoricalRawRevisionStream]] = {} + for revision in revisions: + _, digest = size_and_hash[revision.raw_id] + by_hash.setdefault(digest, []).append(revision) + + representative_stream: dict[str, HistoricalRawRevisionStream] = {} + representative_size: dict[str, int] = {} + members_of: dict[str, list[str]] = {} + for members in by_hash.values(): + representative = min(members, key=lambda revision: revision.raw_id) + representative_stream[representative.raw_id] = representative + representative_size[representative.raw_id] = size_and_hash[representative.raw_id][0] + members_of[representative.raw_id] = sorted(member.raw_id for member in members) + + def is_prefix(parent: str, child: str) -> bool: + return _stream_is_prefix( + representative_stream[parent], + representative_stream[child], + parent_size=representative_size[parent], + child_size=representative_size[child], ) - previous = current - return decisions + + rep_decisions = _classify_deduped_nodes(list(representative_stream), representative_size, is_prefix) + return _expand_duplicate_decisions(rep_decisions, members_of, representative_size) __all__ = [ diff --git a/polylogue/storage/sqlite/archive_tiers/revision_governance.py b/polylogue/storage/sqlite/archive_tiers/revision_governance.py index 40085ebdbc..e9a299fa89 100644 --- a/polylogue/storage/sqlite/archive_tiers/revision_governance.py +++ b/polylogue/storage/sqlite/archive_tiers/revision_governance.py @@ -994,7 +994,15 @@ def open_payload(blob_hash: str = blob_hash) -> BinaryIO: HistoricalRawRevisionStream(raw_id=raw_id, payload_size=blob_size, open_payload=open_payload) ) decisions = classify_historical_full_revision_streams(streams) - if not decisions or decisions[0].authority is not RawRevisionAuthority.BYTE_PROVEN: + # A duplicate/predecessor/baseline decision is always BYTE_PROVEN; any + # QUARANTINED entry means the fork-localization in + # classify_historical_full_revision_streams (I4/I5, polylogue-lb39z) + # found a genuine divergence or an unrelated sibling somewhere in this + # cohort. This caller's contract requires the WHOLE cohort to be one + # provable chain (it uses the result to skip *parsing* older members + # outright), so any partial verdict here still falls back to parsing + # every member, exactly as an all-ambiguous verdict always did. + if not decisions or any(decision.authority is not RawRevisionAuthority.BYTE_PROVEN for decision in decisions): continue groups[decisions[-1].raw_id] = tuple(decision.raw_id for decision in decisions[:-1]) return groups diff --git a/tests/unit/storage/test_raw_revision_authority.py b/tests/unit/storage/test_raw_revision_authority.py index fd544906d8..5aab6bf1d3 100644 --- a/tests/unit/storage/test_raw_revision_authority.py +++ b/tests/unit/storage/test_raw_revision_authority.py @@ -30,6 +30,13 @@ from polylogue.storage.sqlite.archive_tiers.source_write import bind_source_raw_revision, write_source_raw_session +def _payload_opener(payload: bytes) -> Callable[[], BinaryIO]: + def open_payload() -> BinaryIO: + return BytesIO(payload) + + return open_payload + + def test_historical_full_classifier_proves_unique_prefix_chain_independent_of_order() -> None: revisions = [ HistoricalRawRevision("middle", b"one\ntwo\n"), @@ -81,11 +88,25 @@ def open_payload() -> BinaryIO: assert {(item.raw_id, item.predecessor_raw_id, item.authority) for item in streamed} == { (item.raw_id, item.predecessor_raw_id, item.authority) for item in eager } - assert len(opened) == 7 # three size passes plus two adjacent prefix comparisons + # polylogue-lb39z (I4/I5): the classifier now hashes every stream once + # (for byte-equal dedup) and compares every smaller/larger pair (not just + # size-adjacent ones) so it can localize a fork instead of nuking the + # whole cohort -- more opens than the old adjacent-only walk, but still + # bounded and never re-reading a stream's full payload more than once per + # comparison it actually participates in. + assert opened.count("oldest") + opened.count("middle") + opened.count("newest") >= 3 + assert set(opened) == {"oldest", "middle", "newest"} -@pytest.mark.parametrize("payloads", [[b"same", b"same"], [b"left", b"right"], [b"root", b"root-left", b"root-right"]]) +@pytest.mark.parametrize("payloads", [[b"left", b"right"]]) def test_historical_classifier_quarantines_unprovable_authority(payloads: list[bytes]) -> None: + """Two same-size, byte-*different* captures share no structural order. + + Unlike a byte-equal duplicate (I4) or a shared-root fork (I5, see + ``test_historical_classifier_localizes_fork_to_divergent_children``), + there is no anchor at all here -- two unrelated roots -- so the whole + cohort stays genuinely unprovable. + """ decisions = classify_historical_full_revisions( [HistoricalRawRevision(f"raw-{index}", payload) for index, payload in enumerate(payloads)] ) @@ -93,7 +114,7 @@ def test_historical_classifier_quarantines_unprovable_authority(payloads: list[b assert {decision.authority for decision in decisions} == {RawRevisionAuthority.QUARANTINED} -@pytest.mark.parametrize("payloads", [[b"same", b"same"], [b"left", b"right"], [b"root", b"root-left", b"root-right"]]) +@pytest.mark.parametrize("payloads", [[b"left", b"right"]]) def test_streamed_historical_classifier_matches_eager_ambiguous_authority(payloads: list[bytes]) -> None: def stream_revision(index: int, payload: bytes) -> HistoricalRawRevisionStream: return HistoricalRawRevisionStream( @@ -114,6 +135,118 @@ def stream_revision(index: int, payload: bytes) -> HistoricalRawRevisionStream: } +def test_historical_classifier_collapses_byte_equal_duplicates_i4() -> None: + """polylogue-lb39z I4: byte-identical captures are one evidence node, not competing revisions. + + Anti-vacuity: under the pre-fix classifier this exact pair (two raws, + identical bytes) classified BOTH as QUARANTINED/ambiguous -- the size-tie + check treated equal payloads as an unprovable fork. Duplicates must + collapse onto one BYTE_PROVEN representative instead. + """ + decisions = classify_historical_full_revisions( + [HistoricalRawRevision("raw-b", b"same-bytes"), HistoricalRawRevision("raw-a", b"same-bytes")] + ) + by_id = {decision.raw_id: decision for decision in decisions} + assert by_id["raw-a"].authority is RawRevisionAuthority.BYTE_PROVEN + assert by_id["raw-a"].relation == "baseline" + assert by_id["raw-b"].authority is RawRevisionAuthority.BYTE_PROVEN + assert by_id["raw-b"].relation == "duplicate" + + +def test_streamed_classifier_collapses_byte_equal_duplicates_i4() -> None: + payloads = {"raw-b": b"same-bytes", "raw-a": b"same-bytes"} + decisions = classify_historical_full_revision_streams( + [ + HistoricalRawRevisionStream(raw_id, len(payload), _payload_opener(payload)) + for raw_id, payload in payloads.items() + ] + ) + by_id = {decision.raw_id: decision for decision in decisions} + assert by_id["raw-a"].authority is RawRevisionAuthority.BYTE_PROVEN + assert by_id["raw-a"].relation == "baseline" + assert by_id["raw-b"].authority is RawRevisionAuthority.BYTE_PROVEN + assert by_id["raw-b"].relation == "duplicate" + + +def test_historical_classifier_duplicate_of_the_fork_tip_stays_quarantined() -> None: + """A byte-equal duplicate of a genuinely-ambiguous revision mirrors its verdict.""" + decisions = classify_historical_full_revisions( + [ + HistoricalRawRevision("root", b"root"), + HistoricalRawRevision("left", b"root-left"), + HistoricalRawRevision("left-dup", b"root-left"), + HistoricalRawRevision("right", b"root-right"), + ] + ) + by_id = {decision.raw_id: decision for decision in decisions} + assert by_id["root"].authority is RawRevisionAuthority.BYTE_PROVEN + assert by_id["root"].relation == "baseline" + assert by_id["left"].authority is RawRevisionAuthority.QUARANTINED + assert by_id["left-dup"].authority is RawRevisionAuthority.QUARANTINED + assert by_id["left-dup"].relation == "duplicate" + assert by_id["right"].authority is RawRevisionAuthority.QUARANTINED + + +def test_historical_classifier_localizes_fork_to_divergent_children_i5() -> None: + """polylogue-lb39z I5: a divergence quarantines only the divergent suffix. + + Anti-vacuity: under the pre-fix classifier this exact 3-member cohort + (one shared root, two mutually-incomparable children) classified ALL + THREE as QUARANTINED -- cohort-atomic quarantine (Step 2 of the report's + causal engine). The fix must keep ``root`` BYTE_PROVEN (it really is an + unambiguous common ancestor of both children) and quarantine only the + fork tip. + """ + decisions = classify_historical_full_revisions( + [ + HistoricalRawRevision("root", b"root"), + HistoricalRawRevision("left", b"root-left"), + HistoricalRawRevision("right", b"root-right"), + ] + ) + by_id = {decision.raw_id: decision for decision in decisions} + assert by_id["root"].authority is RawRevisionAuthority.BYTE_PROVEN + assert by_id["root"].relation == "baseline" + assert by_id["left"].authority is RawRevisionAuthority.QUARANTINED + assert by_id["left"].relation == "ambiguous" + assert by_id["right"].authority is RawRevisionAuthority.QUARANTINED + assert by_id["right"].relation == "ambiguous" + + +def test_streamed_classifier_localizes_fork_to_divergent_children_i5() -> None: + payloads = {"root": b"root", "left": b"root-left", "right": b"root-right"} + decisions = classify_historical_full_revision_streams( + [ + HistoricalRawRevisionStream(raw_id, len(payload), _payload_opener(payload)) + for raw_id, payload in payloads.items() + ] + ) + by_id = {decision.raw_id: decision for decision in decisions} + assert by_id["root"].authority is RawRevisionAuthority.BYTE_PROVEN + assert by_id["left"].authority is RawRevisionAuthority.QUARANTINED + assert by_id["right"].authority is RawRevisionAuthority.QUARANTINED + + +def test_historical_classifier_proves_chain_up_to_a_downstream_fork() -> None: + """A clean root->middle chain stays provable even though *middle* later forks.""" + decisions = classify_historical_full_revisions( + [ + HistoricalRawRevision("root", b"one\n"), + HistoricalRawRevision("middle", b"one\ntwo\n"), + HistoricalRawRevision("fork-a", b"one\ntwo\nthree-a\n"), + HistoricalRawRevision("fork-b", b"one\ntwo\nthree-b\n"), + ] + ) + by_id = {decision.raw_id: decision for decision in decisions} + assert by_id["root"].authority is RawRevisionAuthority.BYTE_PROVEN + assert by_id["root"].relation == "baseline" + assert by_id["middle"].authority is RawRevisionAuthority.BYTE_PROVEN + assert by_id["middle"].relation == "predecessor" + assert by_id["middle"].predecessor_raw_id == "root" + assert by_id["fork-a"].authority is RawRevisionAuthority.QUARANTINED + assert by_id["fork-b"].authority is RawRevisionAuthority.QUARANTINED + + def test_append_envelope_requires_predecessor_revision_and_exact_forward_offsets() -> None: with pytest.raises(ValueError, match="predecessor revision and offsets"): RawRevisionEnvelope("codex:session", RawRevisionKind.APPEND, "rev-2", 2) From 0f73b837da4bb4706649f5eaa7137f7451bcede7 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 2 Aug 2026 18:52:26 +0200 Subject: [PATCH 2/2] fix: eliminate quadratic streamed-blob-read amplification in maximal-parent filter Address CodeRabbit finding on PR #3574: the "maximal" candidate filter recomputed is_prefix for every candidate pair, re-opening and re-reading parent blobs through _stream_is_prefix. Byte-prefix is transitive over a totally-ordered-by-size candidate set (if parent1 and parent2 are both prefixes of the same child and sizes[parent1] < sizes[parent2], parent1 is necessarily a prefix of parent2), so the unique maximal candidate is simply the largest one -- no further is_prefix comparisons needed. This turns what was O(n^3) streamed reads on a linear revision chain back into the O(n^2) the classifier's docstring already claims. Also strengthens a vacuous test assertion (opened.count(...) >= 3, trivially implied by set membership already checked on the next line) into a real upper-bound pin on read amplification (9 opens for the 3-member fixture, verified empirically against the fixed classifier). Deliberately NOT applying CodeRabbit's second suggestion (mirror a duplicate's predecessor_raw_id onto the duplicate decision) -- verified it would introduce a dict-key collision in revision_governance.py's generation walk (children = {predecessor_raw_id: raw_id, ...} keyed by predecessor; giving a duplicate the same predecessor_raw_id as its representative makes both compete for the same dict key, risking silently dropping the real downstream chain from generation numbering). That finding is real but needs a properly tested fix, not a one-line change under merge-train time pressure -- left as a follow-up. Co-Authored-By: Claude --- polylogue/archive/revision_authority.py | 17 +++++++++-------- .../unit/storage/test_raw_revision_authority.py | 6 +++++- 2 files changed, 14 insertions(+), 9 deletions(-) diff --git a/polylogue/archive/revision_authority.py b/polylogue/archive/revision_authority.py index ba9822fb25..52c49f657b 100644 --- a/polylogue/archive/revision_authority.py +++ b/polylogue/archive/revision_authority.py @@ -157,14 +157,15 @@ def _classify_deduped_nodes( for parent in ordered if parent != child and sizes[parent] < sizes[child] and is_prefix(parent, child) ] - maximal = [ - candidate - for candidate in candidates - if not any( - candidate != other and sizes[candidate] < sizes[other] and is_prefix(candidate, other) - for other in candidates - ) - ] + # Prefix is transitive: if parent1 and parent2 are both prefixes of + # child and sizes[parent1] < sizes[parent2], then parent1 is + # necessarily a prefix of parent2 (parent1's bytes equal child's + # first N bytes, which equal parent2's first N bytes since parent2 + # is itself a prefix of child of length >= N). Candidates therefore + # form a totally ordered chain and the unique maximal element is + # simply the largest one -- no further is_prefix comparisons (and + # their streamed blob re-reads) are needed here. + maximal = [max(candidates, key=lambda raw_id: (sizes[raw_id], raw_id))] if candidates else [] parents[child] = maximal for parent in maximal: children[parent].append(child) diff --git a/tests/unit/storage/test_raw_revision_authority.py b/tests/unit/storage/test_raw_revision_authority.py index 5aab6bf1d3..05c9e6dcc2 100644 --- a/tests/unit/storage/test_raw_revision_authority.py +++ b/tests/unit/storage/test_raw_revision_authority.py @@ -94,8 +94,12 @@ def open_payload() -> BinaryIO: # whole cohort -- more opens than the old adjacent-only walk, but still # bounded and never re-reading a stream's full payload more than once per # comparison it actually participates in. - assert opened.count("oldest") + opened.count("middle") + opened.count("newest") >= 3 assert set(opened) == {"oldest", "middle", "newest"} + # Pin the read amplification instead of only lower-bounding it: 3 hashing + # opens plus one is_prefix comparison per smaller/larger candidate pair + # (2 opens each) across this 3-member chain. Tighten this number if the + # comparison strategy changes; do not relax it silently. + assert len(opened) == 9 @pytest.mark.parametrize("payloads", [[b"left", b"right"]])