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

@@ -362,6 +362,52 @@ 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 +587,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,6 +24,7 @@
     resource_mod = None
 
 from . import __version__
+from .diagnostics import health_record, write_health_records
 from .exporters import OpenTelemetryConfigError, load_opentelemetry_config
 from .exporters.opentelemetry import OpenTelemetryExportError, OpenTelemetryLogExporter
 from .observer import AgentObserver
@@ -88,7 +89,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 +136,7 @@ def main():
 
     success = True
     observer = None
+    stage = "startup"
 
     try:
         # Determine max_age_days
@@ -140,6 +147,7 @@ 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)
 
         # Apply incremental filtering
@@ -154,6 +162,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 +180,56 @@ 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(entries, system_config_data)
+                otel_exporter.export_diagnostics(observer.get_diagnostic_records())
+            finally:
+                otel_exporter.shutdown()
+            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 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,122 @@
+"""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"})
+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

Sensor/adr_sensor/exporters/opentelemetry.py

@@ -6,6 +6,7 @@
 from datetime import datetime, timezone
 from typing import Any, List, Optional, Tuple
 
+from ..diagnostics import sanitize_health_record
 from ..schemas.agent_event_schema import AgentEvent
 from ..schemas.system_config_schema import SystemConfiguration
 from .config import OpenTelemetryConfig
@@ -63,6 +64,7 @@ def __init__(
         self._provider.add_log_record_processor(processor)
         self._logger = self._provider.get_logger("adr_sensor", service_version)
         self._info_severity = severity_number_cls.INFO
+        self._warning_severity = severity_number_cls.WARN
         self._flush_timeout_millis = int(config.flush_timeout_seconds * 1000)
         self._closed = False
 
@@ -101,6 +103,28 @@ def export(
 
         return len(entries) + len(system_config_data)
 
+    def export_diagnostics(self, records: List[dict]) -> int:
+        """Send bounded health summaries through the explicitly configured sink."""
+        for record in records:
+            body = sanitize_health_record(record)
+            degraded = body["status"] in {"partial", "failed"}
+            self._logger.emit(
+                timestamp=_datetime_to_unix_nanos(datetime.fromisoformat(body["timestamp"])),
+                observed_timestamp=time.time_ns(),
+                severity_number=self._warning_severity if degraded else self._info_severity,
+                severity_text="WARN" if degraded else "INFO",
+                body=body,
+                attributes={
+                    "adr.event.type": "sensor_health",
+                    "adr.schema.version": SCHEMA_VERSION,
+                    "adr.source": body["source"],
+                    "adr.sensor.stage": body["stage"],
+                    "adr.sensor.status": body["status"],
+                },
+                event_name="adr.sensor.health",
+            )
+        return len(records)
+
     def shutdown(self) -> None:
         """Flush pending records and stop the provider's worker thread."""
         if self._closed: