feat(sources): report per-type sidecar coverage for Claude Code parse

polylogue/sources/parsers/claude/code_parser.py

@@ -586,28 +586,32 @@ def _accumulate_delegation_progress(
     item: dict[str, object],
     timestamp: str | None,
     accumulator: dict[str, _DelegationProgressStats],
-) -> None:
+) -> bool:
     """Fold one ``progress``/``agent_progress`` record into its dispatch edge.
 
     Only ``data.type == "agent_progress"`` carries a genuine dispatcher edge
     (see the classification comment above ``_NON_MESSAGE_SIDECAR_RECORD_TYPES``);
     every occurrence under the same ``parentToolUseID`` is one streaming tick
     of the same subagent dispatch, so ticks are counted and time-bounded
-    rather than persisted as one row apiece.
+    rather than persisted as one row apiece. Returns whether this record
+    instance folded into a dispatch edge (polylogue-pbuh AC5 coverage
+    counter) -- the other six ``progress`` subtypes return ``False`` and stay
+    genuinely transient, see the classification comment.
     """
     data = item.get("data")
     if not isinstance(data, dict) or data.get("type") != "agent_progress":
-        return
+        return False
     parent_tool_use_id = _string_field(item, "parentToolUseID")
     if not parent_tool_use_id:
-        return
+        return False
     entry = accumulator.setdefault(parent_tool_use_id, _DelegationProgressStats())
     entry.count += 1
     if timestamp:
         if entry.first_seen is None or timestamp < entry.first_seen:
             entry.first_seen = timestamp
         if entry.last_seen is None or timestamp > entry.last_seen:
             entry.last_seen = timestamp
+    return True
 
 
 def _safe_float(value: object) -> float:
@@ -1293,6 +1297,16 @@ def _parse_code_records(
     # non-empty value wins, since it is constant within one file.
     session_slug_value: str | None = None
     session_refs: list[ParsedSessionRef] = []
+    # polylogue-pbuh AC5: per-record-type seen/persisted counts for the
+    # sidecar types this parser used to drop wholesale, plus a bounded
+    # sample of record types that fell all the way through ordinary message
+    # parsing to the empty-content drop below with no text/blocks -- so a
+    # *future* silently-dropped type is visible in the archive (one
+    # ``claude_parse_coverage`` session_event) instead of requiring another
+    # rg-the-corpus audit to notice.
+    sidecar_seen_counts: dict[str, int] = {}
+    sidecar_persisted_counts: dict[str, int] = {}
+    empty_drop_counts: dict[str, int] = {}
 
     for index, item in enumerate(records, start=record_index_start + 1):
         if not isinstance(item, dict):
@@ -1401,23 +1415,28 @@ def _parse_code_records(
         # produced empty ``tool_result``-shaped message rows before the
         # record-type check below existed; see #1617 for that forensic.
         if record_type in _NON_MESSAGE_SIDECAR_RECORD_TYPES:
+            sidecar_seen_counts[record_type] = sidecar_seen_counts.get(record_type, 0) + 1
+            persisted_this_record = False
             if notification is not None:
                 background_notifications.append((notification, record_uuid, timestamp))
             if record_type == "progress":
-                _accumulate_delegation_progress(item, timestamp, delegation_progress)
+                persisted_this_record = _accumulate_delegation_progress(item, timestamp, delegation_progress)
             else:
                 if record_type == "ai-title":
                     ai_title_text = _string_field(item, "aiTitle")
                     if ai_title_text:
                         latest_ai_title = ai_title_text
+                        persisted_this_record = True
                 elif record_type == "custom-title":
                     custom_title_text = _string_field(item, "customTitle")
                     if custom_title_text:
                         latest_custom_title = custom_title_text
+                        persisted_this_record = True
                 elif record_type == "agent-name":
                     agent_name_text = _string_field(item, "agentName")
                     if agent_name_text:
                         latest_agent_name = agent_name_text
+                        persisted_this_record = True
                 elif record_type == "pr-link":
                     # polylogue-2qx.4 / polylogue-cgfy: generalized,
                     # tracker-agnostic evidence alongside the existing
@@ -1438,6 +1457,7 @@ def _parse_code_records(
                                 else None,
                             )
                         )
+                        persisted_this_record = True
                 elif record_type == "attachment":
                     # Subtype-dispatched -- see ``_attachment_sidecar_event``
                     # and ``_ATTACHMENT_SUBTYPE_EVENT_TYPES`` above. Handled
@@ -1453,6 +1473,7 @@ def _parse_code_records(
                 if event_type is not None:
                     evidence_payload = _sidecar_evidence_payload(record_type, item)
                     if evidence_payload is not None:
+                        persisted_this_record = True
                         # Message linkage belongs in the typed field, not buried
                         # in the payload dict: source_message_provider_id is what
                         # the archive joins on, and every other claude_* emitter
@@ -1470,6 +1491,8 @@ def _parse_code_records(
                                 ),
                             )
                         )
