fix(sources): classify sidecars before live session admission

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)

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,

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(

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

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)