From c6190d7db179af95cea6c150bac676ed04ec6e9b Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 31 Jul 2026 23:39:20 +0200 Subject: [PATCH 1/3] fix(archive): let record content override the analysis/ dir path guess Problem: tests/unit/sources/test_revision_backfill.py::test_parse_one_still_replays_real_claude_code_sessions_with_no_path_rule failed on clean master (`assert 0 == 1`). classify_artifact's content gate added toward polylogue-9ykn (PR #3428) consulted classify_artifact_path first and returned unconditionally: any source path with an "analysis" directory segment was refused as a self-generated side-output artifact regardless of record content, so a genuine Claude Code session record (role=user, real sessionId, timestamp) sitting under such a path was silently dropped on replay. Solution: split classify_artifact_path into the weak, content-blind "analysis/" directory heuristic (_self_generated_artifact_dir_classification) and the strong, definitive path rules (_classify_artifact_path_strong: OriginSpec rules, known sidecar filenames, Hermes/Antigravity path markers). Path-only callers (decoder_zip, source_walk skip-listing, schema sampling) keep the exact prior behavior via classify_artifact_path, which still checks the weak heuristic first. The content-aware classify_artifact now checks strong path rules first (unconditional, as before), then content classification, and only falls back to the weak analysis/ heuristic when content shows no positive session evidence -- so a relationship-index pointer record (no envelope markers) is still refused, but a genuine session record wins on its content regardless of directory naming. Verification: python -m devtools test tests/unit/sources/test_revision_backfill.py -> 56 passed python -m devtools test tests/unit/sources/test_artifact_taxonomy.py -> 9 passed Ref polylogue-6mpy Co-Authored-By: Claude --- .../archive/artifact_taxonomy/runtime.py | 113 ++++++++++++++---- 1 file changed, 90 insertions(+), 23 deletions(-) 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( From 18606faf99b279ca833ab247d37a39fa373da716 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 1 Aug 2026 00:26:54 +0200 Subject: [PATCH 2/3] feat(sources): require positive conversational evidence before a session write Problem: the ingest path's default disposition was "this is a session" -- anything not positively recognised as something else still became one. Measured against the live archive (2026-07-31): 5,193 zero-message claude-code-session rows (22.6% of the archive), reconciled via read-only queries against index.db/source.db joined to raw_sessions.source_path: 4,945 agent-*.meta sidecars (already refused by classify_artifact's path rule but pre-dating it), 228+ JSONL files containing only non- conversational envelope records (file-history-snapshot/progress/ bridge-session/custom-title/agent-name), 3 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 (2026-07-31) 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. Solution: polylogue/sources/dispatch.py gains require_positive_conversational_evidence() + message_carries_authored_ content() -- a session is kept only if at least one message has real text or a content block. Deliberately NOT folded into parse_payload/ parse_stream_payload themselves: those two functions are pure provider- routing dispatch pinned by a large "law" test surface (test_source_laws.py and friends) that monkeypatches provider parsers with zero-message stubs specifically to test routing independent of content -- folding the gate in there broke 20+ of those tests. Instead every real production write path applies the filter explicitly after calling parse_payload/parse_stream_payload: - sources/live/batch.py's full-ingest loop (reuses the existing mark_raw_parse_failed "recorded, bounded failure" mechanism a filtered-to-empty result already trips) - pipeline/services/ingest_worker.py's _parse_plan_sessions (subprocess decode/parse worker) - sources/live/append_ingest.py (incremental append) - sources/revision_backfill.py's _parse_one/_parse_stream (offline replay/rebuild, alongside its own polylogue-6mpy path/shape gate -- this filter catches the sibling case where the shape is recognized but the parsed content still carries no message) Checking message *content*, not just message *count*, also closes a narrower sibling gap found while testing this bead: an unrecognized single-record document with no known provider markers previously fell through Claude Code's generic single-document lowering into a one-message session whose sole message had empty text and no blocks. Verification so far: python -m devtools test tests/unit/sources/test_dispatch_payloads.py tests/unit/sources/test_dispatch_ordering.py tests/unit/sources/test_revision_backfill.py tests/unit/sources/test_artifact_taxonomy.py -> 197 passed, 1 pre-existing unrelated failure (test_parse_stream_ payload_codex_long_rollout_with_repeated_session_meta_yields_messages, confirmed failing on clean master before this change) Residual: a broader targeted run surfaced ~15 more tests in test_live_batch_support.py/test_resilience.py/test_synthetic_semantics.py etc. whose fixtures assume the old "session_meta-only stream succeeds" or "zero-byte capture becomes a legitimately empty session" behavior -- being triaged/updated in a follow-up commit on this branch (some are fixture updates matching the new invariant, at least one -- test_full_ingest_empty_jsonl_is_not_misclassified_as_truncated -- is a considered pre-existing behavior decision that conflicts with this bead's invariant and needs an explicit call, documented in the PR body). Ref polylogue-9ykn Co-Authored-By: Claude --- polylogue/pipeline/services/ingest_worker.py | 29 +++-- polylogue/sources/dispatch.py | 99 ++++++++++++++- polylogue/sources/live/append_ingest.py | 18 ++- polylogue/sources/live/batch.py | 14 ++- polylogue/sources/revision_backfill.py | 38 +++++- tests/unit/sources/test_dispatch_payloads.py | 123 +++++++++++++++++++ tests/unit/sources/test_revision_backfill.py | 20 ++- 7 files changed, 321 insertions(+), 20 deletions(-) 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/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_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, From d7759a5415cafe5abe733ecd87b359082455a5e1 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 1 Aug 2026 00:46:18 +0200 Subject: [PATCH 3/3] test(sources): update fixtures/expectations for the positive-evidence gate Problem: the broader targeted regression run for the polylogue-9ykn gate surfaced 15 test failures beyond the previous commit's verified set. Triaged each by reproducing against a clean-master baseline worktree (commit 4477b961a): 6 were genuinely caused by this bead's new require_positive_conversational_evidence filter (fixtures built session_meta-only or empty-mapping payloads that used to silently materialize a zero-message session), and 9 were pre-existing failures already broken on master, unrelated to this change (confirmed by reproducing the identical failure against the baseline worktree with none of this branch's commits applied). Solution (the 6 real fixes): - tests/unit/sources/test_live_batch_support.py: appended a real response_item message record to 6 session_meta-only Codex JSONL fixtures (test_large_full_ingest_uses_archive, test_streaming_sized_full_ingest_uses_archive, test_full_ingest_heartbeats_small_file_groups_with_current_path, test_busy_full_prefix_proof_defers_to_archived_cursor_reconciliation, test_archive_cursor_reconciliation_rejects_restored_mtime_rewrite) and gave mocked zero-message ParsedSession fixtures a real message (test_append_multi_session_payload_is_rejected_before_index_write's sibling test_full_multi_session_failure_retries_without_success_ mapping) so each test keeps exercising the mechanics it is named for, not the now-refused empty shape. - test_full_ingest_empty_jsonl_is_not_misclassified_as_truncated: this one is a considered behavior change, not a fixture patch. Its original assertion was "a zero-byte capture cleanly materializes as a legitimately empty parsed session" -- exactly the silent-inflation default polylogue-9ykn eliminates. Updated to assert the new, correct outcome: refused with a bounded, honest "no positive conversational evidence" parse_error (never the misleading truncation-boundary error the original fix targeted, which the test still pins). - tests/unit/pipeline/test_resilience.py:: test_ingest_worker_decodes_and_dispatches_provider: its own docstring admitted the pre-existing default -- "ingest_record returns a materializable SessionWritePayload even when the source has no messages yet". Updated to assert the new outcome: refused via the existing "session artifact produced no materializable sessions" ingest_worker.py error path (unchanged mechanism, now reached for the message-content-empty case too). The 9 pre-existing failures are left untouched (out of scope for this bead): test_full_ingest_writes_archive_with_route_observability, test_full_ingest_skips_durably_excised_content_without_aborting_batch, test_append_multi_session_payload_is_rejected_before_index_write (a DIFFERENT still-broken assertion than the one this commit fixed in its sibling -- see below), test_archive_filter_kwargs_cover_every_storage_ lowerable_spec_field, test_validation_law_matches_mode_and_payload_ contract, TestParseRoundtrip::test_synthetic_parses_to_sessions[antigravity], TestCorpusParseRoundtrip::test_generated_data_parses[antigravity], test_family_fixture_detector_and_streaming_paths_preserve_one_normalized_identity, test_parse_stream_payload_codex_long_rollout_with_repeated_session_meta_ yields_messages. Note: test_append_multi_session_payload_is_rejected_before_index_write fails on BOTH master and this branch, but for two different reasons -- the fixture's `sessions` list mock is never reached either way because `_is_declared_non_session_artifact` refuses `{}`-shaped raw bytes before `parse_payload` runs (pre-existing, unrelated to this bead). Gave the mock real content anyway since it is otherwise correct for what the test intends to verify. Verification: python -m devtools test tests/infra/pipeline_roundtrip.py tests/unit/sources/test_dispatch_payloads.py tests/unit/sources/test_dispatch_ordering.py tests/unit/sources/test_source_laws.py tests/unit/sources/test_revision_backfill.py tests/unit/sources/test_artifact_taxonomy.py tests/unit/sources/test_live_batch_support.py tests/unit/pipeline/test_resilience.py tests/unit/daemon/test_ingest_worker_handoff.py tests/unit/pipeline/test_ingest_worker_assembly.py -> 334 passed, 5 failed (all 5 confirmed pre-existing against a clean-master baseline worktree at 4477b961a) Ref polylogue-9ykn Co-Authored-By: Claude --- tests/unit/pipeline/test_resilience.py | 14 +- tests/unit/sources/test_live_batch_support.py | 141 +++++++++++++++--- 2 files changed, 129 insertions(+), 26 deletions(-) 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_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),