From 6424723e75840404d5c8b1573f77bf1bb7eb92d2 Mon Sep 17 00:00:00 2001
From: Sinity <ezo.dev@gmail.com>
Date: Fri, 17 Jul 2026 04:29:42 +0200
Subject: [PATCH 1/3] fix(storage): stream safe raw authority replay

Problem:
Raw replay classified and parsed retained JSONL through read_all(), so a
single oversized but otherwise safe authority component could never reach
the bounded executor.

What changed:
Add chunked full-snapshot prefix proof and retained-raw stream access.
Stream-record providers now parse directly from blob handles, and admission
permits an oversized component only when every member is on that route.
Full/append fold verification and metadata-only application steps no longer
materialize raw blobs. Other providers remain fail-closed behind the existing
resource envelope.

Verification:
devtools test tests/unit/storage/test_raw_revision_authority.py \
  tests/unit/sources/test_revision_backfill.py tests/unit/storage/test_repair.py
79 passed.
devtools verify --quick: ruff, mypy, generated surfaces, layering, schema
roundtrip, manifests, and CI workflow checks passed.

Ref polylogue-hjpx.2.
---
 polylogue/archive/revision_authority.py       | 105 +++++++++++++++++-
 polylogue/sources/revision_backfill.py        |  28 ++++-
 polylogue/storage/repair.py                   |   7 +-
 .../storage/sqlite/archive_tiers/archive.py   | 101 ++++++++++++-----
 tests/unit/sources/test_revision_backfill.py  |  25 +++++
 .../storage/test_raw_revision_authority.py    |  35 +++++-
 tests/unit/storage/test_repair.py             |  66 ++++-------
 7 files changed, 286 insertions(+), 81 deletions(-)

diff --git a/polylogue/archive/revision_authority.py b/polylogue/archive/revision_authority.py
index b07d8ab3a8..53096ea1ff 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,105 @@ 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 []
+    parents: dict[str, list[str]] = {}
+    children: dict[str, list[str]] = {revision.raw_id: [] for revision in revisions}
+    actual_sizes = {revision.raw_id: _stream_size(revision) for revision in revisions}
+    prefix_pairs = {
+        (parent.raw_id, child.raw_id)
+        for child in revisions
+        for parent in revisions
+        if parent.raw_id != child.raw_id
+        and _stream_is_prefix(
+            parent,
+            child,
+            parent_size=actual_sizes[parent.raw_id],
+            child_size=actual_sizes[child.raw_id],
+        )
+    }
+    for child in revisions:
+        candidates = [parent.raw_id for parent in revisions if (parent.raw_id, child.raw_id) in prefix_pairs]
+        maximal = [
+            candidate
+            for candidate in candidates
+            if not any(candidate != other and (candidate, other) in prefix_pairs for other in candidates)
+        ]
+        parents[child.raw_id] = 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"
+            )
+            for revision in revisions
+        ]
+    decisions: list[HistoricalRevisionDecision] = []
+    current: str | None = roots[0]
+    while current is not None:
+        predecessor = parents[current][0] if parents[current] else None
+        relation: Literal["baseline", "predecessor"] = "baseline" if predecessor is None else "predecessor"
+        decisions.append(
+            HistoricalRevisionDecision(
+                raw_id=current,
+                authority=RawRevisionAuthority.BYTE_PROVEN,
+                relation=relation,
+                predecessor_raw_id=predecessor,
+            )
+        )
+        current = children[current][0] if children[current] else None
+    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/revision_backfill.py b/polylogue/sources/revision_backfill.py
index c557b7f4a9..7c8a4d5765 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,
@@ -182,7 +183,8 @@ def _census_historical_revision_evidence(
                 payload_sizes = archive.raw_payload_sizes([raw_id for raw_id, _index in rows])
                 total_payload_bytes = sum(payload_sizes.values())
                 oversized = [raw_id for raw_id, size in payload_sizes.items() if size > max_payload_bytes]
-                if oversized or total_payload_bytes > max_payload_bytes:
+                stream_safe = all(_retained_raw_is_stream_safe(archive, raw_id) for raw_id in payload_sizes)
+                if (oversized or total_payload_bytes > max_payload_bytes) and not stream_safe:
                     blocked_ids = oversized or list(payload_sizes)
                     raise RawRevisionReplayResourceBlockedError(
                         sorted(blocked_ids), max_payload_bytes, total_payload_bytes
@@ -420,8 +422,12 @@ 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)
+    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), payload_size, stream_kind
+    _provider, eager_payload, _source_path, _kind = archive.raw_revision_material(raw_id)
+    return _parse_one(provider, eager_payload, source_path), len(eager_payload), kind
 
 
 class _ParsedSessionSpill:
