feat(sensor): report bounded parser health and suspected format drift

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

@@ -354,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
@@ -541,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()

Sensor/adr_sensor/diagnostics.py

@@ -0,0 +1,124 @@
+"""Bounded, content-free operational records, separate from captured sessions."""
+
+import json
+import logging
+import sys
+from datetime import datetime, timezone
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+from typing import Dict, Iterable, Optional
+
+from . import __version__
+from .parsers.base_parser import BaseParser
+
+DIAGNOSTIC_SOURCES = frozenset(
+    {"sensor", "claude", "claude_desktop", "cursor", "cline", "codex", "copilot", "dsh", "gemini", "opencode", "warp"}
+)
+DIAGNOSTIC_STAGES = frozenset({"parse", "save", "save_session", "export", "startup"})
+OPERATIONAL_REASONS = frozenset(
+    {"parser_error", "write_error", "export_error", "startup_error", "checkpoint_read_error", "checkpoint_write_error"}
+)
+COUNT_FIELDS = frozenset({"events_returned", "events_emitted", "events_filtered", "attempted", "succeeded", "failed"})
+MAX_LOG_BYTES = 1024 * 1024
+LOG_BACKUP_COUNT = 2
+MAX_COUNT = 2**63 - 1
+
+
+def _counts(values: Dict[str, int], allowed: Iterable[str]) -> Dict[str, int]:
+    """Accept only fixed keys and bounded integers, never caller-provided text."""
+    allowed = frozenset(allowed)
+    if not isinstance(values, dict):
+        return {}
+    return {
+        key: min(value, MAX_COUNT)
+        for key, value in values.items()
+        if key in allowed and isinstance(value, int) and not isinstance(value, bool) and value >= 0
+    }
+
+
+def health_record(
+    source: str,
+    stage: str,
+    *,
+    counts: Optional[Dict[str, int]] = None,
+    reasons: Optional[Dict[str, int]] = None,
+) -> dict:
+    """Summarize observed issues without claiming that every issue is schema drift."""
+    safe_counts = _counts(counts or {}, COUNT_FIELDS)
+    safe_reasons = _counts(reasons or {}, BaseParser.DIAGNOSTIC_CODES | OPERATIONAL_REASONS)
+    issues = sum(count for reason, count in safe_reasons.items() if reason not in BaseParser.EXPECTED_DIAGNOSTIC_CODES)
+    emitted = safe_counts.get("events_emitted", safe_counts.get("succeeded", 0))
+    if issues:
+        status = "partial" if emitted else "failed"
+    elif safe_reasons.get("input_missing") and not emitted:
+        status = "no_input"
+    elif not emitted and stage == "parse":
+        status = "empty"
+    else:
+        status = "ok"
+    return {
+        "schema_version": 1,
+        "event": "adr.sensor.health",
+        "timestamp": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
+        "sensor_version": __version__,
+        "source": source if isinstance(source, str) and source in DIAGNOSTIC_SOURCES else "sensor",
+        "stage": stage if isinstance(stage, str) and stage in DIAGNOSTIC_STAGES else "startup",
+        "status": status,
+        "suspected_schema_drift": any(
+            count and reason in {"unsupported_schema", "unsupported_record_type", "unsupported_content_block"}
+            for reason, count in safe_reasons.items()
+        ),
+        "counts": safe_counts,
+        "reasons": safe_reasons,
+    }
+
+
+def sanitize_health_record(record: dict) -> dict:
+    """Revalidate the fixed schema at each serialization boundary."""
+    safe = health_record(
+        record.get("source"), record.get("stage"), counts=record.get("counts"), reasons=record.get("reasons")
+    )
+    try:
+        timestamp = datetime.fromisoformat(record.get("timestamp", ""))
+        if timestamp.tzinfo is not None:
+            safe["timestamp"] = timestamp.astimezone(timezone.utc).isoformat(timespec="milliseconds")
+    except (TypeError, ValueError):
+        pass
+    return safe
+
+
+def write_health_records(output_dir: Path, records: Iterable[dict]) -> bool:
+    """Append rotating JSONL diagnostics; logging failures never erase capture."""
+    handlers = []
+    try:
+        output_dir.mkdir(parents=True, exist_ok=True)
+        for name in ("diagnostics.jsonl", "error.log"):
+            handler = RotatingFileHandler(
+                output_dir / name, maxBytes=MAX_LOG_BYTES, backupCount=LOG_BACKUP_COUNT, encoding="utf-8", delay=True
+            )
+            handler.setFormatter(logging.Formatter("%(message)s"))
+
+            # Handler.emit normally suppresses write failures. Surface them to
+            # the single bounded fallback below instead of logging record data.
+            def handle_error(record):
+                raise OSError("diagnostic write failed")
+
+            handler.handleError = handle_error
+            handlers.append(handler)
+        for record in records:
+            record = sanitize_health_record(record)
+            message = json.dumps(record, ensure_ascii=True, separators=(",", ":"))
+            item = logging.LogRecord("adr_sensor.health", logging.INFO, "", 0, message, (), None)
+            handlers[0].handle(item)
+            if record["status"] in {"partial", "failed"}:
+                handlers[1].handle(item)
+        return True
+    except Exception:
+        print("[ADR] Unable to write sensor diagnostics; captured session data is unaffected.", file=sys.stderr)
+        return False
+    finally:
+        for handler in handlers:
+            try:
+                handler.close()
+            except Exception:
+                pass