fix: second-round P1 and risk-item fixes from code review

Status unification (3.4-34):
- New src/status_constants.py with POINT_OK/POINT_FAILED/TASK_* constants
- robust_motorcad.py and task_executor.py: lowercase ok/failed -> uppercase OK/FAILED

Session management (3.4-25):
- task_manager.py: 7x next(get_db()) -> with SessionLocal() as db:

API security (E4):
- main.py: optional API key middleware (AFM_API_KEY env var)
- ai.py ChatRequest: max_tokens Field(le=8192) upper bound

Solver stability (3.1-1):
- solver_core.py: point_timeout=600s via daemon thread
- solver_core.py: max_consecutive_failures=3 auto reconnect

Data safety:
- plans.py: plan_id adds 6-char random hex suffix to avoid collisions
- projects.py + experience.py: list endpoints limit Query(le=200), skip Query(ge=0)
- experience.py: create_experience dedup by source_plan_id + params_json

Verification:
- 75 Python files pass py_compile
- All .py files pure ASCII compliant
- Key module imports verified

Updated docs/CODE_REVIEW_RESPONSE.md with second-round fix record

README.md

@@ -117,6 +117,17 @@
 
 详见 [docs/CODE_REVIEW_RESPONSE.md](docs/CODE_REVIEW_RESPONSE.md)。
 
+**第二轮修复（P1 + 风险项）**：
+
+| 类别 | 修复项 |
+|---|---|
+| 状态统一 | 3.4-34 新建 status_constants.py，全仓库仿真点状态统一为 "OK"/"FAILED" |
+| Session | 3.4-25 task_manager.py 7处 next(get_db()) 改为 with SessionLocal() |
+| API安全 | E4 可选API Key中间件 + max_tokens上限(8192) |
+| 求解稳定 | 3.1-1 solver_core 单点超时(600s) + 连续失败3次自动重连 |
+| 数据安全 | plan_id加随机后缀防碰撞；分页limit上限200；经验库去重 |
+| 验证 | 75个Python文件语法通过，纯ASCII合规，关键模块导入正常 |
+
 ---
 
 ## 快速开始

docs/CODE_REVIEW_RESPONSE.md

@@ -141,3 +141,36 @@
 ---
 
 *应审总结完成，立即开始 P0 修复。*
+
+---
+
+## 六、第二轮修复（P1 + 风险项）
+
+第一轮 P0 修复完成后，继续修复 P1 项和高优先级风险项：
+
+| # | 问题 | 修复内容 |
+|---|---|---|
+| 1 | 3.4-34 状态枚举全仓库不统一 | 新建 `src/status_constants.py` 定义 POINT_OK/POINT_FAILED/TASK_* 常量；robust_motorcad.py 和 task_executor.py 中小写 "ok"/"failed" 统一改为大写 "OK"/"FAILED" |
+| 2 | 3.4-25 Session 管理不统一 | task_manager.py 中 7 处 `next(get_db())` 改为 `with SessionLocal() as db:`，确保 Session 自动关闭 |
+| 3 | E4 API 无认证 + max_tokens 无上限 | main.py 添加可选 API Key 中间件（AFM_API_KEY 环境变量，未设置时不启用）；ai.py ChatRequest.max_tokens 加 Field(le=8192) 上限 |
+| 4 | 3.1-1 求解无超时 + 无连续失败重连 | solver_core.py MotorCADSolver 添加 point_timeout=600s 和 max_consecutive_failures=3 参数；run_single 用 daemon thread 实现超时，连续失败达阈值后自动 disconnect+connect 重连 |
+| 5 | F1 断点续扫 | scan_engine.py 已有完整的 _load_completed_indices + CSV 追加模式实现，确认功能正常，无需额外修改 |
+| 6 | plan_id 碰撞风险 | plans.py _generate_plan_id 从时间戳改为时间戳+6位随机hex后缀（uuid），避免同秒碰撞 |
+| 7 | 分页参数无上限 | projects.py 和 experience.py 的 list 接口 limit 参数加 Query(le=200) 上限，skip 加 Query(ge=0) |
+| 8 | 经验库无去重 | experience.py create_experience 添加 source_plan_id + params_json 去重检查，重复时返回已存在记录 |
+
+### 验证结果
+
+- 75 个 Python 文件全部通过 py_compile 语法检查
+- 所有 .py 文件纯 ASCII 合规（临时脚本除外）
+- 关键模块导入正常（status_constants / solver_core / metrics_constants）
+- 状态枚举统一：全仓库仿真点状态只用 "OK"/"FAILED"
+
+### 仍为已知限制（P2，后续迭代）
+
+- A6: 自适应闭环本地执行桥（需在 task_executor 中集成 feasibility_search，工作量约 1-2 天）
+- N+1 查询优化（analytics 等服务的批量查询）
+- 测试体系重建（离线单元测试 + CI）
+- 其余 🟡 风险项中的低优先级问题
+
+*第二轮修复完成，提交推送。*

