Refactor: session-per-interaction — DB sessions never survive network calls

server/app/api/misc.py

@@ -165,7 +165,7 @@ def telegram_bot_info(user: User = Depends(current_user), db: Session = Depends(
 
 @router.post("/settings/test-notify")
 def test_notify(user: User = Depends(require_admin), db: Session = Depends(get_db)):
-    results = notify_all(db, "JarvisQwen test push", "If you can read this, push notifications are configured correctly ✅")
+    results = notify_all("JarvisQwen test push", "If you can read this, push notifications are configured correctly ✅")
     if not results:
         raise HTTPException(400, "No push channel enabled")
     return {"results": results}
@@ -258,7 +258,7 @@ def library_qa(body: QaIn, user: User = Depends(current_user), db: Session = Dep
         f"Cite sources as [number]; do not invent anything not present in the evidence.\n\n"
         f"Question: {body.question}\n\nEvidence:\n{evidence}"
     )
-    result, escalated = cascade.complete_cascade(db, prompt, step="library_qa", max_tokens=1200)
+    result, escalated = cascade.complete_cascade(prompt, step="library_qa", max_tokens=1200)
     return {"answer": result.text, "escalated": escalated,
             "cited": [{"id": p.id, "title": p.title, "url": p.url} for _, p, _ in top]}
 

server/app/connectors/notify.py

@@ -1,24 +1,28 @@
 """外部推送：Telegram / 邮件（SMTP）。简报生成完成、预算熔断时触发。
 
 失败静默降级（推送不是关键路径，不应让任务失败）。
+
+会话纪律：本模块自管短会话——先短会话读配置并关闭，再做网络发送。
+调用方不需要（也不应该）把自己的会话传进来。
 """
 from __future__ import annotations
 
 import smtplib
 from email.mime.text import MIMEText
 
 import httpx
-from sqlalchemy.orm import Session
 
 from ..core.settings_store import get_setting
+from ..db import session
 
 
-def push_telegram(db: Session, text: str) -> tuple[bool, str]:
-    token = str(get_setting(db, "telegram_bot_token") or "")
-    chat_id = str(get_setting(db, "telegram_chat_id") or "")
+def push_telegram(text: str) -> tuple[bool, str]:
+    with session() as db:  # ① 短会话读配置
+        token = str(get_setting(db, "telegram_bot_token") or "")
+        chat_id = str(get_setting(db, "telegram_chat_id") or "")
     if not token or not chat_id:
         return False, "Telegram bot token / chat ID not configured"
-    try:
+    try:  # ② 网络发送（无会话）
         resp = httpx.post(
             f"https://api.telegram.org/bot{token}/sendMessage",
             json={"chat_id": chat_id, "text": text[:4000], "parse_mode": "Markdown"},
@@ -31,16 +35,17 @@ def push_telegram(db: Session, text: str) -> tuple[bool, str]:
         return False, f"Telegram push error: {e}"
 
 
-def push_email(db: Session, subject: str, body: str) -> tuple[bool, str]:
-    host = str(get_setting(db, "smtp_host") or "")
-    to_addr = str(get_setting(db, "smtp_to") or "")
+def push_email(subject: str, body: str) -> tuple[bool, str]:
+    with session() as db:  # ① 短会话读配置
+        host = str(get_setting(db, "smtp_host") or "")
+        to_addr = str(get_setting(db, "smtp_to") or "")
+        port = int(get_setting(db, "smtp_port") or 587)
+        user = str(get_setting(db, "smtp_user") or "")
+        password = str(get_setting(db, "smtp_password") or "")
+        from_addr = str(get_setting(db, "smtp_from") or user)
     if not host or not to_addr:
         return False, "SMTP server / recipient not configured"
-    port = int(get_setting(db, "smtp_port") or 587)
-    user = str(get_setting(db, "smtp_user") or "")
-    password = str(get_setting(db, "smtp_password") or "")
-    from_addr = str(get_setting(db, "smtp_from") or user)
-    try:
+    try:  # ② 网络发送（无会话）
         msg = MIMEText(body, "plain", "utf-8")
         msg["Subject"] = subject
         msg["From"] = from_addr
@@ -55,13 +60,16 @@ def push_email(db: Session, subject: str, body: str) -> tuple[bool, str]:
         return False, f"Email send error: {e}"
 
 
-def notify_all(db: Session, subject: str, text: str) -> list[str]:
+def notify_all(subject: str, text: str) -> list[str]:
     """按已启用的渠道推送，返回结果消息列表（供审计/调试）。"""
+    with session() as db:
+        tg_enabled = bool(get_setting(db, "notify_telegram_enabled"))
+        email_enabled = bool(get_setting(db, "notify_email_enabled"))
     results = []
-    if get_setting(db, "notify_telegram_enabled"):
-        ok, msg = push_telegram(db, f"*{subject}*\n\n{text}")
+    if tg_enabled:
+        ok, msg = push_telegram(f"*{subject}*\n\n{text}")
         results.append(msg)
-    if get_setting(db, "notify_email_enabled"):
-        ok, msg = push_email(db, subject, text)
+    if email_enabled:
+        ok, msg = push_email(subject, text)
         results.append(msg)
     return results

server/app/core/budget/guard.py

@@ -41,8 +41,8 @@ def check(db: Session, task: Task | None = None, upcoming_estimate: float = 0.01
         if get_setting(db, "notify_on_budget_cutoff") and _last_cutoff_notified_date != today:
             _last_cutoff_notified_date = today
             from ...connectors.notify import notify_all
-            db.commit()  # 释放写锁再做外部推送（Telegram/SMTP 网络调用期间不得持锁）
-            notify_all(db, "JarvisQwen budget cutoff", f"Today's spend reached ${spent:.2f}/${daily_limit:.2f}. Tasks suspended until tomorrow or a higher cap.")
+            db.commit()  # 释放写锁再做外部推送（notify_all 自管短会话 + 网络在会话外）
+            notify_all("JarvisQwen budget cutoff", f"Today's spend reached ${spent:.2f}/${daily_limit:.2f}. Tasks suspended until tomorrow or a higher cap.")
         raise BudgetExceeded(f"Daily budget exhausted (${spent:.2f}/${daily_limit:.2f})")
     if spent >= 0.8 * daily_limit:
         bus.publish("budget_alert", {"level": "warn", "spent": spent, "limit": daily_limit})

server/app/core/engine/engine.py

@@ -8,6 +8,11 @@
 - 步骤耗时写入 step_stats（EMA），驱动进度条与 ETA。
 - 步骤可抛 NeedApproval（高危操作 → 审批队列挂起）、引擎捕获 BudgetExceeded（→ 挂起）。
 - state["artifacts"] 是"验证伪影"：既是 Web 流水线视图的进度证据，也是重跑核验点。
+
+会话纪律：引擎不再持有贯穿任务的长会话。所有引擎簿记（状态/进度/检查点）
+各自开短会话即写即提交；步骤函数通过 ctx.session() 开短会话，且必须保证
+任何网络调用（LLM/下载/推送）都发生在会话之外——SQLite 单写者模型下，
+跨网络调用持有的写锁会堵死全站写操作（这是本项目曾全站瘫痪的根因）。
 """
 from __future__ import annotations
 
@@ -20,6 +25,7 @@
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
+from ...db import session as db_session
 from ...models import Approval, Checkpoint, StepStat, Task
 from ..budget.guard import BudgetExceeded
 from ..bus import bus
@@ -46,10 +52,13 @@ class StepDef:
 
 @dataclass
 class TaskContext:
-    db: Session
-    task: Task
+    task: Task  # 分离态快照：属性只读可用（id/type/owner_id/params_json...）；写库走 ctx.session()
     state: dict = field(default_factory=dict)
 
+    def session(self):
+        """短会话工厂：with ctx.session() as db: ...（退出即提交关闭，不得跨网络调用）。"""
+        return db_session()
+
     def artifact(self, name: str, content: str) -> None:
         """产出一个验证伪影（清单/摘要/文件列表），流水线视图可见。"""
         self.state.setdefault("artifacts", []).append(
@@ -64,10 +73,12 @@ def require_approval(self, desc: str, risk: str = "high") -> None:
         """高危操作前调用：已批准则通过，否则挂起进审批队列。"""
         approval_id = self.state.get("_approval_ids", {}).get(desc)
         if approval_id:
-            row = self.db.execute(select(Approval).where(Approval.id == approval_id)).scalar_one_or_none()
-            if row and row.status == "approved":
+            with db_session() as db:
+                row = db.execute(select(Approval).where(Approval.id == approval_id)).scalar_one_or_none()
+                status = row.status if row else ""
+            if status == "approved":
                 return
-            if row and row.status == "rejected":
+            if status == "rejected":
                 raise TaskFailed(f"Action rejected: {desc}")
         raise NeedApproval(desc, risk)
 
@@ -126,75 +137,94 @@ def latest_checkpoint(db: Session, task_id: str) -> Checkpoint | None:
     ).scalars().first()
 
 
-def run_task(db: Session, task: Task) -> None:
-    """执行任务：从最后一个检查点续跑。由调度器工作线程调用。"""
-    steps = REGISTRY.get(task.type)
-    if not steps:
-        task.status = "FAILED"
-        task.error = f"Unknown task type: {task.type}"
-        return
+def _fetch_task(db: Session, task_id: str) -> Task:
+    return db.execute(select(Task).where(Task.id == task_id)).scalar_one()
 
-    cp = latest_checkpoint(db, task.id)
-    if cp is not None:
-        state: dict = json.loads(cp.state_json)
-        start_index = cp.step_index + 1
-    else:
-        state = {"params": json.loads(task.params_json)}
-        start_index = 0
 
-    ctx = TaskContext(db=db, task=task, state=state)
-    task.status = "RUNNING"
+def run_task(task_id: str) -> None:
+    """执行任务：从最后一个检查点续跑。由调度器工作线程调用。
+
+    引擎自管会话：每次簿记开短会话即写即提交，步骤执行期间（含其中的
+    LLM/网络调用）不持有任何会话。
+    """
+    with db_session() as db:
+        task = _fetch_task(db, task_id)
+        steps = REGISTRY.get(task.type)
+        if not steps:
+            task.status = "FAILED"
+            task.error = f"Unknown task type: {task.type}"
+            return
+        cp = latest_checkpoint(db, task.id)
+        if cp is not None:
+            state: dict = json.loads(cp.state_json)
+            start_index = cp.step_index + 1
+        else:
+            state = {"params": json.loads(task.params_json)}
+            start_index = 0
+        task.status = "RUNNING"
+    # 会话已提交关闭；task 是分离态快照（expire_on_commit=False，属性可读）
+
+    ctx = TaskContext(task=task, state=state)
 
     for i in range(start_index, len(steps)):
         step = steps[i]
         state["_current_step"] = step.name
-        _update_progress(db, task, steps, i)
-        db.commit()  # 进度先落库：①不留脏数据给步内 autoflush 拿锁 ②进度条对 API 即时可见
+        with db_session() as db:  # 进度即写即提交：对 API 实时可见，且不留脏数据给步内查询 autoflush
+            _update_progress(db, _fetch_task(db, task_id), steps, i)
         t0 = time.time()
         try:
             new_state = step.fn(ctx, state)
             state = new_state if new_state is not None else state
             ctx.state = state
         except NeedApproval as e:
-            approval = Approval(task_id=task.id, action_desc=e.desc, risk_level=e.risk)
-            db.add(approval)
-            db.flush()
-            state.setdefault("_approval_ids", {})[e.desc] = approval.id
-            _save_checkpoint(db, task, i - 1, state)  # 审批通过后从本步重跑
-            task.status = "WAITING_APPROVAL"
-            bus.publish("approval_needed", {"task_id": task.id, "desc": e.desc, "approval_id": approval.id})
+            with db_session() as db:
+                approval = Approval(task_id=task_id, action_desc=e.desc, risk_level=e.risk)
+                db.add(approval)
+                db.flush()
+                state.setdefault("_approval_ids", {})[e.desc] = approval.id
+                _save_checkpoint(db, task, i - 1, state)  # 审批通过后从本步重跑
+                _fetch_task(db, task_id).status = "WAITING_APPROVAL"
+                approval_id = approval.id
+            bus.publish("approval_needed", {"task_id": task_id, "desc": e.desc, "approval_id": approval_id})
             return
         except BudgetExceeded as e:
-            _save_checkpoint(db, task, i - 1, state)
-            task.status = "SUSPENDED"
-            task.error = str(e)
-            bus.publish("task_suspended", {"task_id": task.id, "reason": str(e)})
+            with db_session() as db:
+                _save_checkpoint(db, task, i - 1, state)
+                row = _fetch_task(db, task_id)
+                row.status = "SUSPENDED"
+                row.error = str(e)
+            bus.publish("task_suspended", {"task_id": task_id, "reason": str(e)})
             return
         except (RedactionBlocked, TaskFailed) as e:
-            task.status = "FAILED"
-            task.error = str(e)
-            task.finished_at = time.time()
-            bus.publish("task_failed", {"task_id": task.id, "error": str(e)})
+            with db_session() as db:
+                row = _fetch_task(db, task_id)
+                row.status = "FAILED"
+                row.error = str(e)
+                row.finished_at = time.time()
+            bus.publish("task_failed", {"task_id": task_id, "error": str(e)})
             return
         except Exception as e:  # noqa: BLE001  未知错误：保留检查点，标记失败可重跑
-            _save_checkpoint(db, task, i - 1, state)
-            task.status = "FAILED"
-            task.error = f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}"
-            task.finished_at = time.time()
-            bus.publish("task_failed", {"task_id": task.id, "error": str(e)})
+            with db_session() as db:
+                _save_checkpoint(db, task, i - 1, state)
+                row = _fetch_task(db, task_id)
+                row.status = "FAILED"
+                row.error = f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}"
+                row.finished_at = time.time()
+            bus.publish("task_failed", {"task_id": task_id, "error": str(e)})
             return
 
-        _record_duration(db, task.type, step.name, time.time() - t0)
-        _save_checkpoint(db, task, i, state)
-        db.commit()  # 每步提交：释放 SQLite 写锁，让下一步的 LLM/网络调用期间不占锁；
-        #             同时让 artifacts/审计/检查点对 API 读连接（WAL 快照）即时可见。
-
-    task.status = "DONE"
-    task.progress = 1.0
-    task.eta_ts = 0
-    task.finished_at = time.time()
-    _update_progress(db, task, steps, len(steps))
-    bus.publish("task_done", {"task_id": task.id})
+        with db_session() as db:
+            _record_duration(db, task.type, step.name, time.time() - t0)
+            _save_checkpoint(db, task, i, state)
+
+    with db_session() as db:
+        row = _fetch_task(db, task_id)
+        row.status = "DONE"
+        row.progress = 1.0
+        row.eta_ts = 0
+        row.finished_at = time.time()
+        _update_progress(db, row, steps, len(steps))
+    bus.publish("task_done", {"task_id": task_id})
 
 
 def _save_checkpoint(db: Session, task: Task, step_index: int, state: dict) -> None: