fix(sensor): preserve Claude transcript content and incremental updates

Sensor/CONTRIBUTING.md

@@ -62,6 +62,7 @@ class MyAgentParser(BaseParser):
         entries = []
 
         if not self.base_path.exists():
+            self.record_diagnostic("input_missing")
             print(f"[MY_AGENT] No logs found at {self.base_path}")
             return entries
 
@@ -96,6 +97,16 @@ class AgentObserver:
 `self.<source>_parser`, so no per-source branch is needed — it handles the
 `has_meaningful_content()` filter, error isolation and `error.log` reporting for you.
 
+Register the source in `DIAGNOSTIC_SOURCES` in `adr_sensor/diagnostics.py` as well.
+At recovery points, call `self.record_diagnostic()` with a fixed code from
+`BaseParser.DIAGNOSTIC_CODES`, for example `record_decode_error` or `file_read_error`.
+Do not pass paths, input values, exception strings, or dynamically observed type
+names. The observer resets and aggregates counters per run, including zero-output
+runs. Standalone parser callers can use `reset_diagnostics()` and `get_diagnostics()`.
+Use `unsupported_*` codes only for explicit supported-format contracts; missing
+input, age skips, and incomplete live tails are not evidence of schema drift.
+Test both the recovered telemetry and diagnostic counts using synthetic data.
+
 If the agent only exists on some operating systems, add it to
 `PLATFORM_RESTRICTED_SOURCES` so it is skipped elsewhere instead of failing:
 

Sensor/README.md

@@ -23,6 +23,32 @@ ADR Sensor is a Python library that collects telemetry from AI coding agents to
 | **opencode**               | `opencode`       | SQLite (`opencode.db`) or JSON tree | macOS, Linux           |
 | **Gemini CLI**             | `gemini`         | JSONL journals + legacy JSON chats | macOS, Linux, Windows  |
 
