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..d156ca1 100644 --- a/Sensor/README.md +++ b/Sensor/README.md @@ -362,6 +362,52 @@ The one-shot Sensor process flushes and shuts down the exporter before exiting. Use an OpenTelemetry Collector when vendor-specific routing, transformation, retry, or persistent queuing is needed. +### Sensor health and parser diagnostics + +Every ingestion run writes a content-free summary for each attempted source, +including runs that produce no sessions. `diagnostics.jsonl` contains all summaries; +`error.log` contains only `partial` and `failed` summaries. Both live under +`--output-dir` (default `./output`), even when `--save-sessions` uses its separate +default cache directory or `--no-save` suppresses captured session files. Each log +rotates at 1 MiB with two backups. Use one active sensor process per output directory +to avoid concurrent rotation races. Diagnostic write failures produce a fixed stderr +warning and do not discard captured sessions. + +The versioned `adr.sensor.health` schema contains timestamp, sensor version, source, +stage, status, fixed reason codes, and aggregate counts. For example: + +```json +{"schema_version":1,"event":"adr.sensor.health","timestamp":"2026-01-01T00:00:00.000+00:00","sensor_version":"0.0.0","source":"claude","stage":"parse","status":"partial","suspected_schema_drift":false,"counts":{"events_returned":2,"events_emitted":2,"events_filtered":0},"reasons":{"record_decode_error":1}} +``` + +Statuses distinguish successful capture (`ok`), no meaningful output (`empty`), +absent input (`no_input`), usable output with observed errors (`partial`), and +errors without usable output (`failed`). Age filtering and an incomplete live +tail are expected skips, not errors. `suspected_schema_drift` is a triage hint for +explicitly unsupported record/content/schema shapes, not proof of an upstream +format change. Generic corruption is reported separately. + +All ten parsers report observed recovery failures, but coverage is not exhaustive: +some optional metadata/timestamp fallbacks, unknown record kinds, and compressed +DSH tail recovery are not classified. Counts describe observed recovery operations, +not necessarily unique damaged records. A healthy summary does not prove complete +capture; a missing summary also cannot distinguish an idle endpoint from a sensor +that never ran. Schedule runs and monitor last-seen health externally. + +With `--otel-config`, health is also sent as OTLP logs (`adr.event.type=sensor_health`), +including when there are no session records. Health errors use WARN severity; +expected skips use INFO. No OTLP exporter is created without that argument. A failed +export is recorded locally because a broken destination cannot receive its own alert. +`--fail-on-error` exits nonzero after preserving available capture when an observed +parse/save/diagnostic failure occurs; by default these partial failures are reported +without changing the existing continue-on-error behavior. OTLP failures remain nonzero. +When `--resource` is enabled, `resource.log` also marks partial runs unsuccessful. + +New structured diagnostics never include prompts, tool arguments/results, paths, +session IDs, exception messages, or tracebacks. This is a separate operational +schema, **not redaction of captured telemetry**. Legacy console previews/errors and +older entries already present in `error.log` are not sanitized by this change. + ## Output Schema ### AgentEvent @@ -541,8 +587,9 @@ cannot run on the current platform are skipped rather than failing. | `APPDATA` | Cursor, Cline, Claude Desktop parsers | Windows roaming app-data root. Consulted first so redirected/roaming profiles resolve correctly (default `~/AppData/Roaming`) | | `LOCALAPPDATA` | Warp parser | Windows local app-data root, same redirected-profile handling (default `~/AppData/Local`) | -Errors during ingestion never abort the run: each source is isolated, and failures -are appended as single-line JSON records to `error.log` in the output directory. +Each source is isolated during ingestion. See +[Sensor health and parser diagnostics](#sensor-health-and-parser-diagnostics) for +structured logs, partial-failure exit behavior, and monitoring limitations. ## Security Use Cases diff --git a/Sensor/adr_sensor/cli.py b/Sensor/adr_sensor/cli.py index 511b763..ddf6dda 100644 --- a/Sensor/adr_sensor/cli.py +++ b/Sensor/adr_sensor/cli.py @@ -24,6 +24,7 @@ resource_mod = None from . import __version__ +from .diagnostics import health_record, write_health_records from .exporters import OpenTelemetryConfigError, load_opentelemetry_config from .exporters.opentelemetry import OpenTelemetryExportError, OpenTelemetryLogExporter from .observer import AgentObserver @@ -88,7 +89,12 @@ def main(): help="Directory to save output files (default: ./output)", ) parser.add_argument("--limit", type=_non_negative_int, default=2, help="Number of entries to display") - parser.add_argument("--no-save", action="store_true", help="Do not save to file") + parser.add_argument( + "--no-save", action="store_true", help="Do not save captured sessions (diagnostics still written)" + ) + parser.add_argument( + "--fail-on-error", action="store_true", help="Exit nonzero after partial capture or output failures" + ) parser.add_argument( "--save-sessions", action="store_true", @@ -130,6 +136,7 @@ def main(): success = True observer = None + stage = "startup" try: # Determine max_age_days @@ -140,6 +147,7 @@ def main(): observer = AgentObserver(output_dir=args.output_dir, max_age_days=max_age_days) # Ingest logs + stage = "parse" entries, system_config_data = observer.ingest_all(args.source) # Apply incremental filtering @@ -154,6 +162,7 @@ def main(): observer.display_summary(entries, system_config_data, limit=args.limit) # Save + stage = "save" if entries or system_config_data: if not args.no_save: if args.save_sessions: @@ -171,26 +180,56 @@ def main(): entries, system_config_data, output_format=args.output_format, output_dir=project_output_dir ) - if otel_config is not None: - otel_exporter = OpenTelemetryLogExporter(otel_config, service_version=get_version()) - try: - exported_count = otel_exporter.export(entries, system_config_data) - finally: - otel_exporter.shutdown() - print(f"\nOpenTelemetry logs sent: {exported_count}") - - print("\nADR Sensor complete!\n") + if otel_config is not None: + # Health records must reach monitoring even when parsing produced + # no sessions, or all local session snapshots were unchanged. + stage = "export" + otel_exporter = OpenTelemetryLogExporter(otel_config, service_version=get_version()) + try: + exported_count = otel_exporter.export(entries, system_config_data) + otel_exporter.export_diagnostics(observer.get_diagnostic_records()) + finally: + otel_exporter.shutdown() + print(f"\nOpenTelemetry session/configuration logs sent: {exported_count}") + + success = observer.has_errors is not True + if success: + print("\nADR Sensor complete!\n") + else: + print("\nADR Sensor completed with errors; see diagnostics.jsonl.\n") + if args.fail_on_error: + raise SystemExit(1) except OpenTelemetryExportError as exc: success = False + if observer is not None: + observer.record_failure("export", "export_error") print(f"OpenTelemetry export failed: {exc}", file=sys.stderr) raise SystemExit(1) + except Exception: + success = False + if observer is not None: + reason = {"parse": "parser_error", "save": "write_error", "export": "export_error"}.get( + stage, "startup_error" + ) + observer.record_failure(stage, reason) + else: + write_health_records( + args.output_dir or Path.cwd() / "output", + [health_record("sensor", "startup", reasons={"startup_error": 1})], + ) + raise + except BaseException: success = False raise finally: + if observer is not None: + observer.flush_diagnostics() + if observer.has_errors is True: + success = False if capture_resource: try: end_time = time.monotonic() diff --git a/Sensor/adr_sensor/diagnostics.py b/Sensor/adr_sensor/diagnostics.py new file mode 100644 index 0000000..3b756dc --- /dev/null +++ b/Sensor/adr_sensor/diagnostics.py @@ -0,0 +1,122 @@ +"""Bounded, content-free operational records, separate from captured sessions.""" + +import json +import logging +import sys +from datetime import datetime, timezone +from logging.handlers import RotatingFileHandler +from pathlib import Path +from typing import Dict, Iterable, Optional + +from . import __version__ +from .parsers.base_parser import BaseParser + +DIAGNOSTIC_SOURCES = frozenset( + {"sensor", "claude", "claude_desktop", "cursor", "cline", "codex", "copilot", "dsh", "gemini", "opencode", "warp"} +) +DIAGNOSTIC_STAGES = frozenset({"parse", "save", "save_session", "export", "startup"}) +OPERATIONAL_REASONS = frozenset({"parser_error", "write_error", "export_error", "startup_error"}) +COUNT_FIELDS = frozenset({"events_returned", "events_emitted", "events_filtered", "attempted", "succeeded", "failed"}) +MAX_LOG_BYTES = 1024 * 1024 +LOG_BACKUP_COUNT = 2 +MAX_COUNT = 2**63 - 1 + + +def _counts(values: Dict[str, int], allowed: Iterable[str]) -> Dict[str, int]: + """Accept only fixed keys and bounded integers, never caller-provided text.""" + allowed = frozenset(allowed) + if not isinstance(values, dict): + return {} + return { + key: min(value, MAX_COUNT) + for key, value in values.items() + if key in allowed and isinstance(value, int) and not isinstance(value, bool) and value >= 0 + } + + +def health_record( + source: str, + stage: str, + *, + counts: Optional[Dict[str, int]] = None, + reasons: Optional[Dict[str, int]] = None, +) -> dict: + """Summarize observed issues without claiming that every issue is schema drift.""" + safe_counts = _counts(counts or {}, COUNT_FIELDS) + safe_reasons = _counts(reasons or {}, BaseParser.DIAGNOSTIC_CODES | OPERATIONAL_REASONS) + issues = sum(count for reason, count in safe_reasons.items() if reason not in BaseParser.EXPECTED_DIAGNOSTIC_CODES) + emitted = safe_counts.get("events_emitted", safe_counts.get("succeeded", 0)) + if issues: + status = "partial" if emitted else "failed" + elif safe_reasons.get("input_missing") and not emitted: + status = "no_input" + elif not emitted and stage == "parse": + status = "empty" + else: + status = "ok" + return { + "schema_version": 1, + "event": "adr.sensor.health", + "timestamp": datetime.now(timezone.utc).isoformat(timespec="milliseconds"), + "sensor_version": __version__, + "source": source if isinstance(source, str) and source in DIAGNOSTIC_SOURCES else "sensor", + "stage": stage if isinstance(stage, str) and stage in DIAGNOSTIC_STAGES else "startup", + "status": status, + "suspected_schema_drift": any( + count and reason in {"unsupported_schema", "unsupported_record_type", "unsupported_content_block"} + for reason, count in safe_reasons.items() + ), + "counts": safe_counts, + "reasons": safe_reasons, + } + + +def sanitize_health_record(record: dict) -> dict: + """Revalidate the fixed schema at each serialization boundary.""" + safe = health_record( + record.get("source"), record.get("stage"), counts=record.get("counts"), reasons=record.get("reasons") + ) + try: + timestamp = datetime.fromisoformat(record.get("timestamp", "")) + if timestamp.tzinfo is not None: + safe["timestamp"] = timestamp.astimezone(timezone.utc).isoformat(timespec="milliseconds") + except (TypeError, ValueError): + pass + return safe + + +def write_health_records(output_dir: Path, records: Iterable[dict]) -> bool: + """Append rotating JSONL diagnostics; logging failures never erase capture.""" + handlers = [] + try: + output_dir.mkdir(parents=True, exist_ok=True) + for name in ("diagnostics.jsonl", "error.log"): + handler = RotatingFileHandler( + output_dir / name, maxBytes=MAX_LOG_BYTES, backupCount=LOG_BACKUP_COUNT, encoding="utf-8", delay=True + ) + handler.setFormatter(logging.Formatter("%(message)s")) + + # Handler.emit normally suppresses write failures. Surface them to + # the single bounded fallback below instead of logging record data. + def handle_error(record): + raise OSError("diagnostic write failed") + + handler.handleError = handle_error + handlers.append(handler) + for record in records: + record = sanitize_health_record(record) + message = json.dumps(record, ensure_ascii=True, separators=(",", ":")) + item = logging.LogRecord("adr_sensor.health", logging.INFO, "", 0, message, (), None) + handlers[0].handle(item) + if record["status"] in {"partial", "failed"}: + handlers[1].handle(item) + return True + except Exception: + print("[ADR] Unable to write sensor diagnostics; captured session data is unaffected.", file=sys.stderr) + return False + finally: + for handler in handlers: + try: + handler.close() + except Exception: + pass diff --git a/Sensor/adr_sensor/exporters/opentelemetry.py b/Sensor/adr_sensor/exporters/opentelemetry.py index d4bd14e..7037df9 100644 --- a/Sensor/adr_sensor/exporters/opentelemetry.py +++ b/Sensor/adr_sensor/exporters/opentelemetry.py @@ -6,6 +6,7 @@ from datetime import datetime, timezone from typing import Any, List, Optional, Tuple +from ..diagnostics import sanitize_health_record from ..schemas.agent_event_schema import AgentEvent from ..schemas.system_config_schema import SystemConfiguration from .config import OpenTelemetryConfig @@ -63,6 +64,7 @@ def __init__( self._provider.add_log_record_processor(processor) self._logger = self._provider.get_logger("adr_sensor", service_version) self._info_severity = severity_number_cls.INFO + self._warning_severity = severity_number_cls.WARN self._flush_timeout_millis = int(config.flush_timeout_seconds * 1000) self._closed = False @@ -101,6 +103,28 @@ def export( return len(entries) + len(system_config_data) + def export_diagnostics(self, records: List[dict]) -> int: + """Send bounded health summaries through the explicitly configured sink.""" + for record in records: + body = sanitize_health_record(record) + degraded = body["status"] in {"partial", "failed"} + self._logger.emit( + timestamp=_datetime_to_unix_nanos(datetime.fromisoformat(body["timestamp"])), + observed_timestamp=time.time_ns(), + severity_number=self._warning_severity if degraded else self._info_severity, + severity_text="WARN" if degraded else "INFO", + body=body, + attributes={ + "adr.event.type": "sensor_health", + "adr.schema.version": SCHEMA_VERSION, + "adr.source": body["source"], + "adr.sensor.stage": body["stage"], + "adr.sensor.status": body["status"], + }, + event_name="adr.sensor.health", + ) + return len(records) + def shutdown(self) -> None: """Flush pending records and stop the provider's worker thread.""" if self._closed: 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_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_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}