diff --git a/polylogue/archive/revision_authority.py b/polylogue/archive/revision_authority.py index b07d8ab3a8..0cdb9766fa 100644 --- a/polylogue/archive/revision_authority.py +++ b/polylogue/archive/revision_authority.py @@ -2,10 +2,11 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass from enum import StrEnum from hashlib import sha256 -from typing import Literal +from typing import BinaryIO, Literal class RawRevisionKind(StrEnum): @@ -66,6 +67,15 @@ class HistoricalRawRevision: payload: bytes +@dataclass(frozen=True) +class HistoricalRawRevisionStream: + """A retained full revision whose bytes can be compared without loading it.""" + + raw_id: str + payload_size: int + open_payload: Callable[[], BinaryIO] + + @dataclass(frozen=True) class HistoricalRevisionDecision: raw_id: str @@ -147,12 +157,86 @@ def classify_historical_full_revisions( return decisions +def _stream_size(revision: HistoricalRawRevisionStream) -> int: + size = 0 + with revision.open_payload() as handle: + while chunk := handle.read(1024 * 1024): + size += len(chunk) + return size + + +def _stream_is_prefix( + parent: HistoricalRawRevisionStream, + child: HistoricalRawRevisionStream, + *, + parent_size: int, + child_size: int, +) -> bool: + """Return whether *parent* is an exact proper byte prefix of *child*.""" + if parent_size >= child_size: + return False + remaining = parent_size + with parent.open_payload() as parent_handle, child.open_payload() as child_handle: + while remaining: + chunk = parent_handle.read(min(1024 * 1024, remaining)) + if not chunk or child_handle.read(len(chunk)) != chunk: + return False + remaining -= len(chunk) + return parent_handle.read(1) == b"" + + +def classify_historical_full_revision_streams( + revisions: list[HistoricalRawRevisionStream], +) -> list[HistoricalRevisionDecision]: + """Stream the same unique-prefix proof as the eager byte classifier.""" + 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, + ) + ) + previous = current + return decisions + + __all__ = [ "HistoricalRawRevision", + "HistoricalRawRevisionStream", "HistoricalRevisionDecision", "RawRevisionAuthority", "RawRevisionEnvelope", "RawRevisionKind", "append_source_revision", "classify_historical_full_revisions", + "classify_historical_full_revision_streams", ] diff --git a/polylogue/sources/live/append_ingest.py b/polylogue/sources/live/append_ingest.py index fde80a7a0a..e45187883a 100644 --- a/polylogue/sources/live/append_ingest.py +++ b/polylogue/sources/live/append_ingest.py @@ -77,6 +77,7 @@ def _ingest_append_plans_archive( t0 = time.perf_counter() from polylogue.sources.decoders import _iter_json_stream from polylogue.sources.dispatch import parse_payload + from polylogue.sources.revision_backfill import parse_retained_raw_sessions from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore _add_timing(timings, "append.imports", t0) @@ -185,15 +186,7 @@ def _ingest_append_plans_archive( continue parsed_by_raw_id: dict[str, Any] = {} for replay_raw_id in replay_plan.accepted_raw_ids: - replay_provider, replay_payload, replay_source_path, _kind = archive.raw_revision_material( - replay_raw_id - ) - replay_sessions = parse_payload( - replay_provider, - list(_iter_json_stream(BytesIO(replay_payload), Path(replay_source_path).name)), - Path(replay_source_path).stem, - source_path=replay_source_path, - ) + replay_sessions = parse_retained_raw_sessions(archive, replay_raw_id) if len(replay_sessions) != 1: raise RuntimeError(f"raw revision {replay_raw_id} did not replay to exactly one session") parsed_by_raw_id[replay_raw_id] = replay_sessions[0] diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 1e9643644a..ca9887176f 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -111,6 +111,7 @@ from polylogue.sources.live.metrics import LiveBatchMetrics, LiveFullIngestAggregate from polylogue.sources.live.sqlite_locking import is_transient_sqlite_lock from polylogue.sources.parsers import hermes_state +from polylogue.sources.revision_backfill import parse_retained_raw_sessions from polylogue.sources.source_acquisition_components import ( ZipEntryReadContext, iter_zip_entry_raw_data, @@ -1945,22 +1946,7 @@ def _apply_membership_sessions( @staticmethod def _parse_retained_raw_sessions(archive: Any, raw_id: str) -> list[Any]: - provider, payload, source_path, _kind = archive.raw_revision_material(raw_id) - source_name = Path(source_path).name - fallback_id = Path(source_path).stem - if is_stream_record_provider(source_path, str(provider)): - return parse_stream_payload( - provider, - _iter_json_stream(BytesIO(payload), source_name), - fallback_id, - source_path=source_path, - ) - return parse_payload( - provider, - list(_iter_json_stream(BytesIO(payload), source_name)), - fallback_id, - source_path=source_path, - ) + return parse_retained_raw_sessions(archive, raw_id) def _extract_zip_member_records( self, diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index c557b7f4a9..f373c5ba23 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -12,6 +12,7 @@ from io import BytesIO from pathlib import Path from types import TracebackType +from typing import BinaryIO from polylogue.archive.revision_authority import ( BYTE_AUTHORITY_CENSUS_DETAIL, @@ -420,8 +421,24 @@ def backfill_historical_revision_evidence( def _parse_retained_raw(archive: ArchiveStore, raw_id: str) -> tuple[list[ParsedSession], int, RawRevisionKind]: - provider, payload, source_path, kind = archive.raw_revision_material(raw_id) - return _parse_one(provider, payload, source_path), len(payload), kind + provider, _blob_hash, source_path, kind, payload_size = archive.raw_revision_descriptor(raw_id) + return parse_retained_raw_sessions(archive, raw_id), payload_size, kind + + +def parse_retained_raw_sessions(archive: ArchiveStore, raw_id: str) -> list[ParsedSession]: + """Parse retained raw evidence without eagerly loading stream records. + + Raw-revision replay is shared by historical repair and the live full and + append routes. Keeping the provider-shape decision here prevents a + seemingly harmless live replay helper from reintroducing ``read_all()`` + for Codex/Claude JSONL evidence. + """ + provider, _blob_hash, source_path, _kind, _payload_size = archive.raw_revision_descriptor(raw_id) + if is_stream_record_provider(source_path, str(provider)): + with archive.open_raw_revision_material(raw_id) as (stream_provider, payload, stream_path, _stream_kind): + return _parse_stream(stream_provider, payload, stream_path) + _provider, eager_payload, _source_path, _eager_kind = archive.raw_revision_material(raw_id) + return _parse_one(provider, eager_payload, source_path) class _ParsedSessionSpill: @@ -505,10 +522,22 @@ def _parse_one(provider: Provider, payload: bytes, source_path: str) -> list[Par ) +def _parse_stream(provider: Provider, payload: BinaryIO, source_path: str) -> list[ParsedSession]: + source_name = Path(source_path).name + fallback_id = Path(source_path).stem + return parse_stream_payload( + provider, + _iter_json_stream(payload, source_name), + fallback_id, + source_path=source_path, + ) + + __all__ = [ "RawRevisionReplayResourceBlockedError", "RevisionBackfillResult", "RevisionCensusResult", "backfill_historical_revision_evidence", "census_historical_revision_evidence", + "parse_retained_raw_sessions", ] diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py index 3a2eac4294..712ac312a9 100644 --- a/polylogue/storage/repair.py +++ b/polylogue/storage/repair.py @@ -5685,8 +5685,6 @@ def repair_raw_materialization( oversized_stream_safe_raw_ids = [ raw_id for raw_id in oversized_candidate_raw_ids if _raw_materialization_stream_safe(candidates, raw_id) ] - # The retained-raw reader materializes bytes before stream parsing, so - # stream-capable format is diagnostic only until that reader is replaced. oversized_raw_ids = oversized_candidate_raw_ids if oversized_raw_ids: metrics["raw_materialization_oversized_count"] = float(len(oversized_raw_ids)) diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 1481cf2a84..fdca79a8eb 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -11,13 +11,13 @@ import math import sqlite3 import time -from collections.abc import Callable, Mapping, Sequence -from contextlib import closing +from collections.abc import Callable, Iterator, Mapping, Sequence +from contextlib import closing, contextmanager from dataclasses import dataclass, field, replace from datetime import UTC, datetime from pathlib import Path from types import TracebackType -from typing import Any, Literal, TypedDict, cast +from typing import Any, BinaryIO, Literal, TypedDict, cast from polylogue.annotations.batch import AnnotationBatch from polylogue.annotations.schema import AnnotationSchema @@ -38,12 +38,12 @@ QueryTextPredicate, ) from polylogue.archive.revision_authority import ( - HistoricalRawRevision, + HistoricalRawRevisionStream, RawRevisionAuthority, RawRevisionEnvelope, RawRevisionKind, append_source_revision, - classify_historical_full_revisions, + classify_historical_full_revision_streams, ) from polylogue.archive.revision_replay import ( ApplicationDecision, @@ -1799,21 +1799,27 @@ def classify_raw_revision_cohort(self, logical_source_key: str) -> RevisionRepla source_conn = self._ensure_source_conn() full_rows = source_conn.execute( """ - SELECT raw_id, lower(hex(blob_hash)) AS blob_hash + SELECT raw_id, lower(hex(blob_hash)) AS blob_hash, blob_size FROM raw_sessions WHERE logical_source_key = ? AND revision_kind = 'full' """, (logical_source_key,), ).fetchall() - historical: list[HistoricalRawRevision] = [] + historical: list[HistoricalRawRevisionStream] = [] for row in full_rows: + + def open_payload(blob_hash: str = str(row[1])) -> BinaryIO: + assert self._blob_publisher is not None + return self._blob_publisher.open(blob_hash) + historical.append( - HistoricalRawRevision( + HistoricalRawRevisionStream( raw_id=str(row[0]), - payload=self._blob_publisher.read_all(str(row[1])), + payload_size=int(row[2]), + open_payload=open_payload, ) ) - decisions = classify_historical_full_revisions(historical) + decisions = classify_historical_full_revision_streams(historical) by_raw_id = {decision.raw_id: decision for decision in decisions} baseline_ids = [decision.raw_id for decision in decisions if decision.relation == "baseline"] baseline_raw_id = baseline_ids[0] if len(baseline_ids) == 1 else None @@ -1970,11 +1976,11 @@ def _authorize_full_snapshot_fold( or accepted_head.append_end_offset != frontier ): return None - _provider, full_payload, _path, _kind = self.raw_revision_material(full_candidate.raw_id) - if len(full_payload) != frontier: + _full_digest, full_size = self._raw_revision_payload_digest_and_size(full_candidate.raw_id) + if full_size != frontier: return None - tail_payloads: list[bytes] = [] + tail_raw_ids: list[str] = [] current = accepted_head baseline_raw_id = current.baseline_raw_id expected_end = frontier @@ -1991,17 +1997,16 @@ def _authorize_full_snapshot_fold( ): return None visited.add(current.raw_id) - _provider, tail_payload, _path, _kind = self.raw_revision_material(current.raw_id) + tail_digest, tail_size = self._raw_revision_payload_digest_and_size(current.raw_id) assert current.append_end_offset is not None assert current.append_start_offset is not None - if len(tail_payload) != current.append_end_offset - current.append_start_offset: + if tail_size != current.append_end_offset - current.append_start_offset: return None predecessor = candidates.get(current.predecessor_raw_id) if ( predecessor is None or predecessor.source_revision != current.predecessor_source_revision - or current.source_revision - != append_source_revision(predecessor.source_revision, hashlib.sha256(tail_payload).hexdigest()) + or current.source_revision != append_source_revision(predecessor.source_revision, tail_digest) ): return None predecessor_end = ( @@ -2009,7 +2014,7 @@ def _authorize_full_snapshot_fold( ) if predecessor_end != current.append_start_offset: return None - tail_payloads.append(tail_payload) + tail_raw_ids.append(current.raw_id) expected_end = current.append_start_offset current = predecessor if ( @@ -2019,10 +2024,10 @@ def _authorize_full_snapshot_fold( or current.blob_size != expected_end ): return None - _provider, baseline_payload, _path, _kind = self.raw_revision_material(current.raw_id) - if ( - len(baseline_payload) != current.blob_size - or baseline_payload + b"".join(reversed(tail_payloads)) != full_payload + _baseline_digest, baseline_size = self._raw_revision_payload_digest_and_size(current.raw_id) + if baseline_size != current.blob_size or not self._raw_revision_matches_segments( + full_candidate.raw_id, + [current.raw_id, *reversed(tail_raw_ids)], ): return None return FullSnapshotFoldAuthorization( @@ -2036,15 +2041,15 @@ def _authorize_full_snapshot_fold( full_source_revision=full_candidate.source_revision, ) - def raw_revision_material(self, raw_id: str) -> tuple[Provider, bytes, str, RawRevisionKind]: - """Read one retained revision with its parsing identity.""" + def raw_revision_descriptor(self, raw_id: str) -> tuple[Provider, str, str, RawRevisionKind, int]: + """Return one retained revision's identity without materializing its blob.""" if self._blob_publisher is None: raise RuntimeError("raw revision replay requires a writable blob publisher") row = ( self._ensure_source_conn() .execute( """ - SELECT origin, capture_mode, lower(hex(blob_hash)), source_path, revision_kind + SELECT origin, capture_mode, lower(hex(blob_hash)), source_path, revision_kind, blob_size FROM raw_sessions WHERE raw_id = ? """, (raw_id,), @@ -2055,11 +2060,47 @@ def raw_revision_material(self, raw_id: str) -> tuple[Provider, bytes, str, RawR raise KeyError(raw_id) return ( provider_from_origin(Origin.from_string(str(row[0])), family_hint=row[1]), - self._blob_publisher.read_all(str(row[2])), + str(row[2]), str(row[3]), RawRevisionKind(str(row[4])), + int(row[5]), ) + @contextmanager + def open_raw_revision_material(self, raw_id: str) -> Iterator[tuple[Provider, BinaryIO, str, RawRevisionKind]]: + """Open a retained revision for bounded streaming consumption.""" + provider, blob_hash, source_path, kind, _blob_size = self.raw_revision_descriptor(raw_id) + assert self._blob_publisher is not None + with self._blob_publisher.open(blob_hash) as payload: + yield provider, payload, source_path, kind + + def raw_revision_material(self, raw_id: str) -> tuple[Provider, bytes, str, RawRevisionKind]: + """Read one retained revision with its parsing identity. + + Use ``open_raw_revision_material`` for potentially large blobs. + """ + provider, blob_hash, source_path, kind, _blob_size = self.raw_revision_descriptor(raw_id) + assert self._blob_publisher is not None + return provider, self._blob_publisher.read_all(blob_hash), source_path, kind + + def _raw_revision_payload_digest_and_size(self, raw_id: str) -> tuple[str, int]: + digest = hashlib.sha256() + size = 0 + with self.open_raw_revision_material(raw_id) as (_provider, payload, _source_path, _kind): + while chunk := payload.read(1024 * 1024): + digest.update(chunk) + size += len(chunk) + return digest.hexdigest(), size + + def _raw_revision_matches_segments(self, full_raw_id: str, segment_raw_ids: Sequence[str]) -> bool: + with self.open_raw_revision_material(full_raw_id) as (_provider, full, _source_path, _kind): + for raw_id in segment_raw_ids: + with self.open_raw_revision_material(raw_id) as (_provider, segment, _source_path, _kind): + while chunk := segment.read(1024 * 1024): + if full.read(len(chunk)) != chunk: + return False + return full.read(1) == b"" + def unclassified_raw_revision_rows(self) -> tuple[tuple[str, int], ...]: """Return legacy rows that have no durable logical revision identity.""" rows = ( @@ -2582,7 +2623,7 @@ def apply_raw_revision_replay( attachments_by_raw_id: dict[str, dict[int, tuple[bytes | None, int, str]]] = {} attachment_refs_by_raw_id: dict[str, tuple[ArchiveSourceBlobRef, ...]] = {} for raw_id in plan.accepted_raw_ids: - _provider, _payload, source_path, _kind = self.raw_revision_material(raw_id) + _provider, _blob_hash, source_path, _kind, _blob_size = self.raw_revision_descriptor(raw_id) acquired, refs = self._preacquire_attachment_blobs( parsed_by_raw_id[raw_id], source_path=source_path, @@ -2699,7 +2740,7 @@ def apply_raw_revision_replay( } } for raw_id in terminal_raw_ids: - provider, _payload, _source_path, _kind = self.raw_revision_material(raw_id) + provider, _blob_hash, _source_path, _kind, _blob_size = self.raw_revision_descriptor(raw_id) self.mark_raw_parse_succeeded(raw_id, provider=provider) return session_id, plan.accepted_raw_ids @@ -2728,7 +2769,7 @@ def apply_raw_membership_classification( if classification.accepted_raw_ids: accepted_raw_id = classification.accepted_raw_ids[-1] accepted_session = parsed_by_raw_id[accepted_raw_id] - _provider, _payload, source_path, _kind = self.raw_revision_material(accepted_raw_id) + _provider, _blob_hash, source_path, _kind, _blob_size = self.raw_revision_descriptor(accepted_raw_id) attachments, refs = self._preacquire_attachment_blobs( accepted_session, source_path=source_path, @@ -2873,7 +2914,7 @@ def apply_raw_membership_classification( (raw_id,), ).fetchone() if complete is not None and bool(complete[0]): - provider, _payload, _source_path, _kind = self.raw_revision_material(raw_id) + provider, _blob_hash, _source_path, _kind, _blob_size = self.raw_revision_descriptor(raw_id) self.mark_raw_parse_succeeded(raw_id, provider=provider) else: with conn: diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index f266b2ae87..cde9bb3c53 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -141,6 +141,66 @@ def _seed_live_append_plan( return path, plan, _append_owner(archive_root) +def test_live_append_replay_streams_retained_jsonl_raw( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Append replay must not resurrect eager blob materialization.""" + from polylogue.storage.blob_publication import ArchiveBlobPublisher + + _path, plan, owner = _seed_live_append_plan(tmp_path, native_id="streamed-append") + monkeypatch.setattr( + ArchiveBlobPublisher, + "read_all", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("append replay eagerly read raw blob")), + ) + + result = ingest_append_plans(cast(Any, owner), [plan]) + + assert result.succeeded == [plan] + assert result.failed == [] + + +def test_live_full_replay_streams_retained_jsonl_raw( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Full replay must stream its older retained JSONL snapshot.""" + from polylogue.storage.blob_publication import ArchiveBlobPublisher + + root = tmp_path / "sessions" + root.mkdir() + path = root / "streamed-full.jsonl" + path.write_bytes( + b'{"type":"session_meta","payload":{"id":"streamed-full"}}\n' + b'{"type":"response_item","payload":{"type":"message","id":"message-0","role":"user",' + b'"content":[{"type":"input_text","text":"zero"}]}}\n' + ) + index_db = tmp_path / "index.db" + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="codex", root=root),), + cursor=CursorStore(index_db), + parser_fingerprint="test-parser", + ) + assert processor._ingest_full_paths_sync([path], source_name="codex").succeeded == [path] + with path.open("ab") as handle: + handle.write( + b'{"type":"response_item","payload":{"type":"message","id":"message-1",' + b'"role":"assistant","content":[{"type":"output_text","text":"one"}]}}\n' + ) + monkeypatch.setattr( + ArchiveBlobPublisher, + "read_all", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("full replay eagerly read raw blob")), + ) + + result = processor._ingest_full_paths_sync([path], source_name="codex") + + assert result.succeeded == [path] + assert result.failed == [] + + def test_full_ingest_heartbeats_small_file_groups_with_current_path( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/sources/test_revision_backfill.py b/tests/unit/sources/test_revision_backfill.py index f8036a9497..6b1fcbf384 100644 --- a/tests/unit/sources/test_revision_backfill.py +++ b/tests/unit/sources/test_revision_backfill.py @@ -13,6 +13,7 @@ from polylogue.sources.dispatch import parse_payload from polylogue.sources.parsers.base import ParsedSession from polylogue.sources.revision_backfill import _parse_one, backfill_historical_revision_evidence +from polylogue.storage.blob_publication import ArchiveBlobPublisher from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root @@ -68,6 +69,30 @@ def test_revision_reparse_preserves_beads_workspace_identity(tmp_path: Path) -> assert sessions[0].working_directories == [str(source_path.parent.parent.resolve())] +def test_historical_backfill_streams_codex_raw_without_eager_blob_read( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + initialize_active_archive_root(tmp_path) + payload = b'{"type":"session_meta","payload":{"id":"streamed"}}\n' + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + archive.write_raw_payload( + provider=Provider.CODEX, + payload=payload, + source_path="streamed.jsonl", + acquired_at_ms=1, + ) + monkeypatch.setattr( + ArchiveBlobPublisher, + "read_all", + lambda *_args, **_kwargs: pytest.fail("stream-safe revision replay must not eagerly read a blob"), + ) + + result = backfill_historical_revision_evidence(tmp_path) + + assert result.scanned == 1 + assert result.replayed_logical_sources == 1 + + def test_historical_backfill_selects_prefix_newest_independent_of_acquisition_order(tmp_path: Path) -> None: initialize_active_archive_root(tmp_path) baseline = ( diff --git a/tests/unit/storage/test_raw_revision_authority.py b/tests/unit/storage/test_raw_revision_authority.py index 82ccf5f12e..d8c24c83fd 100644 --- a/tests/unit/storage/test_raw_revision_authority.py +++ b/tests/unit/storage/test_raw_revision_authority.py @@ -1,19 +1,23 @@ from __future__ import annotations import sqlite3 +from collections.abc import Callable from hashlib import sha256 +from io import BytesIO from pathlib import Path from types import SimpleNamespace -from typing import Any, cast +from typing import Any, BinaryIO, cast import pytest from polylogue.archive.revision_authority import ( HistoricalRawRevision, + HistoricalRawRevisionStream, RawRevisionAuthority, RawRevisionEnvelope, RawRevisionKind, append_source_revision, + classify_historical_full_revision_streams, classify_historical_full_revisions, ) from polylogue.core.enums import Origin, Provider @@ -50,6 +54,36 @@ def normalized(items: list[HistoricalRawRevision]) -> set[tuple[str, str | None, ) +def test_streamed_historical_full_classifier_matches_byte_proof_without_eager_payloads() -> None: + payloads = { + "oldest": b"one\n", + "middle": b"one\ntwo\n", + "newest": b"one\ntwo\nthree\n", + } + opened: list[str] = [] + + def opener(raw_id: str) -> Callable[[], BinaryIO]: + def open_payload() -> BinaryIO: + opened.append(raw_id) + from io import BytesIO + + return BytesIO(payloads[raw_id]) + + return open_payload + + streamed = classify_historical_full_revision_streams( + [HistoricalRawRevisionStream(raw_id, len(payload), opener(raw_id)) for raw_id, payload in payloads.items()] + ) + eager = classify_historical_full_revisions( + [HistoricalRawRevision(raw_id, payload) for raw_id, payload in payloads.items()] + ) + + 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 + + @pytest.mark.parametrize("payloads", [[b"same", b"same"], [b"left", b"right"], [b"root", b"root-left", b"root-right"]]) def test_historical_classifier_quarantines_unprovable_authority(payloads: list[bytes]) -> None: decisions = classify_historical_full_revisions( @@ -59,6 +93,27 @@ 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"]]) +def test_streamed_historical_classifier_matches_eager_ambiguous_authority(payloads: list[bytes]) -> None: + def stream_revision(index: int, payload: bytes) -> HistoricalRawRevisionStream: + return HistoricalRawRevisionStream( + raw_id=f"raw-{index}", + payload_size=len(payload), + open_payload=lambda: BytesIO(payload), + ) + + eager = classify_historical_full_revisions( + [HistoricalRawRevision(f"raw-{index}", payload) for index, payload in enumerate(payloads)] + ) + streamed = classify_historical_full_revision_streams( + [stream_revision(index, payload) for index, payload in enumerate(payloads)] + ) + + 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 + } + + 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) diff --git a/tests/unit/storage/test_repair.py b/tests/unit/storage/test_repair.py index cdb07582b0..2b57b21dd8 100644 --- a/tests/unit/storage/test_repair.py +++ b/tests/unit/storage/test_repair.py @@ -13,6 +13,7 @@ from polylogue.maintenance.models import DerivedModelStatus from polylogue.sources.revision_backfill import census_historical_revision_evidence from polylogue.storage import repair as repair_mod +from polylogue.storage.blob_publication import ArchiveBlobPublisher from polylogue.storage.blob_store import BlobStore from polylogue.storage.insights.session.repair_assessment import assess_session_insight_repairs from polylogue.storage.insights.session.runtime import SessionInsightCounts, SessionInsightStatusSnapshot @@ -1677,48 +1678,19 @@ def test_raw_materialization_classifies_oversized_stream_record_replay( conn.execute("UPDATE raw_sessions SET blob_size = ? WHERE raw_id = ?", (oversized, raw_id)) conn.commit() census_historical_revision_evidence(tmp_path, selected_raw_ids=[raw_id]) - config = _config(tmp_path) - calls: dict[str, object] = {} - - class FakeBackend: - def __init__(self, *, db_path: Path) -> None: - calls["db_path"] = db_path - - class FakeRepository: - def __init__(self, *, backend: FakeBackend, archive_root: Path) -> None: - calls["archive_root"] = archive_root - - async def close(self) -> None: - calls["closed"] = True - - class FakeParseResult: - processed_ids = {"session-1"} - parse_failures = 0 - - class FakeParsingService: - def __init__(self, **_kwargs: object) -> None: - pass - - async def parse_from_raw(self, **kwargs: object) -> FakeParseResult: - calls["parse_kwargs"] = kwargs - return FakeParseResult() - - monkeypatch.setattr("polylogue.pipeline.services.parsing.ParsingService", FakeParsingService) - monkeypatch.setattr("polylogue.storage.repository.SessionRepository", FakeRepository) - monkeypatch.setattr("polylogue.storage.sqlite.async_sqlite.SQLiteBackend", FakeBackend) monkeypatch.setattr( - "polylogue.sources.revision_backfill.backfill_historical_revision_evidence", - lambda *_args, **_kwargs: pytest.fail("oversized stream raw must be blocked before backfill"), + ArchiveBlobPublisher, + "read_all", + lambda *_args, **_kwargs: pytest.fail("stream-safe oversized replay must not eagerly read a blob"), ) - result = repair_mod.repair_raw_materialization(config, dry_run=False) + result = repair_mod.repair_raw_materialization(_config(tmp_path), dry_run=False) assert result.success is False assert result.repaired_count == 0 assert result.metrics["raw_materialization_stream_oversized_count"] == 1.0 + assert result.metrics["raw_materialization_resource_blocked_count"] == 1.0 assert "1 replay candidate(s) remain" in result.detail - assert "parse_kwargs" not in calls - assert "closed" not in calls def test_raw_materialization_blocks_oversized_expanded_cohort_before_blob_open( @@ -1741,10 +1713,10 @@ def test_raw_materialization_blocks_oversized_expanded_cohort_before_blob_open( ) with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: small_raw = archive.write_raw_payload( - provider=Provider.CODEX, payload=baseline, source_path="expanded.jsonl", acquired_at_ms=1 + provider=Provider.CODEX, payload=baseline, source_path="expanded.json", acquired_at_ms=1 ) oversized_raw = archive.write_raw_payload( - provider=Provider.CODEX, payload=newest, source_path="expanded.jsonl", acquired_at_ms=2 + provider=Provider.CODEX, payload=newest, source_path="expanded.json", acquired_at_ms=2 ) with sqlite3.connect(tmp_path / "source.db") as source_conn: source_conn.execute( @@ -1783,10 +1755,10 @@ def test_raw_materialization_backlog_expands_to_oversized_materialized_sibling( large_payload = b'{"type":"session_meta","payload":{"id":"large-done"}}\n' with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: small_raw = archive.write_raw_payload( - provider=Provider.CODEX, payload=small_payload, source_path="shared.jsonl", acquired_at_ms=1 + provider=Provider.CODEX, payload=small_payload, source_path="shared.json", acquired_at_ms=1 ) large_raw = archive.write_raw_payload( - provider=Provider.CODEX, payload=large_payload, source_path="shared.jsonl", acquired_at_ms=2 + provider=Provider.CODEX, payload=large_payload, source_path="shared.json", acquired_at_ms=2 ) with sqlite3.connect(tmp_path / "source.db") as source_conn: source_conn.execute( @@ -1831,7 +1803,7 @@ def test_raw_materialization_blocks_aggregate_sub_limit_cohort_before_blob_open( archive.write_raw_payload( provider=Provider.CODEX, payload=f'{{"type":"session_meta","payload":{{"id":"aggregate-{index}"}}}}\n'.encode(), - source_path="aggregate.jsonl", + source_path="aggregate.json", acquired_at_ms=index, ) for index in range(2) @@ -1910,7 +1882,9 @@ def test_raw_materialization_processes_independent_components_across_bounded_pas assert repair_mod.repair_raw_materialization(config, raw_artifact_limit=5).success is True -def test_raw_materialization_durable_ledger_survives_ops_reset_for_fairness(tmp_path: Path) -> None: +def test_raw_materialization_durable_ledger_survives_ops_reset_for_fairness( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: """A retryable oldest component must not monopolize a slot after ops reset.""" from polylogue.core.enums import Provider from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore @@ -1938,6 +1912,13 @@ def test_raw_materialization_durable_ledger_survives_ops_reset_for_fairness(tmp_ conn.commit() census_historical_revision_evidence(tmp_path, selected_raw_ids=raw_ids) + original_stream_safe = repair_mod._raw_materialization_stream_safe + monkeypatch.setattr( + repair_mod, + "_raw_materialization_stream_safe", + lambda candidates, raw_id: raw_id != raw_ids[0] and original_stream_safe(candidates, raw_id), + ) + first = repair_mod.repair_raw_materialization(_config(tmp_path), raw_artifact_limit=1) assert first.plan_outcomes[0].status.value == "retryable" (tmp_path / "ops.db").unlink()