diff --git a/Sensor/adr_sensor/parsers/cursor_parser.py b/Sensor/adr_sensor/parsers/cursor_parser.py index 068ad76..dd3f749 100644 --- a/Sensor/adr_sensor/parsers/cursor_parser.py +++ b/Sensor/adr_sensor/parsers/cursor_parser.py @@ -62,70 +62,71 @@ def parse_conversations_from_bubbles(self) -> List[AgentEvent]: try: conn = sqlite3.connect(self.db_path) - cursor = conn.cursor() + try: + cursor = conn.cursor() + + composer_metadata = self.get_composer_metadata(cursor) + + cutoff_time = datetime.now(timezone.utc) - timedelta(days=self.max_age_days) + recent_conv_ids = set() + skipped_count = 0 + + for conv_id, metadata in composer_metadata.items(): + conv_timestamp = None + if "lastUpdatedAt" in metadata: + try: + conv_timestamp = normalize_timestamp(metadata["lastUpdatedAt"]) + except Exception: + pass + if conv_timestamp is None and "createdAt" in metadata: + try: + conv_timestamp = normalize_timestamp(metadata["createdAt"]) + except Exception: + pass + + if conv_timestamp is None or conv_timestamp >= cutoff_time: + recent_conv_ids.add(conv_id) + else: + skipped_count += 1 - composer_metadata = self.get_composer_metadata(cursor) + if skipped_count > 0: + print(f"[CURSOR] Skipped {skipped_count} conversations older than {self.max_age_days} days") - cutoff_time = datetime.now(timezone.utc) - timedelta(days=self.max_age_days) - recent_conv_ids = set() - skipped_count = 0 + cursor.execute("SELECT key, value FROM cursorDiskKV WHERE key LIKE 'bubbleId:%'") - for conv_id, metadata in composer_metadata.items(): - conv_timestamp = None - if "lastUpdatedAt" in metadata: - try: - conv_timestamp = normalize_timestamp(metadata["lastUpdatedAt"]) - except Exception: - pass - if conv_timestamp is None and "createdAt" in metadata: + conversations: Dict[str, List] = {} + for key, value in self._iter_cursor_batches(cursor): try: - conv_timestamp = normalize_timestamp(metadata["createdAt"]) - except Exception: - pass - - if conv_timestamp is None or conv_timestamp >= cutoff_time: - recent_conv_ids.add(conv_id) - else: - skipped_count += 1 + parts = key.split(":") + if len(parts) >= 3: + conv_id = parts[1] - if skipped_count > 0: - 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:%'") - - conversations: Dict[str, List] = {} - for key, value in self._iter_cursor_batches(cursor): - try: - parts = key.split(":") - if len(parts) >= 3: - conv_id = parts[1] - - if conv_id not in recent_conv_ids: - continue - - if conv_id not in conversations: - conversations[conv_id] = [] - - if isinstance(value, str): - try: - bubble_data = json.loads(value) - conversations[conv_id].append(bubble_data) - except json.JSONDecodeError: + if conv_id not in recent_conv_ids: continue - else: - conversations[conv_id].append(value) - except Exception: - continue - for conv_id, bubbles in conversations.items(): - try: - entry = self.parse_conversation(conv_id, bubbles, composer_metadata) - if entry: - entries.append(entry) - except Exception: - pass + if conv_id not in conversations: + conversations[conv_id] = [] + + if isinstance(value, str): + try: + bubble_data = json.loads(value) + conversations[conv_id].append(bubble_data) + except json.JSONDecodeError: + continue + else: + conversations[conv_id].append(value) + except Exception: + continue - conn.close() + for conv_id, bubbles in conversations.items(): + try: + entry = self.parse_conversation(conv_id, bubbles, composer_metadata) + if entry: + entries.append(entry) + except Exception: + pass + finally: + conn.close() except Exception as e: print(f"[CURSOR] Error parsing conversations: {e}") diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index 4112a14..349d66e 100644 --- a/Sensor/tests/test_parsers.py +++ b/Sensor/tests/test_parsers.py @@ -435,6 +435,50 @@ def test_parse_no_directory(self): assert entries == [] +class TestCursorParser: + def test_parse_closes_real_sqlite_connection(self, tmp_path): + db_path = tmp_path / "state.vscdb" + setup_connection = sqlite3.connect(db_path) + setup_connection.execute("CREATE TABLE cursorDiskKV (key TEXT, value TEXT)") + setup_connection.close() + parser = CursorParser() + parser.db_path = db_path + connections = [] + real_connect = sqlite3.connect + + def capture_connection(path): + connection = real_connect(path) + connections.append(connection) + return connection + + with patch("adr_sensor.parsers.cursor_parser.sqlite3.connect", side_effect=capture_connection): + assert parser.parse_conversations_from_bubbles() == [] + + with pytest.raises(sqlite3.ProgrammingError, match="closed database"): + connections[0].execute("SELECT 1") + + def test_parse_closes_connection_once_on_success(self): + parser = CursorParser() + + with patch("adr_sensor.parsers.cursor_parser.sqlite3.connect") as connect: + connect.return_value.cursor.return_value.fetchmany.return_value = [] + + assert parser.parse_conversations_from_bubbles() == [] + + connect.assert_called_once_with(parser.db_path) + connect.return_value.close.assert_called_once_with() + + def test_parse_closes_connection_once_after_exception(self): + parser = CursorParser() + + with patch("adr_sensor.parsers.cursor_parser.sqlite3.connect") as connect: + connect.return_value.cursor.side_effect = RuntimeError("cursor unavailable") + + assert parser.parse_conversations_from_bubbles() == [] + + connect.assert_called_once_with(parser.db_path) + connect.return_value.close.assert_called_once_with() + def _build_warp_db(db_path: Path, conversations: list) -> None: """Create a synthetic warp.sqlite matching the schema WarpParser queries.