fix(sources): decode artifact sidecars before admission

polylogue/archive/raw_payload/decode.py

@@ -2,9 +2,10 @@
 
 from __future__ import annotations
 
+from collections import deque
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Literal, TypeAlias, cast
+from typing import IO, Literal, TypeAlias, cast
 
 from polylogue.archive.artifact_taxonomy import (
     ArtifactClassification,
@@ -182,6 +183,47 @@ def _sample_jsonl_payload_with_detail(
     return samples, malformed_lines, malformed_detail
 
 
+def jsonl_session_artifact(
+    raw: Path | bytes | str | IO[bytes] | IO[str],
+    *,
+    provider: Provider,
+    jsonl_dict_only: bool = False,
+) -> ArtifactClassification | None:
+    """Stream JSONL until one decoded record proves session eligibility.
+
+    Terminal artifact admission must not let an arbitrary prefix of
+    non-conversational records hide a later session record. This retains a
+    rolling 32-record window, including for blob-backed multi-gigabyte JSONL.
+    """
+    records: deque[JSONValue] = deque(maxlen=32)
+    first_line = True
+    with raw_line_stream(raw) as stream:
+        for raw_line in stream:
+            try:
+                line = _decode_provider_utf8(raw_line) if isinstance(raw_line, bytes) else raw_line
+            except UnicodeDecodeError:
+                continue
+            if first_line:
+                line = line.lstrip("\ufeff")
+                first_line = False
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                payload = _load_json_record(line)
+            except (JSONDecodeError, ValueError):
+                continue
+            if jsonl_dict_only and not isinstance(payload, dict):
+                continue
+            records.append(payload)
+            window = list(records)
+            for start in range(len(window)):
+                artifact = classify_artifact(window[start:], provider=provider)
+                if artifact.parse_as_session:
+                    return artifact
+    return None
+
+
 def sample_jsonl_payload(
     raw: Path | bytes | str,
     *,
@@ -437,5 +479,6 @@ def _hermes_sqlite_marker_payload(
     "RawPayloadEnvelope",
     "WireFormat",
     "build_raw_payload_envelope",
+    "jsonl_session_artifact",
     "sample_jsonl_payload",
 ]

polylogue/archive/raw_payload/streams.py

@@ -12,8 +12,8 @@
 
 
 @contextmanager
-def raw_line_stream(raw: Path | bytes | str) -> Iterator[RawLineStream]:
-    """Yield a line stream for path, bytes, or in-memory text payloads."""
+def raw_line_stream(raw: Path | bytes | str | RawLineStream) -> Iterator[RawLineStream]:
+    """Yield a line stream for a path, payload, or caller-owned stream."""
     if isinstance(raw, Path):
         with raw.open("rb") as stream:
             yield stream
@@ -22,5 +22,8 @@ def raw_line_stream(raw: Path | bytes | str) -> Iterator[RawLineStream]:
         with BytesIO(raw) as stream:
             yield stream
         return
+    if not isinstance(raw, str):
+        yield raw
+        return
     with StringIO(raw) as stream:
         yield stream

polylogue/archive/zip_admission.py

@@ -0,0 +1,149 @@
+"""Shared ZIP admission and bounded-entry opening primitives."""
+
+from __future__ import annotations
+
+import io
+import zipfile
+from collections.abc import Callable, Collection, Iterable
+from pathlib import Path
+from typing import IO
+
+from polylogue.logging import get_logger
+
+logger = get_logger(__name__)
+
+MAX_COMPRESSION_RATIO = 1000
+MAX_UNCOMPRESSED_SIZE = 10 * 1024 * 1024 * 1024
+MAX_AGGREGATE_UNCOMPRESSED_SIZE = 64 * 1024 * 1024 * 1024
+# Kept as a public-to-the-source-layer tuning point for bounded streaming
+# callers that need to exercise a read window in tests.
+_ZIP_READ_CHUNK_SIZE = 1024 * 1024
+ZIP_JSON_SUFFIXES = (".json", ".jsonl", ".jsonl.txt", ".ndjson")
+
+
+class ZipBombError(Exception):
+    """Raised when an entry's real decompressed size exceeds the hard cap."""
+
+
+class _BoundedZipReader(io.RawIOBase):
+    def __init__(self, raw: IO[bytes], *, max_bytes: int, entry_name: str) -> None:
+        super().__init__()
+        self._raw = raw
+        self._max_bytes = max_bytes
+        self._entry_name = entry_name
+        self._total = 0
+
+    def readable(self) -> bool:
+        return True
+
+    def readinto(self, buffer: object) -> int:
+        view = memoryview(buffer)  # type: ignore[arg-type]
+        chunk = self._raw.read(len(view))
+        if not chunk:
+            return 0
+        self._total += len(chunk)
+        if self._total > self._max_bytes:
+            raise ZipBombError(
+                f"ZIP entry {self._entry_name!r} exceeded the {self._max_bytes}-byte decompression ceiling during read"
+            )
+        view[: len(chunk)] = chunk
+        return len(chunk)
+
+    def close(self) -> None:
+        try:
+            self._raw.close()
+        finally:
+            super().close()
+
+
+def open_bounded_zip_entry(
+    zf: zipfile.ZipFile,
+    info: zipfile.ZipInfo,
+    *,
+    max_bytes: int | None = None,
+) -> io.BufferedReader:
+    """Open an admitted ZIP entry with a hard real-byte decompression ceiling."""
+    if max_bytes is None:
+        max_bytes = MAX_UNCOMPRESSED_SIZE
+    raw = zf.open(info)
+    return io.BufferedReader(_BoundedZipReader(raw, max_bytes=max_bytes, entry_name=info.filename))
+
+
+class ZipAdmission:
+    """Admit exact central-directory entries before any decompression."""
+
+    __slots__ = ("_zip_path", "_aggregate_total")
+
+    def __init__(self, *, zip_path: Path) -> None:
+        self._zip_path = zip_path
+        self._aggregate_total = 0
+
+    def filter_entries(
+        self,
+        entries: list[zipfile.ZipInfo],
+        *,
+        allowed_suffixes: Collection[str] = ZIP_JSON_SUFFIXES,
+        on_rejected: Callable[[zipfile.ZipInfo, str], None] | None = None,
+    ) -> Iterable[zipfile.ZipInfo]:
+        """Yield admitted ``ZipInfo`` objects and report rejected entries."""
+        suffixes = tuple(suffix.lower() for suffix in allowed_suffixes)
+
+        def reject(info: zipfile.ZipInfo, reason: str) -> None:
+            if on_rejected is not None:
+                on_rejected(info, reason)
+
+        for info in entries:
+            if info.is_dir():
+                continue
+            name = info.filename
+            lower_name = name.lower()
+            if info.compress_size > 0:
+                ratio = info.file_size / info.compress_size
+                if ratio > MAX_COMPRESSION_RATIO:
+                    logger.warning(
+                        "Skipping suspicious file %s in %s: compression ratio %.1f exceeds limit",
+                        name,
+                        self._zip_path,
+                        ratio,
+                    )
+                    reject(info, f"zip entry compression ratio {ratio:.1f} exceeds limit")
+                    continue
+            if info.file_size > MAX_UNCOMPRESSED_SIZE:
+                logger.warning(
+                    "Skipping oversized file %s in %s: %d bytes exceeds limit",
+                    name,
+                    self._zip_path,
+                    info.file_size,
+                )
+                reject(info, f"zip entry file size {info.file_size} exceeds limit")
+                continue
+            if not lower_name.endswith(suffixes):
+                continue
+            projected_total = self._aggregate_total + info.file_size
+            if projected_total > MAX_AGGREGATE_UNCOMPRESSED_SIZE:
+                logger.warning(
+                    "Skipping %s in %s: aggregate uncompressed size %d would exceed the %d-byte archive-wide limit",
+                    name,
+                    self._zip_path,
+                    projected_total,
+                    MAX_AGGREGATE_UNCOMPRESSED_SIZE,
+                )
+                reject(
+                    info,
+                    f"aggregate uncompressed size {projected_total} exceeds archive-wide limit "
+                    f"{MAX_AGGREGATE_UNCOMPRESSED_SIZE}",
+                )
+                continue
+            self._aggregate_total = projected_total
+            yield info
+
+
+__all__ = [
+    "MAX_AGGREGATE_UNCOMPRESSED_SIZE",
+    "MAX_COMPRESSION_RATIO",
+    "MAX_UNCOMPRESSED_SIZE",
+    "ZIP_JSON_SUFFIXES",
+    "ZipAdmission",
+    "ZipBombError",
+    "open_bounded_zip_entry",
+]

polylogue/insights/claude_workflow_materializer.py

@@ -17,7 +17,11 @@
 from pathlib import Path, PurePosixPath
 from typing import Literal
 
+from polylogue.archive.artifact_taxonomy import classify_artifact
+from polylogue.archive.raw_payload.decode import jsonl_session_artifact
 from polylogue.core.enums import Origin, Provider
+from polylogue.core.json import JSONDecodeError
+from polylogue.core.json import loads as json_loads
 from polylogue.core.refs import EvidenceRef, ObjectRef
 from polylogue.insights.claude_workflow_evidence import (
     ClaudeWorkflowCoordinatorInvocation,
@@ -184,15 +188,15 @@ def _prepare_inputs(archive_root: Path) -> _PreparedInputs:
     if not source_db.exists() or not index_db.exists():
         raise FileNotFoundError("Claude Workflow materialization requires source.db and index.db")
 
+    blob_store = BlobStore(archive_root / "blob")
     with sqlite3.connect(source_db) as source_conn:
         source_conn.row_factory = sqlite3.Row
         source_conn.execute("PRAGMA foreign_keys = ON")
-        _ensure_current_artifact_inventory(source_conn)
+        _ensure_current_artifact_inventory(source_conn, blob_store=blob_store)
         source_conn.commit()
         raw_artifacts = _load_current_artifacts(source_conn)
         retained_revisions = _count_retained_revisions(source_conn)
 
-    blob_store = BlobStore(archive_root / "blob")
     parsed: list[ClaudeOrchestrationArtifact] = []
     artifact_evidence: dict[str, ObjectRef] = {}
     for raw in raw_artifacts:
@@ -254,7 +258,26 @@ def _prepare_inputs(archive_root: Path) -> _PreparedInputs:
     )
 
 
-def _ensure_current_artifact_inventory(conn: sqlite3.Connection) -> None:
+def _raw_payload_has_session_evidence(blob_store: BlobStore, row: sqlite3.Row) -> bool:
+    """Keep session-shaped JSON payloads out of path-only artifact inventory."""
+    path = Path(str(row["source_path"]))
+    blob_hash = bytes(row["blob_hash"]).hex()
+    if path.suffix.lower() == ".jsonl":
+        try:
+            return jsonl_session_artifact(blob_store.blob_path(blob_hash), provider=Provider.CLAUDE_CODE) is not None
+        except (OSError, ValueError):
+            return False
+    if path.suffix.lower() != ".json":
+        return False
+    try:
+        with blob_store.open(blob_hash) as handle:
+            document = json_loads(handle.read())
+    except (OSError, JSONDecodeError, ValueError):
+        return False
+    return classify_artifact(document, provider=Provider.CLAUDE_CODE).parse_as_session
+
+
+def _ensure_current_artifact_inventory(conn: sqlite3.Connection, *, blob_store: BlobStore) -> None:
     """Refresh current pointers for OriginSpec-declared Claude artifacts.
 
     Canonical configured acquisition already writes these rows.  The same
@@ -283,6 +306,12 @@ def _ensure_current_artifact_inventory(conn: sqlite3.Connection) -> None:
         rule = artifact_rule_for_path(Provider.CLAUDE_CODE, str(row["source_path"]))
         if rule is None:
             continue
+        if _raw_payload_has_session_evidence(blob_store, row):
+            conn.execute(
+                "DELETE FROM raw_artifacts WHERE origin = ? AND source_path = ? AND source_index = ?",
+                (row["origin"], row["source_path"], row["source_index"]),
+            )
+            continue
         existing = conn.execute(
             """
             SELECT artifact_id, first_observed_at_ms