diff --git a/Sensor/README.md b/Sensor/README.md index 1b47d69..bebd225 100644 --- a/Sensor/README.md +++ b/Sensor/README.md @@ -103,6 +103,26 @@ Use `--all-history` to include older sessions. This source covers GitHub Copilot CLI session state only; it does not read the separate storage used by the VS Code Copilot Chat extension. +### Cline (Claude Dev) + +The `cline` source reads task conversation histories from the Cline extension +(`saoudrizwan.claude-dev`) across supported VS Code-family editors, including +Visual Studio Code (primary), Cursor, VS Code Insiders, VSCodium, and Windsurf: + +| Host | Default task directories | +| ---- | ------------------------ | +| macOS | `~/Library/Application Support//User/globalStorage/saoudrizwan.claude-dev/tasks/` | +| Linux | `~/.config//User/globalStorage/saoudrizwan.claude-dev/tasks/` | +| Windows | `%APPDATA%\\User\globalStorage\saoudrizwan.claude-dev\tasks\` | + +The parser scans all existing candidate directories, normalizes user prompts and +assistant tool usages (including `` invocations) into the ADR +schema, and deduplicates sessions across editors. Lookback defaults to 14 days by +task file modification time; pass `max_age_days` to adjust. +An inaccessible candidate is reported in diagnostics and the sensor log without +preventing collection from other editors. Python callers can restrict collection +to one task directory with `ClineParser(base_path=...)`. + ### opencode [opencode](https://github.com/sst/opencode) uses the XDG layout on every platform, diff --git a/Sensor/adr_sensor/parsers/cline_parser.py b/Sensor/adr_sensor/parsers/cline_parser.py index 3ecbc9e..6c4928a 100644 --- a/Sensor/adr_sensor/parsers/cline_parser.py +++ b/Sensor/adr_sensor/parsers/cline_parser.py @@ -16,84 +16,156 @@ from ..utils.platform_paths import windows_appdata from .base_parser import BaseParser -# Cline stores task history inside the Cursor extension's global storage, -# relative to the per-platform app-data root. logger = logging.getLogger(__name__) -_CLINE_TASKS_SUFFIX = "Cursor/User/globalStorage/saoudrizwan.claude-dev/tasks" +# Task history subpath for the Cline extension across VS Code-based editors, +# relative to the per-platform app-data root. +_CLINE_TASKS_SUBPATH = "User/globalStorage/saoudrizwan.claude-dev/tasks" + +# Legacy alias for backward compatibility. +_CLINE_TASKS_SUFFIX = f"Cursor/{_CLINE_TASKS_SUBPATH}" + +# Editors known to host the Cline extension, ordered by prominence. +_SUPPORTED_EDITORS = ( + "Code", # Visual Studio Code (primary) + "Cursor", # Cursor + "Code - Insiders", # Visual Studio Code Insiders + "VSCodium", # VSCodium + "Windsurf", # Windsurf +) MAX_LOG_AGE_DAYS = 14 +def _build_candidate_base_paths() -> List[Path]: + """Generate candidate task directories across supported editors and platforms.""" + candidates: List[Path] = [] + for editor in _SUPPORTED_EDITORS: + suffix = f"{editor}/{_CLINE_TASKS_SUBPATH}" + candidates.extend( + [ + Path.home() / "Library/Application Support" / suffix, # macOS + Path.home() / ".config" / suffix, # Linux + windows_appdata() / suffix, # Windows (%APPDATA%) + ] + ) + return candidates + + class ClineParser(BaseParser): """Parser for Cline (Claude Dev) logs.""" - #: Candidate task directories, checked in order. - BASE_PATHS = [ - Path.home() / "Library/Application Support" / _CLINE_TASKS_SUFFIX, # macOS - Path.home() / ".config" / _CLINE_TASKS_SUFFIX, # Linux - windows_appdata() / _CLINE_TASKS_SUFFIX, # Windows (%APPDATA%) - ] - - def __init__(self, max_age_days: int = MAX_LOG_AGE_DAYS): - self.base_path = next((p for p in self.BASE_PATHS if p.exists()), self.BASE_PATHS[0]) + #: Candidate task directories, checked in order across supported editors and platforms. + BASE_PATHS = _build_candidate_base_paths() + + def __init__( + self, + max_age_days: int = MAX_LOG_AGE_DAYS, + base_path: Optional[Path] = None, + ): + self._custom_base_path = Path(base_path) if base_path is not None else None + if self._custom_base_path is not None: + self._base_path = self._custom_base_path + else: + self._base_path = self.BASE_PATHS[0] + for candidate in self.BASE_PATHS: + try: + if candidate.exists(): + self._base_path = candidate + break + except OSError: + # Report inaccessible roots in parse_all(), after the + # observer resets the per-run diagnostic counters. + continue self.max_age_days = max_age_days + @property + def base_path(self) -> Path: + return self._base_path + + @base_path.setter + def base_path(self, value: Path) -> None: + """Preserve callers that select one root by assigning base_path.""" + self._base_path = Path(value) + self._custom_base_path = self._base_path + + def _candidate_base_paths(self) -> List[Path]: + """Return task directories to scan for Cline logs.""" + custom_base_path = getattr(self, "_custom_base_path", None) + if custom_base_path is not None: + return [custom_base_path] + # Probe each root inside parse_all's recovery boundary so a failed + # existence check cannot prevent collection from the other editors. + return list(self.BASE_PATHS) + def parse_all(self) -> List[AgentEvent]: """Parse all available Cline logs.""" - entries = [] - - if not self.base_path.exists(): - self.record_diagnostic("input_missing") - logger.info("[CLINE] No logs found at %s", self.base_path) - return entries - - logger.info("[CLINE] Scanning for logs in %s", self.base_path) + entries: List[AgentEvent] = [] + candidate_paths = self._candidate_base_paths() + scanned_any = False + scan_failed = False + seen_sessions = set() - task_dirs = [d for d in self.base_path.iterdir() if d.is_dir()] - logger.info("[CLINE] Found %s task directories", len(task_dirs)) + for base_path in candidate_paths: + try: + if not base_path.exists(): + continue + scanned_any = True + logger.info("[CLINE] Scanning for logs in %s", base_path) + task_dirs = [d for d in base_path.iterdir() if d.is_dir()] + except OSError as exc: + scan_failed = True + self.record_diagnostic("file_read_error") + logger.error("[CLINE] Error scanning task directory %s: %s", base_path, exc) + continue + logger.info("[CLINE] Found %s task directories", len(task_dirs)) - if self.max_age_days > 0: - cutoff_timestamp = (datetime.now(timezone.utc) - timedelta(days=self.max_age_days)).timestamp() - recent_task_dirs = [] - skipped_count = 0 + if self.max_age_days > 0: + cutoff_timestamp = (datetime.now(timezone.utc) - timedelta(days=self.max_age_days)).timestamp() + recent_task_dirs = [] + skipped_count = 0 - for task_dir in task_dirs: - api_file = task_dir / "api_conversation_history.json" - try: - modified_at = api_file.stat().st_mtime - 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") + for task_dir in task_dirs: + api_file = task_dir / "api_conversation_history.json" try: - modified_at = task_dir.stat().st_mtime - except OSError as e: - self.record_diagnostic("file_stat_error") - logger.warning("[CLINE] Error checking task %s: %s", task_dir, e) + modified_at = api_file.stat().st_mtime + 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") + logger.warning("[CLINE] Error checking task %s: %s", task_dir, e) + recent_task_dirs.append(task_dir) + continue + + if modified_at >= cutoff_timestamp: recent_task_dirs.append(task_dir) - continue - - if modified_at >= cutoff_timestamp: - recent_task_dirs.append(task_dir) - else: - skipped_count += 1 + else: + skipped_count += 1 - task_dirs = recent_task_dirs - if skipped_count > 0: - self.record_diagnostic("file_age_skipped", skipped_count) - logger.info("[CLINE] Skipped %s tasks older than %s days", skipped_count, self.max_age_days) + task_dirs = recent_task_dirs + if skipped_count > 0: + self.record_diagnostic("file_age_skipped", skipped_count) + logger.info("[CLINE] Skipped %s tasks older than %s days", skipped_count, self.max_age_days) - logger.info("[CLINE] Processing %s task directories", len(task_dirs)) + logger.info("[CLINE] Processing %s task directories", len(task_dirs)) - for task_dir in task_dirs: - try: - entry = self.parse_cline_log(task_dir) - if entry: - entries.append(entry) - except Exception as e: - self.record_diagnostic("session_build_error") - logger.warning("[CLINE] Error parsing task %s: %s", task_dir, e) + for task_dir in task_dirs: + try: + entry = self.parse_cline_log(task_dir) + if entry and entry.session_id not in seen_sessions: + seen_sessions.add(entry.session_id) + entries.append(entry) + except Exception as e: + self.record_diagnostic("session_build_error") + logger.warning("[CLINE] Error parsing task %s: %s", task_dir, e) + + if not scanned_any and not scan_failed: + self.record_diagnostic("input_missing") + logger.info("[CLINE] No logs found at %s", self.base_path) return entries diff --git a/Sensor/tests/test_cline_discovery.py b/Sensor/tests/test_cline_discovery.py new file mode 100644 index 0000000..9d70845 --- /dev/null +++ b/Sensor/tests/test_cline_discovery.py @@ -0,0 +1,153 @@ +"""Synthetic checks for Cline discovery across independent editor roots.""" + +import json +import logging +from pathlib import Path + +import pytest + +from adr_sensor import sensor_log +from adr_sensor.observer import AgentObserver +from adr_sensor.parsers.cline_parser import ClineParser + + +def _write_task(root, task_id): + task = root / task_id + task.mkdir(parents=True) + (task / "api_conversation_history.json").write_text( + json.dumps([{"role": "user", "content": [{"type": "text", "text": "Synthetic task"}]}]), + encoding="utf-8", + ) + + +def _make_bad_root(root, failure, monkeypatch): + if failure == "not_directory": + root.write_text("Not a task directory", encoding="utf-8") + return + + root.mkdir() + method = "exists" if failure == "exists" else "iterdir" + original = getattr(Path, method) + + def fail_one_root(path): + if path == root: + raise PermissionError("Synthetic access denied") + return original(path) + + monkeypatch.setattr(Path, method, fail_one_root) + + +@pytest.mark.parametrize("failure", ["exists", "enumeration", "not_directory"]) +@pytest.mark.parametrize("bad_first", [True, False]) +@pytest.mark.parametrize("through_observer", [False, True]) +def test_failed_editor_does_not_discard_healthy_sessions(tmp_path, monkeypatch, failure, bad_first, through_observer): + healthy = tmp_path / "healthy-editor" + _write_task(healthy, "healthy") + bad = tmp_path / "failed-editor" + _make_bad_root(bad, failure, monkeypatch) + roots = [bad, healthy] if bad_first else [healthy, bad] + monkeypatch.setattr(ClineParser, "BASE_PATHS", roots) + parser = ClineParser(max_age_days=0) + + if through_observer: + # Avoid initializing other parsers or inspecting real agent histories. + observer = AgentObserver.__new__(AgentObserver) + observer.cline_parser = parser + observer.output_dir = tmp_path / "output" + observer.output_dir.mkdir() + entries, configs = observer.ingest_all("cline") + assert configs == [] + record = observer.get_diagnostic_records()[0] + assert record["reasons"] == {"file_read_error": 1} + assert record["counts"]["events_emitted"] == 1 + else: + entries = parser.parse_all() + + assert [entry.session_id for entry in entries] == ["cline_healthy"] + assert parser.get_diagnostics() == {"file_read_error": 1} + + +@pytest.mark.parametrize("failure", ["exists", "enumeration", "not_directory"]) +def test_inaccessible_input_is_not_reported_as_missing(tmp_path, monkeypatch, failure): + root = tmp_path / "failed-editor" + _make_bad_root(root, failure, monkeypatch) + monkeypatch.setattr(ClineParser, "BASE_PATHS", [root]) + parser = ClineParser(max_age_days=0) + + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"file_read_error": 1} + + +def test_absent_editors_report_input_missing_once(tmp_path, monkeypatch): + monkeypatch.setattr(ClineParser, "BASE_PATHS", [tmp_path / "missing-code", tmp_path / "missing-cursor"]) + parser = ClineParser() + + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"input_missing": 1} + + +def test_explicit_base_path_does_not_scan_other_editors(tmp_path, monkeypatch): + selected = tmp_path / "selected" + other = tmp_path / "other-editor" + _write_task(selected, "selected") + _write_task(other, "unselected") + monkeypatch.setattr(ClineParser, "BASE_PATHS", [other, selected]) + + entries = ClineParser(base_path=selected, max_age_days=0).parse_all() + + assert [entry.session_id for entry in entries] == ["cline_selected"] + + +@pytest.mark.parametrize("editor", ["code", "cursor"]) +def test_base_path_assignment_restricts_collection_to_a_known_editor(tmp_path, monkeypatch, editor): + code = tmp_path / "code" + cursor = tmp_path / "cursor" + _write_task(code, "code") + _write_task(cursor, "cursor") + monkeypatch.setattr(ClineParser, "BASE_PATHS", [code, cursor]) + parser = ClineParser(max_age_days=0) + assert parser.base_path == code + parser.base_path = tmp_path / editor + + assert [entry.session_id for entry in parser.parse_all()] == [f"cline_{editor}"] + + +def test_repeated_session_is_emitted_once_across_roots(tmp_path, monkeypatch): + code = tmp_path / "code" + cursor = tmp_path / "cursor" + _write_task(code, "shared") + _write_task(cursor, "shared") + monkeypatch.setattr(ClineParser, "BASE_PATHS", [code, code, cursor]) + + entries = ClineParser(max_age_days=0).parse_all() + + assert [entry.session_id for entry in entries] == ["cline_shared"] + + +def test_quiet_collection_keeps_errors_in_stderr_and_runtime_log(tmp_path, monkeypatch, capsys): + healthy = tmp_path / "healthy-editor" + bad = tmp_path / "failed-editor" + _write_task(healthy, "healthy") + _make_bad_root(bad, "enumeration", monkeypatch) + monkeypatch.setattr(ClineParser, "BASE_PATHS", [healthy, bad]) + for handler in sensor_log._console_handlers(): + monkeypatch.setattr(handler, "level", logging.WARNING) + + log_dir = tmp_path / "runtime" + assert sensor_log.enable_runtime_log(log_dir) + try: + entries = ClineParser(max_age_days=0).parse_all() + finally: + sensor_log.disable_runtime_log() + + captured = capsys.readouterr() + assert [entry.session_id for entry in entries] == ["cline_healthy"] + assert captured.out == "" + assert "Error scanning task directory" in captured.err + records = [ + json.loads(line) for line in (log_dir / sensor_log.RUNTIME_ERRORS_LOG).read_text(encoding="utf-8").splitlines() + ] + assert len(records) == 1 + assert records[0]["level"] == "ERROR" + assert records[0]["component"] == "parsers.cline_parser" + assert records[0]["exception_type"] == "PermissionError" diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index 244d764..6259062 100644 --- a/Sensor/tests/test_parsers.py +++ b/Sensor/tests/test_parsers.py @@ -237,6 +237,34 @@ def test_parse_no_directory(self): entries = parser.parse_all() assert entries == [] + def test_init_accepts_custom_base_path(self, tmp_path): + """Test initializing ClineParser with a custom base_path.""" + parser = ClineParser(base_path=tmp_path) + assert parser.base_path == tmp_path + assert parser._candidate_base_paths() == [tmp_path] + + def test_cline_base_paths_covers_multiple_editors(self): + """Verify BASE_PATHS includes VS Code, Cursor, Insiders, VSCodium, and Windsurf.""" + paths = [p.as_posix() for p in ClineParser.BASE_PATHS] + for editor in ("Code", "Cursor", "Code - Insiders", "VSCodium", "Windsurf"): + assert any(f"/{editor}/User/globalStorage/" in p for p in paths) + + def test_parse_all_discovers_tasks_across_multiple_existing_editors(self, tmp_path, monkeypatch): + """Test that parse_all scans all existing candidate directories and merges tasks.""" + vscode_dir = tmp_path / "vscode_tasks" + cursor_dir = tmp_path / "cursor_tasks" + vscode_dir.mkdir() + cursor_dir.mkdir() + self._write_task(vscode_dir, "task_from_vscode", "hello from vscode") + self._write_task(cursor_dir, "task_from_cursor", "hello from cursor") + + monkeypatch.setattr(ClineParser, "BASE_PATHS", [vscode_dir, cursor_dir]) + parser = ClineParser() + entries = parser.parse_all() + + session_ids = {e.session_id for e in entries} + assert session_ids == {"cline_task_from_vscode", "cline_task_from_cursor"} + class TestCodexParser: @staticmethod @@ -2797,7 +2825,7 @@ class TestPlatformPathCoverage: """ def test_cursor_covers_macos_linux_and_windows(self): - paths = [str(p) for p in CursorParser.DB_PATHS] + paths = [p.as_posix() for p in CursorParser.DB_PATHS] assert any("Library/Application Support" in p for p in paths) # macOS assert any("/.config/" in p for p in paths) # Linux # Windows root comes from %APPDATA% (redirected-profile safe), not a hardcoded guess. @@ -2805,9 +2833,14 @@ def test_cursor_covers_macos_linux_and_windows(self): assert all(p.endswith("Cursor/User/globalStorage/state.vscdb") for p in paths) def test_cline_covers_macos_linux_and_windows(self): - paths = [str(p) for p in ClineParser.BASE_PATHS] + paths = [p.as_posix() for p in ClineParser.BASE_PATHS] assert any("Library/Application Support" in p for p in paths) # macOS assert any("/.config/" in p for p in paths) # Linux + # VS Code (primary) and Cursor are both covered on Windows: + assert ( + windows_appdata() / "Code/User/globalStorage/saoudrizwan.claude-dev/tasks" + in ClineParser.BASE_PATHS + ) assert ( windows_appdata() / "Cursor/User/globalStorage/saoudrizwan.claude-dev/tasks" in ClineParser.BASE_PATHS @@ -2815,7 +2848,7 @@ def test_cline_covers_macos_linux_and_windows(self): assert all(p.endswith("saoudrizwan.claude-dev/tasks") for p in paths) def test_warp_covers_both_macos_locations_and_windows(self): - paths = [str(p) for p in WarpParser.DB_PATHS] + paths = [p.as_posix() for p in WarpParser.DB_PATHS] assert any("Group Containers/2BBY89MBSN.dev.warp" in p for p in paths) # macOS sandboxed assert any(p.endswith("Library/Application Support/dev.warp.Warp-Stable/warp.sqlite") for p in paths) assert windows_local_appdata() / "warp/Warp/data/warp.sqlite" in WarpParser.DB_PATHS @@ -2824,12 +2857,13 @@ def test_warp_covers_both_macos_locations_and_windows(self): def test_claude_desktop_covers_macos_and_windows(self): from adr_sensor.parsers.claude_desktop_parser import DEFAULT_BASE_PATHS - assert any("Library/Application Support" in str(p) for p in DEFAULT_BASE_PATHS) # macOS + paths = [p.as_posix() for p in DEFAULT_BASE_PATHS] + assert any("Library/Application Support" in p for p in paths) # macOS assert windows_appdata() / "Claude/local-agent-mode-sessions" in DEFAULT_BASE_PATHS def test_opencode_covers_xdg_and_macos(self, monkeypatch): monkeypatch.delenv("XDG_DATA_HOME", raising=False) - paths = [str(p) for p in OpencodeParser._candidate_base_dirs()] + paths = [p.as_posix() for p in OpencodeParser._candidate_base_dirs()] assert any(p.endswith(".local/share/opencode") for p in paths) # Linux + macOS default assert any("Library/Application Support/opencode" in p for p in paths) # macOS fallback @@ -2837,8 +2871,9 @@ def test_opencode_covers_xdg_and_macos(self, monkeypatch): ("parser_cls", "attr"), [(CursorParser, "DB_PATHS"), (ClineParser, "BASE_PATHS"), (WarpParser, "DB_PATHS")], ) - def test_first_candidate_used_when_none_exist(self, parser_cls, attr): + def test_first_candidate_used_when_none_exist(self, monkeypatch, parser_cls, attr): """With no agent installed the parser reports the primary path, not a crash.""" + monkeypatch.setattr(Path, "exists", lambda self: False) parser = parser_cls() resolved = getattr(parser, "db_path", None) or parser.base_path assert resolved == getattr(parser_cls, attr)[0]