scripts/robust_motorcad.py

@@ -608,7 +608,7 @@ class RobustMotorCADSolver:
                         if not compatible:
                             self._log(f"WARNING: {msg}")
                             result["error"] = msg
-                            result["status"] = "failed"
+                            result["status"] = "FAILED"
                             # B5 fix: break instead of return so the point
                             # is appended to results and written to disk
                             break
@@ -646,7 +646,7 @@ class RobustMotorCADSolver:
 
                     metrics = self._parse_export(raw_file)
                     result["metrics"] = metrics
-                    result["status"] = "ok"
+                    result["status"] = "OK"
                     break
 
                 except MotorCADError as e:
@@ -657,7 +657,7 @@ class RobustMotorCADSolver:
                         time.sleep(2)
                         self._reconnect_if_needed()
                     else:
-                        result["status"] = "failed"
+                        result["status"] = "FAILED"
                 except Exception as e:
                     result["error"] = f"{type(e).__name__}: {e}"
                     self._log(f"Point {point_index} attempt {attempt+1} failed: {e}")
@@ -666,7 +666,7 @@ class RobustMotorCADSolver:
                         time.sleep(2)
                         self._reconnect_if_needed()
                     else:
-                        result["status"] = "failed"
+                        result["status"] = "FAILED"
 
         finally:
             # Restore popup state (CRITICAL: must happen even on error)
@@ -838,8 +838,8 @@ class RobustMotorCADSolver:
 
     def get_summary(self) -> Dict[str, Any]:
         """Get run summary."""
