diff --git a/Sensor/CONTRIBUTING.md b/Sensor/CONTRIBUTING.md index 898c610..1931b3e 100644 --- a/Sensor/CONTRIBUTING.md +++ b/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._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: diff --git a/Sensor/README.md b/Sensor/README.md index 7e3e92c..44d71bf 100644 --- a/Sensor/README.md +++ b/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..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 diff --git a/Sensor/adr_sensor/cli.py b/Sensor/adr_sensor/cli.py index 511b763..0ed99b5 100644 --- a/Sensor/adr_sensor/cli.py +++ b/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() diff --git a/Sensor/adr_sensor/diagnostics.py b/Sensor/adr_sensor/diagnostics.py new file mode 100644 index 0000000..08e8ddb --- /dev/null +++ b/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 diff --git a/Sensor/adr_sensor/exporters/delivery_checkpoint.py b/Sensor/adr_sensor/exporters/delivery_checkpoint.py new file mode 100644 index 0000000..cadeef8 --- /dev/null +++ b/Sensor/adr_sensor/exporters/delivery_checkpoint.py @@ -0,0 +1,120 @@ +"""Hash-only acknowledgements for incremental OTLP session delivery.""" + +import hashlib +import json +import os +import re +import tempfile +from dataclasses import asdict +from pathlib import Path +from typing import Dict, List + +from ..schemas.agent_event_schema import AgentEvent +from .config import OpenTelemetryConfig +from .opentelemetry import SCHEMA_VERSION, _validate_credential_provider + +_DIGEST = re.compile(r"[0-9a-f]{64}\Z") + + +class DeliveryCheckpointError(RuntimeError): + """Raised when a successful export cannot be durably checkpointed.""" + + +def _fingerprint(value: object) -> str: + serialized = json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest() + + +class DeliveryCheckpoint: + """Remember delivered session snapshots separately from their local JSON files. + + Select pending entries before export, then call commit only after flush and + shutdown succeed. Losing or corrupting this cache causes retries, never skips. + """ + + def __init__(self, output_dir: Path, config: OpenTelemetryConfig): + _validate_credential_provider() + destination = { + "config": asdict(config), + "schema_version": SCHEMA_VERSION, + "checkpoint_version": 1, + } + if not config.headers: + # The HTTP exporter falls back to these variables for empty headers. + destination["environment_headers"] = os.environ.get( + "OTEL_EXPORTER_OTLP_LOGS_HEADERS", os.environ.get("OTEL_EXPORTER_OTLP_HEADERS", "") + ) + for setting in ("CLIENT_CERTIFICATE", "CLIENT_KEY"): + destination[setting] = os.environ.get( + f"OTEL_EXPORTER_OTLP_LOGS_{setting}", os.environ.get(f"OTEL_EXPORTER_OTLP_{setting}", "") + ) + self.path = Path(output_dir) / f".adr-otel-delivery.{_fingerprint(destination)}.json" + self.load_failed = False + self._delivered = self._load() + self._pending: Dict[str, str] = {} + + def _load(self) -> Dict[str, str]: + try: + with self.path.open(encoding="utf-8") as checkpoint_file: + data = json.load(checkpoint_file) + if not isinstance(data, dict) or not all( + isinstance(key, str) and _DIGEST.fullmatch(key) and isinstance(value, str) and _DIGEST.fullmatch(value) + for key, value in data.items() + ): + raise ValueError("invalid checkpoint fingerprints") + return data + except FileNotFoundError: + return {} + except (OSError, ValueError): + self.load_failed = True + return {} + + def pending_entries(self, entries: List[AgentEvent]) -> List[AgentEvent]: + """Select snapshots not acknowledged for this destination, including results.""" + self._pending = {} + pending_entries = [] + for entry in entries: + identity = _fingerprint([entry.source, entry.session_id, entry.hostname, entry.username]) + payload = _fingerprint(entry.get_non_null_fields()) + if self._delivered.get(identity) != payload: + pending_entries.append(entry) + self._pending[identity] = payload + return pending_entries + + def commit(self) -> None: + """Atomically persist only the snapshots selected for a successful export.""" + if not self._pending: + return + temporary_path = None + try: + self.path.parent.mkdir(parents=True, exist_ok=True) + # Merge recent acknowledgements; concurrent writers may cause extra + # retries, but can never acknowledge a payload they have not exported. + delivered = self._load() + delivered.update(self._pending) + fd, temporary_name = tempfile.mkstemp(prefix=f"{self.path.name}.", suffix=".tmp", dir=self.path.parent) + temporary_path = Path(temporary_name) + with os.fdopen(fd, "w", encoding="utf-8") as checkpoint_file: + json.dump(delivered, checkpoint_file, sort_keys=True, separators=(",", ":")) + checkpoint_file.flush() + os.fsync(checkpoint_file.fileno()) + os.replace(temporary_path, self.path) + temporary_path = None + if os.name != "nt": + directory_fd = os.open(self.path.parent, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + self._delivered = delivered + self._pending = {} + except OSError as exc: + raise DeliveryCheckpointError( + "could not save the OpenTelemetry delivery checkpoint; a later run may resend delivered sessions" + ) from exc + finally: + if temporary_path is not None: + try: + temporary_path.unlink() + except OSError: + pass diff --git a/Sensor/adr_sensor/exporters/opentelemetry.py b/Sensor/adr_sensor/exporters/opentelemetry.py index d4bd14e..a3618fe 100644 --- a/Sensor/adr_sensor/exporters/opentelemetry.py +++ b/Sensor/adr_sensor/exporters/opentelemetry.py @@ -1,16 +1,19 @@ """OTLP/HTTP logs exporter for normalized ADR Sensor records.""" +import os import socket import threading import time 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 SCHEMA_VERSION = "1" +_MAX_BATCH_SIZE = 512 class OpenTelemetryExportError(RuntimeError): @@ -28,6 +31,7 @@ def __init__( _log_record_exporter: Optional[Any] = None, _processor_factory: Optional[Any] = None, ): + _validate_credential_provider() components = _load_opentelemetry_components() ( logger_provider_cls, @@ -54,24 +58,37 @@ def __init__( certificate_file=config.certificate_file, headers=config.headers, timeout=config.timeout_seconds, + session=_create_otlp_session(), ) self._record_exporter = _ExportStatusTracker(record_exporter, export_result_cls.SUCCESS) - processor_factory = _processor_factory or processor_cls - processor = processor_factory(self._record_exporter) + if _processor_factory is None: + # Bound the queue explicitly instead of inheriting environment defaults. + # _emit() drains it before submitting another batch, providing backpressure. + processor = processor_cls( + self._record_exporter, + max_queue_size=_MAX_BATCH_SIZE, + max_export_batch_size=_MAX_BATCH_SIZE, + ) + else: + processor = _processor_factory(self._record_exporter) 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 + self._submitted_count = 0 + self._flushed_count = 0 + self._emit_lock = threading.Lock() def export( self, entries: List[AgentEvent], system_config_data: List[SystemConfiguration], ) -> int: - """Queue complete, unredacted Sensor records for OTLP export.""" + """Submit complete Sensor records; shutdown must succeed to confirm delivery.""" for entry in entries: self._emit( body=entry.get_non_null_fields(), @@ -101,33 +118,88 @@ 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._emit( + timestamp=datetime.fromisoformat(body["timestamp"]), + 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: - return + with self._emit_lock: + if self._closed: + return + + self._closed = True + try: + self._flush() + finally: + try: + self._provider.shutdown() + except Exception as exc: + raise OpenTelemetryExportError("OpenTelemetry exporter shutdown failed") from exc + self._check_delivery() + + def _check_delivery(self) -> None: + if self._record_exporter.failed: + raise OpenTelemetryExportError("the OTLP endpoint did not accept one or more log batches") + exported_count = self._record_exporter.exported_count + if exported_count != self._submitted_count: + raise OpenTelemetryExportError( + f"OpenTelemetry delivery count mismatch: submitted {self._submitted_count}, " + f"successfully exported {exported_count}" + ) - self._closed = True - flushed = False + def _flush(self) -> None: try: flushed = self._provider.force_flush(timeout_millis=self._flush_timeout_millis) - finally: - self._provider.shutdown() - + except Exception as exc: + raise OpenTelemetryExportError("OpenTelemetry logs could not be flushed") from exc if not flushed: raise OpenTelemetryExportError(f"OpenTelemetry logs did not flush within {self._flush_timeout_millis} ms") - if self._record_exporter.failed: - raise OpenTelemetryExportError("the OTLP endpoint did not accept one or more log batches") + self._check_delivery() + self._flushed_count = self._submitted_count - def _emit(self, body: dict, timestamp: datetime, event_name: str, attributes: dict) -> None: - self._logger.emit( - timestamp=_datetime_to_unix_nanos(timestamp), - observed_timestamp=time.time_ns(), - severity_number=self._info_severity, - severity_text="INFO", - body=body, - attributes=attributes, - event_name=event_name, - ) + def _emit( + self, + body: dict, + timestamp: datetime, + event_name: str, + attributes: dict, + *, + severity_number: Optional[Any] = None, + severity_text: str = "INFO", + ) -> None: + with self._emit_lock: + if self._closed: + raise OpenTelemetryExportError("OpenTelemetry exporter is already shut down") + self._logger.emit( + timestamp=_datetime_to_unix_nanos(timestamp), + observed_timestamp=time.time_ns(), + severity_number=self._info_severity if severity_number is None else severity_number, + severity_text=severity_text, + body=body, + attributes=attributes, + event_name=event_name, + ) + self._submitted_count += 1 + if self._submitted_count - self._flushed_count >= _MAX_BATCH_SIZE: + self._flush() def _datetime_to_unix_nanos(value: datetime) -> int: @@ -139,13 +211,61 @@ def _datetime_to_unix_nanos(value: datetime) -> int: return ((delta.days * 86400 + delta.seconds) * 1_000_000_000) + (delta.microseconds * 1000) +def _validate_credential_provider() -> None: + """Reject opaque authentication that cannot be identified in checkpoints.""" + if os.environ.get("OTEL_PYTHON_EXPORTER_OTLP_HTTP_CREDENTIAL_PROVIDER") or os.environ.get( + "OTEL_PYTHON_EXPORTER_OTLP_HTTP_LOGS_CREDENTIAL_PROVIDER" + ): + raise OpenTelemetryExportError( + "OpenTelemetry HTTP credential-provider plugins are unsupported; use explicit headers or mTLS settings" + ) + + +def _create_otlp_session() -> Any: + """Validate the collector acknowledgement before the SDK counts HTTP success. + + The pinned SDK treats any successful HTTP response as full batch success, + including OTLP partial rejection. Its public session hook lets us check the + protobuf response without overriding the SDK's private transport methods. + """ + import requests + from google.protobuf.message import DecodeError + from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import ExportLogsServiceResponse + + def validate_response(response: Any, *args: Any, **kwargs: Any) -> Any: + if response.is_redirect: + return response + if not 200 <= response.status_code < 300: + # Requests follows real redirects; other 3xx responses are not OTLP + # acknowledgements even though the SDK's Response.ok accepts them. + if response.ok: + raise OpenTelemetryExportError("the OTLP endpoint returned an invalid acknowledgement") + return response + if not response.content: + return response + acknowledgement = ExportLogsServiceResponse() + try: + acknowledgement.ParseFromString(response.content) + except DecodeError: + raise OpenTelemetryExportError("the OTLP endpoint returned an invalid acknowledgement") from None + if acknowledgement.partial_success.rejected_log_records != 0: + # Do not log the collector's error_message; it may contain payloads. + raise OpenTelemetryExportError("the OTLP endpoint rejected one or more log records") + return response + + session = requests.Session() + session.hooks["response"].append(validate_response) + return session + + class _ExportStatusTracker: - """Track exporter failures that OpenTelemetry's batch processor otherwise ignores.""" + """Count successful records and track failures the batch processor ignores.""" def __init__(self, exporter: Any, success_result: Any): self._exporter = exporter self._success_result = success_result self._failed = False + self._exported_count = 0 self._lock = threading.Lock() @property @@ -153,6 +273,11 @@ def failed(self) -> bool: with self._lock: return self._failed + @property + def exported_count(self) -> int: + with self._lock: + return self._exported_count + def export(self, batch: Any) -> Any: try: result = self._exporter.export(batch) @@ -161,8 +286,10 @@ def export(self, batch: Any) -> Any: self._failed = True raise - if result != self._success_result: - with self._lock: + with self._lock: + if result == self._success_result: + self._exported_count += len(batch) + else: self._failed = True return result diff --git a/Sensor/adr_sensor/observer.py b/Sensor/adr_sensor/observer.py index f8fbdb3..2917dbe 100644 --- a/Sensor/adr_sensor/observer.py +++ b/Sensor/adr_sensor/observer.py @@ -13,15 +13,15 @@ import re import secrets import stat -import sys import time -import traceback from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from tabulate import tabulate +from .diagnostics import DIAGNOSTIC_SOURCES, MAX_COUNT, health_record, write_health_records +from .parsers.base_parser import BaseParser from .parsers.claude_desktop_parser import ClaudeDesktopParser from .parsers.claude_parser import ClaudeParser from .parsers.cline_parser import ClineParser @@ -88,9 +88,7 @@ def __init__(self, output_dir: Optional[Path] = None, max_age_days: Optional[int ClaudeDesktopParser(max_age_days=max_age_days) if max_age_days is not None else ClaudeDesktopParser() ) self.codex_parser = CodexParser(max_age_days=max_age_days) if max_age_days is not None else CodexParser() - self.copilot_parser = ( - CopilotParser(max_age_days=max_age_days) if max_age_days is not None else CopilotParser() - ) + self.copilot_parser = CopilotParser(max_age_days=max_age_days) if max_age_days is not None else CopilotParser() self.dsh_parser = DshParser(max_age_days=max_age_days) if max_age_days is not None else DshParser() self.cline_parser = ClineParser(max_age_days=max_age_days) if max_age_days is not None else ClineParser() self.warp_parser = WarpParser(max_age_days=max_age_days) if max_age_days is not None else WarpParser() @@ -101,22 +99,64 @@ def __init__(self, output_dir: Optional[Path] = None, max_age_days: Optional[int self.output_dir = output_dir if output_dir else Path("output") self.gemini_parser = GeminiParser(max_age_days=max_age_days) if max_age_days is not None else GeminiParser() self.output_dir.mkdir(exist_ok=True) + self._diagnostic_records: List[dict] = [] + self._diagnostics_flushed = 0 + self._diagnostic_write_failed = False def _emit_error(self, error_payload: Dict[str, Any]) -> None: - """Append a single-line JSON error record to error.log. Best-effort, never raises.""" - try: - record = { - "timestamp": datetime.utcnow().isoformat(timespec="milliseconds") + "Z", - "host_os": platform.system(), - "python_version": sys.version.split()[0], - "pid": os.getpid(), - } - record.update(error_payload) - log_path = self.output_dir / "error.log" - with open(log_path, "a", encoding="utf-8") as f: - f.write(json.dumps(record, separators=(",", ":"), ensure_ascii=False) + "\n") - except Exception: - pass + """Compatibility adapter: never retain exception text or session data.""" + stage = error_payload.get("stage", "startup") + if stage in {"compare_session", "remove_stale_session"}: + stage = "save" + reason = {"parse": "parser_error", "save_session": "write_error", "export": "export_error"}.get( + stage, "startup_error" + ) + if stage == "save": + reason = "write_error" + self.record_failure(stage, reason, source=error_payload.get("source", "sensor")) + + @property + def has_errors(self) -> bool: + """Whether this run observed incomplete capture, output, or delivery.""" + return self._diagnostic_write_failed or any( + record["status"] in {"partial", "failed"} for record in self._diagnostic_records + ) + + def get_diagnostic_records(self) -> List[dict]: + """Return content-free health records for optional remote monitoring.""" + return [ + {**record, "counts": dict(record["counts"]), "reasons": dict(record["reasons"])} + for record in self._diagnostic_records + ] + + def record_failure(self, stage: str, reason: str, *, source: str = "sensor") -> None: + record = health_record(source, stage, counts={"failed": 1}, reasons={reason: 1}) + for pending in self._diagnostic_records[self._diagnostics_flushed :]: + if (pending["source"], pending["stage"], set(pending["reasons"])) == ( + record["source"], + record["stage"], + set(record["reasons"]), + ) and set(pending["counts"]) == {"failed"}: + pending["counts"]["failed"] = min(pending["counts"]["failed"] + 1, MAX_COUNT) + for code in record["reasons"]: + pending["reasons"][code] = min(pending["reasons"][code] + 1, MAX_COUNT) + return + self._diagnostic_records.append(record) + + def flush_diagnostics(self) -> bool: + """Persist each summary once; keep failures visible without stopping capture.""" + pending = self._diagnostic_records[self._diagnostics_flushed :] + if self._diagnostic_write_failed: + return False + if not pending: + return True + written = write_health_records(self.output_dir, pending) + if not written: + self._diagnostic_write_failed = True + self.record_failure("save", "write_error") + # Do not repeatedly flood stderr if the diagnostic destination is broken. + self._diagnostics_flushed = len(self._diagnostic_records) + return written def _get_default_session_dir(self) -> Path: """Get the default directory for session files.""" @@ -128,9 +168,7 @@ def _get_default_session_dir(self) -> Path: return cache_dir / "adr_sensor" - def ingest_all( - self, source_filter: str = "all" - ) -> Tuple[List[AgentEvent], List[SystemConfiguration]]: + def ingest_all(self, source_filter: str = "all") -> Tuple[List[AgentEvent], List[SystemConfiguration]]: """Ingest logs from all supported sources. Args: @@ -142,6 +180,9 @@ def ingest_all( """ all_entries: List[AgentEvent] = [] system_config_data: List[SystemConfiguration] = [] + self._diagnostic_records = [] + self._diagnostics_flushed = 0 + self._diagnostic_write_failed = False print("\n" + "=" * 80) print("ADR Sensor Starting...") @@ -158,21 +199,41 @@ def ingest_all( continue print(f"Ingesting {label} logs...") + parser_instance = getattr(self, f"{source}_parser") + if isinstance(parser_instance, BaseParser): + parser_instance.reset_diagnostics() + entries = [] + filtered = [] + parser_failed = False try: - entries = getattr(self, f"{source}_parser").parse_all() + parsed = parser_instance.parse_all() + if not isinstance(parsed, list): + raise TypeError("parser must return a list of events") + entries = parsed filtered = [e for e in entries if e.has_meaningful_content()] all_entries.extend(filtered) print(f"Found {len(filtered)} entries\n") except Exception as e: print(f"Error ingesting {label} logs: {e}") - self._emit_error({ - "source": source, - "stage": "parse", - "error_type": e.__class__.__name__, - "message": str(e), - "trace": traceback.format_exc(limit=5), - }) + parser_failed = True + finally: + reasons = parser_instance.get_diagnostics() if isinstance(parser_instance, BaseParser) else {} + if parser_failed: + reasons["parser_error"] = reasons.get("parser_error", 0) + 1 + self._diagnostic_records.append( + health_record( + source, + "parse", + counts={ + "events_returned": len(entries), + "events_emitted": len(filtered), + "events_filtered": len(entries) - len(filtered), + }, + reasons=reasons, + ) + ) + self.flush_diagnostics() return all_entries, system_config_data def display_summary( @@ -206,8 +267,7 @@ def display_summary( else: msg_count = sum(len(e.chat_history) for e in source_entries) tool_count = sum( - sum(len(msg.tools) for msg in e.chat_history if msg.role == "assistant") - for e in source_entries + sum(len(msg.tools) for msg in e.chat_history if msg.role == "assistant") for e in source_entries ) summary_data.append([source.upper(), len(source_entries), msg_count, tool_count]) @@ -309,6 +369,8 @@ def save_sessions_to_individual_files( output_dir.mkdir(parents=True, exist_ok=True) saved_files = [] + save_failures: Dict[str, int] = {} + save_successes: Dict[str, int] = {} session_file_index = ( self._build_session_file_index(output_dir) if any(entry.source in self.CONTENT_AWARE_INCREMENTAL_SOURCES for entry in entries) @@ -338,10 +400,7 @@ def save_sessions_to_individual_files( ) filename = file_path.name fresh_target = self._session_file_info(file_path) - if ( - fresh_target is not None - and fresh_target["data"].get("session_id") == entry.session_id - ): + if fresh_target is not None and fresh_target["data"].get("session_id") == entry.session_id: existing_info = self._newer_session_file(existing_info, fresh_target) if self._session_revision_regresses(entry, existing_info): print(f"Skipped stale session: {filename}") @@ -368,18 +427,13 @@ def save_sessions_to_individual_files( self._index_session_file(session_file_index, file_path, entry_data) saved_files.append(file_path) + source = entry.source if entry.source in DIAGNOSTIC_SOURCES else "sensor" + save_successes[source] = save_successes.get(source, 0) + 1 print(f"Saved session: {filename}") except Exception as e: print(f"Error saving session {filename}: {e}") - self._emit_error( - { - "source": entry.source, - "stage": "save_session", - "error_type": e.__class__.__name__, - "message": str(e), - "session_id": entry.session_id, - } - ) + source = entry.source if entry.source in DIAGNOSTIC_SOURCES else "sensor" + save_failures[source] = save_failures.get(source, 0) + 1 finally: if temp_path is not None and temp_path.exists(): try: @@ -389,6 +443,18 @@ def save_sessions_to_individual_files( if lock_fd is not None and lock_path is not None: self._release_session_lock(lock_fd) + for source in sorted(set(save_successes) | set(save_failures)): + failed = save_failures.get(source, 0) + succeeded = save_successes.get(source, 0) + self._diagnostic_records.append( + health_record( + source, + "save_session", + counts={"attempted": failed + succeeded, "succeeded": succeeded, "failed": failed}, + reasons={"write_error": failed} if failed else {}, + ) + ) + self.flush_diagnostics() print(f"\nSaved {len(saved_files)} sessions to: {output_dir}") return saved_files @@ -518,18 +584,19 @@ def _newer_session_file( candidate_event_count = AgentObserver._session_file_event_count(candidate) current_event_count = AgentObserver._session_file_event_count(current) if ( - candidate_revision is not None - and (current_revision is None or candidate_revision > current_revision) - ) or ( - candidate_revision == current_revision - and ( - candidate_event_count is not None - and (current_event_count is None or candidate_event_count > current_event_count) + (candidate_revision is not None and (current_revision is None or candidate_revision > current_revision)) + or ( + candidate_revision == current_revision + and ( + candidate_event_count is not None + and (current_event_count is None or candidate_event_count > current_event_count) + ) + ) + or ( + candidate_revision == current_revision + and candidate_event_count == current_event_count + and candidate["timestamp"] > current["timestamp"] ) - ) or ( - candidate_revision == current_revision - and candidate_event_count == current_event_count - and candidate["timestamp"] > current["timestamp"] ): return candidate return current @@ -635,9 +702,8 @@ def _resolve_session_file_path( preferred = output_dir / f"adr.{filename_session_id}.{timestamp_str}.json" if existing_info is not None and format_timestamp_for_filename(existing_info["timestamp"]) == timestamp_str: existing_session_part = existing_info["file_path"].name[4:-5].rsplit(".", 1)[0] - if ( - existing_session_part == filename_session_id - or existing_session_part.startswith(f"{filename_session_id}_") + if existing_session_part == filename_session_id or existing_session_part.startswith( + f"{filename_session_id}_" ): return existing_info["file_path"] @@ -657,9 +723,7 @@ def _resolve_session_file_path( return alternate counter += 1 - def _session_revision_regresses( - self, entry: AgentEvent, existing_info: Optional[Dict[str, Any]] - ) -> bool: + def _session_revision_regresses(self, entry: AgentEvent, existing_info: Optional[Dict[str, Any]]) -> bool: """Prevent an older concurrent parse from replacing a newer snapshot.""" if existing_info is None: return False @@ -862,9 +926,19 @@ def _get_existing_session_files(self, output_dir: Optional[Path] = None) -> Dict def _clean_filename(self, session_id: str) -> str: """Clean session_id for use in filename.""" replacements = { - "\n": "_", "\r": "_", "\t": "_", - "/": "_", "\\": "_", ":": "_", "*": "_", "?": "_", - '"': "_", "<": "_", ">": "_", "|": "_", " ": "_", + "\n": "_", + "\r": "_", + "\t": "_", + "/": "_", + "\\": "_", + ":": "_", + "*": "_", + "?": "_", + '"': "_", + "<": "_", + ">": "_", + "|": "_", + " ": "_", } clean_id = session_id diff --git a/Sensor/adr_sensor/parsers/base_parser.py b/Sensor/adr_sensor/parsers/base_parser.py index 2cf6586..3cdfc5d 100644 --- a/Sensor/adr_sensor/parsers/base_parser.py +++ b/Sensor/adr_sensor/parsers/base_parser.py @@ -6,7 +6,7 @@ """ from abc import ABC, abstractmethod -from typing import List +from typing import Dict, List from ..schemas.agent_event_schema import AgentEvent @@ -24,6 +24,44 @@ def parse_all(self) -> List[AgentEvent]: ... """ + # Closed vocabulary keeps diagnostic size bounded and prevents input content + # (paths, exception messages, record types, or credentials) becoming labels. + EXPECTED_DIAGNOSTIC_CODES = frozenset({"input_missing", "file_age_skipped", "incomplete_record"}) + DIAGNOSTIC_CODES = EXPECTED_DIAGNOSTIC_CODES | frozenset( + { + "file_read_error", + "file_stat_error", + "record_decode_error", + "record_shape_error", + "unsupported_record_type", + "unsupported_schema", + "unsupported_content_block", + "invalid_timestamp", + "session_build_error", + "database_error", + "parser_error", + } + ) + + def reset_diagnostics(self) -> None: + """Start a new observation window without changing captured telemetry.""" + self._diagnostics: Dict[str, int] = {} + + def record_diagnostic(self, code: str, count: int = 1) -> None: + """Count an expected skip or recovery using a fixed, content-free code.""" + if not isinstance(code, str) or code not in self.DIAGNOSTIC_CODES: + raise ValueError("unknown parser diagnostic code") + if isinstance(count, bool) or not isinstance(count, int) or count < 1: + raise ValueError("parser diagnostic count must be a positive integer") + # Existing parser constructors do not need to call super().__init__(). + if not hasattr(self, "_diagnostics"): + self.reset_diagnostics() + self._diagnostics[code] = min(self._diagnostics.get(code, 0) + count, 2**63 - 1) + + def get_diagnostics(self) -> Dict[str, int]: + """Return a snapshot; callers cannot mutate the parser's counters.""" + return dict(getattr(self, "_diagnostics", {})) + @abstractmethod def parse_all(self) -> List[AgentEvent]: """Parse all available logs and return a list of AgentEvent objects. diff --git a/Sensor/adr_sensor/parsers/claude_desktop_parser.py b/Sensor/adr_sensor/parsers/claude_desktop_parser.py index af937c2..37bf380 100644 --- a/Sensor/adr_sensor/parsers/claude_desktop_parser.py +++ b/Sensor/adr_sensor/parsers/claude_desktop_parser.py @@ -64,6 +64,7 @@ def parse_all(self) -> List[AgentEvent]: entries: List[AgentEvent] = [] if not self.base_path.exists(): + self.record_diagnostic("input_missing") print(f"[CLAUDE_DESKTOP] No sessions found at {self.base_path}") return entries @@ -86,9 +87,11 @@ def parse_all(self) -> List[AgentEvent]: try: activity_time = datetime.fromtimestamp(last_activity / 1000, tz=timezone.utc) if activity_time < cutoff_time: + self.record_diagnostic("file_age_skipped") skipped_count += 1 continue except (ValueError, OSError, OverflowError, TypeError): + self.record_diagnostic("invalid_timestamp") pass # If we cannot parse it, process the session anyway. audit_path = session_dir / "audit.jsonl" @@ -101,6 +104,7 @@ def parse_all(self) -> List[AgentEvent]: processed_count += 1 except Exception as e: + self.record_diagnostic("session_build_error") print(f"[CLAUDE_DESKTOP] Error parsing session {session_dir}: {e}") if skipped_count > 0: @@ -143,12 +147,12 @@ def _discover_sessions(self) -> List[Tuple[Path, Path]]: sessions.extend(self._collect_sessions(agent_dir, DISPATCH_DIR_PREFIX)) except (PermissionError, OSError) as e: + self.record_diagnostic("file_read_error") print(f"[CLAUDE_DESKTOP] Error scanning base path {self.base_path}: {e}") return sessions - @staticmethod - def _collect_sessions(parent: Path, prefix: str) -> List[Tuple[Path, Path]]: + def _collect_sessions(self, parent: Path, prefix: str) -> List[Tuple[Path, Path]]: """Collect (session_dir, metadata_path) pairs directly under `parent`.""" found: List[Tuple[Path, Path]] = [] try: @@ -159,11 +163,11 @@ def _collect_sessions(parent: Path, prefix: str) -> List[Tuple[Path, Path]]: continue found.append((item, parent / f"{item.name}.json")) except (PermissionError, OSError) as e: + self.record_diagnostic("file_read_error") print(f"[CLAUDE_DESKTOP] Error scanning {parent}: {e}") return found - @staticmethod - def _read_session_metadata(metadata_path: Path) -> Optional[Dict[str, Any]]: + def _read_session_metadata(self, metadata_path: Path) -> Optional[Dict[str, Any]]: """Read a session metadata JSON file. Returns an empty dict (rather than None) when the file is missing or @@ -175,8 +179,13 @@ def _read_session_metadata(metadata_path: Path) -> Optional[Dict[str, Any]]: try: with open(metadata_path, encoding="utf-8") as f: data = json.load(f) + if not isinstance(data, dict): + self.record_diagnostic("record_shape_error") return data if isinstance(data, dict) else {} - except (json.JSONDecodeError, OSError, PermissionError): + except (json.JSONDecodeError, OSError, PermissionError) as exc: + self.record_diagnostic( + "record_decode_error" if isinstance(exc, json.JSONDecodeError) else "file_read_error" + ) return {} @staticmethod @@ -231,11 +240,13 @@ def _resolve_timestamp(self, audit_path: Path, metadata: Dict[str, Any]) -> date try: return datetime.fromtimestamp(value / 1000, tz=timezone.utc) except (ValueError, OSError, OverflowError, TypeError) as e: + self.record_diagnostic("invalid_timestamp") print(f"[CLAUDE_DESKTOP] Error parsing timestamp from {key}={value}: {e}") try: return datetime.fromtimestamp(audit_path.stat().st_mtime, tz=timezone.utc) except OSError: + self.record_diagnostic("file_stat_error") return datetime.now(timezone.utc) def _build_session_context( @@ -318,6 +329,7 @@ def _parse_session(self, audit_path: Path, metadata: Dict[str, Any]) -> Optional try: obj = json.loads(line) except json.JSONDecodeError: + self.record_diagnostic("record_decode_error") continue extracted = self._extract_message_data(obj) @@ -329,6 +341,7 @@ def _parse_session(self, audit_path: Path, metadata: Dict[str, Any]) -> Optional del obj except (OSError, PermissionError) as e: + self.record_diagnostic("file_read_error") print(f"[CLAUDE_DESKTOP] Error reading {audit_path}: {e}") return None diff --git a/Sensor/adr_sensor/parsers/claude_parser.py b/Sensor/adr_sensor/parsers/claude_parser.py index 32c9a93..d9b77b3 100644 --- a/Sensor/adr_sensor/parsers/claude_parser.py +++ b/Sensor/adr_sensor/parsers/claude_parser.py @@ -21,6 +21,53 @@ MAX_LOG_AGE_DAYS = 14 +# Transcript bookkeeping is expected even when it has no sessionId or message. +# Keep these kinds separate from unexpected envelopes; never use input types as +# diagnostic labels. New kinds still follow the existing extraction behavior. +_METADATA_RECORD_TYPES = frozenset( + { + "system", + "progress", + "attachment", + "summary", + "file-history-snapshot", + "queue-operation", + "custom-title", + "ai-title", + "tag", + "agent-name", + "agent-color", + "last-prompt", + "permission-mode", + "pr-link", + "content-replacement", + } +) +_KNOWN_CONTENT_TYPES = frozenset( + { + "text", + "tool_use", + "tool_result", + "image", + "document", + "thinking", + "redacted_thinking", + "tool_reference", + "search_result", + "server_tool_use", + "web_search_tool_result", + "web_fetch_tool_result", + "code_execution_tool_result", + "bash_code_execution_tool_result", + "text_editor_code_execution_tool_result", + "container_upload", + "compaction", + "resource", + "resource_link", + "audio", + } +) + class ClaudeParser(BaseParser): """Parser for Claude Code JSONL log files.""" @@ -33,11 +80,21 @@ def parse_all(self) -> List[AgentEvent]: """Parse all available Claude Code logs.""" entries = [] - if not self.base_path.exists(): + try: + base_exists = self.base_path.exists() + except OSError: + self.record_diagnostic("file_stat_error") + raise + if not base_exists: + self.record_diagnostic("input_missing") print(f"[CLAUDE] No logs found at {self.base_path}") return entries - jsonl_files = list(self.base_path.glob("**/*.jsonl")) + try: + jsonl_files = list(self.base_path.glob("**/*.jsonl")) + except OSError: + self.record_diagnostic("file_read_error") + raise print(f"[CLAUDE] Found {len(jsonl_files)} JSONL files") cutoff_time = datetime.now(timezone.utc) - timedelta(days=self.max_age_days) @@ -50,8 +107,10 @@ def parse_all(self) -> List[AgentEvent]: if mtime >= cutoff_time: filtered_files.append(jsonl_file) else: + self.record_diagnostic("file_age_skipped") skipped_count += 1 except (OSError, PermissionError): + self.record_diagnostic("file_stat_error") skipped_count += 1 if skipped_count > 0: @@ -64,6 +123,7 @@ def parse_all(self) -> List[AgentEvent]: file_entries = self.parse_jsonl_file(jsonl_file) entries.extend(file_entries) except Exception as e: + self.record_diagnostic("parser_error") print(f"[CLAUDE] Error parsing {jsonl_file}: {e}") return entries @@ -74,6 +134,7 @@ def _normalize_result_content(self, result_content: Any) -> str: return result_content if isinstance(result_content, list): + self._diagnose_content_blocks(result_content) text_parts = [] for item in result_content: if isinstance(item, dict): @@ -97,8 +158,7 @@ def _truncate_large_arguments(self, arguments: Dict[str, Any]) -> Dict[str, Any] return truncated - @staticmethod - def _decode_jsonl_line(line: str) -> Iterator[Any]: + def _decode_jsonl_line(self, line: str) -> Iterator[Any]: """Decode complete values on one physical line, retaining a valid prefix. NUL padding is accepted only between values, never inside JSON strings. @@ -114,10 +174,45 @@ def _decode_jsonl_line(line: str) -> Iterator[Any]: return try: value, offset = decoder.raw_decode(line, offset) - except (ValueError, RecursionError): + except (ValueError, RecursionError) as exc: + # A writer may not have finished its final physical line yet. + # Only recognizable JSON prefixes without a newline are expected + # tails; terminated malformed records remain corruption signals. + incomplete = ( + isinstance(exc, json.JSONDecodeError) + and not line.endswith(("\n", "\r")) + and self._is_incomplete_json(exc) + ) + self.record_diagnostic("incomplete_record" if incomplete else "record_decode_error") return yield value + @staticmethod + def _is_incomplete_json(error: json.JSONDecodeError) -> bool: + """Recognize common interrupted JSON writes without repairing content.""" + suffix = error.doc[error.pos :].rstrip(" \t") + if not suffix or error.msg.startswith("Unterminated string"): + return True + if error.msg == "Expecting value" and ( + suffix == "-" or any(token.startswith(suffix) for token in ("true", "false", "null")) + ): + return True + if error.msg == "Expecting ',' delimiter" and suffix in (".", "e", "e+", "e-", "E", "E+", "E-"): + return True + if error.msg == "Invalid \\uXXXX escape": + return suffix.startswith("u") and len(suffix) < 5 and all(c in "0123456789abcdefABCDEF" for c in suffix[1:]) + return False + + def _diagnose_content_blocks(self, content: List[Any]) -> None: + """Observe ignored shapes/types without changing captured message data.""" + for item in content: + if not isinstance(item, dict) or not isinstance(item.get("type"), str): + self.record_diagnostic("record_shape_error") + elif item["type"] not in _KNOWN_CONTENT_TYPES: + self.record_diagnostic("unsupported_content_block") + elif item["type"] == "text" and not isinstance(item.get("text"), str): + self.record_diagnostic("record_shape_error") + @staticmethod def _agent_id(obj: Dict[str, Any], file_path: Path) -> Optional[str]: """Identify documented subagent paths, including nested workflow logs.""" @@ -136,16 +231,24 @@ def parse_jsonl_file(self, file_path: Path) -> List[AgentEvent]: for line in file: for obj in self._decode_jsonl_line(line): if not isinstance(obj, dict): + self.record_diagnostic("record_shape_error") continue + msg_type = obj.get("type") + is_metadata = isinstance(msg_type, str) and msg_type in _METADATA_RECORD_TYPES + if isinstance(msg_type, str) and msg_type not in ("user", "assistant") and not is_metadata: + self.record_diagnostic("unsupported_record_type") session_id = obj.get("sessionId") if not isinstance(session_id, str) or not session_id: + if not is_metadata or "sessionId" in obj: + self.record_diagnostic("record_shape_error") continue - msg_type = obj.get("type") if not isinstance(msg_type, str): + self.record_diagnostic("record_shape_error") continue if msg_type in ("user", "assistant"): message = obj.get("message") if not isinstance(message, dict) or not isinstance(message.get("content", ""), (str, list)): + self.record_diagnostic("record_shape_error") continue agent_id = self._agent_id(obj, file_path) @@ -171,7 +274,9 @@ def parse_jsonl_file(self, file_path: Path) -> List[AgentEvent]: session["timestamp"] = min(session["timestamp"] or ts, ts) session["last_event_at"] = max(session["last_event_at"] or ts, ts) except (TypeError, ValueError, OverflowError, OSError): - pass + self.record_diagnostic("invalid_timestamp") + elif isinstance(obj.get("timestamp"), bool): + self.record_diagnostic("invalid_timestamp") if msg_type == "assistant" and isinstance(obj["message"].get("model"), str): session["model"] = obj["message"]["model"] @@ -181,6 +286,7 @@ def parse_jsonl_file(self, file_path: Path) -> List[AgentEvent]: session["messages"].append(extracted_msg) except (OSError, UnicodeError) as e: + self.record_diagnostic("file_read_error") print(f"[CLAUDE] Error reading {file_path}: {e}") for (session_id, _), session_data in sessions.items(): @@ -209,6 +315,7 @@ def _extract_message_data(self, obj: Dict[str, Any]) -> Optional[Dict[str, Any]] if isinstance(content, str): text_parts.append(content) elif isinstance(content, list): + self._diagnose_content_blocks(content) text_parts.extend( item["text"] for item in content @@ -223,6 +330,7 @@ def _extract_message_data(self, obj: Dict[str, Any]) -> Optional[Dict[str, Any]] if isinstance(item, dict) and item.get("type") == "tool_result": tool_use_id = item.get("tool_use_id") if not isinstance(tool_use_id, str) or not tool_use_id: + self.record_diagnostic("record_shape_error") continue result_content = item.get("content", "") if "toolUseResult" in obj and isinstance(obj["toolUseResult"], dict): @@ -245,8 +353,11 @@ def _extract_message_data(self, obj: Dict[str, Any]) -> Optional[Dict[str, Any]] raw_input = item.get("input", {}) name = item.get("name", "unknown") if not isinstance(raw_input, dict) or not isinstance(name, str): + self.record_diagnostic("record_shape_error") continue tool_id = item.get("id") + if not isinstance(tool_id, str) or not tool_id: + self.record_diagnostic("record_shape_error") tools.append( { "id": tool_id if isinstance(tool_id, str) else None, @@ -344,5 +455,6 @@ def _create_entry_from_extracted_session( ) except Exception as e: + self.record_diagnostic("file_stat_error" if isinstance(e, OSError) else "session_build_error") print(f"[CLAUDE] Error creating entry for session {session_id}: {e}") return None diff --git a/Sensor/adr_sensor/parsers/cline_parser.py b/Sensor/adr_sensor/parsers/cline_parser.py index 8a2c474..d5bdd6b 100644 --- a/Sensor/adr_sensor/parsers/cline_parser.py +++ b/Sensor/adr_sensor/parsers/cline_parser.py @@ -40,6 +40,7 @@ def parse_all(self) -> List[AgentEvent]: entries = [] if not self.base_path.exists(): + self.record_diagnostic("input_missing") print(f"[CLINE] No logs found at {self.base_path}") return entries @@ -57,10 +58,15 @@ def parse_all(self) -> List[AgentEvent]: api_file = task_dir / "api_conversation_history.json" try: modified_at = api_file.stat().st_mtime - except OSError: + except OSError as exc: + # A missing conversation file uses the task timestamp; an + # inaccessible task below is a separate inspection failure. + if not isinstance(exc, FileNotFoundError): + self.record_diagnostic("file_stat_error") try: modified_at = task_dir.stat().st_mtime except OSError as e: + self.record_diagnostic("file_stat_error") print(f"[CLINE] Error checking task {task_dir}: {e}") recent_task_dirs.append(task_dir) continue @@ -72,6 +78,7 @@ def parse_all(self) -> List[AgentEvent]: task_dirs = recent_task_dirs if skipped_count > 0: + self.record_diagnostic("file_age_skipped", skipped_count) print(f"[CLINE] Skipped {skipped_count} tasks older than {self.max_age_days} days") print(f"[CLINE] Processing {len(task_dirs)} task directories") @@ -82,6 +89,7 @@ def parse_all(self) -> List[AgentEvent]: if entry: entries.append(entry) except Exception as e: + self.record_diagnostic("session_build_error") print(f"[CLINE] Error parsing task {task_dir}: {e}") return entries @@ -132,6 +140,12 @@ def parse_cline_log(self, task_dir: Path) -> Optional[AgentEvent]: return entry if entry.has_meaningful_content() else None except Exception as e: + if isinstance(e, json.JSONDecodeError): + self.record_diagnostic("record_decode_error") + elif isinstance(e, (OSError, UnicodeError)): + self.record_diagnostic("file_read_error") + else: + self.record_diagnostic("session_build_error") print(f"[CLINE] Error parsing task {task_dir}: {e}") return None @@ -175,6 +189,7 @@ def extract_mcp_tools(self, text: str) -> List[ToolUsage]: ) tools.append(tool) except json.JSONDecodeError: + self.record_diagnostic("record_decode_error") pass return tools diff --git a/Sensor/adr_sensor/parsers/codex_parser.py b/Sensor/adr_sensor/parsers/codex_parser.py index 3a33881..d3ebcf8 100644 --- a/Sensor/adr_sensor/parsers/codex_parser.py +++ b/Sensor/adr_sensor/parsers/codex_parser.py @@ -72,6 +72,7 @@ def parse_all(self) -> List[AgentEvent]: rollout_candidates = self._discover_rollout_files() if not rollout_candidates: + self.record_diagnostic("input_missing") print(f"[CODEX] No logs found under {self.codex_home}") return entries @@ -90,6 +91,7 @@ def parse_all(self) -> List[AgentEvent]: skipped_count += 1 if skipped_count > 0: + self.record_diagnostic("file_age_skipped", skipped_count) print(f"[CODEX] Skipped {skipped_count} files older than {self.max_age_days} days") print(f"[CODEX] Processing {len(rollout_files)} files") @@ -100,6 +102,7 @@ def parse_all(self) -> List[AgentEvent]: if entry and entry.has_meaningful_content(): entries.append(entry) except Exception as e: + self.record_diagnostic("session_build_error") print(f"[CODEX] Error parsing {jsonl_file}: {e}") return entries @@ -112,6 +115,7 @@ def _discover_rollout_files(self) -> Dict[Path, datetime]: for rollout_path in self.base_path.glob("**/*.jsonl"): self._add_rollout_candidate(candidates, rollout_path) except OSError as e: + self.record_diagnostic("file_read_error") # Keep any files yielded before an inaccessible directory interrupted discovery. print(f"[CODEX] Error discovering logs under {self.base_path}: {e}") @@ -119,12 +123,13 @@ def _discover_rollout_files(self) -> Dict[Path, datetime]: for catalog_path in self.codex_home.glob("state_*.sqlite"): self._add_catalog_rollouts(candidates, catalog_path) except OSError as e: + self.record_diagnostic("file_read_error") print(f"[CODEX] Error discovering state catalogs under {self.codex_home}: {e}") return candidates - @staticmethod def _add_rollout_candidate( + self, candidates: Dict[Path, datetime], rollout_path: Path, catalog_timestamp: Optional[datetime] = None, @@ -140,6 +145,7 @@ def _add_rollout_candidate( return file_mtime = datetime.fromtimestamp(file_stat.st_mtime, tz=timezone.utc) except (OSError, RuntimeError, ValueError, OverflowError): + self.record_diagnostic("file_stat_error") return activity_time = file_mtime @@ -162,6 +168,7 @@ def _add_catalog_rollouts(self, candidates: Dict[Path, datetime], catalog_path: columns = {str(row[1]).lower() for row in connection.execute("PRAGMA table_info(threads)")} if not {"id", "rollout_path"}.issubset(columns): + self.record_diagnostic("unsupported_schema") return timestamp_columns = [name for name in ("updated_at", "updated_at_ms") if name in columns] @@ -171,6 +178,7 @@ def _add_catalog_rollouts(self, candidates: Dict[Path, datetime], catalog_path: for row in connection.execute(query): raw_rollout_path = row[1] if not isinstance(raw_rollout_path, str) or not raw_rollout_path: + self.record_diagnostic("record_shape_error") continue rollout_path = Path(raw_rollout_path) @@ -180,11 +188,14 @@ def _add_catalog_rollouts(self, candidates: Dict[Path, datetime], catalog_path: catalog_timestamp = None for column_name, value in zip(timestamp_columns, row[2:]): timestamp = self._parse_catalog_timestamp(value, milliseconds=column_name == "updated_at_ms") + if value is not None and timestamp is None: + self.record_diagnostic("invalid_timestamp") if timestamp is not None and (catalog_timestamp is None or timestamp > catalog_timestamp): catalog_timestamp = timestamp self._add_rollout_candidate(candidates, rollout_path, catalog_timestamp) except (OSError, sqlite3.Error, ValueError) as e: + self.record_diagnostic("database_error") print(f"[CODEX] Error reading state catalog {catalog_path}: {e}") finally: if connection is not None: @@ -256,20 +267,27 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: try: event = json.loads(line) if not isinstance(event, Mapping): + self.record_diagnostic("record_shape_error") continue payload = event.get("payload") if not isinstance(payload, Mapping): + self.record_diagnostic("record_shape_error") continue session_data["event_count"] += 1 self._process_event(event, payload, session_data) - except Exception: + except Exception as exc: + self.record_diagnostic( + "record_decode_error" if isinstance(exc, json.JSONDecodeError) else "record_shape_error" + ) # Rollout records evolve independently; keep a malformed # record from invalidating the rest of the session. continue if not session_data["id"]: + if session_data["event_count"]: + self.record_diagnostic("record_shape_error") return None chat_history = [] @@ -323,6 +341,9 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: ) except Exception as e: + self.record_diagnostic( + "file_read_error" if isinstance(e, (OSError, UnicodeError)) else "session_build_error" + ) print(f"[CODEX] Error reading {file_path}: {e}") traceback.print_exc() return None @@ -602,6 +623,7 @@ def _process_event( if current is None or normalized > current: session_data["last_event_timestamp"] = normalized except Exception: + self.record_diagnostic("invalid_timestamp") pass if evt_type == "session_meta": @@ -610,6 +632,7 @@ def _process_event( session_id = payload.get("id") if not isinstance(session_id, str) or not session_id: + self.record_diagnostic("record_shape_error") return timestamp = payload.get("timestamp") @@ -618,6 +641,7 @@ def _process_event( try: normalized_timestamp = normalize_timestamp(timestamp) except (TypeError, ValueError, OverflowError, OSError): + self.record_diagnostic("invalid_timestamp") pass session_data["id"] = session_id diff --git a/Sensor/adr_sensor/parsers/copilot_parser.py b/Sensor/adr_sensor/parsers/copilot_parser.py index f757d43..afd113b 100644 --- a/Sensor/adr_sensor/parsers/copilot_parser.py +++ b/Sensor/adr_sensor/parsers/copilot_parser.py @@ -43,6 +43,7 @@ def parse_all(self) -> List[AgentEvent]: entries: List[AgentEvent] = [] if not self.base_path.exists(): + self.record_diagnostic("input_missing") print(f"[COPILOT] No logs found at {self.base_path}") return entries @@ -59,10 +60,12 @@ def parse_all(self) -> List[AgentEvent]: continue modified_at = events_path.stat().st_mtime except OSError as exc: + self.record_diagnostic("file_stat_error") print(f"[COPILOT] Unable to inspect {events_path}: {exc}") continue if cutoff_timestamp is not None and modified_at < cutoff_timestamp: + self.record_diagnostic("file_age_skipped") skipped_count += 1 continue candidates.append((path, modified_at)) @@ -79,6 +82,7 @@ def parse_all(self) -> List[AgentEvent]: if entry and entry.has_meaningful_content(): entries.append(entry) except Exception as exc: + self.record_diagnostic("session_build_error") print(f"[COPILOT] Error parsing {session_dir}: {exc}") return entries @@ -128,9 +132,13 @@ def parse_session_dir(self, session_dir: Path) -> Optional[AgentEvent]: try: event = json.loads(line) except json.JSONDecodeError: + self.record_diagnostic("record_decode_error") continue self._process_event(event, session_data) except Exception as exc: + self.record_diagnostic( + "file_read_error" if isinstance(exc, (OSError, UnicodeError)) else "session_build_error" + ) print(f"[COPILOT] Error reading {events_path}: {exc}") traceback.print_exc() return None @@ -529,8 +537,13 @@ def _load_json_file(self, path: Path) -> Dict[str, Any]: try: with open(path, encoding="utf-8") as handle: value = json.load(handle) + if not isinstance(value, dict): + self.record_diagnostic("record_shape_error") return value if isinstance(value, dict) else {} - except Exception: + except Exception as exc: + self.record_diagnostic( + "record_decode_error" if isinstance(exc, json.JSONDecodeError) else "file_read_error" + ) return {} def _load_workspace_yaml(self, path: Path) -> Dict[str, Any]: @@ -548,6 +561,7 @@ def _load_workspace_yaml(self, path: Path) -> Dict[str, Any]: key, value = line.split(":", 1) data[key.strip()] = self._coerce_scalar(value.strip()) except Exception: + self.record_diagnostic("file_read_error") return {} return data @@ -571,6 +585,7 @@ def _normalize_optional_timestamp(self, value: Any) -> Optional[datetime]: try: return normalize_timestamp(value) except Exception: + self.record_diagnostic("invalid_timestamp") return None @staticmethod diff --git a/Sensor/adr_sensor/parsers/cursor_parser.py b/Sensor/adr_sensor/parsers/cursor_parser.py index abf55ed..21afc89 100644 --- a/Sensor/adr_sensor/parsers/cursor_parser.py +++ b/Sensor/adr_sensor/parsers/cursor_parser.py @@ -45,6 +45,7 @@ def parse_all(self) -> List[AgentEvent]: entries = [] if not self.db_path.exists(): + self.record_diagnostic("input_missing") print(f"[CURSOR] No database found at {self.db_path}") return entries @@ -52,6 +53,7 @@ def parse_all(self) -> List[AgentEvent]: entries = self.parse_conversations_from_bubbles() print(f"[CURSOR] Found {len(entries)} entries") except Exception as e: + self.record_diagnostic("database_error") print(f"[CURSOR] Error parsing database: {e}") return entries @@ -77,11 +79,13 @@ def parse_conversations_from_bubbles(self) -> List[AgentEvent]: try: conv_timestamp = normalize_timestamp(metadata["lastUpdatedAt"]) except Exception: + self.record_diagnostic("invalid_timestamp") pass if conv_timestamp is None and "createdAt" in metadata: try: conv_timestamp = normalize_timestamp(metadata["createdAt"]) except Exception: + self.record_diagnostic("invalid_timestamp") pass if conv_timestamp is None or conv_timestamp >= cutoff_time: @@ -90,6 +94,7 @@ def parse_conversations_from_bubbles(self) -> List[AgentEvent]: skipped_count += 1 if skipped_count > 0: + self.record_diagnostic("file_age_skipped", skipped_count) print(f"[CURSOR] Skipped {skipped_count} conversations older than {self.max_age_days} days") cursor.execute("SELECT key, value FROM cursorDiskKV WHERE key LIKE 'bubbleId:%'") @@ -112,10 +117,12 @@ def parse_conversations_from_bubbles(self) -> List[AgentEvent]: bubble_data = json.loads(value) conversations[conv_id].append(bubble_data) except json.JSONDecodeError: + self.record_diagnostic("record_decode_error") continue else: conversations[conv_id].append(value) except Exception: + self.record_diagnostic("record_shape_error") continue for conv_id, bubbles in conversations.items(): @@ -124,11 +131,13 @@ def parse_conversations_from_bubbles(self) -> List[AgentEvent]: if entry: entries.append(entry) except Exception: + self.record_diagnostic("session_build_error") pass finally: conn.close() except Exception as e: + self.record_diagnostic("database_error") print(f"[CURSOR] Error parsing conversations: {e}") return entries @@ -168,13 +177,18 @@ def get_composer_metadata(self, cursor) -> Dict[str, Any]: metadata_entry["lastUpdatedAt"] = data["lastUpdatedAt"] if metadata_entry: metadata[composer_id] = metadata_entry + else: + self.record_diagnostic("record_shape_error") except json.JSONDecodeError: + self.record_diagnostic("record_decode_error") pass except Exception: + self.record_diagnostic("record_shape_error") continue except Exception as e: + self.record_diagnostic("database_error") print(f"[CURSOR] Error getting composer metadata: {e}") return metadata @@ -185,6 +199,8 @@ def parse_conversation( """Parse a single conversation from its bubbles.""" try: valid_bubbles = [b for b in bubbles if isinstance(b, dict)] + if len(valid_bubbles) < len(bubbles): + self.record_diagnostic("record_shape_error", len(bubbles) - len(valid_bubbles)) if not valid_bubbles: return None @@ -198,11 +214,13 @@ def parse_conversation( try: timestamp = normalize_timestamp(metadata["lastUpdatedAt"]) except Exception: + self.record_diagnostic("invalid_timestamp") pass elif "createdAt" in metadata: try: timestamp = normalize_timestamp(metadata["createdAt"]) except Exception: + self.record_diagnostic("invalid_timestamp") pass entry = AgentEvent(timestamp=timestamp, source="cursor", session_id=f"cursor_{conv_id}") @@ -234,6 +252,7 @@ def parse_conversation( return entry except Exception: + self.record_diagnostic("session_build_error") pass return None @@ -250,6 +269,7 @@ def extract_text_from_bubble(self, bubble: Dict[str, Any]) -> str: if extracted_text: return extracted_text.strip() except json.JSONDecodeError: + self.record_diagnostic("record_decode_error") pass if isinstance(text, str): @@ -313,5 +333,6 @@ def _extract_text_recursive(self, node) -> str: elif isinstance(node, list): return " ".join(self._extract_text_recursive(item) for item in node) except Exception: + self.record_diagnostic("record_shape_error") pass return "" diff --git a/Sensor/adr_sensor/parsers/dsh_parser.py b/Sensor/adr_sensor/parsers/dsh_parser.py index d837f08..20bdf2d 100644 --- a/Sensor/adr_sensor/parsers/dsh_parser.py +++ b/Sensor/adr_sensor/parsers/dsh_parser.py @@ -34,6 +34,7 @@ def __init__(self, max_age_days: int = MAX_LOG_AGE_DAYS, base_path: Optional[Pat def parse_all(self) -> List[AgentEvent]: if not self.base_path.is_dir(): + self.record_diagnostic("input_missing") print(f"[DSH] No logs found at {self.base_path}") return [] @@ -44,9 +45,12 @@ def parse_all(self) -> List[AgentEvent]: files = self._select_session_generations() entries: Dict[str, AgentEvent] = {} for path in files: + failure_code = "file_stat_error" try: if cutoff and datetime.fromtimestamp(path.stat().st_mtime, timezone.utc) < cutoff: + self.record_diagnostic("file_age_skipped") continue + failure_code = "file_read_error" entry = self.parse_session_file(path) if not entry or not entry.has_meaningful_content(): continue @@ -54,6 +58,7 @@ def parse_all(self) -> List[AgentEvent]: if previous is None or self._revision(entry) > self._revision(previous): entries[entry.session_id] = entry except (OSError, UnicodeError, ValueError, zstandard.ZstdError) as exc: + self.record_diagnostic(failure_code) print(f"[DSH] Unable to read {path}: {exc}") print(f"[DSH] Found {len(entries)} sessions") return list(entries.values()) @@ -78,6 +83,7 @@ def _select_session_generations(self) -> List[Path]: try: mtime = path.stat().st_mtime except OSError as exc: + self.record_diagnostic("file_stat_error") print(f"[DSH] Error inspecting {path}: {exc}") continue current = selected.get(path.parent) @@ -87,6 +93,7 @@ def _select_session_generations(self) -> List[Path]: current_paths = [] for version, _, path in sorted(selected.values(), key=lambda item: item[1], reverse=True): if version != MAX_SUPPORTED_SESSION_VERSION: + self.record_diagnostic("unsupported_schema") print(f"[DSH] Unsupported session generation v{version} in {path.parent}") continue current_paths.append(path) @@ -122,16 +129,20 @@ def parse_session_file(self, file_path: Path) -> Optional[AgentEvent]: try: event = json.loads(line) except json.JSONDecodeError: + self.record_diagnostic("record_decode_error") data["malformed_records"] += 1 continue if not isinstance(event, dict): + self.record_diagnostic("record_shape_error") data["malformed_records"] += 1 continue if not header_seen: if event.get("type") != "session" or event.get("version") != MAX_SUPPORTED_SESSION_VERSION: + self.record_diagnostic("unsupported_schema") print(f"[DSH] Unsupported or malformed session header in {file_path}") return None if not isinstance(event.get("id"), str) or not event["id"]: + self.record_diagnostic("record_shape_error") return None header_seen = True data["id"] = event["id"] @@ -147,6 +158,7 @@ def parse_session_file(self, file_path: Path) -> Optional[AgentEvent]: event_type = event.get("type") payload = event.get("data") if not isinstance(payload, dict): + self.record_diagnostic("record_shape_error") data["malformed_records"] += 1 continue data["event_count"] += 1 @@ -447,8 +459,7 @@ def _timestamp(value: Any) -> Optional[datetime]: except (TypeError, ValueError, OSError, OverflowError): return None - @staticmethod - def _iter_lines(file_path: Path) -> Iterator[str]: + def _iter_lines(self, file_path: Path) -> Iterator[str]: if not file_path.name.endswith(".zstd"): with open(file_path, "rb") as handle: for line in handle: @@ -456,6 +467,8 @@ def _iter_lines(file_path: Path) -> Iterator[str]: # tail before decoding: it may end inside a UTF-8 character. if line.endswith(b"\n"): yield line.decode("utf-8") + else: + self.record_diagnostic("incomplete_record") return with open(file_path, "rb") as raw: for frame in DshParser._iter_complete_zstd_frames(raw): diff --git a/Sensor/adr_sensor/parsers/gemini_parser.py b/Sensor/adr_sensor/parsers/gemini_parser.py index 5f336f3..7ae0bb9 100644 --- a/Sensor/adr_sensor/parsers/gemini_parser.py +++ b/Sensor/adr_sensor/parsers/gemini_parser.py @@ -36,17 +36,21 @@ def parse_all(self) -> List[AgentEvent]: seen_paths = set() for base in self.base_paths: if not base.is_dir(): + self.record_diagnostic("input_missing") continue for chats in sorted(base.glob("*/chats")): for path in sorted(chats.rglob("*")): if path.suffix not in {".json", ".jsonl"}: continue + failure_code = "file_stat_error" try: if not path.is_file() or path.resolve() in seen_paths: continue seen_paths.add(path.resolve()) if self.max_age_days > 0 and path.stat().st_mtime < cutoff: + self.record_diagnostic("file_age_skipped") continue + failure_code = "file_read_error" entry = self.parse_file(path) if entry is None or not entry.has_meaningful_content(): continue @@ -55,6 +59,7 @@ def parse_all(self) -> List[AgentEvent]: if old is None or self._revision(entry) > self._revision(old): entries[entry.session_id] = entry except (OSError, ValueError) as exc: + self.record_diagnostic(failure_code) print(f"[GEMINI] Unable to read {path}: {exc}") return list(entries.values()) @@ -88,6 +93,7 @@ def parse_file(self, path: Path) -> Optional[AgentEvent]: def add_message(message: Any) -> None: if not isinstance(message, dict) or not isinstance(message.get("id"), str): + self.record_diagnostic("record_shape_error") return messages[message["id"]] = message timestamp = self._timestamp(message.get("timestamp")) @@ -96,6 +102,7 @@ def add_message(message: Any) -> None: calls = message.get("toolCalls") for call in calls if isinstance(calls, list) else []: if not isinstance(call, dict): + self.record_diagnostic("record_shape_error") continue timestamp = self._timestamp(call.get("timestamp")) if timestamp: @@ -105,10 +112,12 @@ def add_message(message: Any) -> None: if permission not in permissions: permissions.append(permission) + failure_code = "file_stat_error" try: # Capture before reading: appended records can advance the revision, # but later writes must not give a partial read a newer file timestamp. modified_at = datetime.fromtimestamp(path.stat().st_mtime, timezone.utc) + failure_code = "file_read_error" with path.open(encoding="utf-8") as handle: if path.suffix == ".json": records = [json.load(handle)] @@ -120,9 +129,11 @@ def add_message(message: Any) -> None: try: records.append(json.loads(line)) except json.JSONDecodeError: + self.record_diagnostic("record_decode_error") malformed += 1 for record in records: if not isinstance(record, dict): + self.record_diagnostic("record_shape_error") malformed += 1 continue event_count += 1 @@ -134,6 +145,7 @@ def add_message(message: Any) -> None: continue update = record.get("$set", record) if not isinstance(update, dict): + self.record_diagnostic("record_shape_error") malformed += 1 continue if "$set" in record: @@ -143,11 +155,16 @@ def add_message(message: Any) -> None: for message in checkpoint if isinstance(checkpoint, list) else []: add_message(message) except (OSError, UnicodeError, ValueError) as exc: + self.record_diagnostic( + "record_decode_error" if isinstance(exc, json.JSONDecodeError) else failure_code + ) print(f"[GEMINI] Unable to parse {path}: {exc}") return None session_id = metadata.get("sessionId") if not isinstance(session_id, str) or not session_id: + if event_count: + self.record_diagnostic("record_shape_error") return None history = [] message_metadata = {} @@ -167,6 +184,9 @@ def add_message(message: Any) -> None: details["tool_metadata"] = [] for call in calls if isinstance(calls, list) else []: if not isinstance(call, dict) or not isinstance(call.get("name"), str): + # Non-object calls were counted while collecting messages. + if isinstance(call, dict): + self.record_diagnostic("record_shape_error") continue tools.append(self._tool(call)) details["tool_metadata"].append({k: v for k, v in call.items() if k not in {"args", "result"}}) @@ -236,8 +256,7 @@ def add_message(message: Any) -> None: else None, ) - @staticmethod - def _project_path(path: Path) -> Optional[str]: + def _project_path(self, path: Path) -> Optional[str]: chats = next((parent for parent in path.parents if parent.name == "chats"), None) if chats is None: return None @@ -246,14 +265,20 @@ def _project_path(path: Path) -> Optional[str]: marker = (project / ".project_root").read_text(encoding="utf-8").strip() if marker: return marker - except (OSError, UnicodeError): + except (OSError, UnicodeError) as exc: + if not isinstance(exc, FileNotFoundError): + self.record_diagnostic("file_read_error") pass try: registry = json.loads((project.parent.parent / "projects.json").read_text(encoding="utf-8")) projects = registry.get("projects", {}) if isinstance(registry, dict) else {} if isinstance(projects, dict): return next((key for key, value in projects.items() if value == project.name), None) - except (OSError, UnicodeError, ValueError): + except (OSError, UnicodeError, ValueError) as exc: + if not isinstance(exc, FileNotFoundError): + self.record_diagnostic( + "record_decode_error" if isinstance(exc, json.JSONDecodeError) else "file_read_error" + ) pass return None diff --git a/Sensor/adr_sensor/parsers/opencode_parser.py b/Sensor/adr_sensor/parsers/opencode_parser.py index 04fea64..21a6f1f 100644 --- a/Sensor/adr_sensor/parsers/opencode_parser.py +++ b/Sensor/adr_sensor/parsers/opencode_parser.py @@ -161,6 +161,7 @@ def parse_all(self) -> List[AgentEvent]: return self._parse_json_storage(storage_dir) print(f"[OPENCODE] No logs found at {self.base_dir}") + self.record_diagnostic("input_missing") return [] # ------------------------------------------------------------------ # @@ -190,8 +191,10 @@ def _parse_sqlite(self, db_path: Path) -> List[AgentEvent]: if entry and entry.has_meaningful_content(): entries.append(entry) except Exception as e: + self.record_diagnostic("session_build_error") print(f"[OPENCODE] Error processing session {session_id}: {e}") except Exception as e: + self.record_diagnostic("database_error") print(f"[OPENCODE] Error reading database: {e}") finally: if conn is not None: @@ -258,9 +261,12 @@ def _parse_json_storage(self, storage_dir: Path) -> List[AgentEvent]: cutoff_ts = time.time() - (self.max_age_days * 86400) if self.max_age_days > 0 else None for session_file in session_files: + failure_code = "file_stat_error" try: if cutoff_ts is not None and session_file.stat().st_mtime < cutoff_ts: + self.record_diagnostic("file_age_skipped") continue + failure_code = "file_read_error" session_meta = self._safe_json_file(session_file) if not session_meta or not session_meta.get("id"): continue @@ -274,6 +280,7 @@ def _parse_json_storage(self, storage_dir: Path) -> List[AgentEvent]: if entry and entry.has_meaningful_content(): entries.append(entry) except Exception as e: + self.record_diagnostic(failure_code if isinstance(e, OSError) else "session_build_error") print(f"[OPENCODE] Error processing session file {session_file}: {e}") return entries @@ -344,6 +351,7 @@ def _session_to_event( ) -> Optional[AgentEvent]: session_id = session_meta.get("id") if not session_id: + self.record_diagnostic("record_shape_error") return None chat_history: List[ChatMessage] = [] @@ -392,6 +400,7 @@ def _build_content_and_tools(self, parts: List[Dict[str, Any]]) -> Tuple[str, Li for part in parts: if not isinstance(part, dict): + self.record_diagnostic("record_shape_error") continue part_type = part.get("type") @@ -532,8 +541,7 @@ def _strip_session_id_prefix(session_id: str) -> str: """ return session_id[len("ses_") :] if session_id.startswith("ses_") else session_id - @staticmethod - def _session_timestamp(session_meta: Dict[str, Any]) -> datetime: + def _session_timestamp(self, session_meta: Dict[str, Any]) -> datetime: """Best-effort session timestamp (uses last-updated when available).""" # SQLite exposes flat epoch-ms columns; JSON nests them under "time". ts = session_meta.get("time_updated") or session_meta.get("time_created") @@ -546,22 +554,27 @@ def _session_timestamp(session_meta: Dict[str, Any]) -> datetime: try: return normalize_timestamp(ts) except (ValueError, TypeError): + self.record_diagnostic("invalid_timestamp") return datetime.now(timezone.utc) - @staticmethod - def _safe_json(text: Optional[str]) -> Optional[Any]: + def _safe_json(self, text: Optional[str]) -> Optional[Any]: if not text: return None try: return json.loads(text) except (json.JSONDecodeError, TypeError): + self.record_diagnostic("record_decode_error") return None - @staticmethod - def _safe_json_file(path: Path) -> Optional[Dict[str, Any]]: + def _safe_json_file(self, path: Path) -> Optional[Dict[str, Any]]: try: with open(path, encoding="utf-8") as f: data = json.load(f) + if not isinstance(data, dict): + self.record_diagnostic("record_shape_error") return data if isinstance(data, dict) else None - except (OSError, json.JSONDecodeError): + except (OSError, json.JSONDecodeError) as exc: + self.record_diagnostic( + "record_decode_error" if isinstance(exc, json.JSONDecodeError) else "file_read_error" + ) return None diff --git a/Sensor/adr_sensor/parsers/warp_parser.py b/Sensor/adr_sensor/parsers/warp_parser.py index f853d69..a0eeb29 100644 --- a/Sensor/adr_sensor/parsers/warp_parser.py +++ b/Sensor/adr_sensor/parsers/warp_parser.py @@ -54,6 +54,7 @@ def parse_all(self) -> List[AgentEvent]: db_path = Path(self.db_path) if isinstance(self.db_path, str) else self.db_path if not db_path.exists(): + self.record_diagnostic("input_missing") print(f"[WARP] No logs found at {db_path}") return entries @@ -70,6 +71,7 @@ def parse_all(self) -> List[AgentEvent]: recent_conversations = self._filter_recent_conversations(conversations) skipped_count = len(conversations) - len(recent_conversations) if skipped_count > 0: + self.record_diagnostic("file_age_skipped", skipped_count) print(f"[WARP] Skipped {skipped_count} conversations older than {self.max_age_days} days") for conversation in recent_conversations: @@ -80,11 +82,13 @@ def parse_all(self) -> List[AgentEvent]: if entry and entry.has_meaningful_content(): entries.append(entry) except Exception as e: + self.record_diagnostic("session_build_error") print(f"[WARP] Error processing conversation {conversation_id}: {e}") conn.close() except Exception as e: + self.record_diagnostic("database_error") print(f"[WARP] Error reading database: {e}") traceback.print_exc() @@ -121,6 +125,7 @@ def _filter_recent_conversations(self, conversations: List[Dict]) -> List[Dict]: try: conv_timestamp = normalize_timestamp(last_modified) except Exception: + self.record_diagnostic("invalid_timestamp") pass if conv_timestamp is None or conv_timestamp >= cutoff_time: @@ -168,7 +173,7 @@ def _create_entry_from_exchanges( timestamp = normalize_timestamp(most_recent["start_ts"]) model_id = most_recent.get("model_id") if isinstance(model_id, str): - parsed_model_id = self._parse_json_safely(model_id) + parsed_model_id = self._parse_json_safely(model_id, report_failure=False) if isinstance(parsed_model_id, str): model_id = parsed_model_id @@ -220,17 +225,20 @@ def _create_entry_from_exchanges( return entry except Exception as e: + self.record_diagnostic("session_build_error") print(f"[WARP] Error creating entry for conversation {conversation_id}: {e}") traceback.print_exc() return None - def _parse_json_safely(self, json_str: str) -> Optional[Any]: + def _parse_json_safely(self, json_str: str, report_failure: bool = True) -> Optional[Any]: """Safely parse JSON string.""" if not json_str: return None try: return json.loads(json_str) except json.JSONDecodeError: + if report_failure: + self.record_diagnostic("record_decode_error") return None def _parse_tool_usage(self, action_result: Dict[str, Any]) -> Optional[ToolUsage]: @@ -272,6 +280,7 @@ def _parse_tool_usage(self, action_result: Dict[str, Any]) -> Optional[ToolUsage ) except Exception: + self.record_diagnostic("record_shape_error") return None def _extract_content_from_action(self, action_result: Dict[str, Any]) -> str: @@ -308,4 +317,5 @@ def _extract_llm_text(self, llm_output: Optional[Dict[str, Any]]) -> str: return "\n".join(text_parts) except Exception: + self.record_diagnostic("record_shape_error") return "" diff --git a/Sensor/tests/test_claude_diagnostics.py b/Sensor/tests/test_claude_diagnostics.py new file mode 100644 index 0000000..b73ee0d --- /dev/null +++ b/Sensor/tests/test_claude_diagnostics.py @@ -0,0 +1,212 @@ +"""Claude health counters use synthetic inputs and never copy payloads into labels.""" + +import json +import os +from pathlib import Path +from unittest.mock import patch + +import pytest + +from adr_sensor.parsers.base_parser import BaseParser +from adr_sensor.parsers.claude_parser import ClaudeParser + +CANARY = "private-content-credential-canary" + + +def _record(content=CANARY, **fields): + return { + "type": "user", + "sessionId": "synthetic-session", + "timestamp": "2026-09-19T10:00:00Z", + "message": {"content": content}, + **fields, + } + + +def _parse(tmp_path, text): + path = tmp_path / "private-source-path.jsonl" + path.write_text(text, encoding="utf-8") + parser = ClaudeParser() + entries = parser.parse_jsonl_file(path) + assert path.read_text(encoding="utf-8") == text + return parser, entries + + +def test_absent_claude_source_is_an_expected_skip(tmp_path): + parser = ClaudeParser() + parser.base_path = tmp_path / "missing" + + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"input_missing": 1} + + +@pytest.mark.parametrize("operation,reason", [("exists", "file_stat_error"), ("glob", "file_read_error")]) +def test_discovery_errors_remain_visible_to_caller_and_are_counted(tmp_path, operation, reason): + parser = ClaudeParser() + parser.base_path = tmp_path + with patch.object(Path, operation, side_effect=PermissionError(CANARY)): + with pytest.raises(PermissionError): + parser.parse_all() + + assert parser.get_diagnostics() == {reason: 1} + + +def test_old_transcript_and_failed_stat_are_distinct(tmp_path, monkeypatch): + path = tmp_path / "old.jsonl" + path.write_text(json.dumps(_record()), encoding="utf-8") + os.utime(path, (1, 1)) + parser = ClaudeParser() + parser.base_path = tmp_path + + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"file_age_skipped": 1} + parser.reset_diagnostics() + original_stat = Path.stat + + def fail_selected_path(candidate, *args, **kwargs): + if candidate == path: + raise PermissionError(CANARY) + return original_stat(candidate, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", fail_selected_path) + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"file_stat_error": 1} + + +def test_read_failure_reports_no_payload(tmp_path): + parser = ClaudeParser() + with patch("builtins.open", side_effect=PermissionError(CANARY)): + assert parser.parse_jsonl_file(tmp_path / CANARY) == [] + + assert parser.get_diagnostics() == {"file_read_error": 1} + assert CANARY not in json.dumps(parser.get_diagnostics()) + + +def test_malformed_records_are_counted_and_valid_content_is_unchanged(tmp_path): + parser, entries = _parse( + tmp_path, + "invalid-json-" + CANARY + "\nnull\n" + json.dumps(_record()) + "\n", + ) + + assert len(entries) == 1 + assert entries[0].chat_history[0].content == CANARY + assert parser.get_diagnostics() == {"record_decode_error": 1, "record_shape_error": 1} + diagnostics = json.dumps(parser.get_diagnostics()) + assert CANARY not in diagnostics + assert "private-source-path" not in diagnostics + + +@pytest.mark.parametrize("suffix", ['{"unfinished":', '{"unfinished":"text', '{"unfinished":tru', '{"value":1e']) +def test_unfinished_final_write_is_expected_and_complete_prefix_survives(tmp_path, suffix): + parser, entries = _parse(tmp_path, json.dumps(_record()) + suffix) + + assert entries[0].chat_history[0].content == CANARY + assert parser.get_diagnostics() == {"incomplete_record": 1} + assert set(parser.get_diagnostics()) <= BaseParser.EXPECTED_DIAGNOSTIC_CODES + + +@pytest.mark.parametrize("malformed", ['{"unfinished":\n', '{"invalid":}\n', '{"invalid":}', "not-json"]) +def test_malformed_terminated_lines_and_invalid_final_tokens_are_corruption(tmp_path, malformed): + parser, entries = _parse(tmp_path, json.dumps(_record()) + "\n" + malformed) + + assert len(entries) == 1 + assert parser.get_diagnostics() == {"record_decode_error": 1} + + +def test_valid_final_record_padding_and_concatenated_objects_have_no_diagnostic(tmp_path): + parser, entries = _parse(tmp_path, "\0" + json.dumps(_record()) + "\0" + json.dumps(_record("Second message"))) + + assert [message.content for message in entries[0].chat_history] == [CANARY, "Second message"] + assert parser.get_diagnostics() == {} + + +@pytest.mark.parametrize( + "record", [_record(message=None), _record(sessionId=[]), _record(type=[]), _record(content={})] +) +def test_invalid_envelope_and_message_shapes_are_counted_once(tmp_path, record): + parser, entries = _parse(tmp_path, json.dumps(record) + "\n" + json.dumps(_record()) + "\n") + + assert entries[0].chat_history[0].content == CANARY + assert parser.get_diagnostics() == {"record_shape_error": 1} + + +@pytest.mark.parametrize("timestamp", [None, True, "invalid-" + CANARY, [], 10**100]) +def test_invalid_timestamp_keeps_content_and_reports_fixed_reason(tmp_path, timestamp): + parser, entries = _parse(tmp_path, json.dumps(_record(timestamp=timestamp)) + "\n") + + assert entries[0].chat_history[0].content == CANARY + assert parser.get_diagnostics() == {"invalid_timestamp": 1} + + +def test_known_metadata_without_session_ids_and_expected_nontext_blocks_are_quiet(tmp_path): + metadata = [ + {"type": kind, "private_value": CANARY} + for kind in ( + "summary", + "file-history-snapshot", + "queue-operation", + "custom-title", + "tag", + "content-replacement", + ) + ] + content = [ + {"type": kind, "private_value": CANARY} + for kind in ("image", "document", "thinking", "redacted_thinking", "tool_reference") + ] + [{"type": "text", "text": CANARY}] + records = metadata + [_record(content), {"type": "system", "sessionId": "synthetic-session", "subtype": CANARY}] + parser, entries = _parse(tmp_path, "\n".join(map(json.dumps, records)) + "\n") + + assert entries[0].chat_history[0].content == CANARY + assert parser.get_diagnostics() == {} + + +def test_unknown_kinds_use_fixed_labels_without_changing_known_content(tmp_path): + records = [ + _record(type="future-envelope-" + CANARY), + _record([{"type": "future-block-" + CANARY}, {"type": "text", "text": CANARY}]), + ] + parser, entries = _parse(tmp_path, "\n".join(map(json.dumps, records)) + "\n") + + assert [message.content for message in entries[0].chat_history] == [CANARY] + assert parser.get_diagnostics() == {"unsupported_record_type": 1, "unsupported_content_block": 1} + assert CANARY not in json.dumps(parser.get_diagnostics()) + + +def test_invalid_blocks_and_tool_identifiers_keep_valid_call_and_result(tmp_path): + records = [ + _record( + [ + {"type": "text", "text": None}, + {"type": "tool_use", "id": "invalid", "name": "Read", "input": []}, + {"type": "tool_use", "id": "call", "name": "Read", "input": {}}, + ], + type="assistant", + ), + _record( + [ + {"type": "tool_result", "tool_use_id": [], "content": CANARY}, + { + "type": "tool_result", + "tool_use_id": "call", + "content": [None, {"type": "image"}, {"type": "text", "text": CANARY}], + }, + ] + ), + ] + parser, entries = _parse(tmp_path, "\n".join(map(json.dumps, records)) + "\n") + + assert len(entries[0].chat_history[0].tools) == 1 + assert entries[0].chat_history[0].tools[0].result == CANARY + assert parser.get_diagnostics() == {"record_shape_error": 4} + + +def test_session_build_error_is_counted_without_exception_text(tmp_path): + path = tmp_path / "session.jsonl" + path.write_text(json.dumps(_record()), encoding="utf-8") + parser = ClaudeParser() + with patch("adr_sensor.parsers.claude_parser.AgentEvent", side_effect=ValueError(CANARY)): + assert parser.parse_jsonl_file(path) == [] + + assert parser.get_diagnostics() == {"session_build_error": 1} + assert CANARY not in json.dumps(parser.get_diagnostics()) diff --git a/Sensor/tests/test_delivery_checkpoint.py b/Sensor/tests/test_delivery_checkpoint.py new file mode 100644 index 0000000..6f2ccad --- /dev/null +++ b/Sensor/tests/test_delivery_checkpoint.py @@ -0,0 +1,332 @@ +"""Synthetic incremental-delivery tests; no collector or real sessions needed.""" + +import json +import re +from dataclasses import replace +from datetime import datetime, timezone +from unittest.mock import patch + +import pytest +from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import ExportLogsServiceRequest, ExportLogsServiceResponse +from requests import Response + +from adr_sensor.cli import main +from adr_sensor.diagnostics import health_record +from adr_sensor.exporters.config import OpenTelemetryConfig +from adr_sensor.exporters.delivery_checkpoint import DeliveryCheckpoint, DeliveryCheckpointError +from adr_sensor.exporters.opentelemetry import OpenTelemetryExportError +from adr_sensor.schemas.agent_event_schema import AgentEvent, ChatMessage, ToolUsage + + +@pytest.fixture +def event(): + return AgentEvent( + timestamp=datetime(2026, 9, 9, tzinfo=timezone.utc), + source="codex", + session_id="synthetic-private-session", + hostname="synthetic-host", + username="synthetic-user", + chat_history=[ + ChatMessage( + role="assistant", + content="synthetic private text", + tools=[ToolUsage(tool_name="shell", tool_type="custom", result="first synthetic result")], + ) + ], + ) + + +@pytest.fixture +def config(): + return OpenTelemetryConfig( + endpoint="https://collector.example.invalid/v1/logs", + headers={"Authorization": "Bearer synthetic-secret"}, + ) + + +def test_checkpoint_skips_only_successfully_committed_payloads(tmp_path, config, event): + checkpoint = DeliveryCheckpoint(tmp_path, config) + assert checkpoint.pending_entries([event]) == [event] + assert not checkpoint.path.exists() + assert DeliveryCheckpoint(tmp_path, config).pending_entries([event]) == [event] + + checkpoint.commit() + + assert DeliveryCheckpoint(tmp_path, config).pending_entries([event]) == [] + + +@pytest.mark.parametrize("changed_field", ["tool_result", "tool_status", "model", "session_context"]) +def test_checkpoint_hashes_complete_normalized_payload(tmp_path, config, event, changed_field): + checkpoint = DeliveryCheckpoint(tmp_path, config) + checkpoint.pending_entries([event]) + checkpoint.commit() + original_uuid = event.uuid + if changed_field == "tool_result": + event.chat_history[0].tools[0] = replace(event.chat_history[0].tools[0], result="updated synthetic result") + elif changed_field == "tool_status": + event.chat_history[0].tools[0] = replace(event.chat_history[0].tools[0], status="error") + elif changed_field == "model": + event = replace(event, model="changed-model") + else: + event = replace(event, session_context={"changed": True}) + + assert event.uuid == original_uuid + assert DeliveryCheckpoint(tmp_path, config).pending_entries([event]) == [event] + + +@pytest.mark.parametrize( + "settings", + [ + {"endpoint": "https://other.example.invalid/v1/logs"}, + {"headers": {"Authorization": "Bearer other-synthetic-secret"}}, + {"service_name": "another-service"}, + ], +) +def test_destination_or_authentication_change_resends_sessions(tmp_path, config, event, settings): + checkpoint = DeliveryCheckpoint(tmp_path, config) + checkpoint.pending_entries([event]) + checkpoint.commit() + changed = DeliveryCheckpoint(tmp_path, replace(config, **settings)) + + assert changed.path != checkpoint.path + assert changed.pending_entries([event]) == [event] + + +def test_environment_authentication_change_resends_sessions(tmp_path, event, monkeypatch): + config = OpenTelemetryConfig(endpoint="https://collector.example.invalid/v1/logs") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_LOGS_HEADERS", "Authorization=first-synthetic-token") + checkpoint = DeliveryCheckpoint(tmp_path, config) + checkpoint.pending_entries([event]) + checkpoint.commit() + monkeypatch.setenv("OTEL_EXPORTER_OTLP_LOGS_HEADERS", "Authorization=second-synthetic-token") + + assert DeliveryCheckpoint(tmp_path, config).pending_entries([event]) == [event] + + +@pytest.mark.parametrize("prefix", ["OTEL_EXPORTER_OTLP", "OTEL_EXPORTER_OTLP_LOGS"]) +@pytest.mark.parametrize("setting", ["CLIENT_CERTIFICATE", "CLIENT_KEY"]) +def test_environment_mtls_path_change_resends_sessions(tmp_path, config, event, monkeypatch, prefix, setting): + variable = f"{prefix}_{setting}" + monkeypatch.setenv(variable, "/synthetic/first-credential.pem") + checkpoint = DeliveryCheckpoint(tmp_path, config) + checkpoint.pending_entries([event]) + checkpoint.commit() + monkeypatch.setenv(variable, "/synthetic/second-credential.pem") + changed = DeliveryCheckpoint(tmp_path, config) + + assert changed.path != checkpoint.path + assert changed.pending_entries([event]) == [event] + + +def test_checkpoint_uses_effective_mtls_settings(tmp_path, config, monkeypatch): + monkeypatch.setenv("OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE", "/synthetic/generic-cert.pem") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_LOGS_CLIENT_CERTIFICATE", "/synthetic/logs-cert.pem") + checkpoint = DeliveryCheckpoint(tmp_path, config) + monkeypatch.setenv("OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE", "/synthetic/unused-cert.pem") + + assert DeliveryCheckpoint(tmp_path, config).path == checkpoint.path + + +def test_opaque_credential_provider_cannot_skip_previously_delivered_sessions(tmp_path, config, event, monkeypatch): + checkpoint = DeliveryCheckpoint(tmp_path, config) + checkpoint.pending_entries([event]) + checkpoint.commit() + monkeypatch.setenv("OTEL_PYTHON_EXPORTER_OTLP_HTTP_LOGS_CREDENTIAL_PROVIDER", "synthetic-private-provider") + + with pytest.raises(OpenTelemetryExportError, match="credential-provider plugins are unsupported"): + DeliveryCheckpoint(tmp_path, config) + + +def test_checkpoint_persists_only_hashes(tmp_path, config, event): + checkpoint = DeliveryCheckpoint(tmp_path, config) + checkpoint.pending_entries([event]) + checkpoint.commit() + + assert re.fullmatch(r"\.adr-otel-delivery\.[0-9a-f]{64}\.json", checkpoint.path.name) + data = json.loads(checkpoint.path.read_text()) + assert len(data) == 1 + assert all(re.fullmatch(r"[0-9a-f]{64}", digest) for pair in data.items() for digest in pair) + + +@pytest.mark.parametrize("contents", ["broken JSON", "[]", '{"raw-session": "raw-content"}', "\udcff"]) +def test_corrupt_checkpoint_retries_and_can_be_replaced(tmp_path, config, event, contents): + path = DeliveryCheckpoint(tmp_path, config).path + path.write_bytes(contents.encode("utf-8", errors="surrogateescape")) + + checkpoint = DeliveryCheckpoint(tmp_path, config) + assert checkpoint.load_failed + assert checkpoint.pending_entries([event]) == [event] + checkpoint.commit() + assert DeliveryCheckpoint(tmp_path, config).pending_entries([event]) == [] + + +def test_unreadable_checkpoint_retries(tmp_path, config, event): + with patch("pathlib.Path.open", side_effect=PermissionError("synthetic private path")): + checkpoint = DeliveryCheckpoint(tmp_path, config) + assert checkpoint.load_failed + assert checkpoint.pending_entries([event]) == [event] + + +def test_atomic_replace_failure_keeps_previous_checkpoint_and_retries(tmp_path, config, event): + checkpoint = DeliveryCheckpoint(tmp_path, config) + checkpoint.pending_entries([event]) + checkpoint.commit() + previous = checkpoint.path.read_bytes() + event.chat_history[0].tools[0] = replace(event.chat_history[0].tools[0], result="changed result") + checkpoint.pending_entries([event]) + + with ( + patch("adr_sensor.exporters.delivery_checkpoint.os.replace", side_effect=OSError("synthetic private path")), + pytest.raises(DeliveryCheckpointError, match="could not save") as error, + ): + checkpoint.commit() + + assert "synthetic private path" not in str(error.value) + assert checkpoint.path.read_bytes() == previous + assert not list(tmp_path.glob("*.tmp")) + assert DeliveryCheckpoint(tmp_path, config).pending_entries([event]) == [event] + + +def _prepare_cli(observer, event, tmp_path, monkeypatch, extra_args=()): + observer.ingest_all.return_value = ([event], []) + observer.get_diagnostic_records.return_value = [health_record("codex", "parse", counts={"events_emitted": 1})] + local_file = tmp_path / "synthetic-session.json" + observer.filter_entries_by_existing_files.side_effect = lambda entries, _: [] if local_file.exists() else entries + + def save_sessions(entries, output_dir): + local_file.write_text(json.dumps(entries[0].get_non_null_fields())) + return [local_file] + + observer.save_sessions_to_individual_files.side_effect = save_sessions + monkeypatch.setattr( + "sys.argv", + [ + "adr-sensor", + "--save-sessions", + "--output-dir", + str(tmp_path), + "--otel-config", + "synthetic.json", + *extra_args, + ], + ) + return local_file + + +@patch("adr_sensor.cli.OpenTelemetryLogExporter") +@patch("adr_sensor.cli.load_opentelemetry_config") +@patch("adr_sensor.cli.AgentObserver") +def test_cli_failed_delivery_retries_then_skips_unchanged( + observer_cls, load_config, exporter_cls, tmp_path, config, event, monkeypatch +): + load_config.return_value = config + observer = observer_cls.return_value + local_file = _prepare_cli(observer, event, tmp_path, monkeypatch) + exporter = exporter_cls.return_value + exporter.export.return_value = 1 + exporter.shutdown.side_effect = [OpenTelemetryExportError("synthetic delivery failure"), None, None] + + with pytest.raises(SystemExit) as error: + main() + assert error.value.code == 1 + assert local_file.exists() + assert not list(tmp_path.glob(".adr-otel-delivery.*.json")) + + main() + main() + + assert exporter_cls.call_count == 3 + assert exporter.export.call_count == 3 + assert [call.args for call in exporter.export.call_args_list] == [([event], []), ([event], []), ([], [])] + assert exporter.export_diagnostics.call_count == 3 + assert observer.save_sessions_to_individual_files.call_count == 1 + assert DeliveryCheckpoint(tmp_path, config).pending_entries([event]) == [] + + +@patch("adr_sensor.cli.OpenTelemetryLogExporter") +@patch("adr_sensor.cli.load_opentelemetry_config") +@patch("adr_sensor.cli.AgentObserver") +def test_cli_checkpoint_write_failure_is_reported_and_retried( + observer_cls, load_config, exporter_cls, tmp_path, config, event, monkeypatch, capsys +): + load_config.return_value = config + _prepare_cli(observer_cls.return_value, event, tmp_path, monkeypatch) + exporter_cls.return_value.export.return_value = 1 + with patch("adr_sensor.exporters.delivery_checkpoint.os.replace", side_effect=OSError("synthetic private path")): + with pytest.raises(SystemExit) as error: + main() + + assert error.value.code == 1 + stderr = capsys.readouterr().err + assert "OpenTelemetry checkpoint failed" in stderr + assert "synthetic private path" not in stderr + assert not list(tmp_path.glob(".adr-otel-delivery.*.json")) + main() + assert exporter_cls.return_value.export.call_count == 2 + + +@patch("adr_sensor.cli.OpenTelemetryLogExporter") +@patch("adr_sensor.cli.load_opentelemetry_config") +@patch("adr_sensor.cli.AgentObserver") +def test_cli_no_save_does_not_read_or_write_delivery_state( + observer_cls, load_config, exporter_cls, tmp_path, config, event, monkeypatch +): + load_config.return_value = config + _prepare_cli(observer_cls.return_value, event, tmp_path, monkeypatch, extra_args=["--no-save"]) + checkpoint = DeliveryCheckpoint(tmp_path, config) + checkpoint.pending_entries([event]) + checkpoint.commit() + original = checkpoint.path.read_bytes() + exporter_cls.return_value.export.return_value = 1 + + main() + + exporter_cls.return_value.export.assert_called_once_with([event], []) + observer_cls.return_value.save_sessions_to_individual_files.assert_not_called() + assert checkpoint.path.read_bytes() == original + + +@patch("adr_sensor.cli.load_opentelemetry_config") +@patch("adr_sensor.cli.AgentObserver") +def test_cli_partial_rejection_does_not_checkpoint_and_next_run_retries( + observer_cls, load_config, tmp_path, config, event, monkeypatch, caplog +): + load_config.return_value = config + local_file = _prepare_cli(observer_cls.return_value, event, tmp_path, monkeypatch) + rejected = Response() + rejected.status_code = 200 + rejected._content = ExportLogsServiceResponse( + partial_success={"rejected_log_records": 1, "error_message": "synthetic private collector response"} + ).SerializeToString() + accepted = Response() + accepted.status_code = 200 + accepted._content = b"" + + with patch("requests.adapters.HTTPAdapter.send", side_effect=[rejected, accepted, accepted]) as transport: + with pytest.raises(SystemExit) as error: + main() + assert error.value.code == 1 + assert local_file.exists() + assert not list(tmp_path.glob(".adr-otel-delivery.*.json")) + + main() + main() + + assert transport.call_count == 3 + requests = [ExportLogsServiceRequest.FromString(call.args[0].body) for call in transport.call_args_list] + event_names = [ + [ + record.event_name + for resource in request.resource_logs + for scope in resource.scope_logs + for record in scope.log_records + ] + for request in requests + ] + assert event_names == [ + ["adr.agent.session", "adr.sensor.health"], + ["adr.agent.session", "adr.sensor.health"], + ["adr.sensor.health"], + ] + assert DeliveryCheckpoint(tmp_path, config).pending_entries([event]) == [] + assert "synthetic private collector response" not in caplog.text diff --git a/Sensor/tests/test_diagnostics.py b/Sensor/tests/test_diagnostics.py new file mode 100644 index 0000000..811aee4 --- /dev/null +++ b/Sensor/tests/test_diagnostics.py @@ -0,0 +1,271 @@ +"""Health summaries are bounded and remain useful when capture yields no data.""" + +import json +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest +from opentelemetry.sdk._logs.export import InMemoryLogRecordExporter, SimpleLogRecordProcessor + +from adr_sensor import diagnostics +from adr_sensor.cli import main +from adr_sensor.diagnostics import health_record, sanitize_health_record, write_health_records +from adr_sensor.exporters.config import OpenTelemetryConfig +from adr_sensor.exporters.opentelemetry import OpenTelemetryExportError, OpenTelemetryLogExporter +from adr_sensor.observer import AgentObserver +from adr_sensor.parsers.base_parser import BaseParser +from adr_sensor.schemas.agent_event_schema import AgentEvent, ChatMessage + + +def _event(): + return AgentEvent( + timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), + source="claude", + session_id="synthetic-session", + chat_history=[ChatMessage(role="user", content="Keep the full synthetic prompt SECRET_CANARY")], + username="synthetic-user", + hostname="synthetic-host", + ) + + +class _Parser(BaseParser): + def __init__(self, records=(), reason=None, error=False): + self.records = list(records) + self.reason = reason + self.error = error + + def parse_all(self): + if self.reason: + self.record_diagnostic(self.reason, 4) + if self.error: + raise ValueError("SECRET_CANARY /private/project/secret.txt") + return self.records + + +def _observer(tmp_path, parser): + observer = AgentObserver(output_dir=tmp_path) + observer.SOURCES = (("claude", "Claude Code"),) + observer.claude_parser = parser + return observer + + +@pytest.mark.parametrize( + ("reasons", "status", "drift"), + [ + ({}, "empty", False), + ({"input_missing": 1}, "no_input", False), + ({"file_age_skipped": 4}, "empty", False), + ({"incomplete_record": 1}, "empty", False), + ({"record_decode_error": 2}, "failed", False), + ({"unsupported_schema": 1}, "failed", True), + ], +) +def test_health_distinguishes_no_input_corruption_and_suspected_drift(reasons, status, drift): + record = health_record("claude", "parse", reasons=reasons) + assert record["status"] == status + assert record["suspected_schema_drift"] is drift + + +def test_fixed_schema_rejects_payloads_paths_and_arbitrary_label_values(): + record = health_record( + "SECRET_CANARY", + "SECRET_CANARY", + counts={"SECRET_CANARY": 1, "events_emitted": "SECRET_CANARY", "failed": -1, "attempted": True}, + reasons={"SECRET_CANARY": 1, "record_decode_error": 10**100}, + ) + record.update(message="SECRET_CANARY", trace="SECRET_CANARY", session_id="SECRET_CANARY") + record["timestamp"] = "SECRET_CANARY" + safe = sanitize_health_record(record) + assert "SECRET_CANARY" not in json.dumps(safe) + assert safe["source"] == "sensor" + assert safe["counts"] == {} + assert safe["reasons"] == {"record_decode_error": 2**63 - 1} + + +def test_parser_counts_are_written_when_no_session_survives(tmp_path): + observer = _observer(tmp_path, _Parser(reason="record_decode_error")) + assert observer.ingest_all("claude") == ([], []) + record = json.loads((tmp_path / "diagnostics.jsonl").read_text()) + assert record["reasons"] == {"record_decode_error": 4} + assert record["status"] == "failed" + assert observer.has_errors + assert json.loads((tmp_path / "error.log").read_text()) == record + + +def test_partial_capture_preserves_payload_and_resets_counters_between_runs(tmp_path): + event = _event() + observer = _observer(tmp_path, _Parser([event], reason="record_shape_error")) + for _ in range(2): + entries, _ = observer.ingest_all("claude") + assert entries == [event] + assert "SECRET_CANARY" in entries[0].chat_history[0].content + records = [json.loads(line) for line in (tmp_path / "diagnostics.jsonl").read_text().splitlines()] + assert len(records) == 2 + assert all(record["reasons"] == {"record_shape_error": 4} for record in records) + assert all(record["status"] == "partial" for record in records) + assert "SECRET_CANARY" not in json.dumps(records) + + +def test_escaping_parser_error_does_not_log_exception_values(tmp_path): + observer = _observer(tmp_path, _Parser(error=True)) + observer.ingest_all("claude") + assert observer.has_errors + assert "SECRET_CANARY" not in (tmp_path / "error.log").read_text() + assert observer.get_diagnostic_records()[0]["reasons"] == {"parser_error": 1} + + +def test_invalid_parser_return_is_isolated_and_reported(tmp_path): + parser = _Parser() + parser.parse_all = lambda: None + observer = _observer(tmp_path, parser) + assert observer.ingest_all("claude") == ([], []) + assert observer.get_diagnostic_records()[0]["reasons"] == {"parser_error": 1} + + +def test_repeated_output_errors_are_coalesced_without_retaining_details(tmp_path): + observer = _observer(tmp_path, _Parser()) + for _ in range(1000): + observer._emit_error({"stage": "compare_session", "source": "claude", "message": "SECRET_CANARY"}) + records = observer.get_diagnostic_records() + assert len(records) == 1 + assert records[0]["stage"] == "save" + assert records[0]["reasons"] == {"write_error": 1000} + records[0]["reasons"]["write_error"] = 0 + assert observer.get_diagnostic_records()[0]["reasons"] == {"write_error": 1000} + observer.flush_diagnostics() + assert "SECRET_CANARY" not in (tmp_path / "error.log").read_text() + + +def test_diagnostic_files_rotate_and_keep_a_bounded_number_of_backups(tmp_path, monkeypatch): + monkeypatch.setattr(diagnostics, "MAX_LOG_BYTES", 650) + records = [health_record("claude", "parse", reasons={"record_shape_error": 1}) for _ in range(25)] + assert write_health_records(tmp_path, records) + assert len(list(tmp_path.glob("diagnostics.jsonl*"))) == 3 + assert len(list(tmp_path.glob("error.log*"))) == 3 + for path in tmp_path.iterdir(): + assert path.stat().st_size <= 650 + for line in path.read_text().splitlines(): + assert json.loads(line)["event"] == "adr.sensor.health" + + +def test_log_failure_warns_once_without_breaking_capture(tmp_path, monkeypatch, capsys): + observer = _observer(tmp_path, _Parser([_event()])) + + def fail(*args, **kwargs): + raise OSError("SECRET_CANARY") + + monkeypatch.setattr(diagnostics, "RotatingFileHandler", fail) + entries, _ = observer.ingest_all("claude") + observer.flush_diagnostics() + observer.flush_diagnostics() + assert len(entries) == 1 + assert observer.has_errors + stderr = capsys.readouterr().err + assert stderr.count("Unable to write sensor diagnostics") == 1 + assert "SECRET_CANARY" not in stderr + + +def test_failed_session_save_is_persisted_and_counted(tmp_path, monkeypatch): + observer = _observer(tmp_path, _Parser()) + + def fail(*args, **kwargs): + raise OSError("SECRET_CANARY") + + monkeypatch.setattr(observer, "_create_session_temp", fail) + assert observer.save_sessions_to_individual_files([_event()], tmp_path) == [] + assert observer.has_errors + record = observer.get_diagnostic_records()[0] + assert record["stage"] == "save_session" + assert record["counts"] == {"attempted": 1, "succeeded": 0, "failed": 1} + assert record["reasons"] == {"write_error": 1} + assert "SECRET_CANARY" not in (tmp_path / "error.log").read_text() + + +def test_otlp_health_record_uses_separate_schema_and_warning_severity(): + memory = InMemoryLogRecordExporter() + exporter = OpenTelemetryLogExporter( + OpenTelemetryConfig(endpoint="http://localhost:4318/v1/logs"), + "test", + _log_record_exporter=memory, + _processor_factory=SimpleLogRecordProcessor, + ) + record = health_record("claude", "parse", reasons={"record_decode_error": 3}) + record["message"] = "SECRET_CANARY" + assert exporter.export_diagnostics([record]) == 1 + exporter.shutdown() + emitted = memory.get_finished_logs()[0].log_record + assert emitted.event_name == "adr.sensor.health" + assert emitted.severity_text == "WARN" + assert emitted.attributes["adr.event.type"] == "sensor_health" + assert emitted.body["reasons"] == {"record_decode_error": 3} + assert "SECRET_CANARY" not in json.dumps(emitted.body) + + +def test_cli_sends_health_even_when_no_session_was_captured(tmp_path, monkeypatch): + observer = _observer(tmp_path, _Parser(reason="unsupported_schema")) + exporter = MagicMock() + exporter.export.return_value = 0 + monkeypatch.setattr("sys.argv", ["adr-sensor", "--no-save", "--otel-config", "synthetic.json"]) + with ( + patch("adr_sensor.cli.AgentObserver", return_value=observer), + patch("adr_sensor.cli.load_opentelemetry_config", return_value=MagicMock()), + patch("adr_sensor.cli.OpenTelemetryLogExporter", return_value=exporter), + ): + main() + exporter.export.assert_called_once_with([], []) + assert exporter.export_diagnostics.call_args.args[0][0]["suspected_schema_drift"] is True + exporter.shutdown.assert_called_once() + + +def test_cli_can_fail_after_preserving_partial_capture(tmp_path, monkeypatch, capsys): + observer = _observer(tmp_path, _Parser([_event()], reason="record_shape_error")) + monkeypatch.setattr("sys.argv", ["adr-sensor", "--no-save", "--fail-on-error"]) + with patch("adr_sensor.cli.AgentObserver", return_value=observer), pytest.raises(SystemExit) as failure: + main() + assert failure.value.code == 1 + assert "completed with errors" in capsys.readouterr().out + assert (tmp_path / "diagnostics.jsonl").exists() + + +def test_export_failure_is_recorded_locally(tmp_path, monkeypatch): + observer = _observer(tmp_path, _Parser()) + exporter = MagicMock() + exporter.shutdown.side_effect = OpenTelemetryExportError("synthetic delivery failure") + monkeypatch.setattr("sys.argv", ["adr-sensor", "--no-save", "--otel-config", "synthetic.json"]) + with ( + patch("adr_sensor.cli.AgentObserver", return_value=observer), + patch("adr_sensor.cli.load_opentelemetry_config", return_value=MagicMock()), + patch("adr_sensor.cli.OpenTelemetryLogExporter", return_value=exporter), + pytest.raises(SystemExit), + ): + main() + records = [json.loads(line) for line in (tmp_path / "error.log").read_text().splitlines()] + assert records[-1]["reasons"] == {"export_error": 1} + + +def test_resource_log_marks_observed_partial_failure_unsuccessful(tmp_path, monkeypatch): + observer = _observer(tmp_path, _Parser([_event()], reason="record_shape_error")) + usage = SimpleNamespace(ru_utime=0, ru_stime=0, ru_maxrss=0) + resources = MagicMock() + resources.getrusage.return_value = usage + monkeypatch.setattr("sys.argv", ["adr-sensor", "--no-save", "--resource", "--output-dir", str(tmp_path)]) + with ( + patch("adr_sensor.cli.AgentObserver", return_value=observer), + patch("adr_sensor.cli.resource_mod", resources), + patch("adr_sensor.cli.platform.system", return_value="Linux"), + ): + main() + assert json.loads((tmp_path / "resource.log").read_text())["success"] is False + + +def test_startup_failure_has_content_free_local_record(tmp_path, monkeypatch): + monkeypatch.setattr("sys.argv", ["adr-sensor", "--no-save", "--output-dir", str(tmp_path)]) + with ( + patch("adr_sensor.cli.AgentObserver", side_effect=RuntimeError("SECRET_CANARY")), + pytest.raises(RuntimeError), + ): + main() + record = json.loads((tmp_path / "error.log").read_text()) + assert record["reasons"] == {"startup_error": 1} + assert "SECRET_CANARY" not in json.dumps(record) diff --git a/Sensor/tests/test_opentelemetry_exporter.py b/Sensor/tests/test_opentelemetry_exporter.py index 4c48759..c06ee41 100644 --- a/Sensor/tests/test_opentelemetry_exporter.py +++ b/Sensor/tests/test_opentelemetry_exporter.py @@ -1,14 +1,19 @@ """Tests for OTLP conversion of ADR Sensor records.""" +import time from datetime import datetime, timezone +from unittest.mock import patch import pytest +from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import ExportLogsServiceResponse from opentelemetry.sdk._logs.export import ( InMemoryLogRecordExporter, LogRecordExportResult, SimpleLogRecordProcessor, ) +from requests import Response +from adr_sensor.diagnostics import health_record from adr_sensor.exporters.config import OpenTelemetryConfig from adr_sensor.exporters.opentelemetry import ( OpenTelemetryExportError, @@ -115,3 +120,172 @@ def test_datetime_to_unix_nanos_treats_naive_datetime_as_utc(): naive = aware.replace(tzinfo=None) assert _datetime_to_unix_nanos(naive) == _datetime_to_unix_nanos(aware) + + +def test_large_export_drains_bounded_batches_without_losing_records(monkeypatch): + # The old default queue lost records above 2048 when its worker was busy. + # Environment settings must not silently shrink our explicit queue bound. + monkeypatch.setenv("OTEL_BLRP_MAX_QUEUE_SIZE", "1") + monkeypatch.setenv("OTEL_BLRP_MAX_EXPORT_BATCH_SIZE", "1") + memory_exporter = InMemoryLogRecordExporter() + batch_sizes = [] + original_export = memory_exporter.export + + def record_batch(batch): + time.sleep(0.005) + batch_sizes.append(len(batch)) + return original_export(batch) + + memory_exporter.export = record_batch + exporter = OpenTelemetryLogExporter( + OpenTelemetryConfig(endpoint="http://localhost:4318/v1/logs"), + service_version="1.2.3", + _log_record_exporter=memory_exporter, + ) + entries = [_event()] * 4097 + health = [health_record("codex", "parse", counts={"events_emitted": 1})] * 1025 + + assert exporter.export(entries, []) == len(entries) + assert exporter.export_diagnostics(health) == len(health) + exporter.shutdown() + + assert len(memory_exporter.get_finished_logs()) == len(entries) + len(health) + assert sum(batch_sizes) == len(entries) + len(health) + assert max(batch_sizes) <= 512 + + +def test_shutdown_detects_silently_dropped_records(): + class DroppingProcessor(SimpleLogRecordProcessor): + def on_emit(self, log_record): + pass + + exporter = OpenTelemetryLogExporter( + OpenTelemetryConfig(endpoint="http://localhost:4318/v1/logs"), + service_version="1.2.3", + _log_record_exporter=InMemoryLogRecordExporter(), + _processor_factory=DroppingProcessor, + ) + exporter.export([_event()], []) + + with pytest.raises(OpenTelemetryExportError, match="submitted 1, successfully exported 0"): + exporter.shutdown() + + +def test_failed_intermediate_batch_stops_large_exports(): + class FailingExporter(InMemoryLogRecordExporter): + def export(self, batch): + return LogRecordExportResult.FAILURE + + exporter = OpenTelemetryLogExporter( + OpenTelemetryConfig(endpoint="http://localhost:4318/v1/logs"), + service_version="1.2.3", + _log_record_exporter=FailingExporter(), + ) + with pytest.raises(OpenTelemetryExportError, match="did not accept"): + exporter.export([_event()] * 4097, []) + with pytest.raises(OpenTelemetryExportError, match="did not accept"): + exporter.shutdown() + + +def test_shutdown_reports_flush_timeout_and_still_stops_provider(): + exporter = OpenTelemetryLogExporter( + OpenTelemetryConfig(endpoint="http://localhost:4318/v1/logs"), + service_version="1.2.3", + _log_record_exporter=InMemoryLogRecordExporter(), + ) + with ( + patch.object(exporter._provider, "force_flush", return_value=False), + patch.object(exporter._provider, "shutdown", wraps=exporter._provider.shutdown) as shutdown, + pytest.raises(OpenTelemetryExportError, match="did not flush"), + ): + exporter.shutdown() + shutdown.assert_called_once_with() + + +def test_export_after_shutdown_is_rejected(): + exporter = OpenTelemetryLogExporter( + OpenTelemetryConfig(endpoint="http://localhost:4318/v1/logs"), + service_version="1.2.3", + _log_record_exporter=InMemoryLogRecordExporter(), + ) + exporter.shutdown() + with pytest.raises(OpenTelemetryExportError, match="already shut down"): + exporter.export([_event()], []) + + +def _http_response(body, status_code=200): + response = Response() + response.status_code = status_code + response._content = body + response.headers["Content-Type"] = "application/x-protobuf" + return response + + +@pytest.mark.parametrize( + "body", + [ + b"", + ExportLogsServiceResponse(partial_success={"rejected_log_records": 0}).SerializeToString(), + ExportLogsServiceResponse( + partial_success={"rejected_log_records": 0, "error_message": "synthetic private warning"} + ).SerializeToString(), + b"\x10\x01", # Unknown protobuf fields remain forward compatible. + ], +) +def test_http_acknowledgement_accepts_full_success_and_zero_rejection_warnings(body, caplog): + exporter = OpenTelemetryLogExporter( + OpenTelemetryConfig(endpoint="https://collector.example.invalid/v1/logs"), service_version="1.2.3" + ) + with patch("requests.adapters.HTTPAdapter.send", return_value=_http_response(body)) as transport: + assert exporter.export([_event()], []) == 1 + exporter.shutdown() + + transport.assert_called_once() + assert exporter._record_exporter.exported_count == 1 + assert "synthetic private warning" not in caplog.text + + +@pytest.mark.parametrize( + "body,status_code", + [ + ( + ExportLogsServiceResponse( + partial_success={"rejected_log_records": 1, "error_message": "synthetic private rejection"} + ).SerializeToString(), + 200, + ), + (ExportLogsServiceResponse(partial_success={"rejected_log_records": -1}).SerializeToString(), 200), + (b"synthetic private invalid response", 200), + (b"", 304), + ], +) +def test_http_acknowledgement_rejects_partial_or_invalid_success_without_private_text(body, status_code, caplog): + exporter = OpenTelemetryLogExporter( + OpenTelemetryConfig(endpoint="https://collector.example.invalid/v1/logs"), service_version="1.2.3" + ) + with patch("requests.adapters.HTTPAdapter.send", return_value=_http_response(body, status_code)) as transport: + exporter.export([_event()], []) + with pytest.raises(OpenTelemetryExportError, match="did not accept") as error: + exporter.shutdown() + + transport.assert_called_once() + assert exporter._record_exporter.exported_count == 0 + assert "synthetic private" not in caplog.text + assert "synthetic private" not in str(error.value) + + +@pytest.mark.parametrize( + "variable", + ["OTEL_PYTHON_EXPORTER_OTLP_HTTP_CREDENTIAL_PROVIDER", "OTEL_PYTHON_EXPORTER_OTLP_HTTP_LOGS_CREDENTIAL_PROVIDER"], +) +def test_opaque_credential_providers_fail_explicitly_without_loading_plugin(variable, monkeypatch): + monkeypatch.setenv(variable, "synthetic-private-provider") + with ( + patch("adr_sensor.exporters.opentelemetry._load_opentelemetry_components") as load_components, + pytest.raises(OpenTelemetryExportError, match="credential-provider plugins are unsupported") as error, + ): + OpenTelemetryLogExporter( + OpenTelemetryConfig(endpoint="https://collector.example.invalid/v1/logs"), service_version="1.2.3" + ) + load_components.assert_not_called() + assert "synthetic-private-provider" not in str(error.value) diff --git a/Sensor/tests/test_parser_diagnostics.py b/Sensor/tests/test_parser_diagnostics.py new file mode 100644 index 0000000..7c79a0e --- /dev/null +++ b/Sensor/tests/test_parser_diagnostics.py @@ -0,0 +1,259 @@ +"""Content-free parser diagnostics exercised only with synthetic inputs.""" + +import json +import os +import sqlite3 +from pathlib import Path + +import pytest + +from adr_sensor.parsers.base_parser import BaseParser +from adr_sensor.parsers.claude_desktop_parser import ClaudeDesktopParser +from adr_sensor.parsers.cline_parser import ClineParser +from adr_sensor.parsers.codex_parser import CodexParser +from adr_sensor.parsers.copilot_parser import CopilotParser +from adr_sensor.parsers.cursor_parser import CursorParser +from adr_sensor.parsers.dsh_parser import DshParser +from adr_sensor.parsers.gemini_parser import GeminiParser +from adr_sensor.parsers.opencode_parser import OpencodeParser +from adr_sensor.parsers.warp_parser import WarpParser + +PARSERS = ( + ClaudeDesktopParser, + ClineParser, + CodexParser, + CopilotParser, + CursorParser, + DshParser, + GeminiParser, + OpencodeParser, + WarpParser, +) +CANARY = "diagnostic-private-payload-credential" + + +def isolated_parser(parser_class, tmp_path): + """Avoid constructors probing real installed-agent paths.""" + parser = parser_class.__new__(parser_class) + parser.max_age_days = 0 + parser.base_path = tmp_path / "missing" + parser.base_paths = [parser.base_path] + parser.codex_home = parser.base_path + parser.db_path = parser.base_path / "missing.db" + parser.base_dir = parser.base_path + parser.backend = None + return parser + + +def test_diagnostics_are_lazy_isolated_bounded_and_resettable(tmp_path): + first = isolated_parser(CodexParser, tmp_path) + second = isolated_parser(CodexParser, tmp_path) + assert first.get_diagnostics() == {} + for _ in range(10_000): + first.record_diagnostic("record_decode_error") + assert first.get_diagnostics() == {"record_decode_error": 10_000} + snapshot = first.get_diagnostics() + snapshot["record_decode_error"] = 0 + assert first.get_diagnostics()["record_decode_error"] == 10_000 + assert second.get_diagnostics() == {} + first.reset_diagnostics() + assert first.get_diagnostics() == {} + + +@pytest.mark.parametrize( + "code,count", + [ + (CANARY, 1), + (None, 1), + ([], 1), + ("parser_error", 0), + ("parser_error", -1), + ("parser_error", True), + ("parser_error", 1.5), + ], +) +def test_diagnostics_reject_unbounded_labels_and_invalid_counts(tmp_path, code, count): + parser = isolated_parser(CodexParser, tmp_path) + with pytest.raises(ValueError) as error: + parser.record_diagnostic(code, count) + assert CANARY not in str(error.value) + assert parser.get_diagnostics() == {} + + +@pytest.mark.parametrize("parser_class", PARSERS) +def test_absent_source_is_an_expected_skip(tmp_path, parser_class): + parser = isolated_parser(parser_class, tmp_path) + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"input_missing": 1} + assert set(parser.get_diagnostics()) <= BaseParser.EXPECTED_DIAGNOSTIC_CODES + + +@pytest.mark.parametrize("source", ["codex", "copilot", "dsh", "gemini", "claude_desktop"]) +def test_jsonl_recovery_reports_decode_error_without_copying_content(tmp_path, source): + classes = { + "codex": CodexParser, + "copilot": CopilotParser, + "dsh": DshParser, + "gemini": GeminiParser, + "claude_desktop": ClaudeDesktopParser, + } + parser = isolated_parser(classes[source], tmp_path) + directory = tmp_path / "private-source-path" + directory.mkdir() + path = directory / ("events.jsonl" if source == "copilot" else "audit.jsonl") + records = { + "codex": [ + {"type": "session_meta", "payload": {"id": "synthetic-session"}}, + {"type": "response_item", "payload": {"type": "message", "role": "user", "content": CANARY}}, + ], + "copilot": [{"type": "user.message", "data": {"content": CANARY}}], + "dsh": [ + {"type": "session", "version": 3, "id": "synthetic-session"}, + { + "type": "user/message", + "data": {"id": "user1", "role": "user", "content": [{"type": "text", "text": CANARY}]}, + }, + ], + "gemini": [{"sessionId": "synthetic-session"}, {"id": "user1", "type": "user", "content": CANARY}], + "claude_desktop": [{"type": "user", "uuid": "user1", "message": {"content": CANARY}}], + }[source] + path.write_text("invalid-json-" + CANARY + "\n" + "\n".join(map(json.dumps, records)) + "\n", encoding="utf-8") + before = path.read_bytes() + if source == "copilot": + entry = parser.parse_session_dir(directory) + elif source == "dsh": + entry = parser.parse_session_file(path) + elif source == "gemini": + entry = parser.parse_file(path) + elif source == "claude_desktop": + entry = parser._parse_session(path, {}) + else: + entry = parser.parse_jsonl_file(path) + + assert entry is not None + assert entry.chat_history[0].content == CANARY + assert path.read_bytes() == before + assert parser.get_diagnostics() == {"record_decode_error": 1} + encoded_diagnostics = json.dumps(parser.get_diagnostics()) + assert CANARY not in encoded_diagnostics + assert str(path) not in encoded_diagnostics + assert "private-source-path" not in encoded_diagnostics + + +@pytest.mark.parametrize("parser_class", [DshParser, GeminiParser]) +def test_diagnostics_survive_when_every_record_is_rejected(tmp_path, parser_class): + parser = isolated_parser(parser_class, tmp_path) + path = tmp_path / "session.jsonl" + path.write_text("not json\n[]\n", encoding="utf-8") + entry = parser.parse_session_file(path) if parser_class is DshParser else parser.parse_file(path) + assert entry is None + assert parser.get_diagnostics()["record_decode_error"] == 1 + assert parser.get_diagnostics()["record_shape_error"] >= 1 + + +@pytest.mark.parametrize("parser_class", [CursorParser, OpencodeParser, WarpParser]) +def test_incompatible_database_reports_failure(tmp_path, parser_class): + parser = isolated_parser(parser_class, tmp_path) + parser.db_path = tmp_path / "synthetic.db" + parser.backend = "sqlite" + with sqlite3.connect(parser.db_path): + pass + assert parser.parse_all() == [] + assert parser.get_diagnostics()["database_error"] >= 1 + + +def test_dsh_new_generation_reports_schema_drift_without_reading_payload(tmp_path): + parser = isolated_parser(DshParser, tmp_path) + parser.base_path = tmp_path + (tmp_path / "session.v999.jsonl").write_text(CANARY, encoding="utf-8") + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"unsupported_schema": 1} + + +def test_dsh_uncommitted_tail_is_not_reported_as_corrupt_json(tmp_path): + parser = isolated_parser(DshParser, tmp_path) + path = tmp_path / "session.v3.jsonl" + path.write_text('{"unfinished":', encoding="utf-8") + assert parser.parse_session_file(path) is None + assert parser.get_diagnostics() == {"incomplete_record": 1} + assert set(parser.get_diagnostics()) <= BaseParser.EXPECTED_DIAGNOSTIC_CODES + + +def test_codex_unsupported_catalog_and_missing_optional_catalog(tmp_path): + parser = isolated_parser(CodexParser, tmp_path) + parser._add_catalog_rollouts({}, tmp_path / "missing.sqlite") + assert parser.get_diagnostics() == {} + path = tmp_path / "catalog.sqlite" + with sqlite3.connect(path) as connection: + connection.execute("CREATE TABLE threads (id TEXT)") + parser._add_catalog_rollouts({}, path) + assert parser.get_diagnostics() == {"unsupported_schema": 1} + + +def test_stat_failure_is_not_reported_as_age_filtering(tmp_path, monkeypatch): + parser = isolated_parser(CodexParser, tmp_path) + path = tmp_path / "session.jsonl" + path.write_text("{}", encoding="utf-8") + original_stat = Path.stat + + def fail_selected_path(candidate, *args, **kwargs): + if candidate == path: + raise PermissionError(CANARY) + return original_stat(candidate, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", fail_selected_path) + parser._add_rollout_candidate({}, path) + assert parser.get_diagnostics() == {"file_stat_error": 1} + + +def test_cline_malformed_file_and_expected_old_file_are_distinct(tmp_path): + parser = isolated_parser(ClineParser, tmp_path) + parser.base_path = tmp_path + task = tmp_path / "synthetic-task" + task.mkdir() + path = task / "api_conversation_history.json" + path.write_text("invalid-json-" + CANARY, encoding="utf-8") + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"record_decode_error": 1} + parser.reset_diagnostics() + parser.max_age_days = 14 + os.utime(path, (1, 1)) + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"file_age_skipped": 1} + + +def test_opencode_skipped_json_rows_and_warp_plain_model_fallback(tmp_path): + parser = isolated_parser(OpencodeParser, tmp_path) + assert parser._safe_json("invalid-json-" + CANARY) is None + assert parser.get_diagnostics() == {"record_decode_error": 1} + warp = isolated_parser(WarpParser, tmp_path) + assert warp._parse_json_safely("plain-model-name", report_failure=False) is None + assert warp.get_diagnostics() == {} + assert warp._parse_json_safely("invalid-json-" + CANARY) is None + assert warp.get_diagnostics() == {"record_decode_error": 1} + + +def test_cline_malformed_tool_arguments_report_loss(tmp_path): + parser = isolated_parser(ClineParser, tmp_path) + assert ( + parser.extract_mcp_tools( + "testread" + '{"invalid":}' + ) + == [] + ) + assert parser.get_diagnostics() == {"record_decode_error": 1} + + +def test_gemini_optional_metadata_absence_is_not_a_read_failure(tmp_path): + parser = isolated_parser(GeminiParser, tmp_path) + project = tmp_path / "tmp" / "project" + chats = project / "chats" + chats.mkdir(parents=True) + path = chats / "session.jsonl" + assert parser._project_path(path) is None + assert parser.get_diagnostics() == {} + (project / ".project_root").write_bytes(b"\xff") + (tmp_path / "projects.json").write_text("invalid-json-" + CANARY, encoding="utf-8") + assert parser._project_path(path) is None + assert parser.get_diagnostics() == {"file_read_error": 1, "record_decode_error": 1}