fix(sources): require positive conversational evidence for session creation

polylogue/archive/artifact_taxonomy/runtime.py

@@ -47,12 +47,70 @@ def _has_self_generated_artifact_dir_segment(normalized_path: str) -> bool:
     return any(part in _SELF_GENERATED_ARTIFACT_DIR_SEGMENTS for part in Path(inner).parts[:-1])
 
 
+def _self_generated_artifact_dir_classification(
+    source_path: str | Path | None,
+    *,
+    provider: str | Provider,
+) -> ArtifactClassification | None:
+    """Weak, content-blind path heuristic: refuse anything under an
+    ``analysis/`` directory segment.
+
+    Deliberately split out of ``classify_artifact_path`` (polylogue-6mpy):
+    this heuristic exists to catch self-generated side-output that never
+    carries genuine conversation evidence (e.g. a sinex
+    ``conversation_relationships.jsonl`` pointer index) when no content is
+    available to classify (pre-decode, path-only filtering routes such as
+    ``decoder_zip``/``source_walk`` skip-listing). But it is a *location*
+    guess, not conversation evidence, and a genuine Claude Code session
+    JSONL file can legitimately be re-homed or replayed from a path that
+    happens to include an ``analysis`` segment. ``classify_artifact`` (the
+    content-aware entry point) must let positive record content override
+    this heuristic rather than let it win unconditionally -- see its own
+    call site for the tie-break order.
+    """
+    provider_token = Provider.from_string(provider)
+    normalized = normalize_source_path(source_path)
+    if not normalized or not _has_self_generated_artifact_dir_segment(normalized):
+        return None
+    return ArtifactClassification(
+        provider=provider_token,
+        kind=ArtifactKind.METADATA_DOCUMENT,
+        parse_as_session=False,
+        schema_eligible=False,
+        default_priority=0,
+        reason="self-generated analysis artifact under an 'analysis/' directory "
+        "(agent side-output, not conversation content; mirrors source_walk _SKIP_DIRS)",
+    )
+
+
 def classify_artifact_path(
     source_path: str | Path | None,
     *,
     provider: str | Provider,
 ) -> ArtifactClassification | None:
-    """Classify obvious sidecars using only the source path."""
+    """Classify obvious sidecars using only the source path.
+
+    Path-only callers (pre-decode filtering: ``decoder_zip``, ``source_walk``
+    skip-listing, schema sampling) get the weak ``analysis/`` directory
+    heuristic first, same as always -- no content is available for them to
+    weigh against it. ``classify_artifact`` (content-aware) instead calls
+    ``_classify_artifact_path_strong`` directly and only falls back to the
+    weak heuristic when content classification finds no positive evidence;
+    see that function's call site.
+    """
+    if weak := _self_generated_artifact_dir_classification(source_path, provider=provider):
+        return weak
+    return _classify_artifact_path_strong(source_path, provider=provider)
+
+
+def _classify_artifact_path_strong(
+    source_path: str | Path | None,
+    *,
+    provider: str | Provider,
+) -> ArtifactClassification | None:
+    """Classify obvious sidecars by path, excluding the weak ``analysis/``
+    directory heuristic (split out so ``classify_artifact`` can let positive
+    record content override that one heuristic; polylogue-6mpy)."""
     provider_token = Provider.from_string(provider)
     normalized = normalize_source_path(source_path)
     if not normalized:
@@ -64,16 +122,6 @@ def classify_artifact_path(
     from polylogue.sources.origin_specs import artifact_rule_for_path
 
     inner_name = Path(normalized.rsplit(":", 1)[-1]).name.lower()
-    if _has_self_generated_artifact_dir_segment(normalized):
-        return ArtifactClassification(
-            provider=provider_token,
-            kind=ArtifactKind.METADATA_DOCUMENT,
-            parse_as_session=False,
-            schema_eligible=False,
-            default_priority=0,
-            reason="self-generated analysis artifact under an 'analysis/' directory "
-            "(agent side-output, not conversation content; mirrors source_walk _SKIP_DIRS)",
-        )
     if rule := artifact_rule_for_path(provider_token, normalized):
         return ArtifactClassification(
             provider=provider_token,
@@ -200,22 +248,41 @@ def classify_artifact(
         if marker_classification is not None:
             return marker_classification
 
-    explicit = classify_artifact_path(source_path, provider=provider_token)
+    # ``_classify_artifact_path_strong`` covers the definitive, content-blind
+    # path rules (OriginSpec artifact rules, known sidecar filenames, Hermes/
+    # Antigravity path markers) -- these always win regardless of content.
+    explicit = _classify_artifact_path_strong(source_path, provider=provider_token)
     if explicit is not None:
         return explicit
 
     if isinstance(payload, Sequence) and not isinstance(payload, str | bytes | bytearray):
-        return _classify_list(payload, provider=provider_token, source_path=source_path)
-    if isinstance(payload, dict):
-        return _classify_dict(payload, provider=provider_token, source_path=source_path)
-    return ArtifactClassification(
-        provider=provider_token,
-        kind=ArtifactKind.UNKNOWN,
-        parse_as_session=False,
-        schema_eligible=False,
-        default_priority=0,
-        reason="non-object payload",
-    )
+        content_classification = _classify_list(payload, provider=provider_token, source_path=source_path)
+    elif isinstance(payload, dict):
+        content_classification = _classify_dict(payload, provider=provider_token, source_path=source_path)
+    else:
+        content_classification = ArtifactClassification(
+            provider=provider_token,
+            kind=ArtifactKind.UNKNOWN,
+            parse_as_session=False,
+            schema_eligible=False,
+            default_priority=0,
+            reason="non-object payload",
+        )
+
+    # polylogue-6mpy: positive conversational evidence in the record content
+    # (recognised session/record shape) outranks the weak, content-blind
+    # ``analysis/`` directory heuristic -- a genuine session record must not
+    # be refused merely because its replay/backfill path happens to route
+    # through a directory segment named "analysis". The heuristic still wins
+    # when content classification found no positive evidence at all, which
+    # is exactly the polylogue-9ykn direction: an unrecognised record stays
+    # refused, never defaults to a session.
+    if content_classification.parse_as_session:
+        return content_classification
+    weak = _self_generated_artifact_dir_classification(source_path, provider=provider_token)
+    if weak is not None:
+        return weak
+    return content_classification
 
 
 def _classify_list(

polylogue/pipeline/services/ingest_worker.py

@@ -515,7 +515,11 @@ def _parse_plan_sessions(
     context: _IngestContext,
     plan: _ParsePlan,
 ) -> list[ParsedSession]:
-    from polylogue.sources.dispatch import parse_payload, parse_stream_payload
+    from polylogue.sources.dispatch import (
+        parse_payload,
+        parse_stream_payload,
+        require_positive_conversational_evidence,
+    )
 
     fallback_id = _fallback_id(context.raw_record.source_path, context.raw_record.raw_id)
     if plan.mode == "stream":
@@ -538,13 +542,24 @@ def counted_stream() -> Iterable[object]:
             )
             if valid_record_count == 0:
                 raise ValueError(f"no valid JSON records in {stream_name}")
-            return sessions
+            # polylogue-9ykn: a session requires positive conversational
+            # evidence -- applied here (the subprocess decode/parse worker's
+            # own chokepoint) so this ingest route can't create a
+            # zero-message session even though it never touches
+            # sources/live/batch.py's or revision_backfill.py's call sites.
+            return require_positive_conversational_evidence(
+                sessions, provider=plan.provider, source_path=context.raw_record.source_path
+            )
 
-    return parse_payload(
-        plan.provider,
-        plan.payload,
-        fallback_id,
-        schema_resolution=plan.schema_resolution,
+    return require_positive_conversational_evidence(
+        parse_payload(
+            plan.provider,
+            plan.payload,
+            fallback_id,
+            schema_resolution=plan.schema_resolution,
+            source_path=context.raw_record.source_path,
+        ),
+        provider=plan.provider,
         source_path=context.raw_record.source_path,
     )
 

polylogue/sources/dispatch.py

@@ -31,7 +31,7 @@
     hermes_verification,
     local_agent,
 )
-from .parsers.base import ParsedSession, extract_messages_from_list
+from .parsers.base import ParsedMessage, ParsedSession, extract_messages_from_list
 from .parsers.claude.code_parser import apply_tool_result_sidecars
 
 if TYPE_CHECKING:
@@ -1502,6 +1502,89 @@ def _parse_lowered_spec(spec: LoweredPayloadSpec) -> list[ParsedSession]:
     return []
 
 
+def message_carries_authored_content(message: ParsedMessage) -> bool:
+    """A message counts as positive conversational evidence when it carries
+    any real text or content block. A message row that exists structurally
+    (a provider_message_id, a role) but has neither -- e.g. a generic
+    unrecognized-record fallback that manufactures a placeholder message --
+    is not evidence of a conversation."""
+    if message.text is not None and message.text.strip():
+        return True
+    return bool(message.blocks)
+
+
+def require_positive_conversational_evidence(
+    sessions: list[ParsedSession],
+    *,
+    provider: str | Provider,
+    source_path: str | None,
+) -> list[ParsedSession]:
+    """polylogue-9ykn: a session requires positive evidence of a conversation
+    -- at minimum one message carrying authored content -- or it is refused
+    loudly rather than written.
+
+    Deliberately NOT folded into ``parse_payload``/``parse_stream_payload``
+    themselves: those two functions are pure provider-routing dispatch, and
+    a large "law" test surface (``tests/unit/sources/test_source_laws.py``
+    and friends) monkeypatches the underlying provider parsers with
+    zero-message stubs specifically to pin *routing* behavior (which parser
+    got called, with what arguments, how many times) independent of parsed
+    content -- folding a content gate into the dispatch functions silently
+    broke 20+ of those tests by deleting the stubbed sessions before the
+    test could observe them. Instead, every real production write path
+    calls this filter explicitly right after it gets a ``parse_payload``/
+    ``parse_stream_payload`` result back: ``pipeline/services/
+    ingest_worker.py`` (subprocess decode/parse worker),
+    ``sources/live/batch.py`` (in-process daemon full-ingest convergence,
+    which already treats an empty session list as a recorded, bounded
+    ``mark_raw_parse_failed`` outcome -- this filter reuses that existing
+    "refused loudly" mechanism rather than inventing a new one),
+    ``sources/live/append_ingest.py`` (incremental append), and
+    ``sources/revision_backfill.py`` (offline replay/rebuild, alongside its
+    own OriginSpec/``classify_artifact`` path-and-shape gate from
+    polylogue-6mpy -- this filter catches the sibling case where the shape
+    is recognized but the parsed *content* still carries no message).
+
+    Measured against the live archive (2026-07-31, read-only query against
+    ``index.db``/``source.db``): every verified zero-message
+    ``claude-code-session`` row was one of a handful of artifact classes --
+    ``agent-*.meta`` sidecars (4,945, already refused by
+    ``classify_artifact``'s path rule but pre-dating it), JSONL files
+    containing only non-conversational envelope records
+    (file-history-snapshot/progress/bridge-session/custom-title/agent-name,
+    228+ rows), and ``tool-results/*.json`` sidecars mis-dispatched as
+    sessions -- plus 47 ``claude-ai-export`` conversations with a real title
+    but ``chat_messages: []``. polylogue-ne6k's own investigation concluded
+    no "genuinely empty but legitimate" session construct exists in this
+    corpus: every zero-message row was retained only because the prior
+    repair predicate could not distinguish it from one, not because it was
+    verified worth keeping. This filter stops the population from growing;
+    the existing rows are purged by the already-planned index rebuild
+    (polylogue-x1gd), not by this change.
+
+    Checking message *content*, not just message *count*, also closes a
+    narrower sibling gap found while testing this bead: an unrecognized
+    single-record document (no ``mapping``/``messages``/envelope markers at
+    all) previously fell through Claude Code's generic single-document
+    lowering into a one-message session whose sole message had an empty
+    ``text`` and no blocks -- structurally "has a message" but zero actual
+    conversational evidence.
+    """
+    kept: list[ParsedSession] = []
+    for session in sessions:
+        if any(message_carries_authored_content(message) for message in session.messages):
+            kept.append(session)
+            continue
+        logger.warning(
+            "polylogue-9ykn: refusing session %s (%s, source_path=%s) -- "
+            "no messages, no positive conversational evidence",
+            session.provider_session_id,
+            Provider.from_string(provider),
+            source_path,
+        )
+    return kept
+
+
 def parse_payload(
     provider: str | Provider,
     payload: object,
@@ -1511,7 +1594,13 @@ def parse_payload(
     schema_resolution: SchemaResolution | None = None,
     source_path: str | None = None,
 ) -> list[ParsedSession]:
-    """Dispatch parsed payload to the appropriate provider parser."""
+    """Dispatch parsed payload to the appropriate provider parser.
+
+    Pure routing: returns whatever the selected provider parser reports,
+    including a zero-message session. Production write paths must apply
+    ``require_positive_conversational_evidence`` to the result themselves
+    (see that function's docstring for why it is not applied here).
+    """
     lowered_specs = _lower_payload_specs(
         provider,
         payload,
@@ -1533,7 +1622,11 @@ def parse_stream_payload(
     *,
     source_path: str | None = None,
 ) -> list[ParsedSession]:
-    """Parse a grouped record stream."""
+    """Parse a grouped record stream.
+
+    Pure routing, same contract as ``parse_payload`` -- see
+    ``require_positive_conversational_evidence``'s docstring.
+    """
     runtime_provider = Provider.from_string(provider)
     if runtime_provider is Provider.CLAUDE_CODE:
         return merge_parsed_session_chunks(_claude_code_stream_sessions(payloads, fallback_id, source_path=source_path))

polylogue/sources/live/append_ingest.py

@@ -76,7 +76,7 @@ def _ingest_append_plans_archive(
 
     t0 = time.perf_counter()
     from polylogue.sources.decoders import _iter_json_stream
-    from polylogue.sources.dispatch import parse_payload
+    from polylogue.sources.dispatch import parse_payload, require_positive_conversational_evidence
     from polylogue.sources.revision_backfill import (
         _is_declared_non_session_artifact,
         parse_retained_raw_sessions,
@@ -146,18 +146,24 @@ def _ingest_append_plans_archive(
                         failed.append(plan)
                         continue
                     t0 = time.perf_counter()
-                    sessions = parse_payload(
-                        provider,
-                        payloads,
-                        plan.path.stem,
+                    sessions = require_positive_conversational_evidence(
+                        parse_payload(
+                            provider,
+                            payloads,
+                            plan.path.stem,
+                            source_path=str(plan.path),
+                        ),
+                        provider=provider,
                         source_path=str(plan.path),
                     )
                     _add_timing(timings, "append.provider_parse", t0)
                     if not sessions:
                         archive.mark_raw_parse_failed(
                             raw_id,
                             provider=provider,
-                            error=ValueError("parsed raw payload produced no sessions"),
+                            error=ValueError(
+                                "parsed raw payload produced no sessions with positive conversational evidence"
+                            ),
                         )
                         failed.append(plan)
                         continue

polylogue/sources/live/batch.py

@@ -56,6 +56,7 @@
     is_stream_record_provider,
     parse_payload,
     parse_stream_payload,
+    require_positive_conversational_evidence,
 )
 from polylogue.sources.live.append_ingest import ingest_append_plans, reset_transient_raw_parse_state
 from polylogue.sources.live.batch_observability import (
@@ -2076,14 +2077,25 @@ def _ingest_full_records_archive(
                             fallback_id,
                             source_path=record.source_path,
                         )
+                    # polylogue-9ykn: a session requires positive
+                    # conversational evidence -- a parse that produced only
+                    # zero-message sessions is treated exactly like a parse
+                    # that produced none: a recorded, bounded
+                    # mark_raw_parse_failed outcome below, never a silently
+                    # written phantom session.
+                    sessions = require_positive_conversational_evidence(
+                        sessions, provider=provider, source_path=record.source_path
+                    )
                     record_timings["full.provider_parse"] = record_timings.get("full.provider_parse", 0.0) + (
                         time.perf_counter() - t0
                     )
                     if not sessions:
                         archive.mark_raw_parse_failed(
                             source_raw_id,
                             provider=provider,
-                            error=ValueError("parsed raw payload produced no sessions"),
+                            error=ValueError(
+                                "parsed raw payload produced no sessions with positive conversational evidence"
+                            ),
                         )
                         continue
                     record_raw_id = source_raw_id