-        ok = [r for r in self._all_results if r["status"] == "ok"]
-        failed = [r for r in self._all_results if r["status"] == "failed"]
+        ok = [r for r in self._all_results if r["status"] == "OK"]
+        failed = [r for r in self._all_results if r["status"] == "FAILED"]
         return {
             "total": len(self._all_results),
             "ok": len(ok),

scripts/task_executor.py

@@ -211,7 +211,7 @@ class TaskExecutor:
                 results.append({
                     "point_index": idx,
                     "params": params,
-                    "status": "failed",
+                    "status": "FAILED",
                     "error": str(e),
                 })
                 if self.on_error:
@@ -222,9 +222,9 @@ class TaskExecutor:
         # Status reflects actual outcome: completed/cancelled/failed
         if self._stop_event.is_set():
             status = "cancelled"
-        elif any(r.get("status") == "failed" for r in results):
+        elif any(r.get("status") == "FAILED" for r in results):
             status = "completed_with_errors" if any(
-                r.get("status") == "ok" for r in results
+                r.get("status") == "OK" for r in results
             ) else "failed"
         else:
             status = "completed"
@@ -262,13 +262,13 @@ class TaskExecutor:
             "efficiency_pct": round(88 + rng.gauss(0, 2), 2),
             "total_losses_w": round(50 + rng.gauss(0, 10), 2),
             "winding_temp_c": round(90 + rng.gauss(0, 10), 1),
-            "status": "ok",
+            "status": "OK",
             "source": "mock",
         }
 
     def _compute_metrics(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
         """Compute aggregated metrics from results."""
-        ok_results = [r for r in results if r.get("status") == "ok"]
+        ok_results = [r for r in results if r.get("status") == "OK"]
         if not ok_results:
             return {
                 "total_points": len(results),

src/solver_core.py

@@ -314,10 +314,14 @@ class MotorCADSolver:
     log events.
     """
 
-    def __init__(self, log_cb=None):
+    def __init__(self, log_cb=None, point_timeout: float = 600.0, max_consecutive_failures: int = 3):
         self.log_cb = log_cb
         self.mc = None
         self.model_path = None
+        # 3.1-1 fix: per-point timeout and consecutive-failure reconnect
+        self.point_timeout = point_timeout
+        self.max_consecutive_failures = max_consecutive_failures
+        self._consecutive_failures = 0
 
     def _log(self, text: str) -> None:
         if self.log_cb:
@@ -450,9 +454,11 @@ class MotorCADSolver:
         output_dir: str | Path = "output",
         tag: str = "",
     ) -> dict:
-        """Run a complete single-point simulation.
+        """Run a complete single-point simulation with timeout and reconnect.
 
         Steps: load model -> write params -> magnetic calc -> export+extract.
+        Includes per-point timeout (default 600s) and automatic reconnect
+        after consecutive failures (default 3).
 
         Args:
             model_path: path to .mot baseline model.
@@ -463,6 +469,8 @@ class MotorCADSolver:
         Returns:
             dict with metrics, status, error, raw_path, solve_time_s.
         """
+        import threading
+
         started = time.time()
         result = {
             "metrics": {},
@@ -472,18 +480,60 @@ class MotorCADSolver:
             "solve_time_s": 0,
             "params": params or {},
         }
-        try:
-            self.load_model(model_path)
-            if params:
-                self.write_parameters(params)
-            self.run_magnetic()
-            ext = self.export_and_extract(output_dir, tag)
-            result.update(ext)
-        except Exception as exc:
+
+        # 3.1-1 fix: consecutive failure reconnect
+        if self._consecutive_failures >= self.max_consecutive_failures:
+            self._log(f"Consecutive failures={self._consecutive_failures}, reconnecting Motor-CAD...")
+            try:
+                self.disconnect()
+            except Exception:
+                pass
+            try:
+                self.connect()
+                self._consecutive_failures = 0
+                self._log("Motor-CAD reconnected successfully")
+            except Exception as exc:
+                result["error"] = f"Reconnect failed: {exc}"
+                result["solve_time_s"] = round(time.time() - started, 1)
+                return result
+
+        # Container for thread result/exception
+        container = {"ext": None, "exc": None}
+
+        def _run_inner():
+            try:
+                self.load_model(model_path)
+                if params:
+                    self.write_parameters(params)
+                self.run_magnetic()
+                container["ext"] = self.export_and_extract(output_dir, tag)
+            except Exception as e:
+                container["exc"] = e
+
+        # 3.1-1 fix: per-point timeout via daemon thread
+        worker = threading.Thread(target=_run_inner, daemon=True)
+        worker.start()
+        worker.join(timeout=self.point_timeout)
+
+        if worker.is_alive():
+            result["status"] = "FAILED"
+            result["error"] = f"Point timed out after {self.point_timeout}s"
+            self._log(result["error"])
+            self._consecutive_failures += 1
+        elif container["exc"] is not None:
+            exc = container["exc"]
             result["status"] = "FAILED"
             result["error"] = f"{type(exc).__name__}: {exc}"
             self._log(result["error"])
             self._log(traceback.format_exc())
+            self._consecutive_failures += 1
+        else:
+            ext = container["ext"]
+            result.update(ext)
+            if ext["status"] == "OK":
+                self._consecutive_failures = 0
+            else:
+                self._consecutive_failures += 1
 
         result["solve_time_s"] = round(time.time() - started, 1)
         return result

src/status_constants.py

@@ -0,0 +1,38 @@
+"""Unified status constants for the simulation system.
+
+Point status (per simulation point):
+  POINT_OK      - point solved successfully
+  POINT_FAILED  - point failed (error recorded)
+
+Task status (per batch/task):
+  TASK_PENDING   - queued, not started
+  TASK_RUNNING   - currently executing
+  TASK_COMPLETED - finished with all points OK
+  TASK_FAILED    - finished with failures
+  TASK_CANCELLED - user cancelled
+"""
+
+# Point status (uppercase, canonical)
+POINT_OK = "OK"
+POINT_FAILED = "FAILED"
+
+# Task status
+TASK_PENDING = "pending"
+TASK_RUNNING = "running"
+TASK_COMPLETED = "completed"
+TASK_FAILED = "failed"
+TASK_CANCELLED = "cancelled"
+
+# Valid status sets
+VALID_POINT_STATUSES = {POINT_OK, POINT_FAILED}
+VALID_TASK_STATUSES = {TASK_PENDING, TASK_RUNNING, TASK_COMPLETED, TASK_FAILED, TASK_CANCELLED}
+
+
+def normalize_point_status(status: str) -> str:
+    """Normalize a point status string to canonical uppercase form."""
+    if not status:
+        return POINT_FAILED
+    s = status.strip().upper()
+    if s in ("OK", "SUCCESS", "PASS", "PASSED"):
+        return POINT_OK
+    return POINT_FAILED

web/backend/app/main.py

@@ -1,5 +1,6 @@
 """FastAPI application entry point."""
-from fastapi import FastAPI
+import os
+from fastapi import FastAPI, Request, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 
 from .config import APP_NAME, APP_VERSION, APP_DESCRIPTION, CORS_ORIGINS
@@ -21,6 +22,25 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+# E4 fix: optional API key authentication middleware.
+# Enabled only when AFM_API_KEY is set in environment.
+_API_KEY = os.environ.get("AFM_API_KEY", "").strip()
+_PUBLIC_PATHS = ("/api/health", "/docs", "/openapi.json", "/redoc", "/")
+
+
+@app.middleware("http")
+async def api_key_auth(request: Request, call_next):
+    """Optional API key authentication (E4 fix)."""
+    if not _API_KEY:
+        return await call_next(request)
+    path = request.url.path
+    if any(path.startswith(p) for p in _PUBLIC_PATHS):
+        return await call_next(request)
+    provided = request.headers.get("X-API-Key", "")
+    if provided != _API_KEY:
+        raise HTTPException(status_code=401, detail="Invalid or missing API key")
+    return await call_next(request)
+
 # Include routers
 app.include_router(projects.router)
 app.include_router(plans.router)

web/backend/app/metrics_constants.py

@@ -1,4 +1,4 @@
-"""Metric key definitions — single source of truth for the Web backend.
+"""Metric key definitions - single source of truth for the Web backend.
 
 This MUST stay in sync with:
 - src/solver_core.py METRIC_DEFINITIONS (local solver)

web/backend/app/routers/experience.py

@@ -66,8 +66,8 @@ def _generate_conclusion(params: dict, metrics: dict) -> str:
 def list_experience(
     topology: str | None = None,
     tag: str | None = None,
-    skip: int = 0,
-    limit: int = 50,
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=200),
     db: Session = Depends(get_db),
 ):
     """List experience cases with filters."""
@@ -86,16 +86,27 @@ def create_experience(
     data: dict,
     db: Session = Depends(get_db),
 ):
-    """Create an experience case from a dict."""
+    """Create an experience case from a dict. Dedup by source_plan_id + params hash."""
+    params_json = json.dumps(data.get("params", {}), ensure_ascii=False, sort_keys=True)
+    source_plan_id = data.get("source_plan_id", "")
+    # Dedup check: if same plan_id and params already exist, return existing
+    if source_plan_id:
+        existing = db.query(ExperienceCase).filter(
+            ExperienceCase.source_plan_id == source_plan_id,
+            ExperienceCase.params_json == params_json,
+        ).first()
+        if existing:
+            return _case_to_dict(existing)
+
     case = ExperienceCase(
-        source_plan_id=data.get("source_plan_id", ""),
+        source_plan_id=source_plan_id,
         topology=data.get("topology", "SSSR"),
         model_path=data.get("model_path", ""),
         conclusion=data.get("conclusion", ""),
         tags=",".join(data.get("tags", [])),
         rating=data.get("rating", 0),
     )
-    case.params_json = json.dumps(data.get("params", {}), ensure_ascii=False)
+    case.params_json = params_json
     case.metrics_json = json.dumps(data.get("metrics", {}), ensure_ascii=False)
     db.add(case)
     db.commit()

web/backend/app/routers/plans.py

@@ -1,5 +1,6 @@
 """Simulation plan API router (CRUD + download + upload results)."""
 import json
+import uuid
 from datetime import datetime
 from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
 from sqlalchemy.orm import Session
@@ -18,7 +19,8 @@ router = APIRouter(prefix="/api/plans", tags=["plans"])
 
 
 def _generate_plan_id() -> str:
-    return f"SP-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
+    """Generate a unique plan ID with timestamp + random suffix to avoid collisions."""
+    return f"SP-{datetime.now().strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}"
 
 
 def _plan_to_response(db: Session, plan: SimulationPlan) -> PlanResponse:

web/backend/app/routers/projects.py

@@ -1,5 +1,5 @@
 """Project API router."""
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 
 from ..database import get_db
@@ -27,8 +27,8 @@ def _project_to_response(db: Session, project: Project) -> ProjectResponse:
 
 @router.get("", response_model=ProjectListResponse)
 def list_projects(
-    skip: int = 0,
-    limit: int = 50,
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=200),
     topology: str | None = None,
     db: Session = Depends(get_db),
 ):

