fix: review findings

Author: bradjin8

CHANGELOG.md

@@ -11,12 +11,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Strict mypy** — `strict = true` in `pyproject.toml`; core TypedDict models
   (`SearchResult`, `ConversationSummary`) and full annotations on API routes and
   `utils/` (#100)
-
-### Changed
-- CI typecheck job runs `mypy .` using pyproject config (strict production code;
-  per-module overrides for `scripts/export.py` and `tests.*`)
-
-### Added
 - **Summary disk cache (Phase 3)** — project list and tab summaries cached under
   `~/.cache/cursor-chat-browser/`, invalidated when global or per-workspace DB
   mtimes change; bypass with `?nocache=1` or `CURSOR_CHAT_BROWSER_NOCACHE=1` (#84)
@@ -47,6 +41,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Unit tests for `determine_project_for_conversation` fallback chain (#87, #89)
 
 ### Changed
+- CI typecheck job runs `mypy .` using pyproject config (strict production code;
+  per-module overrides for `scripts/export.py` and `tests.*`)
 - **List-path performance** — skip full `messageRequestContext` scan unless
   invalid workspace aliases are needed; filter `composerData` in SQL; skip
   `Composer.from_dict` on list/summary paths; cache `composer_id_to_ws` mapping (#84)

api/search.py

@@ -36,9 +36,12 @@ def search() -> tuple[Response, int] | Response:
         query_lower = query.lower()
 
         results: list[SearchResult] = []
-        results.extend(
-            search_global_storage(workspace_path, query, query_lower, rules, parse_warnings)
-        )
+        if search_type != "chat":
+            results.extend(
+                search_global_storage(
+                    workspace_path, query, query_lower, rules, parse_warnings
+                )
+            )
         results.extend(
             search_legacy_workspaces(workspace_path, query, query_lower, search_type, rules)
         )

models/search.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any, TypedDict
+from typing import Any, Literal, TypedDict
 
 
 class ConversationSummary(TypedDict, total=False):
@@ -21,11 +21,11 @@ class _SearchResultRequired(TypedDict):
     chatTitle: str
     timestamp: int | str
     matchingText: str
-    type: str  # "composer" | "chat" | "cli_agent"
+    type: Literal["composer", "chat", "cli_agent"]
 
 
 class _SearchResultOptional(TypedDict, total=False):
-    source: str  # "cli" for CLI agent sessions
+    source: Literal["cli"]
 
 
 class SearchResult(_SearchResultRequired, _SearchResultOptional):

services/search.py

@@ -397,6 +397,7 @@ def search_legacy_workspaces(
                 data = json.loads(chat_row[0])
                 for tab in (data.get("tabs") or []):
                     ct = tab.get("chatTitle") or ""
+                    tab_id = str(tab.get("tabId") or "")
 
                     tab_model_names: list[str] | None = None
                     tab_meta = tab.get("metadata")
@@ -434,8 +435,8 @@ def search_legacy_workspaces(
                     results.append({
                         "workspaceId": name,
                         "workspaceFolder": workspace_folder,
-                        "chatId": tab.get("tabId"),
-                        "chatTitle": ct or f"Chat {(tab.get('tabId') or '')[:8]}",
+                        "chatId": tab_id,
+                        "chatTitle": ct or f"Chat {tab_id[:8]}",
                         "timestamp": tab.get("lastSendTime") or datetime.now().isoformat(),
                         "matchingText": matching_text,
                         "type": "chat",
@@ -560,11 +561,17 @@ def rank_results(results: list[SearchResult]) -> list[SearchResult]:
     """
     def _ts(r: SearchResult) -> float:
         t = r.get("timestamp", 0)
+        if t is None:
+            return 0.0
         if isinstance(t, str):
             try:
                 return datetime.fromisoformat(t.replace("Z", "+00:00")).timestamp()
             except Exception:
                 return 0.0
+        if isinstance(t, bool) or not isinstance(t, (int, float)):
+            return 0.0
+        if t > 1e12:
+            return float(t) / 1000.0
         return float(t) if t else 0.0
 
     return sorted(results, key=_ts, reverse=True)

services/workspace_db.py

@@ -7,7 +7,7 @@
 from collections.abc import Iterator
 from contextlib import closing, contextmanager
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 _logger = logging.getLogger(__name__)
 
@@ -22,6 +22,18 @@
 # corrupt table cannot propagate to callers.
 
 
+def safe_fetchall(
+    conn: sqlite3.Connection,
+    query: str,
+    params: tuple[Any, ...] = (),
+) -> list[sqlite3.Row]:
+    """Run *query* on *conn*; return rows or ``[]`` on sqlite3.Error."""
+    try:
+        return cast(list[sqlite3.Row], conn.execute(query, params).fetchall())
+    except sqlite3.Error:
+        return []
+
+
 def load_bubble_map(global_db: sqlite3.Connection) -> dict[str, dict[str, Any]]:
     """Load all ``bubbleId:*`` KV entries into ``{bubble_id: bubble_dict}``.
 

services/workspace_listing.py

@@ -5,7 +5,7 @@
 import os
 import sqlite3
 from datetime import datetime, timezone
-from typing import Any, cast
+from typing import Any
 
 _logger = logging.getLogger(__name__)
 
@@ -33,6 +33,7 @@
     load_project_layouts_for_composer,
     load_project_layouts_map,
     open_global_db,
+    safe_fetchall,
 )
 from utils.workspace_path import get_cli_chats_path
 from services.workspace_resolver import (
@@ -138,19 +139,8 @@ def _build_workspace_projects_uncached(
 
     with open_global_db(workspace_path) as (global_db, _):
         if global_db:
-            def _safe_fetchall(
-                query: str, params: tuple[Any, ...] = (),
-            ) -> list[sqlite3.Row]:
-                try:
-                    return cast(
-                        list[sqlite3.Row],
-                        global_db.execute(query, params).fetchall(),
-                    )
-                except sqlite3.Error:
-                    return []
-
             try:
-                composer_rows = _safe_fetchall(COMPOSER_ROWS_WITH_HEADERS_SQL)
+                composer_rows = safe_fetchall(global_db, COMPOSER_ROWS_WITH_HEADERS_SQL)
 
                 project_layouts_map: dict[str, list[str]] = {}
                 if invalid_workspace_ids:

services/workspace_tabs.py

@@ -7,7 +7,7 @@
 import sqlite3
 from collections.abc import Mapping
 from datetime import datetime
-from typing import Any, cast
+from typing import Any
 
 _logger = logging.getLogger(__name__)
 
@@ -44,6 +44,7 @@
     load_project_layouts_for_composer,
     load_project_layouts_map,
     open_global_db,
+    safe_fetchall,
 )
 from utils.workspace_path import get_cli_chats_path
 from services.workspace_resolver import (
@@ -548,22 +549,11 @@ def _build_workspace_tab_summaries_uncached(
 
         workspace_display_name = lookup_workspace_display_name(workspace_path, workspace_id)
 
-        def _safe_fetchall(
-            query: str, params: tuple[Any, ...] = (),
-        ) -> list[sqlite3.Row]:
-            try:
-                return cast(
-                    list[sqlite3.Row],
-                    global_db.execute(query, params).fetchall(),
-                )
-            except sqlite3.Error:
-                return []
-
         project_layouts_map: dict[str, list[str]] = {}
         if invalid_workspace_ids:
             project_layouts_map = load_project_layouts_map(global_db)
 
-        composer_rows = _safe_fetchall(COMPOSER_ROWS_WITH_HEADERS_SQL)
+        composer_rows = safe_fetchall(global_db, COMPOSER_ROWS_WITH_HEADERS_SQL)
 
         invalid_workspace_aliases: dict[str, str] = {}
         if invalid_workspace_ids:
@@ -709,18 +699,8 @@ def assemble_single_tab(
 
         workspace_display_name = lookup_workspace_display_name(workspace_path, workspace_id)
 
-        def _safe_fetchall(
-            query: str, params: tuple[Any, ...] = (),
-        ) -> list[sqlite3.Row]:
-            try:
-                return cast(
-                    list[sqlite3.Row],
-                    global_db.execute(query, params).fetchall(),
-                )
-            except sqlite3.Error:
-                return []
-
-        rows = _safe_fetchall(
+        rows = safe_fetchall(
+            global_db,
             "SELECT key, value FROM cursorDiskKV WHERE key = ?",
             (f"composerData:{composer_id}",),
         )
@@ -760,7 +740,7 @@ def _safe_fetchall(
         if invalid_workspace_ids:
             # Alias resolution still needs the composer roster, but project layouts
             # are intentionally limited to this composer (single-tab scope).
-            composer_rows_for_aliases = _safe_fetchall(COMPOSER_ROWS_WITH_HEADERS_SQL)
+            composer_rows_for_aliases = safe_fetchall(global_db, COMPOSER_ROWS_WITH_HEADERS_SQL)
             invalid_workspace_aliases = infer_invalid_workspace_aliases(
                 composer_rows=composer_rows_for_aliases,
                 project_layouts_map=project_layouts_map,
@@ -848,21 +828,11 @@ def assemble_workspace_tabs(
 
         workspace_display_name = lookup_workspace_display_name(workspace_path, workspace_id)
 
-        def _safe_fetchall(
-            query: str, params: tuple[Any, ...] = (),
-        ) -> list[sqlite3.Row]:
-            try:
-                return cast(
-                    list[sqlite3.Row],
-                    global_db.execute(query, params).fetchall(),
-                )
-            except sqlite3.Error:
-                return []
-
         # Load bubbles
-        for row in _safe_fetchall(
+        for row in safe_fetchall(
+            global_db,
             "SELECT key, value FROM cursorDiskKV WHERE key LIKE 'bubbleId:%'"
-            " AND value IS NOT NULL"
+            " AND value IS NOT NULL",
         ):
             parts = row["key"].split(":")
             if len(parts) >= 3:
@@ -901,7 +871,10 @@ def _safe_fetchall(
         # Load messageRequestContext rows once; build both
         # message_request_context_map and project_layouts_map from the same pass.
         project_layouts_map: dict[str, list[str]] = {}
-        for row in _safe_fetchall("SELECT key, value FROM cursorDiskKV WHERE key LIKE 'messageRequestContext:%'"):
+        for row in safe_fetchall(
+            global_db,
+            "SELECT key, value FROM cursorDiskKV WHERE key LIKE 'messageRequestContext:%'",
+        ):
             parts = row["key"].split(":")
             if len(parts) < 2:
                 continue
@@ -934,11 +907,12 @@ def _safe_fetchall(
                         project_layouts_map[chat_id].append(layout["rootPath"])
 
         # Get composer data entries with conversations
-        composer_rows = _safe_fetchall(
+        composer_rows = safe_fetchall(
+            global_db,
             "SELECT key, value FROM cursorDiskKV WHERE key LIKE 'composerData:%'"
             " AND value IS NOT NULL"
             " AND value LIKE '%fullConversationHeadersOnly%'"
-            " AND value NOT LIKE '%fullConversationHeadersOnly\":[]%'"
+            " AND value NOT LIKE '%fullConversationHeadersOnly\":[]%'",
         )
 
         invalid_workspace_aliases = infer_invalid_workspace_aliases(