Add search limit validation, typed models, and mypy strict CI

.github/workflows/ci.yml

@@ -57,3 +57,22 @@ jobs:
 
       - name: Run tests
         run: pytest --tb=short -q
+
+  mypy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+          cache-dependency-path: |
+            requirements.txt
+            requirements-dev.txt
+
+      - name: Install dev dependencies
+        run: pip install -r requirements-dev.txt
+
+      - name: Run mypy (strict)
+        run: mypy

api/_flask_types.py

@@ -0,0 +1,12 @@
+"""Shared Flask handler return types for mypy."""
+
+from typing import Any, Union, cast
+
+from flask import Response, jsonify
+
+FlaskReturn = Union[Response, tuple[Response, int]]
+
+
+def json_ok(*args: Any, **kwargs: Any) -> Response:
+    """Typed wrapper around :func:`flask.jsonify`."""
+    return cast(Response, jsonify(*args, **kwargs))

api/export_api.py

@@ -5,8 +5,12 @@
 import os
 import zipfile
 from datetime import datetime
+from typing import Any
 
-from flask import Blueprint, current_app, jsonify, request, send_file
+from flask import Blueprint, current_app, request, send_file
+
+from api._flask_types import FlaskReturn, json_ok
+from models.export import ExportStateDict
 
 from utils.export_state_store import (
     EXPORT_STATE_FILE,
@@ -33,24 +37,24 @@
 _STATE_FILE = EXPORT_STATE_FILE
 
 
-def _state_lock():
+def _state_lock() -> Any:
     return export_state_lock(_STATE_FILE)
 
 
-def _load_state_from_disk() -> dict:
+def _load_state_from_disk() -> ExportStateDict:
     return load_export_state_from_disk(_STATE_FILE)
 
 
-def _atomic_write_state(state: dict) -> None:
+def _atomic_write_state(state: ExportStateDict) -> None:
     atomic_write_export_state(state, _STATE_FILE)
 
 
-def _read_state() -> dict:
+def _read_state() -> ExportStateDict:
     with _state_lock():
         return _load_state_from_disk()
 
 
-def _write_state(sessions_map: dict, count: int) -> None:
+def _write_state(sessions_map: dict[str, float], count: int) -> None:
     """Persist merge of *sessions_map* and update last-export metadata (*count* = this run only)."""
     with _state_lock():
         state = _load_state_from_disk()
@@ -61,10 +65,10 @@ def _write_state(sessions_map: dict, count: int) -> None:
 
 
 @export_bp.route("/api/export/state")
-def get_export_state():
+def get_export_state() -> FlaskReturn:
     state = _read_state()
     n = state.get("exportedCount", 0)
-    return jsonify(
+    return json_ok(
         {
             "last_export_time": state.get("lastExportTime"),
             # Sessions exported in the last completed bulk export (not a lifetime total).
@@ -75,16 +79,16 @@ def get_export_state():
 
 
 @export_bp.route("/api/export", methods=["POST"])
-def bulk_export():
+def bulk_export() -> FlaskReturn:
     body = request.get_json(silent=True)
     if body is None:
         body = {}
     if not isinstance(body, dict):
-        return jsonify({"error": "Invalid request body"}), 400
+        return json_ok({"error": "Invalid request body"}), 400
 
     since = body.get("since", "all")
     if since not in ("all", "last", "incremental"):
-        return jsonify({"error": "Invalid since mode", "since": since}), 400
+        return json_ok({"error": "Invalid since mode", "since": since}), 400
 
     base = (
         current_app.config.get("CLAUDE_PROJECTS_DIR")
@@ -94,14 +98,14 @@ def bulk_export():
     rules = current_app.config.get("EXCLUSION_RULES") or []
 
     state = _read_state()
-    last_export_sessions: dict = (
+    last_export_sessions: dict[str, float] = (
         state.get("sessions", {}) if since == "incremental" else {}
     )
 
     buf = io.BytesIO()
     count = 0
-    manifest = []
-    new_sessions_map: dict = {}
+    manifest: list[dict[str, Any]] = []
+    new_sessions_map: dict[str, float] = {}
     latest_day = None
 
     with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
@@ -227,7 +231,7 @@ def bulk_export():
 
     if count == 0:
         return (
-            jsonify(
+            json_ok(
                 {
                     "error": "Nothing to export",
                     "since": since,
@@ -251,12 +255,12 @@ def bulk_export():
         buf,
         mimetype="application/zip",
         as_attachment=True,
-        download_name=f"claude-code-export{suffix}-{date_tag}.zip",
+        download_name=f"claude-code-export{suffix}-{date_tag}.zip",  # type: ignore[call-arg]
     )
 
 
 @export_bp.route("/api/export/session/<path:project_name>/<session_id>")
-def export_session(project_name, session_id):
+def export_session(project_name: str, session_id: str) -> FlaskReturn:
     import os
     from utils.session_path import safe_join
 
@@ -267,36 +271,42 @@ def export_session(project_name, session_id):
     try:
         filepath = safe_join(base, project_name, f"{session_id}.jsonl")
     except ValueError:
-        return jsonify({"error": "Invalid path"}), 400
+        return json_ok({"error": "Invalid path"}), 400
 
     if not os.path.isfile(filepath):
-        return jsonify({"error": "Session not found"}), 404
+        return json_ok({"error": "Session not found"}), 404
 
     fmt = request.args.get("format", "md")
-    session = parse_session(filepath)
-    rules = current_app.config.get("EXCLUSION_RULES") or []
-    if is_session_excluded(rules, session, project_name):
-        return jsonify({"error": "Session not found"}), 404
-    stats = compute_stats(session)
-    title_slug = slugify(session["title"], default="session")
-
-    if fmt == "json":
-        content = session_to_json(session, stats)
-        buf = io.BytesIO(content.encode("utf-8"))
+    try:
+        session = parse_session(filepath)
+        rules = current_app.config.get("EXCLUSION_RULES") or []
+        if is_session_excluded(rules, session, project_name):
+            return json_ok({"error": "Session not found"}), 404
+        stats = compute_stats(session)
+        title_slug = slugify(session["title"], default="session")
+
+        if fmt == "json":
+            content = session_to_json(session, stats)
+            buf = io.BytesIO(content.encode("utf-8"))
+            buf.seek(0)
+            return send_file(
+                buf,
+                mimetype="application/json",
+                as_attachment=True,
+                download_name=f"{title_slug}.json",  # type: ignore[call-arg]
+            )
+
+        md = session_to_markdown(session, stats)
+        buf = io.BytesIO(md.encode("utf-8"))
         buf.seek(0)
         return send_file(
             buf,
-            mimetype="application/json",
+            mimetype="text/markdown",
             as_attachment=True,
-            download_name=f"{title_slug}.json",
+            download_name=f"{title_slug}.md",  # type: ignore[call-arg]
         )
-
-    md = session_to_markdown(session, stats)
-    buf = io.BytesIO(md.encode("utf-8"))
-    buf.seek(0)
-    return send_file(
-        buf,
-        mimetype="text/markdown",
-        as_attachment=True,
-        download_name=f"{title_slug}.md",
-    )
+    except Exception:
+        current_app.logger.exception(
+            "Failed to export session %s/%s", project_name, session_id
+        )
+        return json_ok({"error": "Internal server error exporting session"}), 500

api/projects.py

@@ -1,15 +1,19 @@
 """Project listing endpoints."""
 
+from typing import cast
+
 from flask import Blueprint, current_app, jsonify
 
+from api._flask_types import FlaskReturn, json_ok
+from models.project import ProjectSessionRowDict
 from utils.session_path import get_claude_projects_dir, list_projects, list_sessions, safe_join
 from utils.exclusion_rules import is_session_excluded
 
 projects_bp = Blueprint("projects", __name__)
 
 
 @projects_bp.route("/api/projects")
-def get_projects():
+def get_projects() -> FlaskReturn:
     base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
     projects = list_projects(base)
 
@@ -36,21 +40,21 @@ def get_projects():
         if latest_ts:
             project["last_modified"] = latest_ts
 
-    return jsonify(projects)
+    return json_ok(projects)
 
 
 @projects_bp.route("/api/projects/<path:project_name>/sessions")
-def get_project_sessions(project_name):
+def get_project_sessions(project_name: str) -> FlaskReturn:
     base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
     try:
         project_dir = safe_join(base, project_name)
     except ValueError:
-        return jsonify([]), 400
+        return json_ok([]), 400
     sessions = list_sessions(project_dir)
     # Add summary preview for each session
     from utils.jsonl_parser import parse_session
     rules = current_app.config.get("EXCLUSION_RULES") or []
-    result = []
+    result: list[ProjectSessionRowDict] = []
     for s in sessions:
         try:
             parsed = parse_session(s["path"])
@@ -60,20 +64,25 @@ def get_project_sessions(project_name):
                 continue
             if is_session_excluded(rules, parsed, project_name):
                 continue
-            result.append({
+            models = meta.get("models_used", [])
+            result.append(cast(ProjectSessionRowDict, {
                 **s,
                 "title": parsed["title"],
-                "models": meta["models_used"],
+                "models": sorted(models) if isinstance(models, set) else list(models),
                 "tokens": meta["total_input_tokens"] + meta["total_output_tokens"],
                 "tool_calls": meta["total_tool_calls"],
                 "first_timestamp": meta["first_timestamp"],
                 "last_timestamp": meta["last_timestamp"],
-            })
+            }))
         except Exception:
             # Full detail (class, message, traceback) to the server log via
             # logger.exception. The per-session card carries only `error: True`
             # — the class-name+message string was a leak (issue #25). The
             # operator looks at the server log for triage.
             current_app.logger.exception("Failed to parse session %s", s["id"])
-            result.append({**s, "title": "Error parsing session", "error": True})
-    return jsonify(result)
+            result.append(cast(ProjectSessionRowDict, {
+                **s,
+                "title": "Error parsing session",
+                "error": True,
+            }))
+    return json_ok(result)

api/search.py

@@ -4,25 +4,45 @@
 
 from flask import Blueprint, current_app, jsonify, request
 
+from api._flask_types import FlaskReturn, json_ok
+from models.search import SearchHitDict
 from utils.session_path import get_claude_projects_dir, list_projects, list_sessions
 from utils.jsonl_parser import parse_session
 from utils.exclusion_rules import is_session_excluded
 
 search_bp = Blueprint("search", __name__)
 
+_DEFAULT_LIMIT = 50
+
+
+def _parse_limit(raw: str | None, default: int = _DEFAULT_LIMIT) -> int:
+    """Parse a positive integer limit from a query string value."""
+    if raw is None or raw.strip() == "":
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        raise ValueError("Invalid limit: must be a positive integer") from None
+    if value < 1:
+        raise ValueError("Invalid limit: must be a positive integer")
+    return value
+
 
 @search_bp.route("/api/search")
-def search():
+def search() -> FlaskReturn:
     query = request.args.get("q", "").strip().lower()
     if not query:
-        return jsonify([])
+        return json_ok([])
 
-    max_results = int(request.args.get("limit", 50))
+    try:
+        max_results = _parse_limit(request.args.get("limit"))
+    except ValueError as e:
+        return json_ok({"error": str(e)}), 400
     base = current_app.config.get("CLAUDE_PROJECTS_DIR") or get_claude_projects_dir()
     projects = list_projects(base)
 
     rules = current_app.config.get("EXCLUSION_RULES") or []
-    results = []
+    results: list[SearchHitDict] = []
     for project in projects:
         sessions = list_sessions(project["path"])
         for sess_info in sessions:
@@ -56,4 +76,4 @@ def search():
                     if len(results) >= max_results:
                         break
 
-    return jsonify(results)
+    return json_ok(results)