+            if persisted_this_record:
+                sidecar_persisted_counts[record_type] = sidecar_persisted_counts.get(record_type, 0) + 1
             continue
         if timestamp:
             created_at = timestamp if created_at is None or timestamp < created_at else created_at
@@ -1529,6 +1552,7 @@ def _parse_code_records(
                 and material_origin is MaterialOrigin.HUMAN_AUTHORED
             )
             if not keep_empty_human_turn:
+                empty_drop_counts[record_type] = empty_drop_counts.get(record_type, 0) + 1
                 continue
         # Paste markers only appear in user prompts; restricting detection to the
         # user role avoids false positives from assistant text that quotes a marker.
@@ -1716,6 +1740,27 @@ def _parse_code_records(
             )
         )
 
+    # polylogue-pbuh AC5: one bounded coverage event per session so a future
+    # silently-dropped record type is visible without another corpus audit --
+    # counts for the known sidecar types (seen vs. actually persisted as
+    # evidence, distinguishing e.g. an ``attachment`` record whose payload
+    # was empty from one that produced an event) plus a sample of record
+    # types that reached ordinary message parsing but carried no text/blocks
+    # and were dropped there (the pre-#1617 failure mode this bead's method
+    # note warns future readers not to repeat by assumption).
+    if sidecar_seen_counts or empty_drop_counts:
+        session_events.append(
+            ParsedSessionEvent(
+                event_type="claude_parse_coverage",
+                timestamp=updated_at,
+                payload={
+                    "sidecar_seen": dict(sorted(sidecar_seen_counts.items())),
+                    "sidecar_persisted": dict(sorted(sidecar_persisted_counts.items())),
+                    "empty_dropped_by_record_type": dict(sorted(empty_drop_counts.items())),
+                },
+            )
+        )
+
     title = str(composed_session_id)
     title_source: TitleSource | None = None
     title_ref: str | None = None

tests/unit/sources/test_claude_code_sidecar_evidence.py

@@ -12,8 +12,22 @@
 from __future__ import annotations
 
 from polylogue.core.enums import TitleSource
+from polylogue.sources.parsers.base import ParsedSession, ParsedSessionEvent
 from polylogue.sources.parsers.claude import parse_code
 