@@ -505,6 +511,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,
+    )
+
+
+def _retained_raw_is_stream_safe(archive: ArchiveStore, raw_id: str) -> bool:
+    provider, _blob_hash, source_path, _kind, _payload_size = archive.raw_revision_descriptor(raw_id)
+    return is_stream_record_provider(source_path, str(provider))
+
+
 __all__ = [
     "RawRevisionReplayResourceBlockedError",
     "RevisionBackfillResult",
diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py
index 3a2eac4294..8448dc3d39 100644
--- a/polylogue/storage/repair.py
+++ b/polylogue/storage/repair.py
@@ -5625,6 +5625,7 @@ def repair_raw_materialization(
         for component in ordered_components
         if sum(_raw_materialization_component_blob_bytes(candidates, member) for member in component)
         > RAW_MATERIALIZATION_EXECUTE_BLOB_LIMIT_BYTES
+        and not all(_raw_materialization_stream_safe(candidates, member) for member in component)
     ]
     all_blocked_component_raw_ids = {raw_id for component in all_blocked_components for raw_id in component}
     selected_components = (
@@ -5685,9 +5686,9 @@ 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
+    oversized_raw_ids = [
+        raw_id for raw_id in oversized_candidate_raw_ids if not _raw_materialization_stream_safe(candidates, raw_id)
+    ]
     if oversized_raw_ids:
         metrics["raw_materialization_oversized_count"] = float(len(oversized_raw_ids))
         metrics["raw_materialization_resource_blocked_count"] = max(
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_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..83593d95d6 100644
--- a/tests/unit/storage/test_raw_revision_authority.py
+++ b/tests/unit/storage/test_raw_revision_authority.py
@@ -1,19 +1,22 @@
 from __future__ import annotations
 
 import sqlite3
+from collections.abc import Callable
 from hashlib import sha256
 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 +53,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 opened
+
+
 @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(
diff --git a/tests/unit/storage/test_repair.py b/tests/unit/storage/test_repair.py
index cdb07582b0..65399a54e5 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,18 @@ 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.success is True
+    assert result.repaired_count == 1
     assert result.metrics["raw_materialization_stream_oversized_count"] == 1.0
-    assert "1 replay candidate(s) remain" in result.detail
-    assert "parse_kwargs" not in calls
-    assert "closed" not in calls
+    assert result.metrics.get("raw_materialization_resource_blocked_count", 0.0) == 0.0
 
 
 def test_raw_materialization_blocks_oversized_expanded_cohort_before_blob_open(
@@ -1741,10 +1712,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 +1754,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 +1802,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 +1881,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 +1911,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()

From 8be68f9df2fd6826e11f05bd1f644b1e81182bbb Mon Sep 17 00:00:00 2001
From: Sinity <ezo.dev@gmail.com>
Date: Fri, 17 Jul 2026 04:35:48 +0200
Subject: [PATCH 2/3] fix(storage): retain raw replay resource admission

Problem:
Streaming retained bytes does not bound the parser's complete ParsedSession
construction. Admitting arbitrary multi-GiB record streams could still exceed
the daemon resource envelope before the spill cache is reached.

What changed:
Retain the existing component-size admission gate while preserving streamed
parsing for admitted JSONL revisions. Expand streamed/eager authority tests
across all ambiguous prefix cohorts.

Verification:
devtools test tests/unit/storage/test_raw_revision_authority.py \
  tests/unit/sources/test_revision_backfill.py tests/unit/storage/test_repair.py
82 passed.

Ref polylogue-hjpx.2.
---
 polylogue/sources/revision_backfill.py        |  8 +------
 polylogue/storage/repair.py                   |  5 +----
 .../storage/test_raw_revision_authority.py    | 22 +++++++++++++++++++
 tests/unit/storage/test_repair.py             |  7 +++---
 4 files changed, 28 insertions(+), 14 deletions(-)

diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py
index 7c8a4d5765..79cfd4636f 100644
--- a/polylogue/sources/revision_backfill.py
+++ b/polylogue/sources/revision_backfill.py
@@ -183,8 +183,7 @@ def _census_historical_revision_evidence(
                 payload_sizes = archive.raw_payload_sizes([raw_id for raw_id, _index in rows])
                 total_payload_bytes = sum(payload_sizes.values())
                 oversized = [raw_id for raw_id, size in payload_sizes.items() if size > max_payload_bytes]
-                stream_safe = all(_retained_raw_is_stream_safe(archive, raw_id) for raw_id in payload_sizes)
-                if (oversized or total_payload_bytes > max_payload_bytes) and not stream_safe:
+                if oversized or total_payload_bytes > max_payload_bytes:
                     blocked_ids = oversized or list(payload_sizes)
                     raise RawRevisionReplayResourceBlockedError(
                         sorted(blocked_ids), max_payload_bytes, total_payload_bytes
@@ -522,11 +521,6 @@ def _parse_stream(provider: Provider, payload: BinaryIO, source_path: str) -> li
     )
 
 
-def _retained_raw_is_stream_safe(archive: ArchiveStore, raw_id: str) -> bool:
-    provider, _blob_hash, source_path, _kind, _payload_size = archive.raw_revision_descriptor(raw_id)
-    return is_stream_record_provider(source_path, str(provider))
-
-
 __all__ = [
     "RawRevisionReplayResourceBlockedError",
     "RevisionBackfillResult",
diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py
index 8448dc3d39..712ac312a9 100644
--- a/polylogue/storage/repair.py
+++ b/polylogue/storage/repair.py
@@ -5625,7 +5625,6 @@ def repair_raw_materialization(
         for component in ordered_components
         if sum(_raw_materialization_component_blob_bytes(candidates, member) for member in component)
         > RAW_MATERIALIZATION_EXECUTE_BLOB_LIMIT_BYTES
-        and not all(_raw_materialization_stream_safe(candidates, member) for member in component)
     ]
     all_blocked_component_raw_ids = {raw_id for component in all_blocked_components for raw_id in component}
     selected_components = (
@@ -5686,9 +5685,7 @@ 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)
     ]
-    oversized_raw_ids = [
-        raw_id for raw_id in oversized_candidate_raw_ids if not _raw_materialization_stream_safe(candidates, raw_id)
-    ]
+    oversized_raw_ids = oversized_candidate_raw_ids
     if oversized_raw_ids:
         metrics["raw_materialization_oversized_count"] = float(len(oversized_raw_ids))
         metrics["raw_materialization_resource_blocked_count"] = max(
diff --git a/tests/unit/storage/test_raw_revision_authority.py b/tests/unit/storage/test_raw_revision_authority.py
index 83593d95d6..21a25cc780 100644
--- a/tests/unit/storage/test_raw_revision_authority.py
+++ b/tests/unit/storage/test_raw_revision_authority.py
@@ -3,6 +3,7 @@
 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, BinaryIO, cast
@@ -92,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 65399a54e5..2b57b21dd8 100644
--- a/tests/unit/storage/test_repair.py
+++ b/tests/unit/storage/test_repair.py
@@ -1686,10 +1686,11 @@ def test_raw_materialization_classifies_oversized_stream_record_replay(
 
     result = repair_mod.repair_raw_materialization(_config(tmp_path), dry_run=False)
 
-    assert result.success is True
-    assert result.repaired_count == 1
+    assert result.success is False
+    assert result.repaired_count == 0
     assert result.metrics["raw_materialization_stream_oversized_count"] == 1.0
-    assert result.metrics.get("raw_materialization_resource_blocked_count", 0.0) == 0.0
+    assert result.metrics["raw_materialization_resource_blocked_count"] == 1.0
+    assert "1 replay candidate(s) remain" in result.detail
 
 
 def test_raw_materialization_blocks_oversized_expanded_cohort_before_blob_open(

From 484bcbc6827a93d79562b120bc20ee954a365a22 Mon Sep 17 00:00:00 2001
From: Sinity <ezo.dev@gmail.com>
Date: Fri, 17 Jul 2026 04:43:35 +0200
Subject: [PATCH 3/3] fix(storage): stream all retained raw replay paths

Problem: live append and full replay still reopened retained JSONL raw blobs through the eager materialization helper, bypassing the streaming authority path. The streamed classifier also compared every pair of historical revisions.

What changed: centralize retained-raw parsing in the source layer; JSONL records now use the blob stream in historical, append, and full replay. Reduce the prefix-chain proof to sorted adjacent comparisons, which is equivalent for a unique strict-prefix chain.

Verification: devtools test tests/unit/storage/test_raw_revision_authority.py tests/unit/sources/test_revision_backfill.py tests/unit/storage/test_repair.py tests/unit/sources/test_live_batch_support.py (150 passed); devtools verify --quick.
---
 polylogue/archive/revision_authority.py       | 57 ++++++------------
 polylogue/sources/live/append_ingest.py       | 11 +---
 polylogue/sources/live/batch.py               | 18 +-----
 polylogue/sources/revision_backfill.py        | 21 +++++--
 tests/unit/sources/test_live_batch_support.py | 60 +++++++++++++++++++
 .../storage/test_raw_revision_authority.py    |  2 +-
 6 files changed, 101 insertions(+), 68 deletions(-)

diff --git a/polylogue/archive/revision_authority.py b/polylogue/archive/revision_authority.py
index 53096ea1ff..0cdb9766fa 100644
--- a/polylogue/archive/revision_authority.py
+++ b/polylogue/archive/revision_authority.py
@@ -191,40 +191,9 @@ def classify_historical_full_revision_streams(
     """Stream the same unique-prefix proof as the eager byte classifier."""
     if not revisions:
         return []
-    parents: dict[str, list[str]] = {}
-    children: dict[str, list[str]] = {revision.raw_id: [] for revision in revisions}
     actual_sizes = {revision.raw_id: _stream_size(revision) for revision in revisions}
-    prefix_pairs = {
-        (parent.raw_id, child.raw_id)
-        for child in revisions
-        for parent in revisions
-        if parent.raw_id != child.raw_id
-        and _stream_is_prefix(
-            parent,
-            child,
-            parent_size=actual_sizes[parent.raw_id],
-            child_size=actual_sizes[child.raw_id],
-        )
-    }
-    for child in revisions:
-        candidates = [parent.raw_id for parent in revisions if (parent.raw_id, child.raw_id) in prefix_pairs]
-        maximal = [
-            candidate
-            for candidate in candidates
-            if not any(candidate != other and (candidate, other) in prefix_pairs for other in candidates)
-        ]
-        parents[child.raw_id] = 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:
+    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"
@@ -232,19 +201,31 @@ def classify_historical_full_revision_streams(
             for revision in revisions
         ]
     decisions: list[HistoricalRevisionDecision] = []
-    current: str | None = roots[0]
-    while current is not None:
-        predecessor = parents[current][0] if parents[current] else None
+    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=current.raw_id,
                 authority=RawRevisionAuthority.BYTE_PROVEN,
                 relation=relation,
                 predecessor_raw_id=predecessor,
             )
         )
-        current = children[current][0] if children[current] else None
+        previous = current
     return decisions
 
 
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 79cfd4636f..f373c5ba23 100644
--- a/polylogue/sources/revision_backfill.py
+++ b/polylogue/sources/revision_backfill.py
@@ -422,11 +422,23 @@ def backfill_historical_revision_evidence(
 
 def _parse_retained_raw(archive: ArchiveStore, raw_id: str) -> tuple[list[ParsedSession], int, RawRevisionKind]:
     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), payload_size, stream_kind
-    _provider, eager_payload, _source_path, _kind = archive.raw_revision_material(raw_id)
-    return _parse_one(provider, eager_payload, source_path), len(eager_payload), kind
+        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:
@@ -527,4 +539,5 @@ def _parse_stream(provider: Provider, payload: BinaryIO, source_path: str) -> li
     "RevisionCensusResult",
     "backfill_historical_revision_evidence",
     "census_historical_revision_evidence",
+    "parse_retained_raw_sessions",
 ]
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/storage/test_raw_revision_authority.py b/tests/unit/storage/test_raw_revision_authority.py
index 21a25cc780..d8c24c83fd 100644
--- a/tests/unit/storage/test_raw_revision_authority.py
+++ b/tests/unit/storage/test_raw_revision_authority.py
@@ -81,7 +81,7 @@ 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 opened
+    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"]])