+### Claude Code
+
+The `claude` source reads transcripts recursively under `~/.claude/projects/`,
+including [subagent transcripts](https://code.claude.com/docs/en/sub-agents#resume-subagents)
+under `<project>/<sessionId>/subagents/agent-<agentId>.jsonl` and nested workflow
+directories. Main sessions keep the `claude_<sessionId>` identity; subagents use
+`claude_<sessionId>_agent_<agentId>` and include `parent_session_id` and `agent_id`
+in `session_context` so their exports do not overwrite the parent conversation.
+
+String and text-block messages are retained, including user text accompanying
+tool results. Results are matched by tool-call ID. Malformed records are skipped
+without discarding surrounding messages. Complete JSON objects concatenated on
+one physical line and NUL padding between objects are accepted; incomplete tails
+are skipped without joining physical lines or repairing text inside a message.
+
+Each event includes `raw_log_path`, a stable conversation-start `timestamp`, and
+`session_context.last_event_at` and `event_count` for incremental updates. A file
+without any valid timestamp uses its modification time. `--save-sessions` updates
+the saved snapshot when a tool completes or a conversation resumes, even within
+the same timestamp second. All recorded branches are retained in file order.
+
+The default lookback is 14 days by file modification time. Existing limits still
+apply: top-level tool argument strings and tool results are truncated at 1,000
+characters; non-text content blocks and separately spilled tool-output files are
+not imported. Contract tests use synthetic transcripts and do not launch Claude.
+
 ### Claude Desktop Agent Mode
 
 The `claude_desktop` source covers Claude Desktop's local agent mode (released as
@@ -328,14 +354,92 @@ no redaction or field projection, so prompts, responses, tool arguments, tool
 results, usernames, hostnames, and local paths can be transmitted. Any
 normalization already performed by a source parser still applies.
 
-System-configuration records are sent as `adr.system.configuration` logs. Runs
-are not checkpointed specifically for OTLP: repeated runs can resend the same
-records, and consumers can use `adr.event.uuid` to deduplicate them.
+System-configuration records are sent as `adr.system.configuration` logs on each
+run. Sensor health logs are also sent on every run, even when all session snapshots
+are already acknowledged. With `--save-sessions`, successful session delivery is tracked independently
+of local session files. A failed export is retried on the next run, even when the
+local JSON already exists. A session is skipped only when its complete normalized
+payload was successfully exported to the same destination configuration. Changes
+to tool results, destination settings, or configured authentication headers cause
+a resend. The checkpoint also accounts for effective OTLP environment headers and
+mTLS client certificate/key paths. It does not read credential files: after
+changing certificate or key contents in place, remove the destination's checkpoint
+to resend sessions. Dynamic HTTP credential-provider plugins
+(`OTEL_PYTHON_EXPORTER_OTLP_HTTP_CREDENTIAL_PROVIDER` and its `LOGS` variant)
+are unsupported and cause an explicit error; use configured headers or mTLS.
+
+Delivery checkpoints are hidden `.adr-otel-delivery.<hash>.json` files in the
+session output directory. They contain only hashes, including a destination hash
+that accounts for authentication headers; they do not store raw URLs, credentials,
+session identifiers, or payloads. Missing, unreadable, or corrupt checkpoints cause
+sessions to be retried. The checkpoint is replaced atomically only after flush and
+shutdown succeed; a checkpoint write failure exits with an error. `--no-save`
+disables checkpoint reads and writes. Without `--save-sessions`, every run exports
+all collected sessions.
+
+The one-shot Sensor process drains bounded batches and reconciles submitted and
+successfully exported counts before reporting success, so a full SDK queue cannot
+silently drop records. HTTP success is also checked for an OTLP acknowledgement:
+partial rejection or a malformed response fails delivery and leaves the affected
+run unacknowledged. Resolve persistent collector rejection before rerunning: OTLP
+does not identify individual rejected records, so retrying can resend accepted
+records too. Export or checkpoint failures exit with a nonzero status.
+Delivery is at least once: a collector may receive data before a timeout, process
+interruption, or checkpoint write failure, so retries can duplicate records.
+Checkpointing only covers sessions that are collected again on a later run; it is
+not a persistent payload queue. Consumers can use `adr.event.uuid` and a full
+payload digest to identify repeated snapshots, since a session UUID alone does not
+necessarily change when tool results change.
 
-The one-shot Sensor process flushes and shuts down the exporter before exiting.
 Use an OpenTelemetry Collector when vendor-specific routing, transformation,
 retry, or persistent queuing is needed.
 
+### Sensor health and parser diagnostics
+
+Every ingestion run writes a content-free summary for each attempted source,
+including runs that produce no sessions. `diagnostics.jsonl` contains all summaries;
+`error.log` contains only `partial` and `failed` summaries. Both live under
+`--output-dir` (default `./output`), even when `--save-sessions` uses its separate
+default cache directory or `--no-save` suppresses captured session files. Each log
+rotates at 1 MiB with two backups. Use one active sensor process per output directory
+to avoid concurrent rotation races. Diagnostic write failures produce a fixed stderr
+warning and do not discard captured sessions.
+
+The versioned `adr.sensor.health` schema contains timestamp, sensor version, source,
+stage, status, fixed reason codes, and aggregate counts. For example:
+
+```json
+{"schema_version":1,"event":"adr.sensor.health","timestamp":"2026-01-01T00:00:00.000+00:00","sensor_version":"0.0.0","source":"claude","stage":"parse","status":"partial","suspected_schema_drift":false,"counts":{"events_returned":2,"events_emitted":2,"events_filtered":0},"reasons":{"record_decode_error":1}}
+```
+
+Statuses distinguish successful capture (`ok`), no meaningful output (`empty`),
+absent input (`no_input`), usable output with observed errors (`partial`), and
+errors without usable output (`failed`). Age filtering and an incomplete live
+tail are expected skips, not errors. `suspected_schema_drift` is a triage hint for
+explicitly unsupported record/content/schema shapes, not proof of an upstream
+format change. Generic corruption is reported separately.
+
+All ten parsers report observed recovery failures, but coverage is not exhaustive:
+some optional metadata/timestamp fallbacks, unknown record kinds, and compressed
+DSH tail recovery are not classified. Counts describe observed recovery operations,
+not necessarily unique damaged records. A healthy summary does not prove complete
+capture; a missing summary also cannot distinguish an idle endpoint from a sensor
+that never ran. Schedule runs and monitor last-seen health externally.
+
+With `--otel-config`, health is also sent as OTLP logs (`adr.event.type=sensor_health`),
+including when there are no session records. Health errors use WARN severity;
+expected skips use INFO. No OTLP exporter is created without that argument. A failed
+export is recorded locally because a broken destination cannot receive its own alert.
+`--fail-on-error` exits nonzero after preserving available capture when an observed
+parse/save/diagnostic failure occurs; by default these partial failures are reported
+without changing the existing continue-on-error behavior. OTLP failures remain nonzero.
+When `--resource` is enabled, `resource.log` also marks partial runs unsuccessful.
+
+New structured diagnostics never include prompts, tool arguments/results, paths,
+session IDs, exception messages, or tracebacks. This is a separate operational
+schema, **not redaction of captured telemetry**. Legacy console previews/errors and
+older entries already present in `error.log` are not sanitized by this change.
+
 ## Output Schema
 
 ### AgentEvent
@@ -515,8 +619,9 @@ cannot run on the current platform are skipped rather than failing.
 | `APPDATA`         | Cursor, Cline, Claude Desktop parsers | Windows roaming app-data root. Consulted first so redirected/roaming profiles resolve correctly (default `~/AppData/Roaming`) |
 | `LOCALAPPDATA`    | Warp parser                | Windows local app-data root, same redirected-profile handling (default `~/AppData/Local`) |
 
-Errors during ingestion never abort the run: each source is isolated, and failures
-are appended as single-line JSON records to `error.log` in the output directory.
+Each source is isolated during ingestion. See
+[Sensor health and parser diagnostics](#sensor-health-and-parser-diagnostics) for
+structured logs, partial-failure exit behavior, and monitoring limitations.
 
 ## Security Use Cases
 

Sensor/adr_sensor/cli.py

@@ -24,7 +24,9 @@
     resource_mod = None
 
 from . import __version__
+from .diagnostics import health_record, write_health_records
 from .exporters import OpenTelemetryConfigError, load_opentelemetry_config
+from .exporters.delivery_checkpoint import DeliveryCheckpoint, DeliveryCheckpointError
 from .exporters.opentelemetry import OpenTelemetryExportError, OpenTelemetryLogExporter
 from .observer import AgentObserver
 
@@ -88,7 +90,12 @@ def main():
         help="Directory to save output files (default: ./output)",
     )
     parser.add_argument("--limit", type=_non_negative_int, default=2, help="Number of entries to display")
-    parser.add_argument("--no-save", action="store_true", help="Do not save to file")
+    parser.add_argument(
+        "--no-save", action="store_true", help="Do not save captured sessions (diagnostics still written)"
+    )
+    parser.add_argument(
+        "--fail-on-error", action="store_true", help="Exit nonzero after partial capture or output failures"
+    )
     parser.add_argument(
         "--save-sessions",
         action="store_true",
@@ -130,6 +137,7 @@ def main():
 
     success = True
     observer = None
+    stage = "startup"
 
     try:
         # Determine max_age_days
@@ -140,9 +148,24 @@ def main():
         observer = AgentObserver(output_dir=args.output_dir, max_age_days=max_age_days)
 
         # Ingest logs
+        stage = "parse"
         entries, system_config_data = observer.ingest_all(args.source)
 
+        # A local file is not an OTLP acknowledgement. Keep remote candidates
+        # independent of local incremental filtering, including after a failed run.
+        otel_entries = entries
+        delivery_checkpoint = None
+        if otel_config is not None and args.save_sessions and not args.no_save:
+            stage = "export"
+            checkpoint_dir = args.output_dir if args.output_dir is not None else observer._get_default_session_dir()
+            delivery_checkpoint = DeliveryCheckpoint(checkpoint_dir, otel_config)
+            otel_entries = delivery_checkpoint.pending_entries(entries)
+            if delivery_checkpoint.load_failed:
+                observer.record_failure("export", "checkpoint_read_error")
+                print("OpenTelemetry delivery checkpoint unreadable or invalid; retrying sessions.", file=sys.stderr)
+
         # Apply incremental filtering
+        stage = "save"
         if args.save_sessions and entries:
             print("\nSession-based incremental mode: Checking existing session files...")
             original_count = len(entries)
@@ -154,6 +177,7 @@ def main():
         observer.display_summary(entries, system_config_data, limit=args.limit)
 
         # Save
+        stage = "save"
         if entries or system_config_data:
             if not args.no_save:
                 if args.save_sessions:
@@ -171,26 +195,65 @@ def main():
                         entries, system_config_data, output_format=args.output_format, output_dir=project_output_dir
                     )
 
-            if otel_config is not None:
-                otel_exporter = OpenTelemetryLogExporter(otel_config, service_version=get_version())
-                try:
-                    exported_count = otel_exporter.export(entries, system_config_data)
-                finally:
-                    otel_exporter.shutdown()
-                print(f"\nOpenTelemetry logs sent: {exported_count}")
-
-        print("\nADR Sensor complete!\n")
+        if otel_config is not None:
+            # Health records must reach monitoring even when parsing produced
+            # no sessions, or all local session snapshots were unchanged.
+            stage = "export"
+            otel_exporter = OpenTelemetryLogExporter(otel_config, service_version=get_version())
+            try:
+                exported_count = otel_exporter.export(otel_entries, system_config_data)
+                otel_exporter.export_diagnostics(observer.get_diagnostic_records())
+            finally:
+                otel_exporter.shutdown()
+            if delivery_checkpoint is not None:
+                delivery_checkpoint.commit()
+            print(f"\nOpenTelemetry session/configuration logs sent: {exported_count}")
+
+        success = observer.has_errors is not True
+        if success:
+            print("\nADR Sensor complete!\n")
+        else:
+            print("\nADR Sensor completed with errors; see diagnostics.jsonl.\n")
+            if args.fail_on_error:
+                raise SystemExit(1)
 
     except OpenTelemetryExportError as exc:
         success = False
+        if observer is not None:
+            observer.record_failure("export", "export_error")
         print(f"OpenTelemetry export failed: {exc}", file=sys.stderr)
         raise SystemExit(1)
 
+    except DeliveryCheckpointError as exc:
+        success = False
+        if observer is not None:
+            observer.record_failure("export", "checkpoint_write_error")
+        print(f"OpenTelemetry checkpoint failed: {exc}", file=sys.stderr)
+        raise SystemExit(1)
+
+    except Exception:
+        success = False
+        if observer is not None:
+            reason = {"parse": "parser_error", "save": "write_error", "export": "export_error"}.get(
+                stage, "startup_error"
+            )
+            observer.record_failure(stage, reason)
+        else:
+            write_health_records(
+                args.output_dir or Path.cwd() / "output",
+                [health_record("sensor", "startup", reasons={"startup_error": 1})],
+            )
+        raise
+
     except BaseException:
         success = False
         raise
 
     finally:
+        if observer is not None:
+            observer.flush_diagnostics()
+            if observer.has_errors is True:
+                success = False
         if capture_resource:
             try:
                 end_time = time.monotonic()