web/backend/app/schemas/ai.py

@@ -14,7 +14,7 @@ class ChatRequest(BaseModel):
     messages: List[ChatMessage]
     model: Optional[str] = None
     temperature: Optional[float] = None
-    max_tokens: Optional[int] = None
+    max_tokens: Optional[int] = Field(default=None, ge=1, le=8192, description="Max output tokens, capped at 8192")
     system_prompt: Optional[str] = None
 
 

web/backend/app/services/task_manager.py

@@ -10,7 +10,7 @@ from datetime import datetime
 from typing import Dict, List, Optional, Any
 from pathlib import Path
 
-from ..database import get_db, Task
+from ..database import SessionLocal, Task
 
 
 class TaskManager:
@@ -76,23 +76,23 @@ class TaskManager:
             json.dump(task_payload, f, ensure_ascii=False, indent=2)
 
         # Save to database
-        db = next(get_db())
-        db_task = Task(
-            task_id=task_uuid,
-            task_name=task_name,
-            plan_id=plan_id,
-            status="pending",
-            priority=priority,
-            total_points=len(parameters),
-            completed_points=0,
-            task_dir=task_dir,
-            task_file=task_file,
-            created_by=created_by,
-            created_at=datetime.now(),
-        )
-        db.add(db_task)
-        db.commit()
-        db.refresh(db_task)
+        with SessionLocal() as db:
+            db_task = Task(
+                task_id=task_uuid,
+                task_name=task_name,
+                plan_id=plan_id,
+                status="pending",
+                priority=priority,
+                total_points=len(parameters),
+                completed_points=0,
+                task_dir=task_dir,
+                task_file=task_file,
+                created_by=created_by,
+                created_at=datetime.now(),
+            )
+            db.add(db_task)
+            db.commit()
+            db.refresh(db_task)
 
         return self._task_to_dict(db_task)
 
