fix(storage): stream safe raw authority replay

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",
 ]

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]

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,

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",
 ]

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))