diff --git a/polylogue/archive/raw_payload/decode.py b/polylogue/archive/raw_payload/decode.py index c897f4c9df..4deb3554bc 100644 --- a/polylogue/archive/raw_payload/decode.py +++ b/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", ] diff --git a/polylogue/archive/raw_payload/streams.py b/polylogue/archive/raw_payload/streams.py index 4dff08e077..79d4205903 100644 --- a/polylogue/archive/raw_payload/streams.py +++ b/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 diff --git a/polylogue/archive/zip_admission.py b/polylogue/archive/zip_admission.py new file mode 100644 index 0000000000..995e08e317 --- /dev/null +++ b/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", +] diff --git a/polylogue/insights/claude_workflow_materializer.py b/polylogue/insights/claude_workflow_materializer.py index 5386718ae0..d619914641 100644 --- a/polylogue/insights/claude_workflow_materializer.py +++ b/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 diff --git a/polylogue/pipeline/services/archive_ingest.py b/polylogue/pipeline/services/archive_ingest.py index aba9f9d26f..8911287371 100644 --- a/polylogue/pipeline/services/archive_ingest.py +++ b/polylogue/pipeline/services/archive_ingest.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import zipfile from concurrent.futures import as_completed from datetime import UTC, datetime from pathlib import Path @@ -18,8 +19,15 @@ process_pool_executor, resolve_archive_ingest_dispatch, ) +from polylogue.sources.decoder_zip import ( + ZipBombError, + ZipEntryValidator, + open_bounded_zip_entry, + zip_entry_session_artifact, +) from polylogue.sources.parsers.base import ParsedSession, RawSessionData from polylogue.sources.source_parsing import ( + has_decoded_session_evidence, iter_antigravity_language_server_sessions, iter_source_sessions_with_raw, parse_one_source_path, @@ -412,9 +420,22 @@ def _admit_non_session_origin_artifacts( ) if walk is None: continue + provider = Provider.from_string(source.name) for candidate, _mtime in walk.paths_to_process: + if candidate.suffix.lower() == ".zip": + admitted += _admit_non_session_zip_artifacts( + archive, + candidate, + provider=provider, + acquired_at_ms=acquired_at_ms, + ) + continue classification = classify_artifact_path(candidate, provider=source.name) - if classification is None or classification.parse_as_session: + if ( + classification is None + or classification.parse_as_session + or has_decoded_session_evidence(candidate, provider=provider) + ): continue try: # polylogue-1fijp arm 4: route through the raw-admission @@ -436,6 +457,51 @@ def _admit_non_session_origin_artifacts( return admitted +def _admit_non_session_zip_artifacts( + archive: ArchiveStore, + zip_path: Path, + *, + provider: Provider, + acquired_at_ms: int, +) -> int: + """Retain ZIP member artifacts only after decoded JSONL evidence is absent.""" + from polylogue.storage.blob_publication import ArchiveBlobPublisher + + admitted = 0 + publisher = ArchiveBlobPublisher(archive.source_db_path, archive.archive_root / "blob") + try: + with zipfile.ZipFile(zip_path) as zf: + validator = ZipEntryValidator(provider, cursor_state=None, zip_path=zip_path) + for info in validator.filter_entries(zf.infolist()): + classification = classify_artifact_path(info.filename, provider=provider) + if ( + classification is None + or classification.parse_as_session + or zip_entry_session_artifact(zf, info, provider=provider) is not None + ): + continue + with open_bounded_zip_entry(zf, info) as payload: + blob_hash, blob_size = publisher.write_from_fileobj(payload) + receipt_id = publisher.receipt_id(blob_hash) + publisher.flush() + archive.admit_raw_artifact_blob_ref( + provider=provider, + blob_hash_hex=blob_hash, + blob_size=blob_size, + source_path=f"{zip_path}:{info.filename}", + source_index=0, + acquired_at_ms=acquired_at_ms, + classification=classification, + blob_publication_receipt_id=receipt_id, + ) + admitted += 1 + except (OSError, ZipBombError, zipfile.BadZipFile): + logger.error("Failed to admit configured ZIP artifacts from %s", zip_path, exc_info=True) + finally: + publisher.discard_pending() + return admitted + + def _record_post_commit_upkeep(archive_root: Path, result: ParseResult, *, reason: str) -> None: """Run bounded archive-tier upkeep after a direct archive ingest commit. diff --git a/polylogue/pipeline/services/ingest_worker.py b/polylogue/pipeline/services/ingest_worker.py index 5f316e88ad..51c18f5b2e 100644 --- a/polylogue/pipeline/services/ingest_worker.py +++ b/polylogue/pipeline/services/ingest_worker.py @@ -19,9 +19,18 @@ from typing_extensions import TypedDict -from polylogue.archive.artifact_taxonomy import ArtifactClassification, ArtifactKind, classify_artifact +from polylogue.archive.artifact_taxonomy import ( + ArtifactClassification, + ArtifactKind, + classify_artifact, + classify_artifact_path, +) from polylogue.archive.artifact_taxonomy.support import is_subagent_path -from polylogue.archive.raw_payload.decode import RawPayloadEnvelope +from polylogue.archive.raw_payload.decode import ( + RawPayloadEnvelope, + _sample_jsonl_payload_with_detail, + jsonl_session_artifact, +) from polylogue.core.common import format_malformed_jsonl_error as _format_malformed_jsonl_error from polylogue.core.enums import IngestOutcome, Provider, ValidationMode, ValidationStatus from polylogue.logging import get_logger @@ -310,7 +319,6 @@ def _build_stream_parse_plan( *, payload_provider: str | None, ) -> _ParsePlan | None: - from polylogue.archive.raw_payload.decode import _sample_jsonl_payload_with_detail from polylogue.sources.dispatch import detect_provider stream_name = context.raw_record.source_path or context.raw_record.raw_id @@ -346,10 +354,21 @@ def _build_stream_parse_plan( return None runtime_provider = detected_provider - artifact = classify_artifact( + decoded_artifact = classify_artifact( sample_payloads, provider=runtime_provider, - source_path=context.raw_record.source_path, + ) + path_artifact = classify_artifact_path( + context.raw_record.source_path, + provider=runtime_provider, + ) + session_artifact = ( + jsonl_session_artifact(context.raw_source, provider=runtime_provider, jsonl_dict_only=True) + if path_artifact is not None and not path_artifact.parse_as_session + else None + ) + artifact = session_artifact or ( + decoded_artifact if decoded_artifact.parse_as_session else path_artifact or decoded_artifact ) return _build_parse_plan( provider=runtime_provider, @@ -374,6 +393,58 @@ def _build_fast_stream_parse_plan( if runtime_provider not in STREAM_RECORD_PROVIDERS: return None + # The validation-off shortcut still has to honor path-declared fact and + # raw-only artifacts. Without this check, a workflow journal's JSONL path + # is replaced by the generic session classification below before the + # payload is decoded, so session-shaped journal records materialize as + # conversations even though the same path is classified as evidence by + # the ordinary envelope route. + path_artifact = classify_artifact_path( + context.raw_record.source_path, + provider=runtime_provider, + ) + if path_artifact is not None and not path_artifact.parse_as_session: + try: + sample_payloads, malformed_lines, malformed_detail = _sample_jsonl_payload_with_detail( + context.raw_source, + max_samples=64, + jsonl_dict_only=True, + scan_full=False, + ) + except Exception: + logger.exception( + "JSONL sample probe failed for %s; retaining path-declared artifact", + context.raw_record.source_path or context.raw_record.raw_id, + ) + else: + decoded_artifact = jsonl_session_artifact( + context.raw_source, + provider=runtime_provider, + jsonl_dict_only=True, + ) or classify_artifact(sample_payloads, provider=runtime_provider) + if decoded_artifact.parse_as_session: + return _build_parse_plan( + provider=runtime_provider, + payload_provider=str(runtime_provider), + artifact=decoded_artifact, + source_path=context.raw_record.source_path, + mode="stream", + payload=sample_payloads, + schema_payload_source=sample_payloads, + stream_name=context.raw_record.source_path or context.raw_record.raw_id, + malformed_jsonl_lines=malformed_lines, + malformed_jsonl_detail=malformed_detail, + ) + return _build_parse_plan( + provider=runtime_provider, + payload_provider=str(runtime_provider), + artifact=path_artifact, + source_path=context.raw_record.source_path, + mode="stream", + schema_payload_source=None, + stream_name=context.raw_record.source_path or context.raw_record.raw_id, + ) + kind = ( ArtifactKind.AGENT_TRANSCRIPT if is_subagent_path(context.raw_record.source_path) diff --git a/polylogue/sources/assembly_chatgpt.py b/polylogue/sources/assembly_chatgpt.py index aa3b03e8ee..cc057ec98b 100644 --- a/polylogue/sources/assembly_chatgpt.py +++ b/polylogue/sources/assembly_chatgpt.py @@ -52,89 +52,81 @@ def _read_json_file(path: Path) -> object | None: return None -def _read_json_zip_member(zip_path: Path, member_name: str) -> object | None: +def _read_json_zip_member(zip_path: Path, info: zipfile.ZipInfo, zf: zipfile.ZipFile) -> object | None: + from .decoder_zip import ZipBombError, open_bounded_zip_entry + try: - with zipfile.ZipFile(zip_path) as zf, zf.open(member_name) as handle: + with open_bounded_zip_entry(zf, info) as handle: data: object = json.load(handle) return data - except (OSError, KeyError, zipfile.BadZipFile, json.JSONDecodeError) as exc: + except (OSError, KeyError, zipfile.BadZipFile, json.JSONDecodeError, ZipBombError) as exc: logger.debug( "chatgpt_sidecar_zip_member_unavailable", zip_path=str(zip_path), - member=member_name, + member=info.filename, error=str(exc), ) return None -def _dat_asset_id(basename: str) -> str: - bare = basename[: -len(_DAT_SUFFIX)] if basename.lower().endswith(_DAT_SUFFIX) else basename - return _normalize_file_id(bare) - - -def _acquire_dat_blobs_from_zip(zip_path: Path, store: BlobStore) -> dict[str, tuple[str, int]]: - """Stream every ``.dat`` ZIP member into *store*, keyed by asset id. +def _read_chatgpt_zip_sidecars( + zip_path: Path, + store: BlobStore | None, +) -> tuple[dict[str, object], dict[str, tuple[str, int]]]: + """Read admitted JSON sidecars and stream admitted ``.dat`` members. - Mirrors ``decoder_zip.py``'s ``capture_raw`` streaming pattern (bounded - decompression via ``open_bounded_zip_entry``, no full-file memory load) - but scans the whole archive up front rather than the main - ``ZipEntryValidator`` per-entry loop, which only ever admits - ``.json``/``.jsonl`` entries (``session_only=True``) and would otherwise - never see a ``.dat`` member at all. + ``ZipInfo`` identity is preserved from central-directory admission through + decompression. In particular, a later duplicate filename cannot replace an + earlier member by making ``ZipFile.open(name)`` resolve through the archive's + name map. One validator accounts for every relevant member in the archive, + so JSON and ``.dat`` payloads share the cumulative limit. """ - from polylogue.storage.blob_publication import flush_blob_publications - - from .decoder_zip import ( - MAX_COMPRESSION_RATIO, - MAX_UNCOMPRESSED_SIZE, - ZipBombError, - open_bounded_zip_entry, - ) + from .decoder_zip import ZIP_JSON_SUFFIXES, ZipBombError, ZipEntryValidator, open_bounded_zip_entry + targets = {_LIBRARY_FILES_NAME, _ASSET_NAMES_NAME} + seen_targets: set[str] = set() + payloads: dict[str, object] = {} acquired: dict[str, tuple[str, int]] = {} try: with zipfile.ZipFile(zip_path) as zf: - for info in zf.infolist(): - if info.is_dir(): - continue - name = info.filename - if not name.lower().endswith(_DAT_SUFFIX): - continue - dat_id = _dat_asset_id(Path(name).name) - if dat_id in acquired: - continue - if info.compress_size > 0 and info.file_size / info.compress_size > MAX_COMPRESSION_RATIO: - logger.warning("chatgpt_dat_suspicious_compression_ratio", path=str(zip_path), member=name) + validator = ZipEntryValidator("chatgpt", cursor_state=None, zip_path=zip_path) + for info in validator.filter_entries(zf.infolist(), allowed_suffixes=(*ZIP_JSON_SUFFIXES, _DAT_SUFFIX)): + if info.filename.lower().endswith(_DAT_SUFFIX): + if store is None: + continue + dat_id = _dat_asset_id(Path(info.filename).name) + if dat_id in acquired: + continue + try: + with open_bounded_zip_entry(zf, info) as handle: + blob_hash, size = store.write_from_fileobj(handle) + except ZipBombError: + logger.warning("chatgpt_dat_zip_bomb", path=str(zip_path), member=info.filename) + continue + except (KeyError, zipfile.BadZipFile, OSError) as exc: + logger.debug( + "chatgpt_dat_read_failed", + path=str(zip_path), + member=info.filename, + error=str(exc), + ) + continue + acquired[dat_id] = (blob_hash, size) continue - if info.file_size > MAX_UNCOMPRESSED_SIZE: - logger.warning( - "chatgpt_dat_oversized", - path=str(zip_path), - member=name, - size=info.file_size, - ) + if info.filename not in targets or info.filename in seen_targets: continue - try: - with open_bounded_zip_entry(zf, name) as handle: - blob_hash, size = store.write_from_fileobj(handle) - except ZipBombError: - logger.warning("chatgpt_dat_zip_bomb", path=str(zip_path), member=name) - continue - except (KeyError, zipfile.BadZipFile, OSError) as exc: - logger.debug( - "chatgpt_dat_read_failed", - path=str(zip_path), - member=name, - error=str(exc), - ) - continue - acquired[dat_id] = (blob_hash, size) + seen_targets.add(info.filename) + payload = _read_json_zip_member(zip_path, info, zf) + if payload is not None: + payloads[info.filename] = payload except (OSError, zipfile.BadZipFile) as exc: - logger.warning("chatgpt_dat_zip_open_failed", path=str(zip_path), error=str(exc)) - return acquired - if acquired: - flush_blob_publications(store) - return acquired + logger.debug("chatgpt_sidecar_zip_open_failed", zip_path=str(zip_path), error=str(exc)) + return payloads, acquired + + +def _dat_asset_id(basename: str) -> str: + bare = basename[: -len(_DAT_SUFFIX)] if basename.lower().endswith(_DAT_SUFFIX) else basename + return _normalize_file_id(bare) def _acquire_dat_blobs_from_directory(directory: Path, store: BlobStore) -> dict[str, tuple[str, int]]: @@ -213,12 +205,16 @@ def discover_sidecars( seen_dirs: set[Path] = set() for path in source_paths: if path.suffix.lower() == ".zip": + zip_sidecars, zip_dat_blobs = _read_chatgpt_zip_sidecars(path, blob_store) if library_files_payload is None: - library_files_payload = _read_json_zip_member(path, _LIBRARY_FILES_NAME) + library_files_payload = zip_sidecars.get(_LIBRARY_FILES_NAME) if asset_names_payload is None: - asset_names_payload = _read_json_zip_member(path, _ASSET_NAMES_NAME) - if blob_store is not None: - dat_blobs.update(_acquire_dat_blobs_from_zip(path, blob_store)) + asset_names_payload = zip_sidecars.get(_ASSET_NAMES_NAME) + dat_blobs.update(zip_dat_blobs) + if zip_dat_blobs and blob_store is not None: + from polylogue.storage.blob_publication import flush_blob_publications + + flush_blob_publications(blob_store) continue directory = path.parent if directory in seen_dirs: @@ -315,7 +311,7 @@ def _resolve_dat_attachment( update["provider_file_id"] = resolved.file_id if blob is not None and attachment.inline_bytes is None and attachment.precomputed_blob is None: # bd polylogue-8ac0: bytes already streamed into the blob store during - # sidecar discovery (`_acquire_dat_blobs_from_zip`/`_from_directory`). + # sidecar discovery (`_read_chatgpt_zip_sidecars`/`_from_directory`). # Recording the (hash, size) pair here -- rather than re-reading the # source bytes -- lets `ingest_batch/_core.py` mark the attachment # acquired without re-hashing already-written bytes. diff --git a/polylogue/sources/decoder_zip.py b/polylogue/sources/decoder_zip.py index 87bc1219bd..06e7f9defb 100644 --- a/polylogue/sources/decoder_zip.py +++ b/polylogue/sources/decoder_zip.py @@ -2,14 +2,24 @@ from __future__ import annotations -import io import zipfile -from collections.abc import Iterable +from collections.abc import Callable, Collection, Iterable from pathlib import Path -from typing import IO -from polylogue.archive.artifact_taxonomy import classify_artifact_path +from polylogue.archive.artifact_taxonomy import ArtifactClassification, classify_artifact_path +from polylogue.archive.zip_admission import ( + _ZIP_READ_CHUNK_SIZE, + MAX_AGGREGATE_UNCOMPRESSED_SIZE, + MAX_COMPRESSION_RATIO, + MAX_UNCOMPRESSED_SIZE, + ZIP_JSON_SUFFIXES, + ZipAdmission, + ZipBombError, + open_bounded_zip_entry, +) from polylogue.core.enums import Provider +from polylogue.core.json import JSONDecodeError +from polylogue.core.json import loads as json_loads from polylogue.logging import get_logger from polylogue.storage.blob_store import BlobStore from polylogue.storage.cursor_state import CursorStatePayload @@ -20,101 +30,11 @@ logger = get_logger(__name__) -MAX_COMPRESSION_RATIO = 1000 -MAX_UNCOMPRESSED_SIZE = 10 * 1024 * 1024 * 1024 - -#: Aggregate ceiling (bytes) on the sum of declared ``file_size`` across every -#: entry admitted from a single ZIP archive (polylogue-lqxx). The per-entry -#: cap above bounds one entry but does nothing to stop a "zip bomb by many -#: entries": thousands of entries each just under the 10 GiB per-entry cap -#: would still sum to an unbounded total. 64 GiB is chosen to comfortably -#: exceed any real single-archive GDPR/Takeout export this repo has observed -#: (the largest full raw corpus recorded across this operator's *entire* -#: archive history, spanning every session ever ingested, is ~52.1 GiB -- -#: see `sources/revision_backfill.py`'s newest-revision-raws comment) while -#: still bounding aggregate decompression well below the terabyte-scale -#: totals a many-small-entries zip bomb would otherwise reach. It is also in -#: the same order of magnitude as the daemon's other whole-pass resource -#: envelopes (e.g. the whale-pass `raw_authority_whale_payload_bytes` -#: default of 8 GiB for a *single* stream-safe-gated component). -MAX_AGGREGATE_UNCOMPRESSED_SIZE = 64 * 1024 * 1024 * 1024 - -#: Chunk size used when bounding ZIP entry decompression. Read in fixed -#: windows so a malicious entry cannot allocate more than this much extra -#: memory beyond the running total before the ceiling check fires. -_ZIP_READ_CHUNK_SIZE = 1024 * 1024 - - -class ZipBombError(Exception): - """Raised when an entry's real decompressed size exceeds the hard cap. - - The declared header sizes (``ZipInfo.file_size`` / ``compress_size``) - are attacker-controllable, so they are used only for an early cheap - skip. The authoritative ceiling is enforced here against the actual - bytes produced by decompression. - """ - - -class _BoundedZipReader(io.RawIOBase): - """Wrap a ZIP entry stream and abort once ``max_bytes`` is exceeded. - - Every read is counted against the real decompressed byte total. If the - total would cross ``max_bytes`` the reader raises :class:`ZipBombError` - instead of returning the bytes, so downstream consumers never receive - an over-cap payload regardless of the entry's declared sizes. - """ - - 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, - name: str, - *, - max_bytes: int = MAX_UNCOMPRESSED_SIZE, -) -> io.BufferedReader: - """Open a ZIP entry with a hard real-byte decompression ceiling. - - Returns a buffered stream that raises :class:`ZipBombError` if the - actual decompressed size would exceed ``max_bytes``. This does not - trust the (forgeable) declared header sizes — the ceiling is enforced - against bytes produced by the decompressor itself. - """ - raw = zf.open(name) - return io.BufferedReader(_BoundedZipReader(raw, max_bytes=max_bytes, entry_name=name)) - class ZipEntryValidator: """Validate ZIP entries for security and relevance.""" - __slots__ = ("_provider_hint", "_cursor_state", "_zip_path", "_session_only", "_aggregate_total") + __slots__ = ("_cursor_state", "_zip_path", "_admission") def __init__( self, @@ -122,98 +42,74 @@ def __init__( *, cursor_state: CursorStatePayload | None, zip_path: Path, - session_only: bool = False, ) -> None: - self._provider_hint = Provider.from_string(provider_hint) + del provider_hint self._cursor_state = cursor_state self._zip_path = zip_path - self._session_only = session_only - self._aggregate_total = 0 - - def filter_entries(self, entries: list[zipfile.ZipInfo]) -> Iterable[zipfile.ZipInfo]: - """Yield safe, relevant entries and record failures in cursor state.""" - for info in entries: - if info.is_dir(): - continue - name = info.filename - lower_name = name.lower() + self._admission = ZipAdmission(zip_path=zip_path) - 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, - ) - _record_cursor_failure( - self._cursor_state, - f"{self._zip_path}:{name}", - f"Suspicious compression ratio: {ratio:.1f}", - ) - 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, - ) - _record_cursor_failure( - self._cursor_state, - f"{self._zip_path}:{name}", - f"File size {info.file_size} exceeds limit", - ) - continue + 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 safe, relevant entries and record failures in cursor state. + + ``allowed_suffixes`` selects which member kinds a caller needs, while + this validator remains the sole owner of the security checks. The + yielded object is the exact central-directory ``ZipInfo`` that was + admitted. Callers must pass it through to ``open_bounded_zip_entry``; + reopening by filename can select a different duplicate member. + + ``on_rejected`` lets read-only surfaces report the same admission + decisions without duplicating the security checks. + """ + + def reject(info: zipfile.ZipInfo, reason: str) -> None: + _record_cursor_failure( + self._cursor_state, + f"{self._zip_path}:{info.filename}", + reason.capitalize() if reason.startswith("aggregate") else reason, + ) + if on_rejected is not None: + on_rejected(info, reason) - if lower_name.endswith((".json", ".jsonl", ".jsonl.txt", ".ndjson")): - if self._session_only: - # Classify on the bare intra-archive relative path, not - # the zip-container-prefixed ``{zip_path}:{name}`` form. - # Every ``OriginArtifactRule.path_pattern`` is anchored - # ``(?:^|/)`` to match a relative filesystem-style path - # (the same convention every non-zip caller of - # ``classify_artifact_path`` already uses, e.g. - # ``sources/live/batch_support.py``): the character - # immediately before a rule's leading path segment must be - # ``/`` or start-of-string. Prefixing with the zip path - # inserts a ``:`` there instead, so no rule could ever - # match and this exclusion was dead code (polylogue-dc1k). - path_classification = classify_artifact_path( - name, - provider=self._provider_hint, - ) - if path_classification is not None and not path_classification.parse_as_session: - continue + yield from self._admission.filter_entries( + entries, + allowed_suffixes=allowed_suffixes, + on_rejected=reject, + ) - # Aggregate cap: the per-entry check above bounds one entry, - # but a zip bomb built from many entries each just under the - # per-entry cap would otherwise sum to an unbounded total. - # Check the running total of declared uncompressed sizes - # against MAX_AGGREGATE_UNCOMPRESSED_SIZE before yielding, so - # rejection happens from central-directory metadata alone -- - # before any entry is opened/decompressed. - 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, - ) - _record_cursor_failure( - self._cursor_state, - f"{self._zip_path}:{name}", - f"Aggregate uncompressed size {projected_total} exceeds archive-wide limit " - f"{MAX_AGGREGATE_UNCOMPRESSED_SIZE}", - ) - continue - self._aggregate_total = projected_total - yield info +def zip_entry_session_artifact( + zf: zipfile.ZipFile, + info: zipfile.ZipInfo, + *, + provider: Provider, +) -> ArtifactClassification | None: + """Decode a member before applying a terminal artifact path rule.""" + from polylogue.archive.raw_payload.decode import jsonl_session_artifact + + lower_name = info.filename.lower() + if lower_name.endswith((".jsonl", ".jsonl.txt", ".ndjson")): + with open_bounded_zip_entry(zf, info) as handle: + return jsonl_session_artifact(handle, provider=provider) + if not lower_name.endswith(".json"): + return None + try: + with open_bounded_zip_entry(zf, info) as handle: + payload = json_loads(handle.read()) + except JSONDecodeError: + return None + # Deliberately omit source_path. The caller is asking whether decoded + # content can override a non-session path rule, so reapplying that rule + # here would make the evidence check circular. + from polylogue.archive.artifact_taxonomy import classify_artifact + + artifact = classify_artifact(payload, provider=provider) + return artifact if artifact.parse_as_session else None def zip_entry_provider_hint(entry_name: str, fallback_provider: str | Provider) -> Provider: @@ -261,13 +157,18 @@ def process_zip( provider_hint, cursor_state=cursor_state, zip_path=zip_path, - session_only=True, ) with zipfile.ZipFile(zip_path) as zf: for info in validator.filter_entries(zf.infolist()): name = info.filename entry_provider_hint = zip_entry_provider_hint(name, provider_hint) + path_classification = classify_artifact_path(name, provider=entry_provider_hint) + session_artifact: ArtifactClassification | None = None + if path_classification is not None and not path_classification.parse_as_session: + session_artifact = zip_entry_session_artifact(zf, info, provider=entry_provider_hint) + if session_artifact is None: + continue entry_should_group = entry_provider_hint in GROUP_PROVIDERS ctx = _ParseContext( provider_hint=entry_provider_hint, @@ -285,7 +186,7 @@ def process_zip( # ``open_bounded_zip_entry`` enforces a hard real-byte # ceiling during decompression, independent of the # entry's (forgeable) declared header sizes. - with open_bounded_zip_entry(zf, name) as handle: + with open_bounded_zip_entry(zf, info) as handle: blob_hash, blob_size = store.write_from_fileobj(handle) receipt_id = publication_receipt_id(store, blob_hash) flush_blob_publications(store) @@ -299,8 +200,13 @@ def process_zip( blob_size=blob_size, blob_publication_receipt_id=receipt_id, ) - with open_bounded_zip_entry(zf, name) as handle: - yield from emitter.emit(handle, name, precomputed_raw=precomputed_raw) + with open_bounded_zip_entry(zf, info) as handle: + yield from emitter.emit( + handle, + name, + precomputed_raw=precomputed_raw, + session_artifact=session_artifact, + ) except ZipBombError as exc: logger.warning( "Skipping ZIP entry %s in %s: %s", @@ -317,12 +223,15 @@ def process_zip( __all__ = [ + "_ZIP_READ_CHUNK_SIZE", "MAX_AGGREGATE_UNCOMPRESSED_SIZE", "MAX_COMPRESSION_RATIO", "MAX_UNCOMPRESSED_SIZE", + "ZIP_JSON_SUFFIXES", "ZipBombError", "ZipEntryValidator", "open_bounded_zip_entry", "process_zip", + "zip_entry_session_artifact", "zip_entry_provider_hint", ] diff --git a/polylogue/sources/decoders.py b/polylogue/sources/decoders.py index 140bc54ede..2d3842a744 100644 --- a/polylogue/sources/decoders.py +++ b/polylogue/sources/decoders.py @@ -17,6 +17,7 @@ MAX_AGGREGATE_UNCOMPRESSED_SIZE, MAX_COMPRESSION_RATIO, MAX_UNCOMPRESSED_SIZE, + open_bounded_zip_entry, ) from polylogue.sources.decoder_zip import ZipEntryValidator as _ZipEntryValidator from polylogue.sources.decoder_zip import process_zip as _process_zip @@ -46,4 +47,5 @@ def _iter_json_stream( "MAX_AGGREGATE_UNCOMPRESSED_SIZE", "MAX_COMPRESSION_RATIO", "MAX_UNCOMPRESSED_SIZE", + "open_bounded_zip_entry", ] diff --git a/polylogue/sources/emitter.py b/polylogue/sources/emitter.py index fd12baef8f..4f26e38530 100644 --- a/polylogue/sources/emitter.py +++ b/polylogue/sources/emitter.py @@ -81,6 +81,7 @@ def emit( *, pre_read_bytes: bytes | None = None, precomputed_raw: RawSessionData | None = None, + session_artifact: ArtifactClassification | None = None, ) -> Iterable[tuple[RawSessionData | None, ParsedSession]]: """Parse a stream and yield ``(raw, conv)`` tuples. @@ -100,6 +101,7 @@ def emit( stream_name, pre_read_bytes, precomputed_raw=precomputed_raw, + session_artifact=session_artifact, ) return @@ -112,7 +114,12 @@ def emit( ) return - yield from self._emit_individual(handle, stream_name, pre_read_bytes=pre_read_bytes) + yield from self._emit_individual( + handle, + stream_name, + pre_read_bytes=pre_read_bytes, + session_artifact=session_artifact, + ) def _emit_grouped( self, @@ -122,6 +129,7 @@ def _emit_grouped( *, precomputed_raw: RawSessionData | None = None, precomputed_payloads: list[JsonValue] | None = None, + session_artifact: ArtifactClassification | None = None, ) -> Iterable[tuple[RawSessionData | None, ParsedSession]]: """Grouped JSONL: entire file = one session.""" if precomputed_raw is not None: @@ -142,6 +150,12 @@ def _emit_grouped( raw_data = precomputed_raw or (self._make_raw(raw_bytes) if raw_bytes else None) resolved = self._resolve_payload(payloads) + if session_artifact is not None: + resolved = _ResolvedPayload( + provider=resolved.provider, + artifact=session_artifact, + schema_resolution=resolved.schema_resolution, + ) if not resolved.artifact.parse_as_session: return for conv in parse_payload( @@ -159,6 +173,7 @@ def _emit_individual( stream_name: str, *, pre_read_bytes: bytes | None = None, + session_artifact: ArtifactClassification | None = None, ) -> Iterable[tuple[RawSessionData | None, ParsedSession]]: """Individual items: each payload = one session.""" unpack = not (stream_name.lower().endswith(".json") and self._ctx.should_group) @@ -171,6 +186,7 @@ def _emit_individual( _iter_json_stream(handle, stream_name, unpack_lists=unpack), stream_name=stream_name, whole_file_raw=whole_file_raw, + session_artifact=session_artifact, ) def _emit_individual_payloads( @@ -179,11 +195,18 @@ def _emit_individual_payloads( *, stream_name: str, whole_file_raw: RawSessionData | None = None, + session_artifact: ArtifactClassification | None = None, ) -> Iterable[tuple[RawSessionData | None, ParsedSession]]: source_index = 0 for payload in payloads: try: resolved = self._resolve_payload(payload) + if session_artifact is not None: + resolved = _ResolvedPayload( + provider=resolved.provider, + artifact=session_artifact, + schema_resolution=resolved.schema_resolution, + ) if not resolved.artifact.parse_as_session: continue @@ -380,11 +403,13 @@ def _resolve_schema( def _resolve_payload(self, payload: JsonValue) -> _ResolvedPayload: provider = detect_provider(payload) or self._ctx.provider_hint - artifact = classify_artifact( - payload, - provider=provider, - source_path=self._ctx.source_path_str, - ) + artifact = classify_artifact(payload, provider=provider) + if not artifact.parse_as_session: + artifact = classify_artifact( + payload, + provider=provider, + source_path=self._ctx.source_path_str, + ) return _ResolvedPayload( provider=provider, artifact=artifact, diff --git a/polylogue/sources/import_explain.py b/polylogue/sources/import_explain.py index ab71ab6713..9933635b18 100644 --- a/polylogue/sources/import_explain.py +++ b/polylogue/sources/import_explain.py @@ -17,11 +17,11 @@ from polylogue.core.json import JSONValue from polylogue.core.sources import origin_from_provider from polylogue.sources.decoder_zip import ( - MAX_AGGREGATE_UNCOMPRESSED_SIZE, - MAX_COMPRESSION_RATIO, MAX_UNCOMPRESSED_SIZE, ZipBombError, + ZipEntryValidator, open_bounded_zip_entry, + zip_entry_session_artifact, ) from polylogue.sources.decoders import _decode_json_bytes, _iter_json_stream from polylogue.sources.dispatch import ( @@ -45,8 +45,6 @@ ImportSkippedRowPayload, ) -_SUPPORTED_ENTRY_SUFFIXES = (".json", ".jsonl", ".jsonl.txt", ".ndjson") - def explain_import_path( path: Path, @@ -548,26 +546,50 @@ def _explain_zip( ), _evidence("zip.container", matched=True, reason="ZIP container"), ] - aggregate_total = 0 try: with zipfile.ZipFile(path) as archive: - for info in archive.infolist(): - skip_reason, aggregate_total = _zip_entry_skip_reason( - info, - aggregate_total=aggregate_total, - zip_path=path, - provider_hint=provider_hint, + validator = ZipEntryValidator(provider_hint, cursor_state=None, zip_path=path) + + def record_rejection(info: zipfile.ZipInfo, reason: str) -> None: + skipped.append( + ImportSkippedRowPayload( + reason=reason, + source_path=f"{path}:{info.filename}", + ) ) - if skip_reason is not None: + + for info in validator.filter_entries(archive.infolist(), on_rejected=record_rejection): + path_classification = classify_artifact_path(info.filename, provider=provider_hint) + decoded_session_artifact: ArtifactClassification | None = None + if path_classification is not None and not path_classification.parse_as_session: + try: + decoded_session_artifact = zip_entry_session_artifact( + archive, + info, + provider=provider_hint, + ) + except ZipBombError as exc: + skipped.append( + ImportSkippedRowPayload( + reason=f"zip entry rejected: {exc}", + source_path=f"{path}:{info.filename}", + ) + ) + continue + if ( + path_classification is not None + and not path_classification.parse_as_session + and decoded_session_artifact is None + ): skipped.append( ImportSkippedRowPayload( - reason=skip_reason, + reason=path_classification.reason or "not a session artifact", source_path=f"{path}:{info.filename}", ) ) continue try: - with open_bounded_zip_entry(archive, info.filename) as handle: + with open_bounded_zip_entry(archive, info) as handle: entry = _explain_bytes( handle.read(MAX_UNCOMPRESSED_SIZE + 1), stream_name=info.filename, @@ -617,51 +639,6 @@ def _explain_zip( ) -def _zip_entry_skip_reason( - info: zipfile.ZipInfo, - *, - aggregate_total: int, - zip_path: Path, - provider_hint: Provider, -) -> tuple[str | None, int]: - """Return a skip reason (if any) plus the aggregate-total that should - carry forward to the next entry. - - Mirrors ``ZipEntryValidator.filter_entries`` in ``decoder_zip.py`` (which - ``process_zip`` always constructs with ``session_only=True``): an entry - that fails the extension/ratio/per-entry-size checks, or that classifies - as a non-session artifact (sidecar/metadata), never contributes to the - running aggregate total -- the real decode path only accumulates entries - that clear every earlier check AND are actually parsed as a session. The - aggregate check itself -- evaluated last, from central-directory metadata - alone -- is what decides whether *this* entry's size gets added to the - total that subsequent entries are checked against. - """ - if info.is_dir() or not info.filename.lower().endswith(_SUPPORTED_ENTRY_SUFFIXES): - return "unsupported ZIP entry", aggregate_total - if info.compress_size > 0 and (info.file_size / info.compress_size) > MAX_COMPRESSION_RATIO: - return f"zip entry compression ratio {info.file_size / info.compress_size:.1f} exceeds limit", aggregate_total - if info.file_size > MAX_UNCOMPRESSED_SIZE: - return f"zip entry file size {info.file_size} exceeds limit", aggregate_total - # Classify on the bare intra-archive relative path (matches - # ``ZipEntryValidator.filter_entries``'s identical fix, polylogue-dc1k): - # every ``OriginArtifactRule.path_pattern`` is anchored ``(?:^|/)``, so a - # ``{zip_path}:{name}`` prefix put a ``:`` immediately before the pattern - # instead of ``/``/start-of-string and no rule could ever match. - del zip_path - path_classification = classify_artifact_path(info.filename, provider=provider_hint) - if path_classification is not None and not path_classification.parse_as_session: - return path_classification.reason or "not a session artifact", aggregate_total - projected_total = aggregate_total + info.file_size - if projected_total > MAX_AGGREGATE_UNCOMPRESSED_SIZE: - return ( - f"aggregate uncompressed size {projected_total} exceeds archive-wide limit " - f"{MAX_AGGREGATE_UNCOMPRESSED_SIZE}", - aggregate_total, - ) - return None, projected_total - - def _explain_bytes( raw_bytes: bytes, *, diff --git a/polylogue/sources/import_preflight.py b/polylogue/sources/import_preflight.py index cd32f101bd..6492ceda32 100644 --- a/polylogue/sources/import_preflight.py +++ b/polylogue/sources/import_preflight.py @@ -18,11 +18,17 @@ from typing import Any from polylogue.core.enums import Provider +from polylogue.sources.decoder_zip import ( + MAX_UNCOMPRESSED_SIZE, + ZIP_JSON_SUFFIXES, + ZipBombError, + ZipEntryValidator, + open_bounded_zip_entry, +) from polylogue.sources.decoders import _decode_json_bytes, _iter_json_stream from polylogue.sources.dispatch import detect_provider _JSON_SUFFIXES = frozenset({".json", ".jsonl", ".ndjson"}) -_ZIP_JSON_SUFFIXES = (".json", ".jsonl", ".ndjson", ".jsonl.txt") _MAX_DIRECTORY_CANDIDATES = 256 _MAX_STREAM_RECORDS = 32 @@ -201,22 +207,35 @@ def _preflight_file(path: Path, acc: _PreflightAccumulator, *, label: str) -> No def _preflight_zip(path: Path, acc: _PreflightAccumulator, *, label: str) -> None: try: with zipfile.ZipFile(path) as zf: - json_entries = [ - info - for info in zf.infolist() - if not info.is_dir() and info.filename.lower().endswith(_ZIP_JSON_SUFFIXES) - ] - if not json_entries: - acc.unsupported(label, "ZIP contains no JSON or JSONL import candidates") - return - for info in json_entries: + validator = ZipEntryValidator("unknown", cursor_state=None, zip_path=path) + admitted = False + rejected = False + + def record_rejection(info: zipfile.ZipInfo, reason: str) -> None: + nonlocal rejected + rejected = True + acc.malformed( + f"{label}:{info.filename}", + f"ZIP entry rejected before read: {reason}", + ) + + for info in validator.filter_entries( + zf.infolist(), + allowed_suffixes=ZIP_JSON_SUFFIXES, + on_rejected=record_rejection, + ): + admitted = True entry_label = f"{label}:{info.filename}" try: - raw = zf.read(info) - except (OSError, zipfile.BadZipFile) as exc: + with open_bounded_zip_entry(zf, info) as handle: + raw = handle.read(MAX_UNCOMPRESSED_SIZE + 1) + except (OSError, KeyError, zipfile.BadZipFile, ZipBombError) as exc: acc.malformed(entry_label, f"could not read ZIP entry: {exc}") continue _preflight_json_bytes(raw, acc, label=entry_label) + + if not admitted and not rejected: + acc.unsupported(label, "ZIP contains no JSON or JSONL import candidates") except zipfile.BadZipFile as exc: acc.malformed(label, f"invalid ZIP archive: {exc}") except OSError as exc: diff --git a/polylogue/sources/live/append_ingest.py b/polylogue/sources/live/append_ingest.py index ce9dd62f18..91fcb4e231 100644 --- a/polylogue/sources/live/append_ingest.py +++ b/polylogue/sources/live/append_ingest.py @@ -9,6 +9,8 @@ from pathlib import Path from typing import Any, Protocol +from polylogue.archive.artifact_taxonomy import classify_artifact, classify_artifact_path +from polylogue.archive.raw_payload.decode import _sample_jsonl_payload_with_detail, jsonl_session_artifact from polylogue.archive.revision_authority import ( RawRevisionAuthority, RawRevisionEnvelope, @@ -79,7 +81,12 @@ def _ingest_append_plans_archive( t0 = time.perf_counter() from polylogue.sources.decoders import _iter_json_stream - from polylogue.sources.dispatch import parse_payload, require_positive_conversational_evidence + from polylogue.sources.dispatch import ( + STREAM_RECORD_PROVIDERS, + parse_payload, + parse_stream_payload, + require_positive_conversational_evidence, + ) from polylogue.sources.revision_backfill import ( _declared_non_session_artifact_classification, parse_retained_raw_sessions, @@ -100,21 +107,41 @@ def _ingest_append_plans_archive( for plan in plans: provider: Provider | None = None raw_id: str | None = None + session_artifact = None try: provider = Provider.from_string(plan.source_name) + path_artifact = classify_artifact_path( + str(plan.path), + provider=provider, + ) json_stream_started = time.perf_counter() try: - payloads = list(_iter_json_stream(BytesIO(plan.payload), plan.path.name)) + payloads, _malformed_lines, _malformed_detail = _sample_jsonl_payload_with_detail( + plan.payload, + max_samples=64, + jsonl_dict_only=True, + scan_full=False, + ) + session_artifact = jsonl_session_artifact( + plan.payload, + provider=provider, + jsonl_dict_only=True, + ) except Exception: # Preserve the pre-parse raw capture for malformed input; # the normal parser path below records the typed failure. payloads = None _add_timing(timings, "append.json_stream", json_stream_started) if payloads is not None: - classification = _declared_non_session_artifact_classification( - provider, - str(plan.path), - sample=payloads[:64], + decoded_artifact = session_artifact or classify_artifact(payloads, provider=provider) + classification = ( + _declared_non_session_artifact_classification( + provider, + str(plan.path), + sample=payloads[:64], + ) + if session_artifact is None and not decoded_artifact.parse_as_session + else None ) if classification is not None: artifact_result = archive.admit_raw_artifact_payload( @@ -129,6 +156,20 @@ def _ingest_append_plans_archive( raise RuntimeError(f"unexpected append artifact admission arm: {artifact_result.arm!r}") succeeded.append(plan) continue + elif path_artifact is not None and not path_artifact.parse_as_session: + artifact_result = archive.admit_raw_artifact_payload( + provider=provider, + payload=plan.payload, + source_path=str(plan.path), + source_index=-1, + acquired_at_ms=acquired_at_ms, + classification=path_artifact, + ) + if artifact_result.arm is not RawAdmissionArm.ARTIFACT: + raise RuntimeError(f"unexpected append artifact admission arm: {artifact_result.arm!r}") + raw_id = artifact_result.raw_id + succeeded.append(plan) + continue t0 = time.perf_counter() raw_id = archive.write_raw_payload( provider=provider, @@ -171,13 +212,22 @@ def _ingest_append_plans_archive( # ``fallback_id`` exactly when its own record stream # carries no session_meta of its own, which is always # true for an append delta. - sessions = require_positive_conversational_evidence( - parse_payload( + if provider in STREAM_RECORD_PROVIDERS: + parsed_sessions = parse_stream_payload( + provider, + _iter_json_stream(BytesIO(plan.payload), plan.path.name), + plan.native_id_hint or plan.path.stem, + source_path=str(plan.path), + ) + else: + parsed_sessions = parse_payload( provider, payloads, plan.native_id_hint or plan.path.stem, source_path=str(plan.path), - ), + ) + sessions = require_positive_conversational_evidence( + parsed_sessions, provider=provider, source_path=str(plan.path), ) diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 23133f1c44..674af7d42d 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -23,6 +23,7 @@ DOM_FALLBACK_INGEST_FLAG, NATIVE_BROWSER_CAPTURE_INGEST_FLAG, ) +from polylogue.archive.raw_payload.decode import jsonl_session_artifact from polylogue.archive.revision_authority import ( HISTORICAL_NON_PREFIX_GOVERNANCE_DETAIL, RawRevisionAuthority, @@ -57,6 +58,7 @@ success_disposition, ) from polylogue.pipeline.services.ingest_batch._models import _IngestBatchSummary +from polylogue.sources.decoder_zip import ZipBombError, open_bounded_zip_entry from polylogue.sources.decoders import _iter_json_stream, _ZipEntryValidator from polylogue.sources.dispatch import ( _detect_provider_from_raw_bytes, @@ -308,6 +310,21 @@ def _iso_to_epoch_ms(value: str) -> int: return int(datetime.fromisoformat(value).timestamp() * 1000) +def _blob_jsonl_has_session_evidence( + blob_store: BlobStore, + blob_hash: str, + *, + provider: Provider, + source_path: str, +) -> bool: + if Path(source_path).suffix.lower() != ".jsonl": + return False + try: + return jsonl_session_artifact(blob_store.blob_path(blob_hash), provider=provider) is not None + except (OSError, ValueError): + return False + + def _live_parse_stage_candidates(paths: list[Path], *, fallback_provider: Provider) -> list[LiveParseCandidate]: """Select and read eligible files for off-writer-hold pre-parse (polylogue-wf8a). @@ -2150,7 +2167,21 @@ def _ingest_full_records_archive( provider, record.source_path, ) - if artifact_classification is not None: + session_evidence = ( + _blob_jsonl_has_session_evidence( + blob_store, + blob_hash, + provider=provider, + source_path=record.source_path, + ) + if payload is None + else _parse_payload_as_session_artifact( + Path(record.source_path), + provider=provider, + payload=payload, + ) + ) + if artifact_classification is not None and not session_evidence: explicit_raw_id = record.raw_id if record.blob_hash is not None else None if payload is None: source_raw_id = archive.admit_raw_artifact_blob_ref( @@ -2443,7 +2474,12 @@ def _ingest_full_records_archive( ) plan = archive.classify_raw_revision_cohort_for_live_watch(logical_source_key) if plan.accepted_raw_ids: - parsed_by_raw_id = self._parse_raw_revision_chain(archive, plan) + parsed_by_raw_id = self._parse_raw_revision_chain( + archive, + plan, + current_raw_id=source_raw_id, + current_session=session, + ) session_id, applied_raw_ids = archive.apply_raw_revision_replay( plan, parsed_by_raw_id, @@ -2625,10 +2661,21 @@ def _ingest_full_records_archive( ) return result - def _parse_raw_revision_chain(self, archive: Any, plan: Any) -> dict[str, Any]: + def _parse_raw_revision_chain( + self, + archive: Any, + plan: Any, + *, + current_raw_id: str | None = None, + current_session: ParsedSession | None = None, + ) -> dict[str, Any]: parsed_by_raw_id: dict[str, Any] = {} for raw_id in plan.accepted_raw_ids: - sessions = self._parse_retained_raw_sessions(archive, raw_id) + sessions = ( + [current_session] + if raw_id == current_raw_id and current_session is not None + else self._parse_retained_raw_sessions(archive, raw_id) + ) if len(sessions) != 1: raise RuntimeError(f"raw revision {raw_id} did not replay to exactly one session") parsed_by_raw_id[raw_id] = sessions[0] @@ -2827,7 +2874,6 @@ def _extract_zip_member_records( fallback_provider, cursor_state=None, zip_path=path, - session_only=False, ) try: with zipfile.ZipFile(path) as zf: @@ -2850,43 +2896,46 @@ def _extract_zip_member_records( for info in entries: if info.file_size == 0: continue - for raw_data in iter_zip_entry_raw_data( - zf, - ZipEntryReadContext( - source=source, - zip_path=path, - entry=info, - file_mtime=file_mtime, - provider_hint=zip_provider_hint, - blob_store=blob_store, - ), - ): - if raw_data.blob_hash is None: - continue - member_provider = raw_data.provider_hint or fallback_provider - member_size = raw_data.blob_size or 0 - total_bytes += member_size - records.append( - ( - raw_data.blob_hash, - RawSessionRecord( - raw_id=raw_data.blob_hash, - payload_provider=member_provider, - capture_mode=( - fallback_provider - if fallback_provider is not Provider.UNKNOWN - else member_provider + try: + for raw_data in iter_zip_entry_raw_data( + zf, + ZipEntryReadContext( + source=source, + zip_path=path, + entry=info, + file_mtime=file_mtime, + provider_hint=zip_provider_hint, + blob_store=blob_store, + ), + ): + if raw_data.blob_hash is None: + continue + member_provider = raw_data.provider_hint or fallback_provider + member_size = raw_data.blob_size or 0 + total_bytes += member_size + records.append( + ( + raw_data.blob_hash, + RawSessionRecord( + raw_id=raw_data.blob_hash, + payload_provider=member_provider, + capture_mode=( + fallback_provider + if fallback_provider is not Provider.UNKNOWN + else member_provider + ), + source_name=member_provider.value, + source_path=raw_data.source_path, + source_index=raw_data.source_index or 0, + blob_size=member_size, + blob_publication_receipt_id=raw_data.blob_publication_receipt_id, + acquired_at=acquired_at, + file_mtime=raw_data.file_mtime, ), - source_name=member_provider.value, - source_path=raw_data.source_path, - source_index=raw_data.source_index or 0, - blob_size=member_size, - blob_publication_receipt_id=raw_data.blob_publication_receipt_id, - acquired_at=acquired_at, - file_mtime=raw_data.file_mtime, - ), + ) ) - ) + except ZipBombError as exc: + logger.warning("Skipping ZIP member %s in %s: %s", info.filename, path, exc) except (zipfile.BadZipFile, OSError) as exc: logger.warning("Failed to expand inbox ZIP %s: %s", path, exc) return [], 0 @@ -2912,9 +2961,9 @@ def _sniff_zip_provider( if not name_lower.endswith((".json", ".jsonl", ".jsonl.txt", ".ndjson")): continue try: - with zf.open(info.filename) as handle: + with open_bounded_zip_entry(zf, info) as handle: prefix = handle.read(_DETECTION_PREFIX_SIZE) - except (zipfile.BadZipFile, OSError): + except (zipfile.BadZipFile, OSError, ZipBombError): continue if not prefix: continue diff --git a/polylogue/sources/live/batch_support.py b/polylogue/sources/live/batch_support.py index ee15bca128..cd4ff8efe7 100644 --- a/polylogue/sources/live/batch_support.py +++ b/polylogue/sources/live/batch_support.py @@ -7,13 +7,13 @@ import time from collections.abc import Callable, Iterable from dataclasses import dataclass, field -from io import BytesIO from pathlib import Path from typing import Protocol import ijson from polylogue.archive.artifact_taxonomy import classify_artifact, classify_artifact_path +from polylogue.archive.raw_payload.decode import jsonl_session_artifact from polylogue.core.enums import Provider from polylogue.core.json import JSONDecodeError, JSONValue from polylogue.core.json import loads as json_loads @@ -577,12 +577,12 @@ def _jsonl_provider_and_session_artifact( ) -> tuple[Provider, bool]: records = _jsonl_sample_from_path(path) provider = (detect_provider(records) if records else None) or fallback_provider + if jsonl_session_artifact(path, provider=provider) is not None: + return provider, True path_classification = classify_artifact_path(path, provider=provider) if path_classification is not None: return provider, path_classification.parse_as_session - if not records: - return provider, False - return provider, classify_artifact(records, provider=provider, source_path=path).parse_as_session + return provider, False def _parse_path_as_session_artifact(path: Path, *, provider: Provider) -> bool: @@ -591,14 +591,14 @@ def _parse_path_as_session_artifact(path: Path, *, provider: Provider) -> bool: or hermes_verification.looks_like_verification_evidence_db_path(path) ): return True + if path.suffix.lower() == ".jsonl": + if jsonl_session_artifact(path, provider=provider) is not None: + return True + path_classification = classify_artifact_path(path, provider=provider) + return path_classification.parse_as_session if path_classification is not None else False path_classification = classify_artifact_path(path, provider=provider) if path_classification is not None: return path_classification.parse_as_session - if path.suffix.lower() == ".jsonl": - records = _jsonl_sample_from_path(path) - if not records: - return False - return classify_artifact(records, provider=provider, source_path=path).parse_as_session if _path_size(path) > _STREAMING_FULL_INGEST_BYTES: browser_capture, _browser_provider = _browser_capture_prefix_probe(path) if browser_capture: @@ -638,24 +638,14 @@ def _parse_payload_as_session_artifact(path: Path, *, provider: Provider, payloa return hermes_state.looks_like_state_db_path( path ) or hermes_verification.looks_like_verification_evidence_db_path(path) + if path.suffix.lower() == ".jsonl": + if jsonl_session_artifact(payload, provider=provider) is not None: + return True + path_classification = classify_artifact_path(path, provider=provider) + return path_classification.parse_as_session if path_classification is not None else False path_classification = classify_artifact_path(path, provider=provider) if path_classification is not None: return path_classification.parse_as_session - if path.suffix.lower() == ".jsonl": - records: list[JSONValue] = [] - for line in BytesIO(payload): - if len(records) >= 32: - break - raw = line.strip() - if not raw: - continue - try: - records.append(json_loads(raw)) - except JSONDecodeError: - continue - if not records: - return False - return classify_artifact(records, provider=provider, source_path=path).parse_as_session try: document = json_loads(payload) except JSONDecodeError: diff --git a/polylogue/sources/source_acquisition.py b/polylogue/sources/source_acquisition.py index 2fe920e138..0252851d41 100644 --- a/polylogue/sources/source_acquisition.py +++ b/polylogue/sources/source_acquisition.py @@ -107,7 +107,6 @@ def iter_source_raw_data( provider_hint, cursor_state=cursor_state, zip_path=path, - session_only=False, ) with zipfile.ZipFile(path) as zf: for info in validator.filter_entries(zf.infolist()): diff --git a/polylogue/sources/source_acquisition_components.py b/polylogue/sources/source_acquisition_components.py index 434ca99044..f3280040ff 100644 --- a/polylogue/sources/source_acquisition_components.py +++ b/polylogue/sources/source_acquisition_components.py @@ -421,7 +421,7 @@ def _stream_preserved_zip_entry( *, provider_hint: Provider, ) -> RawSessionData: - with zf.open(context.entry.filename) as handle: + with _decoders.open_bounded_zip_entry(zf, context.entry) as handle: blob_hash, blob_size = stream_fileobj_to_blob( context.blob_store, handle, @@ -462,7 +462,7 @@ def iter_zip_entry_raw_data( detected_provider = entry_provider_hint split_buffer = SplitPayloadBuffer() - with zf.open(context.entry.filename) as handle: + with _decoders.open_bounded_zip_entry(zf, context.entry) as handle: for detected in iter_entry_payloads( handle, stream_name=context.entry.filename, diff --git a/polylogue/sources/source_parsing.py b/polylogue/sources/source_parsing.py index 9c8c445b0f..21418ebb69 100644 --- a/polylogue/sources/source_parsing.py +++ b/polylogue/sources/source_parsing.py @@ -2,14 +2,16 @@ from __future__ import annotations -import json import zipfile from collections.abc import Iterable from pathlib import Path -from polylogue.archive.artifact_taxonomy import classify_artifact_path +from polylogue.archive.artifact_taxonomy import classify_artifact, classify_artifact_path +from polylogue.archive.raw_payload.decode import jsonl_session_artifact from polylogue.config import Source from polylogue.core.enums import Provider +from polylogue.core.json import JSONDecodeError +from polylogue.core.json import loads as json_loads from polylogue.logging import get_logger from polylogue.sources.assembly import SidecarData from polylogue.storage.blob_store import BlobStore @@ -31,6 +33,20 @@ _decoders.logger = logger +def has_decoded_session_evidence(path: Path, *, provider: Provider) -> bool: + """Return whether decoded JSON content outranks a non-session path rule.""" + if path.suffix.lower() == ".jsonl": + return jsonl_session_artifact(path, provider=provider) is not None + + if path.suffix.lower() != ".json": + return False + try: + document = json_loads(path.read_bytes()) + except (JSONDecodeError, OSError): + return False + return classify_artifact(document, provider=provider).parse_as_session + + def iter_antigravity_language_server_sessions( source: Source, *, @@ -186,7 +202,11 @@ def parse_one_source_path( path = Path(path_str) provider_hint = Provider.from_string(source_name) path_classification = classify_artifact_path(path, provider=source_name) - if path_classification is not None and not path_classification.parse_as_session: + if ( + path_classification is not None + and not path_classification.parse_as_session + and not has_decoded_session_evidence(path, provider=provider_hint) + ): return should_group = provider_hint in _GROUP_PROVIDERS @@ -378,7 +398,7 @@ def iter_source_sessions_with_raw( str(path), f"File not found (may have been deleted): {exc}", ) - except (json.JSONDecodeError, UnicodeDecodeError, zipfile.BadZipFile) as exc: + except (JSONDecodeError, UnicodeDecodeError, zipfile.BadZipFile) as exc: failed_count += 1 logger.warning("Failed to parse %s: %s", path, exc) _record_cursor_failure(cursor_state, str(path), str(exc)) @@ -400,5 +420,6 @@ def iter_source_sessions_with_raw( "iter_antigravity_language_server_sessions", "iter_source_sessions", "iter_source_sessions_with_raw", + "has_decoded_session_evidence", "parse_one_source_path", ] diff --git a/polylogue/storage/blob_integrity.py b/polylogue/storage/blob_integrity.py index 47a43949d1..b8690e417f 100644 --- a/polylogue/storage/blob_integrity.py +++ b/polylogue/storage/blob_integrity.py @@ -23,6 +23,13 @@ from pathlib import Path from typing import Any, Literal +from polylogue.archive.zip_admission import ( + MAX_UNCOMPRESSED_SIZE, + ZIP_JSON_SUFFIXES, + ZipAdmission, + ZipBombError, + open_bounded_zip_entry, +) from polylogue.core.json import JSONDecodeError as CoreJSONDecodeError from polylogue.core.json import dumps_bytes as json_dumps_bytes from polylogue.core.json import loads as json_loads @@ -1117,12 +1124,23 @@ def _current_raw_payload_bytes( if source_bytes_cache is not None and source_path in source_bytes_cache: member_bytes = source_bytes_cache[source_path] else: - with zipfile.ZipFile(zip_path) as archive, archive.open(member) as handle: - member_bytes = handle.read() + with zipfile.ZipFile(zip_path) as archive: + matching = [info for info in archive.infolist() if info.filename == member] + if len(matching) != 1: + return None, "ambiguous_container_member" + admitted = list( + ZipAdmission(zip_path=zip_path).filter_entries(matching, allowed_suffixes=ZIP_JSON_SUFFIXES) + ) + if len(admitted) != 1: + return None, "container_member_rejected" + with open_bounded_zip_entry(archive, admitted[0]) as handle: + member_bytes = handle.read(MAX_UNCOMPRESSED_SIZE + 1) if source_bytes_cache is not None: source_bytes_cache[source_path] = member_bytes except KeyError: return None, "source_missing" + except ZipBombError: + return None, "container_member_rejected" if source_index is None: return None, "source_index_missing" try: diff --git a/tests/integration/test_claude_workflow_admission.py b/tests/integration/test_claude_workflow_admission.py index 2c65379f3f..d304437dda 100644 --- a/tests/integration/test_claude_workflow_admission.py +++ b/tests/integration/test_claude_workflow_admission.py @@ -22,6 +22,7 @@ import pytest +from polylogue.archive.artifact_taxonomy import classify_artifact_path from polylogue.config import Source from polylogue.core.enums import Provider from polylogue.insights.claude_workflow_materializer import ( @@ -30,7 +31,9 @@ materialize_claude_workflow_archive, ) from polylogue.pipeline.services.archive_ingest import parse_sources_archive +from polylogue.storage.blob_store import BlobStore from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root RUN_ID = "wf_54d4fb2e-841" ATTEMPT_COUNT = 91 @@ -259,6 +262,56 @@ async def test_configured_claude_workflow_admission_preserves_raw_revisions_and_ assert any("missing paired agent metadata sidecar" in gap for gap in degraded.gaps) +def test_materializer_streams_large_jsonl_evidence_before_inventory_read( + workspace_env: dict[str, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + """Inventory repair must detect delayed sessions without ``read_all``.""" + archive_root = workspace_env["archive_root"] + initialize_active_archive_root(archive_root) + source_path = workspace_env["data_root"] / ".claude/projects/project/subagents/workflows/wf-large/journal.jsonl" + payload = ( + b'{"contentKey":"' + + b"x" * (2 * 1024 * 1024) + + b'","agentId":"workflow-agent"}\n' + + b"".join( + b'{"contentKey":"artifact-' + str(index).encode() + b'","agentId":"workflow-agent"}\n' + for index in range(1, 64) + ) + + b'{"sessionId":"late-session","parentUuid":null,"type":"user",' + b'"message":{"role":"user","content":"recover this session"},' + b'"uuid":"late-user","timestamp":"2025-01-01T00:00:00Z"}\n' + b'{"sessionId":"late-session","parentUuid":"late-user","type":"assistant",' + b'"message":{"role":"assistant","content":[{"type":"text","text":"recovered"}]},' + b'"uuid":"late-assistant","timestamp":"2025-01-01T00:00:01Z"}\n' + ) + classification = classify_artifact_path(str(source_path), provider=Provider.CLAUDE_CODE) + assert classification is not None and not classification.parse_as_session + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + archive.admit_raw_artifact_payload( + provider=Provider.CLAUDE_CODE, + payload=payload, + source_path=str(source_path), + source_index=0, + acquired_at_ms=2_000_000_000_000, + classification=classification, + ) + + original_read_all = BlobStore.read_all + + def reject_large_read(self: BlobStore, hash_hex: str) -> bytes: + if self.blob_path(hash_hex).stat().st_size > 1024: + raise AssertionError("materializer must detect large JSONL sessions before BlobStore.read_all") + return original_read_all(self, hash_hex) + + monkeypatch.setattr(BlobStore, "read_all", reject_large_read) + + summary = materialize_claude_workflow_archive(archive_root) + + assert summary.current_artifact_count == 0 + with sqlite3.connect(archive_root / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_artifacts").fetchone() == (0,) + + @pytest.mark.asyncio async def test_claude_workflow_convergence_stage_surfaces_gap_through_readiness( workspace_env: dict[str, Path], diff --git a/tests/unit/cli/test_import_explain.py b/tests/unit/cli/test_import_explain.py index 38aae77ea7..91a2674297 100644 --- a/tests/unit/cli/test_import_explain.py +++ b/tests/unit/cli/test_import_explain.py @@ -8,9 +8,9 @@ from click.testing import CliRunner from pytest import MonkeyPatch +from polylogue.archive import zip_admission as zip_admission_module from polylogue.cli.click_app import cli from polylogue.core.enums import Provider -from polylogue.sources import decoder_zip as decoder_zip_module from polylogue.sources import import_explain as import_explain_module from polylogue.sources.decoder_zip import ZipEntryValidator from polylogue.sources.import_explain import explain_import_path @@ -112,6 +112,39 @@ def test_import_explain_zip_propagates_member_decode_skip(tmp_path: Path) -> Non assert payload.skipped[0].reason.startswith("decode failure:") +def test_import_explain_zip_recovers_path_classified_json_record_array(tmp_path: Path) -> None: + """Explain applies decoded-session evidence before a workflow path skip.""" + archive = tmp_path / "workflow-json.zip" + records = [ + { + "sessionId": "explain-json-session", + "parentUuid": None, + "type": "user", + "message": {"role": "user", "content": "explain this session"}, + "uuid": "explain-json-user", + "timestamp": "2025-01-01T00:00:00Z", + }, + { + "sessionId": "explain-json-session", + "parentUuid": "explain-json-user", + "type": "assistant", + "message": {"role": "assistant", "content": [{"type": "text", "text": "explained"}]}, + "uuid": "explain-json-assistant", + "timestamp": "2025-01-01T00:00:01Z", + }, + ] + with zipfile.ZipFile(archive, "w") as zf: + zf.writestr("workflows/explain.json", json.dumps(records)) + + payload = explain_import_path(archive, source_name="claude-code") + + assert payload.produced.sessions >= 1 + assert not any( + row.source_path and row.source_path.endswith("workflow-json.zip:workflows/explain.json") + for row in payload.skipped + ) + + def test_import_explain_zip_rejects_oversized_member_before_read( tmp_path: Path, monkeypatch: MonkeyPatch, @@ -120,6 +153,7 @@ def test_import_explain_zip_rejects_oversized_member_before_read( with zipfile.ZipFile(archive, "w") as zf: zf.writestr("big.json", b"{}") monkeypatch.setattr(import_explain_module, "MAX_UNCOMPRESSED_SIZE", 1) + monkeypatch.setattr(zip_admission_module, "MAX_UNCOMPRESSED_SIZE", 1) payload = explain_import_path(archive) @@ -148,8 +182,7 @@ def test_import_explain_zip_rejects_aggregate_over_cap_before_read( with zipfile.ZipFile(archive, "w") as zf: for name in entry_names: zf.writestr(name, entry_bytes) - monkeypatch.setattr(import_explain_module, "MAX_AGGREGATE_UNCOMPRESSED_SIZE", len(entry_bytes)) - monkeypatch.setattr(decoder_zip_module, "MAX_AGGREGATE_UNCOMPRESSED_SIZE", len(entry_bytes)) + monkeypatch.setattr(zip_admission_module, "MAX_AGGREGATE_UNCOMPRESSED_SIZE", len(entry_bytes)) payload = explain_import_path(archive) @@ -169,71 +202,53 @@ def test_import_explain_zip_rejects_aggregate_over_cap_before_read( assert rejected_by_preview == set(entry_names) - accepted_names -def test_import_explain_zip_excludes_non_session_artifact_from_aggregate( +def test_import_explain_zip_aggregate_admission_precedes_path_session_decode( tmp_path: Path, monkeypatch: MonkeyPatch, ) -> None: - """A non-session-classified entry must not count toward the preview's - aggregate total (CodeRabbit finding on PR #3317): ``process_zip`` always - constructs ``ZipEntryValidator`` with ``session_only=True``, which - excludes non-session-classified entries from the running total entirely - (they ``continue`` before the aggregate check in ``decoder_zip.py``). - The preview must apply the identical exclusion, or it can wrongly - predict an aggregate-cap rejection a real import would never hit. - - Uses a monkeypatched ``classify_artifact_path`` (isolating the exclusion - LOGIC in ``_zip_entry_skip_reason`` from real ``OriginArtifactRule`` - matching, which is covered separately) matching on the bare intra-archive - relative path -- both ``_zip_entry_skip_reason`` and - ``ZipEntryValidator.filter_entries`` classify on that bare path, not a - ``{zip_path}:{name}`` prefix (polylogue-dc1k: every rule's ``(?:^|/)``-anchored - pattern only matches after start-of-string or ``/``, never after the ``:`` - a container prefix would insert). - """ - from polylogue.archive.artifact_taxonomy.models import ArtifactClassification, ArtifactKind + """Aggregate admission rejects a later session-shaped member before decode. + The tiny cap stands in for the production 64 GiB aggregate ceiling. The + central-directory sizes are enough to exercise admission, so this test + does not allocate a hostile payload. + """ archive = tmp_path / "workflow.zip" - session_bytes = b'{"a": 1}' - non_session_bytes = b'{"run": "snapshot"}' * 1000 + first_bytes = b"{}" + later_session_bytes = json.dumps( + [ + { + "sessionId": "later-session", + "type": "user", + "uuid": "later-user", + "message": {"role": "user", "content": "later"}, + } + ] + ).encode() with zipfile.ZipFile(archive, "w") as zf: - zf.writestr("session.json", session_bytes) - zf.writestr("run.json", non_session_bytes) - # Cap sits between the session entry alone and session+non-session - # combined -- if the non-session entry wrongly counted, this would - # falsely reject the accepted session entry too. - monkeypatch.setattr( - import_explain_module, - "MAX_AGGREGATE_UNCOMPRESSED_SIZE", - len(session_bytes) + len(non_session_bytes) // 2, - ) + zf.writestr("safe.json", first_bytes) + zf.writestr("workflows/later.json", later_session_bytes) + monkeypatch.setattr(zip_admission_module, "MAX_AGGREGATE_UNCOMPRESSED_SIZE", len(first_bytes)) - def fake_classify(source_path: object, *, provider: object) -> ArtifactClassification | None: - if str(source_path) == "run.json": - return ArtifactClassification( - provider=Provider.CLAUDE_CODE, - kind=ArtifactKind.WORKFLOW_RUN_SNAPSHOT, - parse_as_session=False, - schema_eligible=False, - default_priority=0, - reason="non-session workflow snapshot (test fixture)", - ) - return None + decoded_members: list[str] = [] - monkeypatch.setattr(import_explain_module, "classify_artifact_path", fake_classify) + def fail_if_later_member_decoded( + _archive: zipfile.ZipFile, + info: zipfile.ZipInfo, + *, + provider: Provider, + ) -> object: + del provider + decoded_members.append(info.filename) + raise AssertionError(f"aggregate admission must reject {info.filename} before decode") + + monkeypatch.setattr(import_explain_module, "zip_entry_session_artifact", fail_if_later_member_decoded) payload = explain_import_path(archive, source_name="claude-code") - assert not any("aggregate uncompressed size" in row.reason for row in payload.skipped) - non_session_skips = [row for row in payload.skipped if row.source_path == f"{archive}:run.json"] - assert len(non_session_skips) == 1 - assert non_session_skips[0].reason == "non-session workflow snapshot (test fixture)" - # session.json is separately skipped as "metadata-oriented document" (its - # trivial fixture bytes aren't a real session shape) -- but crucially - # NOT for an aggregate-size reason, which is the only thing this test - # proves: run.json's bytes never reached the running aggregate total. - session_skips = [row for row in payload.skipped if row.source_path == f"{archive}:session.json"] - assert len(session_skips) == 1 - assert "aggregate uncompressed size" not in session_skips[0].reason + assert decoded_members == [] + assert payload.produced.sessions == 0 + aggregate_skips = [row for row in payload.skipped if "aggregate uncompressed size" in row.reason] + assert [row.source_path for row in aggregate_skips] == [f"{archive}:workflows/later.json"] def test_import_explain_zip_allows_archive_comfortably_under_aggregate_cap( @@ -248,7 +263,7 @@ def test_import_explain_zip_allows_archive_comfortably_under_aggregate_cap( with zipfile.ZipFile(archive, "w") as zf: for i in range(3): zf.writestr(f"entry_{i}.json", entry_bytes) - monkeypatch.setattr(import_explain_module, "MAX_AGGREGATE_UNCOMPRESSED_SIZE", len(entry_bytes) * 10) + monkeypatch.setattr(zip_admission_module, "MAX_AGGREGATE_UNCOMPRESSED_SIZE", len(entry_bytes) * 10) payload = explain_import_path(archive) diff --git a/tests/unit/pipeline/test_archive_ingest_shared_raw.py b/tests/unit/pipeline/test_archive_ingest_shared_raw.py index 6b183e60ed..775aa6a305 100644 --- a/tests/unit/pipeline/test_archive_ingest_shared_raw.py +++ b/tests/unit/pipeline/test_archive_ingest_shared_raw.py @@ -36,10 +36,12 @@ from __future__ import annotations +import json import sqlite3 +import zipfile from collections.abc import Iterator from pathlib import Path -from typing import Any +from typing import Any, cast import pytest @@ -51,6 +53,25 @@ from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +class _RejectUnboundedRead: + """ZIP handle proxy that rejects a full member read in archive admission.""" + + def __init__(self, handle: Any) -> None: + self._handle = handle + + def __enter__(self) -> _RejectUnboundedRead: + self._handle.__enter__() + return self + + def __exit__(self, *args: object) -> None: + self._handle.__exit__(*args) + + def read(self, size: int = -1) -> bytes: + if size < 0: + raise AssertionError("ZIP artifact admission must stream to the blob store") + return cast("bytes", self._handle.read(size)) + + def _write_carryover_chain(root: Path, *, session_prefix: str = "") -> tuple[Path, Path]: """Write parent-session.jsonl (real "parent" session) + child-session.jsonl (a 1-record carryover of parent's tail under `sessionId=parent-session`, @@ -134,6 +155,258 @@ def _membership_rows(source_db: Path, raw_id: str) -> set[tuple[str, str]]: return {(str(row[0]), str(row[1])) for row in rows} +def _workflow_journal_payload(*, malformed: bool = False, delayed: bool = False) -> bytes: + if malformed: + return b'{"contentKey":"broken"\n' + prefix = b"" + if delayed: + prefix = b"".join( + b'{"contentKey":"artifact-' + str(index).encode() + b'","agentId":"workflow-agent"}\n' + for index in range(32) + ) + return prefix + ( + b'{"sessionId":"journal-session","parentUuid":null,"type":"user",' + b'"message":{"role":"user","content":[{"type":"text","text":"recover journal"}]},' + b'"uuid":"journal-user","timestamp":"2025-01-01T00:00:00Z"}\n' + b'{"sessionId":"journal-session","parentUuid":"journal-user","type":"assistant",' + b'"message":{"role":"assistant",' + b'"content":[{"type":"text","text":"repaired reply"}]},"uuid":"journal-assistant",' + b'"timestamp":"2025-01-01T00:00:01Z"}\n' + ) + + +def _write_session_shaped_workflow_journal(root: Path, *, malformed: bool = False) -> Path: + journal = root / "subagents" / "workflows" / "wf-archive" / "journal.jsonl" + journal.parent.mkdir(parents=True) + journal.write_bytes(_workflow_journal_payload(malformed=malformed)) + return journal + + +def _write_workflow_journal_zip(root: Path, *, malformed: bool = False) -> Path: + archive = root / "claude-export.zip" + archive.parent.mkdir(parents=True) + with zipfile.ZipFile(archive, "w") as zf: + zf.writestr( + "subagents/workflows/wf-archive/journal.jsonl", + _workflow_journal_payload(malformed=malformed, delayed=not malformed), + ) + return archive + + +def _write_large_zip_member(root: Path, name: str, payload: bytes) -> Path: + archive = root / "large-export.zip" + archive.parent.mkdir(parents=True) + with zipfile.ZipFile(archive, "w", compression=zipfile.ZIP_STORED) as zf: + zf.writestr(name, payload) + return archive + + +@pytest.mark.asyncio +async def test_archive_ingest_session_shaped_workflow_journal_reaches_parser_idempotently( + tmp_path: Path, workspace_env: dict[str, Path] +) -> None: + """The production one-shot route must decode a journal before path exclusion.""" + archive_root = workspace_env["archive_root"] + journal = _write_session_shaped_workflow_journal(tmp_path / "sessions") + sources = [Source(name="claude-code", path=journal)] + + first = await parse_sources_archive(archive_root, sources, parse_workers=1) + second = await parse_sources_archive(archive_root, sources, parse_workers=1) + + assert first.parse_failures == 0 + assert first.counts["sessions"] == 1 + assert second.parse_failures == 0 + with sqlite3.connect(archive_root / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone() == (1,) + assert conn.execute("SELECT COUNT(*) FROM raw_artifacts").fetchone() == (0,) + with sqlite3.connect(archive_root / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (1,) + + +@pytest.mark.asyncio +async def test_archive_ingest_malformed_workflow_journal_remains_typed_evidence( + tmp_path: Path, workspace_env: dict[str, Path] +) -> None: + """A journal with no decodable session evidence remains a typed artifact.""" + archive_root = workspace_env["archive_root"] + journal = _write_session_shaped_workflow_journal(tmp_path / "sessions", malformed=True) + + result = await parse_sources_archive( + archive_root, + [Source(name="claude-code", path=journal)], + parse_workers=1, + ) + + assert result.parse_failures == 0 + with sqlite3.connect(archive_root / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone() == (1,) + assert conn.execute("SELECT artifact_kind, parse_as_session FROM raw_artifacts").fetchone() == ( + "workflow_journal", + 0, + ) + with sqlite3.connect(archive_root / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (0,) + + +@pytest.mark.asyncio +async def test_archive_ingest_zip_workflow_journal_scans_delayed_session_evidence_idempotently( + tmp_path: Path, workspace_env: dict[str, Path] +) -> None: + """ZIP member routing must decode beyond 32 artifact records before exclusion.""" + archive_root = workspace_env["archive_root"] + journal_zip = _write_workflow_journal_zip(tmp_path / "sessions") + sources = [Source(name="claude-code", path=journal_zip)] + + first = await parse_sources_archive(archive_root, sources, parse_workers=1) + second = await parse_sources_archive(archive_root, sources, parse_workers=1) + + assert first.parse_failures == 0 + assert first.counts["sessions"] == 1 + assert second.parse_failures == 0 + with sqlite3.connect(archive_root / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone() == (1,) + assert conn.execute("SELECT COUNT(*) FROM raw_artifacts").fetchone() == (0,) + with sqlite3.connect(archive_root / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (1,) + + +@pytest.mark.asyncio +async def test_archive_ingest_malformed_zip_workflow_journal_remains_typed_evidence( + tmp_path: Path, workspace_env: dict[str, Path] +) -> None: + """Malformed ZIP journals are retained as typed evidence without sessions.""" + archive_root = workspace_env["archive_root"] + journal_zip = _write_workflow_journal_zip(tmp_path / "sessions", malformed=True) + + result = await parse_sources_archive( + archive_root, + [Source(name="claude-code", path=journal_zip)], + parse_workers=1, + ) + + assert result.parse_failures == 0 + with sqlite3.connect(archive_root / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone() == (1,) + assert conn.execute("SELECT artifact_kind, parse_as_session FROM raw_artifacts").fetchone() == ( + "workflow_journal", + 0, + ) + with sqlite3.connect(archive_root / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (0,) + + +@pytest.mark.asyncio +async def test_archive_ingest_large_zip_artifact_streams_to_blob_reference( + tmp_path: Path, workspace_env: dict[str, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + """A large ZIP journal artifact must not be read into an admission payload.""" + from polylogue.sources.decoder_zip import _ZIP_READ_CHUNK_SIZE, MAX_UNCOMPRESSED_SIZE, open_bounded_zip_entry + + archive_root = workspace_env["archive_root"] + payload = b'{"contentKey":"artifact","agentId":"workflow-agent","body":"' + b"x" * _ZIP_READ_CHUNK_SIZE + b'"}\n' + journal_zip = _write_large_zip_member( + tmp_path / "sessions", + "subagents/workflows/wf-archive/journal.jsonl", + payload, + ) + original_open = open_bounded_zip_entry + + def reject_unbounded_read( + zf: zipfile.ZipFile, + info: zipfile.ZipInfo, + *, + max_bytes: int = MAX_UNCOMPRESSED_SIZE, + ) -> _RejectUnboundedRead: + return _RejectUnboundedRead(original_open(zf, info, max_bytes=max_bytes)) + + monkeypatch.setattr( + "polylogue.pipeline.services.archive_ingest.open_bounded_zip_entry", + reject_unbounded_read, + ) + + result = await parse_sources_archive(archive_root, [Source(name="claude-code", path=journal_zip)], parse_workers=1) + + assert result.parse_failures == 0 + with sqlite3.connect(archive_root / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_artifacts").fetchone() == (1,) + assert conn.execute("SELECT blob_size FROM raw_sessions").fetchone() == (len(payload),) + + +@pytest.mark.asyncio +async def test_archive_ingest_large_ordinary_zip_jsonl_skips_delayed_artifact_scan( + tmp_path: Path, workspace_env: dict[str, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + """Unclassified ZIP JSONL follows normal parsing without a second full scan.""" + from polylogue.sources import decoder_zip + + archive_root = workspace_env["archive_root"] + payload = ( + b'{"sessionId":"ordinary-session","parentUuid":null,"type":"user",' + b'"message":{"role":"user","content":[{"type":"text","text":"' + b"x" * (1024 * 1024) + b'"}]},' + b'"uuid":"ordinary-user","timestamp":"2025-01-01T00:00:00Z"}\n' + b'{"sessionId":"ordinary-session","parentUuid":"ordinary-user","type":"assistant",' + b'"message":{"role":"assistant","content":[{"type":"text","text":"reply"}]},' + b'"uuid":"ordinary-assistant","timestamp":"2025-01-01T00:00:01Z"}\n' + ) + session_zip = _write_large_zip_member(tmp_path / "sessions", "nested/ordinary.jsonl", payload) + + def fail_unexpected_scan(*args: object, **kwargs: object) -> None: + raise AssertionError("ordinary ZIP JSONL must not receive a delayed artifact scan") + + monkeypatch.setattr(decoder_zip, "zip_entry_session_artifact", fail_unexpected_scan) + + result = await parse_sources_archive(archive_root, [Source(name="claude-code", path=session_zip)], parse_workers=1) + + assert result.parse_failures == 0 + assert result.counts["sessions"] == 1 + + +@pytest.mark.asyncio +async def test_archive_ingest_path_classified_zip_json_record_array_reaches_parser( + tmp_path: Path, workspace_env: dict[str, Path] +) -> None: + """Decoded Claude records outrank a non-session workflow snapshot path.""" + archive_root = workspace_env["archive_root"] + journal_zip = _write_large_zip_member( + tmp_path / "sessions", + "workflows/wf-json.json", + json.dumps( + [ + { + "sessionId": "json-array-session", + "parentUuid": None, + "type": "user", + "message": {"role": "user", "content": "recover JSON records"}, + "uuid": "json-array-user", + "timestamp": "2025-01-01T00:00:00Z", + }, + { + "sessionId": "json-array-session", + "parentUuid": "json-array-user", + "type": "assistant", + "message": { + "role": "assistant", + "content": [{"type": "text", "text": "recovered reply"}], + }, + "uuid": "json-array-assistant", + "timestamp": "2025-01-01T00:00:01Z", + }, + ] + ).encode(), + ) + + result = await parse_sources_archive( + archive_root, + [Source(name="claude-code", path=journal_zip)], + parse_workers=1, + ) + + assert result.parse_failures == 0 + assert result.counts["sessions"] >= 1 + with sqlite3.connect(archive_root / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_artifacts").fetchone() == (0,) + + @pytest.mark.asyncio async def test_grouped_carryover_sessions_share_one_raw_row(tmp_path: Path, workspace_env: dict[str, Path]) -> None: """Two sessions split from ONE Claude Code file's bytes must NOT produce diff --git a/tests/unit/pipeline/test_quarantine_fixtures.py b/tests/unit/pipeline/test_quarantine_fixtures.py index 97df45abc8..2ba95a6a00 100644 --- a/tests/unit/pipeline/test_quarantine_fixtures.py +++ b/tests/unit/pipeline/test_quarantine_fixtures.py @@ -99,6 +99,14 @@ def claude_code_malformed_jsonl_bytes() -> bytes: return good_a + b"\n" + bad + b"\n" + good_b + b"\n" +def delayed_claude_code_session_jsonl_bytes() -> bytes: + """Thirty-two workflow rows precede the recoverable Claude session.""" + prefix = b"".join( + b'{"contentKey":"artifact-' + str(index).encode() + b'","agentId":"workflow-agent"}\n' for index in range(32) + ) + return prefix + claude_code_malformed_jsonl_bytes() + + def codex_malformed_jsonl_bytes() -> bytes: """Valid codex JSONL with one record that is not valid JSON. @@ -224,6 +232,43 @@ def test_malformed_jsonl_tolerated_in_validation_off_mode(tmp_path: Path) -> Non assert result.sessions, "valid surrounding records should still parse" +def test_validation_off_fast_path_repairs_session_shaped_workflow_journal(tmp_path: Path) -> None: + """Decoded session evidence must outrank a workflow-journal path. + + This drives the validation-off worker route with a journal path containing + one recoverable Claude Code session record and one malformed line. It must + enter the stream parser, which repairs the usable record, rather than + reporting a successful sidecar admission from the path alone. + """ + payload = delayed_claude_code_session_jsonl_bytes() + record = _make_raw_record( + payload, + "claude-code", + "/tmp/.claude/projects/project/subagents/workflows/wf-run-1/journal.jsonl", + ).model_copy(update={"source_name": "claude-code"}) + + result = ingest_record(record, str(tmp_path / "archive"), "off") + + assert result.error is None + assert len(result.sessions) == 1 + assert result.sessions[0].parsed_session.messages[0].text == "hello" + + +def test_validation_advisory_stream_repairs_session_shaped_workflow_journal(tmp_path: Path) -> None: + """The normal worker stream plan must classify decoded journal records first.""" + record = _make_raw_record( + delayed_claude_code_session_jsonl_bytes(), + "claude-code", + "/tmp/.claude/projects/project/subagents/workflows/wf-run-1/journal.jsonl", + ).model_copy(update={"source_name": "claude-code"}) + + result = ingest_record(record, str(tmp_path / "archive"), "advisory") + + assert result.error is None + assert len(result.sessions) == 1 + assert result.sessions[0].parsed_session.messages[0].text == "hello" + + # --------------------------------------------------------------------------- # Persistence lifecycle — ingest_record → mark_raw_parsed → quarantined # --------------------------------------------------------------------------- diff --git a/tests/unit/sources/test_assembly_chatgpt.py b/tests/unit/sources/test_assembly_chatgpt.py index d303a72a8e..9728ca1b86 100644 --- a/tests/unit/sources/test_assembly_chatgpt.py +++ b/tests/unit/sources/test_assembly_chatgpt.py @@ -12,6 +12,9 @@ import zipfile from pathlib import Path +import pytest + +from polylogue.archive import zip_admission as zip_admission_module from polylogue.archive.message.roles import Role from polylogue.core.enums import Provider from polylogue.sources.assembly_chatgpt import ChatGPTAssemblySpec @@ -95,6 +98,44 @@ def test_zip_missing_sidecars_returns_empty_index(self, tmp_path: Path) -> None: sidecar_data = ChatGPTAssemblySpec().discover_sidecars([zip_path]) assert sidecar_data["chatgpt_asset_index"].is_empty is True + def test_duplicate_sidecar_name_does_not_replace_first_admitted_member(self, tmp_path: Path) -> None: + zip_path = tmp_path / "duplicate-sidecar.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("library_files.json", json.dumps([{"file_id": "file-first", "file_name": "first.md"}])) + zf.writestr("library_files.json", json.dumps([{"file_id": "file-second", "file_name": "second.md"}])) + + sidecar_data = ChatGPTAssemblySpec().discover_sidecars([zip_path]) + index = sidecar_data["chatgpt_asset_index"] + + assert index.resolve_dat("file-first") is not None + assert index.resolve_dat("file-second") is None + + @pytest.mark.parametrize("limit_name", ["MAX_UNCOMPRESSED_SIZE", "MAX_COMPRESSION_RATIO"]) + def test_rejects_json_sidecar_before_open_for_size_and_ratio_limits( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + limit_name: str, + ) -> None: + zip_path = tmp_path / f"rejected-{limit_name}.zip" + sidecar_bytes = b'{"file_id":"file-abc","file_name":"notes.md"}' + (b" " * 2048) + with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: + zf.writestr("library_files.json", sidecar_bytes) + + monkeypatch.setattr(zip_admission_module, limit_name, 1) + opened: list[object] = [] + + def fail_if_open(_archive: zipfile.ZipFile, member: object, *args: object, **kwargs: object) -> object: + opened.append(member) + raise AssertionError("rejected JSON sidecar must not be opened") + + monkeypatch.setattr(zipfile.ZipFile, "open", fail_if_open) + + sidecar_data = ChatGPTAssemblySpec().discover_sidecars([zip_path]) + + assert opened == [] + assert sidecar_data["chatgpt_asset_index"].is_empty is True + class TestEnrichSession: def _index(self) -> ChatGPTAssetIndex: @@ -267,6 +308,72 @@ def test_non_dat_members_are_not_streamed(self, tmp_path: Path) -> None: sidecar_data = ChatGPTAssemblySpec().discover_sidecars([zip_path], blob_store=store) assert "chatgpt_dat_blobs" not in sidecar_data + def test_many_dat_members_obey_aggregate_limit_before_second_read( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + zip_path = tmp_path / "aggregate-dat.zip" + first_bytes = b"first attachment" + second_bytes = b"second attachment" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("file-first.dat", first_bytes) + zf.writestr("file-second.dat", second_bytes) + + monkeypatch.setattr(zip_admission_module, "MAX_AGGREGATE_UNCOMPRESSED_SIZE", len(first_bytes)) + original_open = zipfile.ZipFile.open + opened: list[str] = [] + + def track_open( + archive: zipfile.ZipFile, + member: str | zipfile.ZipInfo, + ) -> object: + info = member if isinstance(member, zipfile.ZipInfo) else archive.getinfo(member) + opened.append(info.filename) + return original_open(archive, member) + + monkeypatch.setattr(zipfile.ZipFile, "open", track_open) + store = BlobStore(tmp_path / "blobs") + + sidecar_data = ChatGPTAssemblySpec().discover_sidecars([zip_path], blob_store=store) + + assert opened == ["file-first.dat"] + dat_blobs = sidecar_data.get("chatgpt_dat_blobs") + assert dat_blobs is not None + assert set(dat_blobs) == {"file-first"} + + def test_json_and_dat_members_share_aggregate_limit_before_dat_read( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + zip_path = tmp_path / "aggregate-cross-type.zip" + json_bytes = b"[]" + dat_bytes = b"attachment" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("library_files.json", json_bytes) + zf.writestr("file-xyz.dat", dat_bytes) + + monkeypatch.setattr(zip_admission_module, "MAX_AGGREGATE_UNCOMPRESSED_SIZE", len(json_bytes)) + original_open = zipfile.ZipFile.open + opened: list[str] = [] + + def track_open( + archive: zipfile.ZipFile, + member: str | zipfile.ZipInfo, + ) -> object: + info = member if isinstance(member, zipfile.ZipInfo) else archive.getinfo(member) + opened.append(info.filename) + return original_open(archive, member) + + monkeypatch.setattr(zipfile.ZipFile, "open", track_open) + store = BlobStore(tmp_path / "blobs") + + sidecar_data = ChatGPTAssemblySpec().discover_sidecars([zip_path], blob_store=store) + + assert opened == ["library_files.json"] + assert "chatgpt_dat_blobs" not in sidecar_data + class TestAcquireDatBlobsFromDirectory: def test_dat_sibling_streamed_into_blob_store(self, tmp_path: Path) -> None: diff --git a/tests/unit/sources/test_decoders.py b/tests/unit/sources/test_decoders.py index 38d968a086..97a20ae5fc 100644 --- a/tests/unit/sources/test_decoders.py +++ b/tests/unit/sources/test_decoders.py @@ -20,6 +20,7 @@ _decode_json_bytes, _iter_json_stream, _ZipEntryValidator, + open_bounded_zip_entry, ) from polylogue.storage.cursor_state import CursorFailurePayload, CursorStatePayload @@ -296,6 +297,18 @@ def test_valid_entry_passes_through(self) -> None: entries = list(validator.filter_entries([normal_entry])) assert len(entries) == 1 + def test_bounded_open_preserves_duplicate_zipinfo_identity(self) -> None: + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as zf: + zf.writestr("duplicate.json", b"first") + zf.writestr("duplicate.json", b"second") + buffer.seek(0) + + with zipfile.ZipFile(buffer) as zf: + infos = zf.infolist() + with open_bounded_zip_entry(zf, infos[0]) as handle: + assert handle.read() == b"first" + def test_cursor_state_records_failures(self) -> None: """Rejected entries record failures in cursor_state.""" cursor_state = _seeded_cursor_state() @@ -370,10 +383,8 @@ def test_aggregate_size_limit_allows_archive_comfortably_under_cap(self) -> None assert len(accepted) == 3 assert sum(info.file_size for info in accepted) == 3 * one_gib - def test_session_only_excludes_non_session_artifact_via_real_classification(self) -> None: - """``session_only=True`` (what ``process_zip`` always passes in - production) must actually exclude a real non-session artifact via a - genuine, non-monkeypatched ``classify_artifact_path`` call. + def test_validator_leaves_terminal_artifact_classification_to_zip_processing(self) -> None: + """ZIP validation must not path-exclude entries before payload decoding. Regression test for polylogue-dc1k: every ``OriginArtifactRule.path_pattern`` in ``origin_specs.py`` is anchored ``(?:^|/)``, but the entry was @@ -392,21 +403,17 @@ def test_session_only_excludes_non_session_artifact_via_real_classification(self # Matches the "agent_transcript" OriginArtifactRule for claude-code # (parse_policy="session" -> parse_as_session=True): must survive. zf.writestr("subagents/agent-1.jsonl", json.dumps({"type": "user"}) + "\n") - # No OriginArtifactRule matches this path at all (classify_artifact_path - # returns None): must also survive -- session_only only excludes on an - # affirmative non-session classification, never on "unclassified". + # No OriginArtifactRule matches this path at all, so ZIP processing + # must leave it available for ordinary payload classification. zf.writestr("sessions.json", json.dumps({"conversations": []})) buffer.seek(0) with zipfile.ZipFile(buffer) as zf: validator = _ZipEntryValidator( - "claude-code", - cursor_state=_seeded_cursor_state(), - zip_path=Path("export.zip"), - session_only=True, + "claude-code", cursor_state=_seeded_cursor_state(), zip_path=Path("export.zip") ) accepted = [info.filename for info in validator.filter_entries(zf.infolist())] - assert "workflows/run.json" not in accepted + assert "workflows/run.json" in accepted assert "subagents/agent-1.jsonl" in accepted assert "sessions.json" in accepted diff --git a/tests/unit/sources/test_import_preflight.py b/tests/unit/sources/test_import_preflight.py index 615436a988..5353ea957b 100644 --- a/tests/unit/sources/test_import_preflight.py +++ b/tests/unit/sources/test_import_preflight.py @@ -6,7 +6,11 @@ import zipfile from pathlib import Path +import pytest + +from polylogue.archive import zip_admission as zip_admission_module from polylogue.core.enums import Provider +from polylogue.sources import import_preflight as import_preflight_module from polylogue.sources.import_preflight import ImportPreflightStatus, preflight_import_source @@ -107,3 +111,25 @@ def test_preflight_rejects_zip_without_parseable_members(tmp_path: Path) -> None assert result.status is ImportPreflightStatus.UNSUPPORTED assert result.admissible is False assert result.error_code == "unsupported_import_source" + + +def test_preflight_rejects_oversized_json_before_open(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + source = tmp_path / "oversized-preflight.zip" + with zipfile.ZipFile(source, "w") as zf: + zf.writestr("conversations.json", b"{}") + + monkeypatch.setattr(zip_admission_module, "MAX_UNCOMPRESSED_SIZE", 1) + monkeypatch.setattr(import_preflight_module, "MAX_UNCOMPRESSED_SIZE", 1) + opened: list[object] = [] + + def fail_if_open(_archive: zipfile.ZipFile, member: object, *args: object, **kwargs: object) -> object: + opened.append(member) + raise AssertionError("preflight must admit JSON before opening it") + + monkeypatch.setattr(zipfile.ZipFile, "open", fail_if_open) + + result = preflight_import_source(source) + + assert opened == [] + assert result.status is ImportPreflightStatus.MALFORMED + assert result.malformed_count == 1 diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index dc68c4438e..4d74f40642 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -1305,6 +1305,133 @@ def fail_read_bytes(_path: Path) -> bytes: assert _parse_path_as_session_artifact(target, provider=Provider.UNKNOWN) is False +def test_full_ingest_retains_sidecar_evidence_and_ingests_genuine_session(tmp_path: Path) -> None: + """Full live acquisition keeps non-session evidence and repairs session-shaped journals.""" + root = tmp_path / ".claude" + metadata_path = root / "projects" / "project" / "subagents" / "agent-a.meta.json" + journal_path = root / "projects" / "project" / "subagents" / "workflows" / "wf-run-1" / "journal.jsonl" + session_path = root / "projects" / "project" / "genuine-session.jsonl" + metadata_path.parent.mkdir(parents=True) + journal_path.parent.mkdir(parents=True) + session_path.parent.mkdir(parents=True, exist_ok=True) + + metadata_payload = b'{"agentId":"agent-a","transcriptPath":"agent-a.jsonl"}' + journal_payload = ( + json.dumps( + { + "type": "user", + "sessionId": "wf-run-1", + "uuid": "journal-message-1", + "message": {"role": "user", "content": "retain this workflow evidence"}, + } + ) + + "\n" + ).encode() + session_payload = ( + b'{"parentUuid":null,"type":"user","message":{"role":"user","content":"real session"},' + b'"uuid":"real-user","timestamp":"2025-01-01T00:00:00Z"}\n' + b'{"parentUuid":"real-user","type":"assistant","message":{"role":"assistant",' + b'"content":[{"type":"text","text":"real reply"}]},"uuid":"real-assistant",' + b'"timestamp":"2025-01-01T00:00:01Z"}\n' + ) + metadata_path.write_bytes(metadata_payload) + journal_path.write_bytes(journal_payload) + session_path.write_bytes(session_payload) + + index_db = tmp_path / "index.db" + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="claude-code", root=root),), + cursor=CursorStore(index_db), + parser_fingerprint="test-parser", + ) + + result = asyncio.run(processor.ingest_files([metadata_path, journal_path, session_path], emit_event=False)) + + assert result.succeeded_file_count == 3 + assert result.failed_file_count == 0 + assert result.ingested_session_count == 2 + with sqlite3.connect(index_db) as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (2,) + with sqlite3.connect(tmp_path / "source.db") as conn: + rows = conn.execute( + """ + SELECT a.source_path, a.artifact_kind, a.support_status, a.parse_as_session, r.blob_hash + FROM raw_artifacts AS a + JOIN raw_sessions AS r ON r.raw_id = a.raw_id + WHERE a.parse_as_session = 0 + ORDER BY a.source_path + """ + ).fetchall() + + assert [(Path(row[0]).name, row[1], row[2], row[3]) for row in rows] == [ + ("agent-a.meta.json", "agent_sidecar_meta", "unknown", 0), + ] + expected_payloads = { + metadata_path.name: metadata_payload, + } + for source_path, _kind, _support_status, _parse_as_session, blob_hash in rows: + blob_hash_hex = bytes(blob_hash).hex() + assert (tmp_path / "blob" / blob_hash_hex[:2] / blob_hash_hex[2:]).read_bytes() == expected_payloads[ + Path(source_path).name + ] + + +def test_append_declared_workflow_journal_retains_evidence_without_a_session(tmp_path: Path) -> None: + """Malformed journals remain typed evidence when decoding cannot recover them.""" + path = tmp_path / ".claude" / "projects" / "project" / "subagents" / "workflows" / "wf-append" / "journal.jsonl" + path.parent.mkdir(parents=True) + payload = b'{"contentKey":"broken"\n' + path.write_bytes(payload) + plan = replace(_append_plan(path, payload, payload_hash="artifact"), source_name="claude-code") + + result = ingest_append_plans(cast(Any, _append_owner(tmp_path)), [plan]) + + assert result.succeeded == [plan] + assert result.failed == [] + with sqlite3.connect(tmp_path / "source.db") as conn: + artifacts = conn.execute( + """ + SELECT artifact_kind, classification_reason, parse_as_session + FROM raw_artifacts + """ + ).fetchall() + assert len(artifacts) == 1 + assert [row[0] for row in artifacts] == ["workflow_journal"] + assert all(row[2] == 0 for row in artifacts) + assert all("OriginSpec" in row[1] for row in artifacts) + with sqlite3.connect(tmp_path / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (0,) + + +def test_append_session_shaped_workflow_journal_enters_revision_repair(tmp_path: Path) -> None: + """Decoded session evidence bypasses path-only workflow-journal admission.""" + path = tmp_path / ".claude" / "projects" / "project" / "subagents" / "workflows" / "wf-append" / "journal.jsonl" + path.parent.mkdir(parents=True) + payload = b"".join( + b'{"contentKey":"artifact-' + str(index).encode() + b'","agentId":"workflow-agent"}\n' for index in range(64) + ) + ( + b'{"parentUuid":null,"type":"user","message":{"role":"user","content":"recover this journal record"},' + b'"uuid":"journal-user","timestamp":"2025-01-01T00:00:00Z"}\n' + b'{"parentUuid":"journal-user","type":"assistant","message":{"role":"assistant",' + b'"content":[{"type":"text","text":"repaired reply"}]},"uuid":"journal-assistant",' + b'"timestamp":"2025-01-01T00:00:01Z"}\n' + ) + path.write_bytes(payload) + plan = replace(_append_plan(path, payload, payload_hash="session-shaped"), source_name="claude-code") + + result = ingest_append_plans(cast(Any, _append_owner(tmp_path)), [plan]) + + assert result.succeeded == [] + assert result.failed == [] + assert result.deferred == [plan] + with sqlite3.connect(tmp_path / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_artifacts").fetchone() == (0,) + assert conn.execute("SELECT revision_kind, revision_authority FROM raw_sessions").fetchall() == [ + ("append", "quarantined") + ] + + def _write_plain_sqlite_db(path: Path) -> None: """A genuine SQLite database with no Hermes state.db/verification_evidence.db shape.""" path.parent.mkdir(parents=True, exist_ok=True) @@ -3832,6 +3959,108 @@ def test_full_batch_declared_artifact_is_admitted_before_pending_raw_write( assert artifact == ("workflow_journal", 0, raw[0]) +def test_full_batch_session_shaped_workflow_journal_reaches_parser_idempotently(tmp_path: Path) -> None: + root = tmp_path / "sessions" + source = root / "subagents" / "workflows" / "wf-batch" / "journal.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes( + b"".join( + b'{"contentKey":"artifact-' + str(index).encode() + b'","agentId":"workflow-agent"}\n' + for index in range(64) + ) + + b'{"parentUuid":null,"type":"user","message":{"role":"user","content":"recover this journal record"},' + b'"uuid":"journal-user","timestamp":"2025-01-01T00:00:00Z"}\n' + b'{"parentUuid":"journal-user","type":"assistant","message":{"role":"assistant",' + b'"content":[{"type":"text","text":"repaired reply"}]},"uuid":"journal-assistant",' + b'"timestamp":"2025-01-01T00:00:01Z"}\n' + ) + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=tmp_path / "index.db"))), + (WatchSource(name="claude-code", root=root),), + cursor=CursorStore(tmp_path / "index.db"), + parser_fingerprint="test-parser", + ) + + first = asyncio.run(processor.ingest_files([source], emit_event=False)) + second = asyncio.run(processor.ingest_files([source], emit_event=False)) + + assert first.ingested_session_count == 1 + assert first.failed_file_count == 0 + assert second.failed_file_count == 0 + with sqlite3.connect(tmp_path / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone() == (1,) + assert conn.execute("SELECT COUNT(*) FROM raw_artifacts").fetchone() == (0,) + with sqlite3.connect(tmp_path / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (1,) + + +def test_large_full_batch_session_shaped_workflow_journal_reaches_parser_idempotently(tmp_path: Path) -> None: + from polylogue.sources.live.batch_support import _STREAMING_FULL_INGEST_BYTES + + root = tmp_path / "sessions" + source = root / "subagents" / "workflows" / "wf-batch" / "journal.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes( + b'{"contentKey":"artifact-0","agentId":"workflow-agent","summary":"' + + b"x" * _STREAMING_FULL_INGEST_BYTES + + b'"}\n' + + b"".join( + b'{"contentKey":"artifact-' + str(index).encode() + b'","agentId":"workflow-agent"}\n' + for index in range(1, 32) + ) + + b'{"parentUuid":null,"type":"user","message":{"role":"user","content":"recover this journal record"},' + b'"uuid":"journal-user","timestamp":"2025-01-01T00:00:00Z"}\n' + + b'{"parentUuid":"journal-user","type":"assistant","message":{"role":"assistant",' + b'"content":[{"type":"text","text":"repaired reply"}]},"uuid":"journal-assistant",' + b'"timestamp":"2025-01-01T00:00:01Z"}\n' + ) + assert source.stat().st_size > _STREAMING_FULL_INGEST_BYTES + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=tmp_path / "index.db"))), + (WatchSource(name="claude-code", root=root),), + cursor=CursorStore(tmp_path / "index.db"), + parser_fingerprint="test-parser", + ) + + first = asyncio.run(processor.ingest_files([source], emit_event=False)) + second = asyncio.run(processor.ingest_files([source], emit_event=False)) + + assert first.ingested_session_count == 1 + assert first.failed_file_count == 0 + assert second.failed_file_count == 0 + with sqlite3.connect(tmp_path / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone() == (1,) + assert conn.execute("SELECT COUNT(*) FROM raw_artifacts").fetchone() == (0,) + with sqlite3.connect(tmp_path / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (1,) + + +def test_full_batch_malformed_workflow_journal_remains_typed_evidence(tmp_path: Path) -> None: + root = tmp_path / "sessions" + source = root / "subagents" / "workflows" / "wf-batch" / "journal.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes(b'{"contentKey":"broken"\n') + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=tmp_path / "index.db"))), + (WatchSource(name="claude-code", root=root),), + cursor=CursorStore(tmp_path / "index.db"), + parser_fingerprint="test-parser", + ) + + metrics = asyncio.run(processor.ingest_files([source], emit_event=False)) + + assert metrics.succeeded_file_count == 1 + assert metrics.failed_file_count == 0 + with sqlite3.connect(tmp_path / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone() == (1,) + assert conn.execute("SELECT artifact_kind, parse_as_session FROM raw_artifacts").fetchone() == ( + "workflow_journal", + 0, + ) + with sqlite3.connect(tmp_path / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (0,) + + def test_append_admission_bind_failure_persists_exact_pending_envelope_and_retries( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/storage/test_blob_integrity.py b/tests/unit/storage/test_blob_integrity.py index 2132ece569..a59be45beb 100644 --- a/tests/unit/storage/test_blob_integrity.py +++ b/tests/unit/storage/test_blob_integrity.py @@ -11,9 +11,11 @@ import pytest +from polylogue.archive import zip_admission from polylogue.archive.message.roles import Role from polylogue.core.enums import BlockType, Provider from polylogue.sources.parsers.base import ParsedAttachment, ParsedContentBlock, ParsedMessage, ParsedSession +from polylogue.storage import blob_integrity from polylogue.storage.blob_gc import run_blob_gc_report from polylogue.storage.blob_integrity import ( classify_blob_reference_debt, @@ -998,6 +1000,51 @@ def test_replace_raw_backed_blob_reference_debt_from_source_updates_raw_refs(tmp assert conn.execute("SELECT COUNT(*) FROM blob_publication_reservations").fetchone()[0] == 0 +def test_blob_recovery_rejects_duplicate_container_member_before_open( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + zip_source = tmp_path / "duplicate.zip" + with zipfile.ZipFile(zip_source, "w") as archive: + archive.writestr("conversations.json", b'{"first": true}') + archive.writestr("conversations.json", b'{"second": true}') + + def fail_open(*args: object, **kwargs: object) -> object: + raise AssertionError("rejected duplicate member must not be opened") + + monkeypatch.setattr(zipfile.ZipFile, "open", fail_open) + payload, reason = blob_integrity._current_raw_payload_bytes( + f"{zip_source}:conversations.json", + 0, + ) + + assert payload is None + assert reason == "ambiguous_container_member" + + +def test_blob_recovery_rejects_oversized_container_member_before_open( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + zip_source = tmp_path / "oversized.zip" + with zipfile.ZipFile(zip_source, "w") as archive: + archive.writestr("conversations.json", b"{}") + + monkeypatch.setattr(zip_admission, "MAX_UNCOMPRESSED_SIZE", 1) + monkeypatch.setattr(blob_integrity, "MAX_UNCOMPRESSED_SIZE", 1) + + def fail_open(*args: object, **kwargs: object) -> object: + raise AssertionError("rejected oversized member must not be opened") + + monkeypatch.setattr(zipfile.ZipFile, "open", fail_open) + payload, reason = blob_integrity._current_raw_payload_bytes( + f"{zip_source}:conversations.json", + 0, + ) + + assert payload is None + assert reason == "container_member_rejected" + + def test_source_replacement_publication_survives_gc_before_reference_commit(tmp_path: Path) -> None: archive_root = tmp_path / "archive" initialize_active_archive_root(archive_root)