diff --git a/polylogue/archive/artifact_taxonomy/runtime.py b/polylogue/archive/artifact_taxonomy/runtime.py
index e2ff476a69..b2e20e8b51 100644
--- a/polylogue/archive/artifact_taxonomy/runtime.py
+++ b/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(
diff --git a/polylogue/pipeline/services/ingest_worker.py b/polylogue/pipeline/services/ingest_worker.py
index 879ebe4dc8..3843b24849 100644
--- a/polylogue/pipeline/services/ingest_worker.py
+++ b/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,
     )
 
diff --git a/polylogue/sources/dispatch.py b/polylogue/sources/dispatch.py
index 6d9d3704eb..b71edae101 100644
--- a/polylogue/sources/dispatch.py
+++ b/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))
diff --git a/polylogue/sources/live/append_ingest.py b/polylogue/sources/live/append_ingest.py
index 65ae195473..b6697d93e3 100644
--- a/polylogue/sources/live/append_ingest.py
+++ b/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,10 +146,14 @@ 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)
@@ -157,7 +161,9 @@ def _ingest_append_plans_archive(
                         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
diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py
index 7b0369c85a..72b768dc14 100644
--- a/polylogue/sources/live/batch.py
+++ b/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,6 +2077,15 @@ 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
                     )
@@ -2083,7 +2093,9 @@ def _ingest_full_records_archive(
                         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
diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py
index fd65f31832..d98deb56e3 100644
--- a/polylogue/sources/revision_backfill.py
+++ b/polylogue/sources/revision_backfill.py
@@ -40,7 +40,12 @@
 from polylogue.pipeline.parsed_tree_size import effective_physical_memory_bytes, estimate_parsed_tree_bytes
 from polylogue.pipeline.services.process_pool import parallel_threads_effective
 from polylogue.sources.decoders import _iter_json_stream
-from polylogue.sources.dispatch import is_stream_record_provider, parse_payload, parse_stream_payload
+from polylogue.sources.dispatch import (
+    is_stream_record_provider,
+    parse_payload,
+    parse_stream_payload,
+    require_positive_conversational_evidence,
+)
 from polylogue.sources.origin_specs import artifact_rule_for_path
 from polylogue.sources.parsers import hermes_state, hermes_verification
 from polylogue.sources.parsers.base import ParsedSession
@@ -2235,6 +2240,27 @@ def _parse_one(
     *,
     payload_path: Path | None = None,
     archive_root: Path | None = None,
+) -> list[ParsedSession]:
+    # polylogue-9ykn: replay must apply the same positive-conversational-
+    # evidence gate the live ingest paths apply, on top of the path/shape
+    # gate above (``_is_declared_non_session_artifact``, polylogue-6mpy) --
+    # a source can pass that gate (its shape IS a recognized Claude Code
+    # JSONL file with no path rule) yet still parse to zero real messages
+    # (e.g. a file containing only file-history-snapshot records).
+    return require_positive_conversational_evidence(
+        _parse_one_raw(provider, payload, source_path, payload_path=payload_path, archive_root=archive_root),
+        provider=provider,
+        source_path=source_path,
+    )
+
+
+def _parse_one_raw(
+    provider: Provider,
+    payload: bytes,
+    source_path: str,
+    *,
+    payload_path: Path | None = None,
+    archive_root: Path | None = None,
 ) -> list[ParsedSession]:
     source_name = Path(source_path).name
     fallback_id = Path(source_path).stem
@@ -2297,6 +2323,16 @@ def _sqlite_payload_path(
 
 
 def _parse_stream(provider: Provider, payload: BinaryIO, source_path: str) -> list[ParsedSession]:
+    # polylogue-9ykn: see ``_parse_one``'s comment -- the same positive-
+    # conversational-evidence gate applies to the streaming replay path.
+    return require_positive_conversational_evidence(
+        _parse_stream_raw(provider, payload, source_path),
+        provider=provider,
+        source_path=source_path,
+    )
+
+
+def _parse_stream_raw(provider: Provider, payload: BinaryIO, source_path: str) -> list[ParsedSession]:
     if _is_declared_non_session_artifact(provider, source_path):
         return []
     source_name = Path(source_path).name
diff --git a/tests/unit/pipeline/test_resilience.py b/tests/unit/pipeline/test_resilience.py
index b335fab546..9dd2244b1e 100644
--- a/tests/unit/pipeline/test_resilience.py
+++ b/tests/unit/pipeline/test_resilience.py
@@ -576,11 +576,17 @@ def test_ingest_worker_decodes_and_dispatches_provider(tmp_path: Path) -> None:
     # Verify the result structure
     assert result.raw_id is not None  # Should be the actual hash
     assert result.payload_provider is not None  # Provider detected
-    # ingest_record returns a materializable SessionWritePayload even when
-    # the source has no messages yet; that still produces an index.db session.
+    # polylogue-9ykn: an empty ``mapping`` carries no positive conversational
+    # evidence (zero messages) -- ingest_record now refuses to materialize a
+    # session for it (via require_positive_conversational_evidence, applied
+    # in _parse_plan_sessions) and records a bounded, honest parse error
+    # instead of the old "materializable session with zero messages"
+    # default. See test_ingest_worker_quarantines_session_artifact_with_no_
+    # sessions below for the equivalent no-sessions-at-all case this now
+    # shares an error shape with.
     assert isinstance(result.sessions, list)
-    assert len(result.sessions) == 1
-    assert result.error is None
+    assert len(result.sessions) == 0
+    assert result.error == "parse: session artifact produced no materializable sessions"
 
 
 def test_ingest_worker_quarantines_session_artifact_with_no_sessions(
diff --git a/tests/unit/sources/test_dispatch_payloads.py b/tests/unit/sources/test_dispatch_payloads.py
index 66678a2b23..1cef55cd2a 100644
--- a/tests/unit/sources/test_dispatch_payloads.py
+++ b/tests/unit/sources/test_dispatch_payloads.py
@@ -16,6 +16,7 @@
     detect_provider,
     parse_payload,
     parse_stream_payload,
+    require_positive_conversational_evidence,
 )
 from polylogue.sources.source_parsing import iter_source_sessions_with_raw
 
@@ -718,3 +719,125 @@ def test_parse_stream_payload_codex_long_rollout_with_repeated_session_meta_yiel
     assert total_messages > 0, "long multi-session_meta Codex rollout must not parse to zero messages"
     # 20 turns * (user + assistant + function_call use/output + custom_tool_call use/output)
     assert total_messages == 20 * 6
+
+
+def test_require_positive_conversational_evidence_refuses_claude_code_stream_with_no_conversational_records(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """polylogue-9ykn regression: a Claude Code JSONL stream consisting
+    entirely of non-conversational envelope records (file-history-snapshot
+    here; the live archive also has this for progress/bridge-session/
+    custom-title/agent-name-only files, all measured 2026-07-31) must not
+    become a session -- it has zero messages, i.e. no positive
+    conversational evidence. This is the dominant real shape found behind
+    the 5,193 zero-message claude-code-session rows in the live archive
+    (228+ of the ~237 non-agent-meta, non-analysis-dir rows were exactly
+    this: a genuine per-session JSONL file whose only records were
+    file-history-snapshot checkpoints).
+
+    ``require_positive_conversational_evidence`` is applied (not by
+    ``parse_stream_payload`` itself, which stays pure routing -- see its
+    caller-facing docstring) by every real production write path:
+    ``sources/live/batch.py``'s full-ingest loop,
+    ``pipeline/services/ingest_worker.py``'s ``_parse_plan_sessions``,
+    ``sources/live/append_ingest.py``, and
+    ``sources/revision_backfill.py``'s ``_parse_stream``. This test pins the
+    filter function itself against the exact ``parse_stream_payload`` output
+    shape those callers see.
+
+    Mutation that fails this: removing the message-content check from
+    ``require_positive_conversational_evidence``, or changing it to keep
+    zero-message sessions.
+    """
+    payload = [
+        {
+            "type": "file-history-snapshot",
+            "messageId": "06a77336-517e-4a27-996c-27547731e76b",
+            "sessionId": "history-only-session",
+            "snapshot": {"messageId": "06a77336-517e-4a27-996c-27547731e76b", "trackedFileBackups": {}},
+        },
+        {
+            "type": "file-history-snapshot",
+            "messageId": "fc6f7a3a-f38e-4f7c-9943-63157eea12c6",
+            "sessionId": "history-only-session",
+            "snapshot": {"messageId": "fc6f7a3a-f38e-4f7c-9943-63157eea12c6", "trackedFileBackups": {}},
+        },
+    ]
+    source_path = "/home/user/.claude/projects/proj/history-only-session.jsonl"
+    parsed = parse_stream_payload(Provider.CLAUDE_CODE, iter(payload), "history-only-session", source_path=source_path)
+    assert len(parsed) == 1
+    assert parsed[0].messages == []
+
+    with caplog.at_level("WARNING", logger="polylogue.sources.dispatch"):
+        sessions = require_positive_conversational_evidence(
+            parsed, provider=Provider.CLAUDE_CODE, source_path=source_path
+        )
+
+    assert sessions == []
+    assert any("polylogue-9ykn" in record.message and "no messages" in record.message for record in caplog.records)
+
+
+def test_require_positive_conversational_evidence_refuses_claude_ai_export_conversation_with_no_messages(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """polylogue-9ykn regression: a claude.ai export conversation with a real
+    title/timestamps but ``chat_messages: []`` (47 such rows measured live
+    2026-07-31) is content that legitimately exists in the export but is not
+    a conversation -- it must not become a session either.
+
+    Mutation that fails this: removing the message-content check from
+    ``require_positive_conversational_evidence``, or changing it to keep
+    zero-message sessions.
+    """
+    payload = {
+        "uuid": "ff9f9372-0ce1-40de-b890-0db7ab7d6917",
+        "name": "An empty conversation",
+        "summary": "",
+        "created_at": "2026-07-13T01:19:06.077895Z",
+        "updated_at": "2026-07-13T01:19:06.077895Z",
+        "chat_messages": [],
+    }
+    parsed = parse_payload(Provider.CLAUDE_AI, payload, "ff9f9372-0ce1-40de-b890-0db7ab7d6917")
+    assert len(parsed) == 1
+    assert parsed[0].messages == []
+
+    with caplog.at_level("WARNING", logger="polylogue.sources.dispatch"):
+        sessions = require_positive_conversational_evidence(
+            parsed, provider=Provider.CLAUDE_AI, source_path="claude-ai-export.zip:conversations.json"
+        )
+
+    assert sessions == []
+    assert any("polylogue-9ykn" in record.message and "no messages" in record.message for record in caplog.records)
+
+
+def test_parse_payload_generic_unrecognized_record_shape_manufactures_only_a_content_free_message(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """polylogue-9ykn AC(a) sibling finding: a dict with no known
+    provider-record marker at all (no ``mapping``/``messages``/
+    ``chat_messages``/envelope keys -- none of the shapes any real parser
+    recognizes; this is the sinex ``conversation_relationships.jsonl``
+    pointer-record shape from polylogue-6mpy/gvgi) is NOT declined by
+    ``parse_payload``'s Claude Code single-document lowering today: it
+    manufactures a one-message session whose sole message has an empty
+    ``text`` and no blocks -- structurally "has a message" but zero actual
+    conversational evidence. ``require_positive_conversational_evidence``
+    (checking message *content*, not just message *count*) is what actually
+    refuses it before any production write path can persist it.
+
+    Mutation that fails this: making the message-content check in
+    ``require_positive_conversational_evidence`` accept a message with empty
+    text and no blocks, or dropping the check back to a message-count test.
+    """
+    payload = {"conversation": "conv-1", "parent": "parent-1", "child": "child-1", "type": "user", "timestamp": "t"}
+
+    parsed = parse_payload(Provider.CLAUDE_CODE, payload, "fallback")
+    assert len(parsed) == 1
+    assert len(parsed[0].messages) == 1
+    assert not parsed[0].messages[0].text
+    assert not parsed[0].messages[0].blocks
+
+    with caplog.at_level("WARNING", logger="polylogue.sources.dispatch"):
+        sessions = require_positive_conversational_evidence(parsed, provider=Provider.CLAUDE_CODE, source_path=None)
+
+    assert sessions == []
diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py
index d09d22067c..69cd240adb 100644
--- a/tests/unit/sources/test_live_batch_support.py
+++ b/tests/unit/sources/test_live_batch_support.py
@@ -228,8 +228,18 @@ def test_full_ingest_empty_jsonl_is_not_misclassified_as_truncated(
     that pre-filter would make on real content, so this test exercises the
     write stage's boundary check exactly as the race does. An empty payload
     has zero records, none complete and none incomplete, so it is trivially
-    at a record boundary; the real outcome is "parsed raw payload produced no
-    sessions", not a boundary error.
+    at a record boundary.
+
+    polylogue-9ykn superseded this test's original outcome: an empty
+    capture used to "cleanly materialize as a legitimately empty parsed
+    session" -- exactly the silent-inflation default that bead eliminates.
+    The correct outcome for a zero-record capture is now the same bounded,
+    recorded ``require_positive_conversational_evidence`` refusal any other
+    zero-message parse gets (``failed``, not ``succeeded``), NOT the
+    misleading truncation-boundary error the original fix targeted. Both
+    halves of the original claim still hold: no misleading truncation
+    error, and no crash/quarantine loop -- just an honest "no positive
+    conversational evidence" outcome instead of a phantom session.
     """
     root = tmp_path / "sessions"
     root.mkdir()
@@ -249,16 +259,16 @@ def test_full_ingest_empty_jsonl_is_not_misclassified_as_truncated(
 
     result = processor._ingest_full_paths_sync([path], source_name="codex")
 
-    assert result.succeeded == [path]
-    assert result.failed == []
+    assert result.succeeded == []
+    assert result.failed == [path]
     parsed_at_ms, parse_error = _raw_parse_state(tmp_path)
     assert parse_error != "captured JSONL payload ends before a complete record boundary"
-    # The Codex stream parser derives session identity from the filename
-    # even with zero records, so an empty capture now cleanly materializes
-    # as a (legitimately empty) parsed session instead of being quarantined
-    # under a misleading truncation error.
-    assert parse_error is None
-    assert parsed_at_ms is not None
+    # polylogue-9ykn: a zero-record capture carries no positive
+    # conversational evidence -- refused with a recorded, honest reason
+    # (never the misleading truncation-boundary error), not silently
+    # accepted as a phantom empty session.
+    assert isinstance(parse_error, str) and "no sessions with positive conversational evidence" in parse_error
+    assert parsed_at_ms is None
 
 
 def test_full_ingest_heartbeats_small_file_groups_with_current_path(
@@ -269,8 +279,22 @@ def test_full_ingest_heartbeats_small_file_groups_with_current_path(
     root.mkdir()
     first = root / "first.jsonl"
     second = root / "second.jsonl"
-    first.write_text('{"type":"session_meta","payload":{"id":"first"}}\n', encoding="utf-8")
-    second.write_text('{"type":"session_meta","payload":{"id":"second"}}\n', encoding="utf-8")
+    # polylogue-9ykn: a session_meta-only stream carries no positive
+    # conversational evidence and is refused -- append one real message
+    # record so these fixtures keep testing heartbeat/byte-scan mechanics,
+    # not the now-refused empty shape.
+    first.write_text(
+        '{"type":"session_meta","payload":{"id":"first"}}\n'
+        '{"type":"response_item","payload":{"type":"message","role":"user",'
+        '"content":[{"type":"input_text","text":"hello"}]}}\n',
+        encoding="utf-8",
+    )
+    second.write_text(
+        '{"type":"session_meta","payload":{"id":"second"}}\n'
+        '{"type":"response_item","payload":{"type":"message","role":"user",'
+        '"content":[{"type":"input_text","text":"hello"}]}}\n',
+        encoding="utf-8",
+    )
     db_path = tmp_path / "archive.sqlite"
     polylogue = SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=db_path))
     cursor = CursorStore(db_path)
@@ -316,7 +340,16 @@ def test_large_full_ingest_uses_archive(
     root = tmp_path / "sessions"
     root.mkdir()
     source = root / "large.jsonl"
-    source.write_text('{"type":"session_meta","payload":{"id":"large"}}\n', encoding="utf-8")
+    # polylogue-9ykn: a session_meta-only stream carries no positive
+    # conversational evidence and is refused -- append one real message
+    # record so this fixture keeps testing full-ingest mechanics, not the
+    # now-refused empty shape.
+    source.write_text(
+        '{"type":"session_meta","payload":{"id":"large"}}\n'
+        '{"type":"response_item","payload":{"type":"message","role":"user",'
+        '"content":[{"type":"input_text","text":"hello"}]}}\n',
+        encoding="utf-8",
+    )
     db_path = tmp_path / "archive.sqlite"
     processor = LiveBatchProcessor(
         cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=db_path))),
@@ -343,7 +376,16 @@ def test_streaming_sized_full_ingest_uses_archive(
     root = tmp_path / "sessions"
     root.mkdir()
     source = root / "large.jsonl"
-    source.write_bytes(b'{"type":"session_meta","payload":{"id":"large"}}\n' + (b" " * (9 * 1024 * 1024)))
+    # polylogue-9ykn: a session_meta-only stream carries no positive
+    # conversational evidence and is refused -- append one real message
+    # record (before the size padding) so this fixture keeps testing the
+    # streaming-vs-eager routing it is named for, not the now-refused empty
+    # shape.
+    source.write_bytes(
+        b'{"type":"session_meta","payload":{"id":"large"}}\n'
+        b'{"type":"response_item","payload":{"type":"message","role":"user",'
+        b'"content":[{"type":"input_text","text":"hello"}]}}\n' + (b" " * (9 * 1024 * 1024))
+    )
     db_path = tmp_path / "archive.sqlite"
     processor = LiveBatchProcessor(
         cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=db_path))),
@@ -2147,7 +2189,14 @@ def grow_on_every_prefix_proof(
     assert second.succeeded_file_count == 1
 
     processor._defer_full_cursor_retry(path, source_name="codex", stat=path.stat())
-    replacement = b'{"type":"session_meta","payload":{"id":"busy-replacement"}}\n'
+    # polylogue-9ykn: a session_meta-only stream carries no positive
+    # conversational evidence and is refused -- append one real message
+    # record so the third ingest below still succeeds.
+    replacement = (
+        b'{"type":"session_meta","payload":{"id":"busy-replacement"}}\n'
+        b'{"type":"response_item","payload":{"type":"message","role":"user",'
+        b'"content":[{"type":"input_text","text":"hello"}]}}\n'
+    )
     path.write_bytes(replacement)
 
     assert watcher._needs_work(path)
@@ -2263,8 +2312,22 @@ def test_archive_cursor_reconciliation_rejects_restored_mtime_rewrite(
     root = tmp_path / "sessions"
     root.mkdir()
     path = root / "archive-reconcile.jsonl"
-    payload_a = b'{"type":"session_meta","payload":{"id":"archive-reconcile-a"}}\n'
-    payload_b = b'{"type":"session_meta","payload":{"id":"archive-reconcile-b"}}\n'
+    # polylogue-9ykn: a session_meta-only stream carries no positive
+    # conversational evidence and is refused -- append one real,
+    # equal-length message record to each payload so this fixture keeps
+    # testing the mtime-restore-reconciliation race, not the now-refused
+    # empty shape (the equal-length invariant below is load-bearing for the
+    # race itself, so both messages must stay identical length too).
+    payload_a = (
+        b'{"type":"session_meta","payload":{"id":"archive-reconcile-a"}}\n'
+        b'{"type":"response_item","payload":{"type":"message","role":"user",'
+        b'"content":[{"type":"input_text","text":"hello"}]}}\n'
+    )
+    payload_b = (
+        b'{"type":"session_meta","payload":{"id":"archive-reconcile-b"}}\n'
+        b'{"type":"response_item","payload":{"type":"message","role":"user",'
+        b'"content":[{"type":"input_text","text":"hello"}]}}\n'
+    )
     assert len(payload_a) == len(payload_b)
     path.write_bytes(payload_a)
     index_db = tmp_path / "index.db"
@@ -3175,9 +3238,21 @@ def test_append_multi_session_payload_is_rejected_before_index_write(
     path.write_bytes(payload)
     plan = _append_plan(path, payload, payload_hash="multi")
     owner = _append_owner(tmp_path)
+    # polylogue-9ykn: a message-less ParsedSession carries no positive
+    # conversational evidence and is refused before this test's own
+    # "more than one session" check ever runs -- give each session one real
+    # message so this fixture keeps testing the multi-session rejection.
     sessions = [
-        ParsedSession(source_name=Provider.CODEX, provider_session_id="multi-1", messages=[]),
-        ParsedSession(source_name=Provider.CODEX, provider_session_id="multi-2", messages=[]),
+        ParsedSession(
+            source_name=Provider.CODEX,
+            provider_session_id="multi-1",
+            messages=[ParsedMessage(provider_message_id="multi-1-0", role=Role.USER, text="hello")],
+        ),
+        ParsedSession(
+            source_name=Provider.CODEX,
+            provider_session_id="multi-2",
+            messages=[ParsedMessage(provider_message_id="multi-2-0", role=Role.USER, text="hello")],
+        ),
     ]
     monkeypatch.setattr("polylogue.sources.dispatch.parse_payload", lambda *_args, **_kwargs: sessions)
     result = ingest_append_plans(cast(Any, owner), [plan])
@@ -3205,9 +3280,21 @@ def test_full_multi_session_failure_retries_without_success_mapping(
         cursor=CursorStore(index_db),
         parser_fingerprint="test-parser",
     )
+    # polylogue-9ykn: a message-less ParsedSession carries no positive
+    # conversational evidence and is refused before this test's own
+    # injected-second-write-failure path ever runs -- give each session one
+    # real message so this fixture keeps testing that failure-handling path.
     sessions = [
-        ParsedSession(source_name=Provider.CODEX, provider_session_id="full-multi-1", messages=[]),
-        ParsedSession(source_name=Provider.CODEX, provider_session_id="full-multi-2", messages=[]),
+        ParsedSession(
+            source_name=Provider.CODEX,
+            provider_session_id="full-multi-1",
+            messages=[ParsedMessage(provider_message_id="full-multi-1-0", role=Role.USER, text="hello")],
+        ),
+        ParsedSession(
+            source_name=Provider.CODEX,
+            provider_session_id="full-multi-2",
+            messages=[ParsedMessage(provider_message_id="full-multi-2-0", role=Role.USER, text="hello")],
+        ),
     ]
     monkeypatch.setattr(
         "polylogue.sources.live.batch._jsonl_provider_and_session_artifact",
@@ -3326,7 +3413,17 @@ def test_full_ingest_skips_durably_excised_content_without_aborting_batch(
         cursor=CursorStore(index_db),
         parser_fingerprint="test-parser",
     )
-    sessions = [ParsedSession(source_name=Provider.CODEX, provider_session_id="normal-1", messages=[])]
+    # polylogue-9ykn: a message-less ParsedSession carries no positive
+    # conversational evidence and is refused before this test's own
+    # durably-excised-content skip path is exercised -- give it one real
+    # message so "normal.jsonl" still succeeds.
+    sessions = [
+        ParsedSession(
+            source_name=Provider.CODEX,
+            provider_session_id="normal-1",
+            messages=[ParsedMessage(provider_message_id="normal-1-0", role=Role.USER, text="hello")],
+        )
+    ]
     monkeypatch.setattr(
         "polylogue.sources.live.batch._jsonl_provider_and_session_artifact",
         lambda _path, fallback_provider: (fallback_provider, True),
diff --git a/tests/unit/sources/test_revision_backfill.py b/tests/unit/sources/test_revision_backfill.py
index 29c665ffce..0161216ce6 100644
--- a/tests/unit/sources/test_revision_backfill.py
+++ b/tests/unit/sources/test_revision_backfill.py
@@ -292,7 +292,15 @@ def test_historical_backfill_streams_codex_raw_without_eager_blob_read(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
     initialize_active_archive_root(tmp_path)
-    payload = b'{"type":"session_meta","payload":{"id":"streamed"}}\n'
+    # polylogue-9ykn: a session_meta-only stream carries no positive
+    # conversational evidence and is refused (never becomes a session) --
+    # append one real message record so this fixture keeps testing what it
+    # means to test (stream-safe blob I/O), not the now-refused empty shape.
+    payload = (
+        b'{"type":"session_meta","payload":{"id":"streamed"}}\n'
+        b'{"type":"response_item","payload":{"type":"message","role":"user",'
+        b'"content":[{"type":"input_text","text":"hello"}]}}\n'
+    )
     with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
         archive.write_raw_payload(
             provider=Provider.CODEX,
@@ -865,7 +873,15 @@ def test_historical_backfill_reparses_multi_gib_shaped_raw_instead_of_spilling_a
 ) -> None:
     """A cache miss reparses durable bytes rather than retaining a giant cohort tree."""
     initialize_active_archive_root(tmp_path)
-    payload = b'{"type":"session_meta","payload":{"id":"multi-gib-shaped"}}\n'
+    # polylogue-9ykn: a session_meta-only stream carries no positive
+    # conversational evidence and is refused -- append one real message
+    # record so this fixture keeps testing the cache/reparse mechanics it is
+    # named for, not the now-refused empty shape.
+    payload = (
+        b'{"type":"session_meta","payload":{"id":"multi-gib-shaped"}}\n'
+        b'{"type":"response_item","payload":{"type":"message","role":"user",'
+        b'"content":[{"type":"input_text","text":"hello"}]}}\n'
+    )
     with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
         raw_id = archive.write_raw_payload(
             provider=Provider.CODEX,