From cabfadd7c56301677ddf1cc376f00d2fdce3faae Mon Sep 17 00:00:00 2001 From: bozbas Date: Mon, 24 Aug 2026 07:00:30 +0000 Subject: [PATCH] feat(sensor): apply log lookback to Cline Summary: Intent: - Bound Cline collection to the configured log lookback. - Avoid dropping otherwise readable tasks when file metadata is unavailable. Changes: - Filter task directories by modification time. - Retain tasks when timestamp lookup fails. --- Sensor/adr_sensor/observer.py | 2 +- Sensor/adr_sensor/parsers/cline_parser.py | 34 +++++++++- Sensor/tests/test_observer.py | 10 +-- Sensor/tests/test_parsers.py | 81 +++++++++++++++++++++++ 4 files changed, 120 insertions(+), 7 deletions(-) diff --git a/Sensor/adr_sensor/observer.py b/Sensor/adr_sensor/observer.py index 97e2109..63bd9a2 100644 --- a/Sensor/adr_sensor/observer.py +++ b/Sensor/adr_sensor/observer.py @@ -72,7 +72,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() - self.cline_parser = ClineParser() + 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() self.opencode_parser = ( OpencodeParser(max_age_days=max_age_days) if max_age_days is not None else OpencodeParser() diff --git a/Sensor/adr_sensor/parsers/cline_parser.py b/Sensor/adr_sensor/parsers/cline_parser.py index b068954..8a2c474 100644 --- a/Sensor/adr_sensor/parsers/cline_parser.py +++ b/Sensor/adr_sensor/parsers/cline_parser.py @@ -7,7 +7,7 @@ import json import re -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Dict, List, Optional @@ -18,6 +18,7 @@ # Cline stores task history inside the Cursor extension's global storage, # relative to the per-platform app-data root. _CLINE_TASKS_SUFFIX = "Cursor/User/globalStorage/saoudrizwan.claude-dev/tasks" +MAX_LOG_AGE_DAYS = 14 class ClineParser(BaseParser): @@ -30,8 +31,9 @@ class ClineParser(BaseParser): windows_appdata() / _CLINE_TASKS_SUFFIX, # Windows (%APPDATA%) ] - def __init__(self): + 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]) + self.max_age_days = max_age_days def parse_all(self) -> List[AgentEvent]: """Parse all available Cline logs.""" @@ -46,6 +48,34 @@ def parse_all(self) -> List[AgentEvent]: task_dirs = [d for d in self.base_path.iterdir() if d.is_dir()] print(f"[CLINE] Found {len(task_dirs)} task directories") + 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: + try: + modified_at = task_dir.stat().st_mtime + except OSError as e: + print(f"[CLINE] Error checking task {task_dir}: {e}") + recent_task_dirs.append(task_dir) + continue + + if modified_at >= cutoff_timestamp: + recent_task_dirs.append(task_dir) + else: + skipped_count += 1 + + task_dirs = recent_task_dirs + if skipped_count > 0: + print(f"[CLINE] Skipped {skipped_count} tasks older than {self.max_age_days} days") + + print(f"[CLINE] Processing {len(task_dirs)} task directories") + for task_dir in task_dirs: try: entry = self.parse_cline_log(task_dir) diff --git a/Sensor/tests/test_observer.py b/Sensor/tests/test_observer.py index 2854b0f..d3e1cd3 100644 --- a/Sensor/tests/test_observer.py +++ b/Sensor/tests/test_observer.py @@ -2,13 +2,10 @@ import json from datetime import datetime, timezone -from pathlib import Path from unittest.mock import MagicMock, patch -import pytest - from adr_sensor.observer import AgentObserver -from adr_sensor.schemas.agent_event_schema import AgentEvent, ChatMessage, ToolUsage +from adr_sensor.schemas.agent_event_schema import AgentEvent, ChatMessage class TestAgentObserver: @@ -17,6 +14,11 @@ def test_init_default(self, tmp_path): observer = AgentObserver(output_dir=tmp_path) assert observer.output_dir == tmp_path + def test_init_propagates_max_age_days_to_cline(self, tmp_path): + observer = AgentObserver(output_dir=tmp_path, max_age_days=30) + + assert observer.cline_parser.max_age_days == 30 + def test_display_summary_empty(self, tmp_path, capsys): """Test display summary with no data.""" observer = AgentObserver(output_dir=tmp_path) diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index 4112a14..87b98f8 100644 --- a/Sensor/tests/test_parsers.py +++ b/Sensor/tests/test_parsers.py @@ -103,6 +103,17 @@ def test_truncate_large_arguments(self): class TestClineParser: + @staticmethod + def _write_task(base_path: Path, name: str, text: str = "hello"): + task_dir = base_path / name + task_dir.mkdir() + api_file = task_dir / "api_conversation_history.json" + api_file.write_text( + json.dumps([{"role": "user", "content": [{"type": "text", "text": text}]}]), + encoding="utf-8", + ) + return task_dir, api_file + def test_parse_cline_log(self, tmp_path): """Test parsing a Cline task directory.""" task_dir = tmp_path / "1234567890" @@ -132,6 +143,76 @@ def test_parse_cline_log(self, tmp_path): assert entry.source == "cline" assert len(entry.chat_history) == 2 + def test_parse_all_filters_tasks_by_conversation_mtime(self, tmp_path): + recent_task, _ = self._write_task(tmp_path, "recent", "recent task") + old_task, old_api_file = self._write_task(tmp_path, "old", "old task") + now = datetime.now(timezone.utc).timestamp() + old = (datetime.now(timezone.utc) - timedelta(days=30)).timestamp() + + # The conversation file, not its task directory, determines a task's age. + os.utime(recent_task, (old, old)) + os.utime(old_api_file, (old, old)) + os.utime(old_task, (now, now)) + + parser = ClineParser(max_age_days=14) + parser.base_path = tmp_path + with patch.object(parser, "parse_cline_log", wraps=parser.parse_cline_log) as parse_log: + entries = parser.parse_all() + + assert {entry.session_id for entry in entries} == {"cline_recent"} + assert [call.args[0] for call in parse_log.call_args_list] == [recent_task] + + def test_parse_all_uses_directory_mtime_when_conversation_is_missing(self, tmp_path): + task_dir = tmp_path / "missing-conversation" + task_dir.mkdir() + old = (datetime.now(timezone.utc) - timedelta(days=30)).timestamp() + os.utime(task_dir, (old, old)) + parser = ClineParser(max_age_days=14) + parser.base_path = tmp_path + + with patch.object(parser, "parse_cline_log", wraps=parser.parse_cline_log) as parse_log: + entries = parser.parse_all() + + assert entries == [] + parse_log.assert_not_called() + + def test_parse_all_falls_back_to_task_mtime_after_file_stat_failure(self, tmp_path, monkeypatch): + _, failing_api_file = self._write_task(tmp_path, "stat-failure", "recovered task") + self._write_task(tmp_path, "healthy", "healthy task") + parser = ClineParser(max_age_days=14) + parser.base_path = tmp_path + original_stat = Path.stat + should_fail = True + + def fail_one_stat(path, *args, **kwargs): + nonlocal should_fail + if path == failing_api_file and should_fail: + should_fail = False + raise OSError("stat unavailable") + return original_stat(path, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", fail_one_stat) + + entries = parser.parse_all() + + assert {entry.session_id for entry in entries} == {"cline_stat-failure", "cline_healthy"} + + @pytest.mark.parametrize("max_age_days", [0, -1]) + def test_parse_all_non_positive_max_age_includes_all_history(self, tmp_path, max_age_days): + task_dir, api_file = self._write_task(tmp_path, "old", "old task") + old = (datetime.now(timezone.utc) - timedelta(days=30)).timestamp() + os.utime(api_file, (old, old)) + os.utime(task_dir, (old, old)) + parser = ClineParser(max_age_days=max_age_days) + parser.base_path = tmp_path + + entries = parser.parse_all() + + assert {entry.session_id for entry in entries} == {"cline_old"} + + def test_default_max_age_days(self): + assert ClineParser().max_age_days == 14 + def test_extract_mcp_tools(self): """Test MCP tool extraction from text.""" parser = ClineParser()