@@ -105,17 +105,17 @@ class TaskManager:
         Returns:
             Updated task dict
         """
-        db = next(get_db())
-        task = db.query(Task).filter(Task.task_id == task_id).first()
-        if not task:
-            raise ValueError(f"Task {task_id} not found")
-        if task.status != "pending":
-            raise ValueError(f"Task {task_id} is not pending (status: {task.status})")
-
-        task.status = "dispatched"
-        task.dispatched_at = datetime.now()
-        db.commit()
-        db.refresh(task)
+        with SessionLocal() as db:
+            task = db.query(Task).filter(Task.task_id == task_id).first()
+            if not task:
+                raise ValueError(f"Task {task_id} not found")
+            if task.status != "pending":
+                raise ValueError(f"Task {task_id} is not pending (status: {task.status})")
+
+            task.status = "dispatched"
+            task.dispatched_at = datetime.now()
+            db.commit()
+            db.refresh(task)
         return self._task_to_dict(task)
 
     def update_progress(
@@ -138,32 +138,32 @@ class TaskManager:
         Returns:
             Updated task dict
         """
-        db = next(get_db())
-        task = db.query(Task).filter(Task.task_id == task_id).first()
-        if not task:
-            raise ValueError(f"Task {task_id} not found")
-
-        if task.status in ("dispatched", "running"):
-            task.status = "running"
-            task.started_at = task.started_at or datetime.now()
-
-        task.completed_points = current_point
-        if total_points:
-            task.total_points = total_points
-
-        # Update progress metadata
-        progress_data = {
-            "current_point": current_point,
-            "current_params": current_params,
-            "elapsed_time": elapsed_time,
-            "updated_at": datetime.now().isoformat(),
-        }
-        existing_progress = json.loads(task.progress_data or "{}")
-        existing_progress.update(progress_data)
-        task.progress_data = json.dumps(existing_progress, ensure_ascii=False)
-
-        db.commit()
-        db.refresh(task)
+        with SessionLocal() as db:
+            task = db.query(Task).filter(Task.task_id == task_id).first()
+            if not task:
+                raise ValueError(f"Task {task_id} not found")
+
+            if task.status in ("dispatched", "running"):
+                task.status = "running"
+                task.started_at = task.started_at or datetime.now()
+
+            task.completed_points = current_point
+            if total_points:
+                task.total_points = total_points
+
+            # Update progress metadata
+            progress_data = {
+                "current_point": current_point,
+                "current_params": current_params,
+                "elapsed_time": elapsed_time,
+                "updated_at": datetime.now().isoformat(),
+            }
+            existing_progress = json.loads(task.progress_data or "{}")
+            existing_progress.update(progress_data)
+            task.progress_data = json.dumps(existing_progress, ensure_ascii=False)
+
+            db.commit()
+            db.refresh(task)
         return self._task_to_dict(task)
 
     def report_results(
@@ -188,36 +188,36 @@ class TaskManager:
         Returns:
             Updated task dict
         """
-        db = next(get_db())
-        task = db.query(Task).filter(Task.task_id == task_id).first()
-        if not task:
-            raise ValueError(f"Task {task_id} not found")
-
-        task.status = status
-        task.completed_at = datetime.now()
-        task.completed_points = len(results)
-        if duration:
-            task.duration = duration
-
-        # Save results to file
-        results_file = os.path.join(task.task_dir, "results.json")
-        with open(results_file, "w", encoding="utf-8") as f:
-            json.dump({"results": results, "metrics": metrics, "logs": logs}, f, ensure_ascii=False, indent=2)
-
-        task.results_file = results_file
-        if metrics:
-            task.result_metrics = json.dumps(metrics, ensure_ascii=False)
-
-        db.commit()
-        db.refresh(task)
+        with SessionLocal() as db:
+            task = db.query(Task).filter(Task.task_id == task_id).first()
+            if not task:
+                raise ValueError(f"Task {task_id} not found")
+
+            task.status = status
+            task.completed_at = datetime.now()
+            task.completed_points = len(results)
+            if duration:
+                task.duration = duration
+
+            # Save results to file
+            results_file = os.path.join(task.task_dir, "results.json")
+            with open(results_file, "w", encoding="utf-8") as f:
+                json.dump({"results": results, "metrics": metrics, "logs": logs}, f, ensure_ascii=False, indent=2)
+
+            task.results_file = results_file
+            if metrics:
+                task.result_metrics = json.dumps(metrics, ensure_ascii=False)
+
+            db.commit()
+            db.refresh(task)
         return self._task_to_dict(task)
 
     def get_task(self, task_id: str) -> Optional[Dict[str, Any]]:
         """Get task by ID."""
-        db = next(get_db())
-        task = db.query(Task).filter(Task.task_id == task_id).first()
-        if not task:
-            return None
+        with SessionLocal() as db:
+            task = db.query(Task).filter(Task.task_id == task_id).first()
+            if not task:
+                return None
         return self._task_to_dict(task)
 
     def list_tasks(
@@ -228,15 +228,15 @@ class TaskManager:
         offset: int = 0,
     ) -> Dict[str, Any]:
         """List tasks with filters."""
-        db = next(get_db())
-        query = db.query(Task)
-        if status:
-            query = query.filter(Task.status == status)
-        if plan_id:
-            query = query.filter(Task.plan_id == plan_id)
+        with SessionLocal() as db:
+            query = db.query(Task)
+            if status:
+                query = query.filter(Task.status == status)
+            if plan_id:
+                query = query.filter(Task.plan_id == plan_id)
 
-        total = query.count()
-        tasks = query.order_by(Task.created_at.desc()).offset(offset).limit(limit).all()
+            total = query.count()
+            tasks = query.order_by(Task.created_at.desc()).offset(offset).limit(limit).all()
 
         return {
             "total": total,
@@ -247,17 +247,17 @@ class TaskManager:
 
     def cancel_task(self, task_id: str) -> Dict[str, Any]:
         """Cancel a pending or running task."""
-        db = next(get_db())
-        task = db.query(Task).filter(Task.task_id == task_id).first()
-        if not task:
-            raise ValueError(f"Task {task_id} not found")
-        if task.status in ("completed", "failed", "cancelled"):
-            raise ValueError(f"Task {task_id} already finished (status: {task.status})")
-
-        task.status = "cancelled"
-        task.completed_at = datetime.now()
-        db.commit()
-        db.refresh(task)
+        with SessionLocal() as db:
+            task = db.query(Task).filter(Task.task_id == task_id).first()
+            if not task:
+                raise ValueError(f"Task {task_id} not found")
+            if task.status in ("completed", "failed", "cancelled"):
+                raise ValueError(f"Task {task_id} already finished (status: {task.status})")
+
+            task.status = "cancelled"
+            task.completed_at = datetime.now()
+            db.commit()
+            db.refresh(task)
         return self._task_to_dict(task)
 
     def get_task_results(self, task_id: str) -> Optional[Dict[str, Any]]: