diff --git a/polylogue/pipeline/services/ingest_worker.py b/polylogue/pipeline/services/ingest_worker.py index 5f316e88ad..a402c48127 100644 --- a/polylogue/pipeline/services/ingest_worker.py +++ b/polylogue/pipeline/services/ingest_worker.py @@ -19,7 +19,12 @@ 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.core.common import format_malformed_jsonl_error as _format_malformed_jsonl_error @@ -374,6 +379,27 @@ 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: + 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/live/append_ingest.py b/polylogue/sources/live/append_ingest.py index 8b01409582..5f8cce1bfd 100644 --- a/polylogue/sources/live/append_ingest.py +++ b/polylogue/sources/live/append_ingest.py @@ -9,6 +9,7 @@ from pathlib import Path from typing import Any, Protocol +from polylogue.archive.artifact_taxonomy import classify_artifact_path from polylogue.archive.revision_authority import ( RawRevisionAuthority, RawRevisionEnvelope, @@ -101,6 +102,22 @@ def _ingest_append_plans_archive( raw_id: str | None = None try: provider = Provider.from_string(plan.source_name) + artifact_classification = classify_artifact_path( + str(plan.path), + provider=provider, + ) + if artifact_classification is not None and not artifact_classification.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=artifact_classification, + ) + raw_id = artifact_result.raw_id + succeeded.append(plan) + continue t0 = time.perf_counter() raw_id = archive.write_raw_payload( provider=provider, diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 2818e1770f..35c53b5d59 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -18,6 +18,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, ParamSpec, TypeVar, cast +from polylogue.archive.artifact_taxonomy import classify_artifact_path from polylogue.archive.ingest_flags import ( COMPACT_BROWSER_CAPTURE_INGEST_FLAG, DOM_FALLBACK_INGEST_FLAG, @@ -2143,6 +2144,27 @@ def _ingest_full_records_archive( fallback_id = Path(record.source_path).stem blob_hash = record.blob_hash or record.raw_id acquired_at_ms = _iso_to_epoch_ms(record.acquired_at) + artifact_classification = classify_artifact_path( + record.source_path, + provider=provider, + ) + if ( + artifact_classification is not None + and not artifact_classification.parse_as_session + and payload is not None + ): + source_raw_id = archive.admit_raw_artifact_payload( + provider=provider, + payload=payload, + source_path=record.source_path, + source_index=record.source_index or 0, + acquired_at_ms=acquired_at_ms, + classification=artifact_classification, + blob_publication_receipt_id=record.blob_publication_receipt_id, + ).raw_id + result.raw_ids[record.raw_id] = source_raw_id + _accumulate_stage_timings(result.stage_timings_s, record_timings) + continue source_write_started = time.perf_counter() if payload is None: source_raw_id = archive.write_raw_blob_ref( diff --git a/tests/unit/pipeline/test_quarantine_fixtures.py b/tests/unit/pipeline/test_quarantine_fixtures.py index 97df45abc8..7bee6647d9 100644 --- a/tests/unit/pipeline/test_quarantine_fixtures.py +++ b/tests/unit/pipeline/test_quarantine_fixtures.py @@ -28,6 +28,7 @@ from __future__ import annotations +import json from collections.abc import Callable from datetime import datetime, timezone from pathlib import Path @@ -224,6 +225,44 @@ 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_refuses_session_shaped_workflow_journal(tmp_path: Path) -> None: + """A fact-artifact path must win over the validation-off stream shortcut. + + This drives the real ``ingest_record`` worker route with a workflow journal + record that is deliberately shaped like a conversational Claude Code + record. Before the guard in ``_build_fast_stream_parse_plan``, validation + ``off`` assigned every known Claude JSONL path a session artifact directly, + so this payload produced one session despite the OriginSpec + ``workflow_journal`` rule declaring ``parse_policy=\"fact\"``. The test + therefore fails if that production classification line is removed or if + the shortcut is allowed to outrank the path classifier again. + """ + payload = ( + json.dumps( + { + "type": "user", + "sessionId": "wf-run-1", + "uuid": "journal-message-1", + "message": { + "role": "user", + "content": "workflow journal evidence is not a conversation", + }, + } + ) + + "\n" + ).encode() + record = _make_raw_record( + payload, + "claude-code", + "/tmp/.claude/projects/project/subagents/workflows/wf-run-1/journal.jsonl", + ) + + result = ingest_record(record, str(tmp_path / "archive"), "off") + + assert result.error is None + assert result.sessions == [] + + # --------------------------------------------------------------------------- # Persistence lifecycle — ingest_record → mark_raw_parsed → quarantined # --------------------------------------------------------------------------- diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index e71a93d0d5..05316f7180 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -5,6 +5,7 @@ import json import os import sqlite3 +from dataclasses import replace from hashlib import sha256 from pathlib import Path from types import SimpleNamespace @@ -1287,6 +1288,101 @@ 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_without_materializing_sessions(tmp_path: Path) -> None: + """Agent metadata and workflow journals remain durable source evidence. + + The fixtures contain one agent metadata sidecar and a session-shaped + workflow journal. The real full-ingest route must classify both from their + native paths, retain their raw rows and blobs, and produce no index + sessions. This protects the metadata-retention half of the b508/ioz7 + cluster while the classification regression protects the worker shortcut. + """ + root = tmp_path / ".claude" / "projects" / "project" / "session" + metadata_path = root / "subagents" / "agent-a.meta.json" + journal_path = root / "subagents" / "workflows" / "wf-run-1" / "journal.jsonl" + metadata_path.parent.mkdir(parents=True) + journal_path.parent.mkdir(parents=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() + metadata_path.write_bytes(metadata_payload) + journal_path.write_bytes(journal_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=tmp_path / ".claude"),), + cursor=CursorStore(index_db), + parser_fingerprint="test-parser", + ) + + result = processor._ingest_full_paths_sync([metadata_path, journal_path], source_name="claude-code") + + assert result.succeeded == [metadata_path, journal_path] + assert result.failed == [] + with sqlite3.connect(index_db) as conn: + assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (0,) + 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 + 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), + ("journal.jsonl", "workflow_journal", "unknown", 0), + ] + expected_payloads = { + metadata_path.name: metadata_payload, + journal_path.name: journal_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: + """Append acquisition uses the same typed sidecar admission as full ingest.""" + path = tmp_path / "subagents" / "workflows" / "wf-append" / "journal.jsonl" + path.parent.mkdir(parents=True) + payload = b'{"contentKey":"call-1","agentId":"agent-a"}\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: + raw = conn.execute( + "SELECT raw_id, logical_source_key, revision_kind, revision_authority FROM raw_sessions" + ).fetchone() + artifact = conn.execute( + "SELECT artifact_kind, classification_reason, parse_as_session, raw_id FROM raw_artifacts" + ).fetchone() + assert raw is not None + assert raw[1:] == (None, "unknown", "quarantined") + assert artifact is not None + assert artifact[0] == "workflow_journal" + assert "OriginSpec" in artifact[1] + assert artifact[2:] == (0, raw[0]) + + 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)