+# polylogue-pbuh AC5: every parse now also emits one bounded
+# ``claude_parse_coverage`` event per session (seen/persisted counts by
+# record type). It is orthogonal to the specific-record-type behavior each
+# test below pins, so the assertions here look through it rather than
+# hard-coding it into every expected event list -- see
+# ``test_parse_coverage_event_reports_seen_and_persisted_counts`` for the
+# dedicated coverage-event test.
+_COVERAGE_EVENT_TYPE = "claude_parse_coverage"
+
+
+def _typed_events(session: ParsedSession) -> list[ParsedSessionEvent]:
+    return [e for e in session.session_events if e.event_type != _COVERAGE_EVENT_TYPE]
+
 
 def test_ai_title_wins_session_title_over_uuid_fallback() -> None:
     """``ai-title`` must resolve the session title, not just exist as a sidecar.
@@ -75,7 +89,7 @@ def test_agent_name_persists_as_typed_event() -> None:
         [{"type": "agent-name", "sessionId": "sess-agent", "agentName": "orchestration-docs-6np"}],
         "sess-agent",
     )
-    events = [(e.event_type, e.payload) for e in parsed.session_events]
+    events = [(e.event_type, e.payload) for e in _typed_events(parsed)]
     assert events == [
         ("claude_agent_name", {"agent_name": "orchestration-docs-6np", "summary": "orchestration-docs-6np"})
     ]
@@ -132,7 +146,7 @@ def test_pr_link_persists_typed_pr_fields() -> None:
         ],
         "sess-pr",
     )
-    events = [(e.event_type, e.payload) for e in parsed.session_events]
+    events = [(e.event_type, e.payload) for e in _typed_events(parsed)]
     assert events == [
         (
             "claude_pr_link",
@@ -158,7 +172,7 @@ def test_bridge_session_persists_cross_session_link() -> None:
         ],
         "sess-bridge",
     )
-    events = [(e.event_type, e.payload) for e in parsed.session_events]
+    events = [(e.event_type, e.payload) for e in _typed_events(parsed)]
     assert events == [
         (
             "claude_bridge_session",
@@ -192,7 +206,7 @@ def test_file_history_snapshot_persists_tracked_file_count() -> None:
         ],
         "sess-fhs",
     )
-    events = [(e.event_type, e.payload) for e in parsed.session_events]
+    events = [(e.event_type, e.payload) for e in _typed_events(parsed)]
     assert events == [
         (
             "claude_file_history_snapshot",
@@ -217,7 +231,7 @@ def test_permission_mode_persists_operational_signal() -> None:
         [{"type": "permission-mode", "sessionId": "sess-perm", "permissionMode": "bypassPermissions"}],
         "sess-perm",
     )
-    events = [(e.event_type, e.payload) for e in parsed.session_events]
+    events = [(e.event_type, e.payload) for e in _typed_events(parsed)]
     assert events == [
         ("claude_permission_mode", {"permission_mode": "bypassPermissions", "summary": "bypassPermissions"})
     ]
@@ -228,7 +242,7 @@ def test_last_prompt_persists_resume_continuity_signal() -> None:
         [{"type": "last-prompt", "sessionId": "sess-lp", "lastPrompt": "hello world."}],
         "sess-lp",
     )
-    events = [(e.event_type, e.payload) for e in parsed.session_events]
+    events = [(e.event_type, e.payload) for e in _typed_events(parsed)]
     assert events == [("claude_last_prompt", {"last_prompt": "hello world.", "summary": "hello world."})]
 
 
@@ -254,7 +268,7 @@ def test_queue_operation_enqueue_persists_content_dequeue_does_not() -> None:
         ],
         "sess-queue",
     )
-    events = [(e.event_type, e.payload) for e in parsed.session_events]
+    events = [(e.event_type, e.payload) for e in _typed_events(parsed)]
     assert events == [
         (
             "claude_queue_operation",
@@ -282,7 +296,7 @@ def test_attachment_file_subtype_gets_its_own_event_type() -> None:
         ],
         "sess-att",
     )
-    events = [(e.event_type, e.payload) for e in parsed.session_events]
+    events = [(e.event_type, e.payload) for e in _typed_events(parsed)]
     assert events == [
         (
             "claude_attachment_file",
@@ -549,7 +563,7 @@ def test_progress_bash_progress_and_hook_progress_stay_transient() -> None:
         },
     ]
     parsed = parse_code(records, "sess-bash")
-    assert parsed.session_events == []
+    assert _typed_events(parsed) == []
     assert parsed.messages == []
 
 
@@ -587,7 +601,7 @@ def test_file_history_delta_persists_tracking_path() -> None:
         ],
         "sess-delta",
     )
-    events = [(e.event_type, e.payload) for e in parsed.session_events]
+    events = [(e.event_type, e.payload) for e in _typed_events(parsed)]
     assert events == [
         (
             "claude_file_history_delta",
@@ -763,5 +777,55 @@ def test_init_and_mode_records_produce_no_events_or_messages() -> None:
         ],
         "sess-init",
     )
-    assert parsed.session_events == []
+    assert _typed_events(parsed) == []
     assert parsed.messages == []
+
+
+def test_parse_coverage_event_reports_seen_and_persisted_counts() -> None:
+    """polylogue-pbuh AC5: coverage is reported per type -- seen vs. actually
+    persisted -- so a future silently-dropped record type is visible in the
+    archive itself rather than requiring another corpus rg audit to notice.
+
+    ``permission-mode`` here always persists (its record always carries a
+    ``permissionMode`` string), while a ``bash_progress`` tick under
+    ``progress`` is seen but never persisted (see the classification comment
+    above ``_SKIPPED_SIDECAR_RECORD_TYPES``) -- pinning that seen and
+    persisted counts can genuinely diverge, not just mirror each other.
+    """
+    parsed = parse_code(
+        [
+            {"type": "permission-mode", "sessionId": "sess-cov", "permissionMode": "plan"},
+            {"type": "permission-mode", "sessionId": "sess-cov", "permissionMode": "acceptEdits"},
+            {
+                "type": "progress",
+                "sessionId": "sess-cov",
+                "toolUseID": "bash-progress-0",
+                "parentToolUseID": "toolu_bash_1",
+                "data": {"type": "bash_progress", "output": "", "elapsedTimeSeconds": 1},
+            },
+        ],
+        "sess-cov",
+    )
+    coverage_events = [e for e in parsed.session_events if e.event_type == _COVERAGE_EVENT_TYPE]
+    assert len(coverage_events) == 1
+    payload = coverage_events[0].payload
+    assert payload["sidecar_seen"] == {"permission-mode": 2, "progress": 1}
+    assert payload["sidecar_persisted"] == {"permission-mode": 2}
+    assert payload["empty_dropped_by_record_type"] == {}
+
+
+def test_parse_coverage_event_absent_when_only_ordinary_messages_parsed() -> None:
+    """A session with no sidecar/empty-drop activity gets no coverage event
+    at all -- keeps the common case from carrying a useless empty payload."""
+    parsed = parse_code(
+        [
+            {
+                "type": "user",
+                "uuid": "u1",
+                "sessionId": "sess-plain",
+                "message": {"role": "user", "content": "plain session, nothing skipped"},
+            },
+        ],
+        "sess-plain",
+    )
+    assert [e for e in parsed.session_events if e.event_type == _COVERAGE_EVENT_TYPE] == []