From 9fd699bd72566bbfd75e7b1a059f66a69fd0a263 Mon Sep 17 00:00:00 2001 From: Vector897 <242271150+Vector897@users.noreply.github.com> Date: Mon, 6 Jul 2026 09:28:03 +0100 Subject: [PATCH] =?UTF-8?q?Refactor:=20session-per-interaction=20=E2=80=94?= =?UTF-8?q?=20no=20DB=20session=20ever=20survives=20a=20network=20call?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root-cause fix for the site-wide POST deadlocks. The old convention threaded one long-lived session through engine -> graphs -> llm -> budget/cache/audit -> notify, with LLM calls, PDF downloads and Telegram/SMTP pushes happening inside the open transaction. Under SQLite's single-writer model any dirty state autoflushed by an in-step SELECT armed a write lock that was then held across up-to-120s network waits, blocking every other writer. Previous commits patched individual sites; this makes the class of bug impossible. New convention (enforced across worker/scheduler paths): - engine.run_task(task_id) owns no session; every bookkeeping write (status/progress/checkpoint/duration) is its own short committed session - TaskContext.db is gone; steps use `with ctx.session() as db:` scoped to DB-only blocks; fetch/download/LLM run sessionless - llm.complete(prompt, task_id=...) is the exemplar: (1) short session for redact/cache/budget and a full routing plan incl. decrypted keys, (2) pure-network litellm call, (3) short session for audit/cache/cost - cascade, notify_all, memory.consolidate/_arbitrate follow the same shape; notify and consolidate now manage their own sessions - runner passes task_id only; nightly consolidation no longer holds a session across per-user LLM calls Tests: updated to new signatures; added a regression guard that runs a task whose step waits 2s "on the network" and asserts a concurrent writer commits in <1s. 15/15 pass. Co-Authored-By: Claude Opus 4.8 --- server/app/api/misc.py | 4 +- server/app/connectors/notify.py | 44 +++--- server/app/core/budget/guard.py | 4 +- server/app/core/engine/engine.py | 138 ++++++++++------- server/app/core/engine/graphs/arxiv_watch.py | 91 ++++++----- server/app/core/engine/graphs/briefing.py | 34 +++-- server/app/core/memory/memory.py | 59 ++++--- server/app/core/router/cascade.py | 10 +- server/app/core/router/llm.py | 153 ++++++++++--------- server/app/core/scheduler/runner.py | 16 +- server/tests/test_core.py | 55 ++++++- server/tests/test_tier_features.py | 5 +- 12 files changed, 359 insertions(+), 254 deletions(-) diff --git a/server/app/api/misc.py b/server/app/api/misc.py index bd8ee32..323ae84 100644 --- a/server/app/api/misc.py +++ b/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]} diff --git a/server/app/connectors/notify.py b/server/app/connectors/notify.py index 44291bb..c8ad67f 100644 --- a/server/app/connectors/notify.py +++ b/server/app/connectors/notify.py @@ -1,6 +1,9 @@ """外部推送：Telegram / 邮件（SMTP）。简报生成完成、预算熔断时触发。 失败静默降级（推送不是关键路径，不应让任务失败）。 + +会话纪律：本模块自管短会话——先短会话读配置并关闭，再做网络发送。 +调用方不需要（也不应该）把自己的会话传进来。 """ from __future__ import annotations @@ -8,17 +11,18 @@ 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 diff --git a/server/app/core/budget/guard.py b/server/app/core/budget/guard.py index 5649044..62f4d60 100644 --- a/server/app/core/budget/guard.py +++ b/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}) diff --git a/server/app/core/engine/engine.py b/server/app/core/engine/engine.py index c4a6ac6..8deac76 100644 --- a/server/app/core/engine/engine.py +++ b/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: diff --git a/server/app/core/engine/graphs/arxiv_watch.py b/server/app/core/engine/graphs/arxiv_watch.py index d188c7d..af6ac68 100644 --- a/server/app/core/engine/graphs/arxiv_watch.py +++ b/server/app/core/engine/graphs/arxiv_watch.py @@ -1,6 +1,9 @@ """学术工作流主图：轮询 → 去重 → 初筛 → 下载归档 → 逐篇总结 → 写记忆。 对应架构文档 3.7 学术工作流。params: {"query": "...", "max_results": 15} + +会话纪律：每步只在纯 DB 操作段开短会话（with ctx.session()）， +LLM 调用与 PDF 下载等网络 I/O 一律在会话之外执行。 """ from __future__ import annotations @@ -28,7 +31,7 @@ def fingerprint(title: str, arxiv_id: str) -> str: def step_fetch(ctx: TaskContext, state: dict) -> dict: """抓取候选条目。学术话题走 arXiv；股票/财经/时事等非学术话题走新闻源； - arXiv 无结果时也自动回退到新闻，保证"任意查询都有结果"。""" + arXiv 无结果时也自动回退到新闻，保证"任意查询都有结果"。纯网络，无会话。""" params = state["params"] query = params.get("query", "LLM agents") n = int(params.get("max_results", 15)) @@ -49,12 +52,13 @@ def step_fetch(ctx: TaskContext, state: dict) -> dict: def step_dedupe(ctx: TaskContext, state: dict) -> dict: fresh = [] - for p in state["found"]: - fp = fingerprint(p["title"], p["arxiv_id"]) - exists = ctx.db.execute(select(Paper).where(Paper.dedup_fingerprint == fp)).first() - if not exists: - p["fingerprint"] = fp - fresh.append(p) + with ctx.session() as db: + for p in state["found"]: + fp = fingerprint(p["title"], p["arxiv_id"]) + exists = db.execute(select(Paper).where(Paper.dedup_fingerprint == fp)).first() + if not exists: + p["fingerprint"] = fp + fresh.append(p) state["fresh"] = fresh ctx.artifact("Dedupe result", f"{len(fresh)} new / {len(state['found'])} fetched") return state @@ -67,17 +71,18 @@ def step_filter(ctx: TaskContext, state: dict) -> dict: state["selected"] = [] return state # 新闻按"查询词"判相关（命中查询的文章都相关）；论文按用户研究画像判相关。 - if state.get("source") == "news": - profile = state["params"].get("query", "") or str(get_setting(ctx.db, "research_profile")) - else: - profile = str(get_setting(ctx.db, "research_profile")) or state["params"].get("query", "") + with ctx.session() as db: + if state.get("source") == "news": + profile = state["params"].get("query", "") or str(get_setting(db, "research_profile")) + else: + profile = str(get_setting(db, "research_profile")) or state["params"].get("query", "") + threshold = float(get_setting(db, "relevance_threshold")) listing = "\n".join(f"{i}. {p['title']} — {p['abstract'][:300]}" for i, p in enumerate(fresh)) prompt = ( f"My research focus: {profile}\n\nBelow are candidate papers. Score each for relevance (0-1). " f'Output ONLY a JSON array like [{{"i":0,"score":0.8}}]:\n\n{wrap_external(listing)}' ) - result = llm.complete(ctx.db, prompt, tier=policy.TIER_LIGHT, task=ctx.task, step="filter", max_tokens=1024) - threshold = float(get_setting(ctx.db, "relevance_threshold")) + result = llm.complete(prompt, tier=policy.TIER_LIGHT, task_id=ctx.task.id, step="filter", max_tokens=1024) scores: dict[int, float] = {} try: text = result.text @@ -97,23 +102,26 @@ def step_filter(ctx: TaskContext, state: dict) -> dict: def step_ingest(ctx: TaskContext, state: dict) -> dict: - """下载 PDF 并入库（纯代码，0 token）。""" + """下载 PDF 并入库（0 token）。下载在会话外，逐篇短会话入库。""" stored_ids = [] for p in state["selected"]: - # 逐篇提交后崩溃重跑的幂等兜底：该指纹已入库则复用，不重复下载/插入 - prev = ctx.db.execute(select(Paper).where(Paper.dedup_fingerprint == p["fingerprint"])).scalar_one_or_none() - if prev is not None: - stored_ids.append(prev.id) + # 崩溃重跑的幂等兜底：该指纹已入库则复用，不重复下载/插入 + with ctx.session() as db: + prev = db.execute(select(Paper).where(Paper.dedup_fingerprint == p["fingerprint"])).scalar_one_or_none() + prev_id = prev.id if prev is not None else None + if prev_id is not None: + stored_ids.append(prev_id) continue - pdf_path = pdf_ingest.download_pdf(p.get("pdf_url", ""), p["arxiv_id"]) - paper = Paper( - arxiv_id=p["arxiv_id"], title=p["title"], authors=p["authors"], - abstract=p["abstract"], published_at=p["published_at"], url=p["url"], - pdf_path=pdf_path, dedup_fingerprint=p["fingerprint"], owner_id=ctx.task.owner_id, - ) - ctx.db.add(paper) - ctx.db.commit() # 逐篇提交而非 flush：flush 拿到的写锁会横跨下一篇的 PDF 下载（网络） - stored_ids.append(paper.id) + pdf_path = pdf_ingest.download_pdf(p.get("pdf_url", ""), p["arxiv_id"]) # 网络，无会话 + with ctx.session() as db: + paper = Paper( + arxiv_id=p["arxiv_id"], title=p["title"], authors=p["authors"], + abstract=p["abstract"], published_at=p["published_at"], url=p["url"], + pdf_path=pdf_path, dedup_fingerprint=p["fingerprint"], owner_id=ctx.task.owner_id, + ) + db.add(paper) + db.flush() + stored_ids.append(paper.id) state["paper_ids"] = stored_ids ctx.artifact("Archive manifest", f"Archived {len(stored_ids)} papers (PDF + metadata)") return state @@ -126,11 +134,12 @@ def step_summarize(ctx: TaskContext, state: dict) -> dict: for idx, pid in enumerate(ids): if pid in done: continue - # 上次运行已总结并提交过 → 幂等跳过，不重复付费（逐篇提交后崩溃重跑的兜底）。 - if ctx.db.execute(select(Summary).where(Summary.paper_id == pid)).first(): - done.append(pid) - continue - paper = ctx.db.execute(select(Paper).where(Paper.id == pid)).scalar_one() + with ctx.session() as db: + # 已有总结（崩溃重跑）→ 幂等跳过，不重复付费 + if db.execute(select(Summary).where(Summary.paper_id == pid)).first(): + done.append(pid) + continue + paper = db.execute(select(Paper).where(Paper.id == pid)).scalar_one() fulltext = pdf_ingest.extract_text(paper.pdf_path, max_chars=40000) body = fulltext if fulltext else paper.abstract if state.get("source") == "news": @@ -146,13 +155,12 @@ def step_summarize(ctx: TaskContext, state: dict) -> dict: "Markdown, under 300 words.\n\n" f"Title: {paper.title}\n\n{wrap_external(body)}" ) - result = llm.complete(ctx.db, prompt, tier=policy.TIER_FRONTIER, task=ctx.task, - step="summarize", max_tokens=1500) - ctx.db.add(Summary(paper_id=pid, model=result.model, content_md=result.text, + result = llm.complete(prompt, tier=policy.TIER_FRONTIER, task_id=ctx.task.id, + step="summarize", max_tokens=1500) # 网络，无会话 + with ctx.session() as db: + db.add(Summary(paper_id=pid, model=result.model, content_md=result.text, cost_usd=result.cost_usd)) done.append(pid) - ctx.db.commit() # 逐篇提交：本步会跑多次 120s LLM 调用，若不提交则写锁横跨整步， - # API POST 全被锁死；提交后每篇总结即时落库、下一篇调用期间不占锁。 bus.publish("task_progress", {"task_id": ctx.task.id, "sub_progress": f"Summarizing {idx + 1}/{len(ids)}"}) ctx.artifact("Summaries done", f"Summarized {len(done)} papers") @@ -162,10 +170,11 @@ def step_summarize(ctx: TaskContext, state: dict) -> dict: def step_memorize(ctx: TaskContext, state: dict) -> dict: n = len(state.get("summarized", [])) q = state["params"].get("query", "") - memory.write_episodic(ctx.db, ctx.task.owner_id, - f"Ran literature watch '{q}': fetched {len(state.get('found', []))}, " - f"{len(state.get('fresh', []))} new, summarized {n}.", - tags=f"arxiv_watch,{q}") + with ctx.session() as db: + memory.write_episodic(db, ctx.task.owner_id, + f"Ran literature watch '{q}': fetched {len(state.get('found', []))}, " + f"{len(state.get('fresh', []))} new, summarized {n}.", + tags=f"arxiv_watch,{q}") return state diff --git a/server/app/core/engine/graphs/briefing.py b/server/app/core/engine/graphs/briefing.py index 6cdc2c8..d7894a0 100644 --- a/server/app/core/engine/graphs/briefing.py +++ b/server/app/core/engine/graphs/briefing.py @@ -1,4 +1,7 @@ -"""晨间简报图：聚合近 24h 总结 → 轻量层撰写 → 入库推送。""" +"""晨间简报图：聚合近 24h 总结 → 轻量层撰写 → 入库推送。 + +会话纪律：DB 读写在短会话内；LLM 撰写与外部推送（Telegram/SMTP）在会话外。 +""" from __future__ import annotations import time @@ -14,14 +17,15 @@ def step_gather(ctx: TaskContext, state: dict) -> dict: cutoff = time.time() - 86400 - rows = ctx.db.execute( - select(Summary, Paper).join(Paper, Summary.paper_id == Paper.id) - .where(Summary.created_at >= cutoff, Paper.owner_id == ctx.task.owner_id) - .order_by(Summary.created_at.desc()) - ).all() - state["items"] = [ - {"title": p.title, "url": p.url, "summary": s.content_md[:1200]} for s, p in rows - ] + with ctx.session() as db: + rows = db.execute( + select(Summary, Paper).join(Paper, Summary.paper_id == Paper.id) + .where(Summary.created_at >= cutoff, Paper.owner_id == ctx.task.owner_id) + .order_by(Summary.created_at.desc()) + ).all() + state["items"] = [ + {"title": p.title, "url": p.url, "summary": s.content_md[:1200]} for s, p in rows + ] ctx.artifact("Source material", f"{len(state['items'])} summaries from the last 24h") return state @@ -38,8 +42,8 @@ def step_compose(ctx: TaskContext, state: dict) -> dict: "each paper give 2-3 sentences of highlights plus one line on why it's worth reading. " "Concise, information-dense, no filler.\n\n" + material ) - result = llm.complete(ctx.db, prompt, tier=policy.TIER_LIGHT, task=ctx.task, - step="compose", max_tokens=2000) + result = llm.complete(prompt, tier=policy.TIER_LIGHT, task_id=ctx.task.id, + step="compose", max_tokens=2000) # 网络，无会话 # 简报末尾附原文链接（本地拼接，0 token） links = "\n".join(f"- [{it['title']}]({it['url']})" for it in items) state["briefing_md"] = result.text + "\n\n## Links\n" + links @@ -48,12 +52,12 @@ def step_compose(ctx: TaskContext, state: dict) -> dict: def step_save(ctx: TaskContext, state: dict) -> dict: date = time.strftime("%Y-%m-%d") - ctx.db.add(Briefing(date=date, content_md=state["briefing_md"], owner_id=ctx.task.owner_id)) + with ctx.session() as db: + db.add(Briefing(date=date, content_md=state["briefing_md"], owner_id=ctx.task.owner_id)) ctx.artifact("Briefing", state["briefing_md"][:2000]) - ctx.db.commit() # 先落库并释放写锁，再做外部推送——Telegram+SMTP 最长 ~30s， - # 期间若持有写锁会堵死全站 POST（settings SELECT 会 autoflush 脏对象） + # 会话已提交关闭，再做外部推送（Telegram+SMTP 最长 ~30s，期间不得持有会话/写锁） bus.publish("briefing_ready", {"date": date, "task_id": ctx.task.id}) - notify_all(ctx.db, f"JarvisQwen briefing {date}", state["briefing_md"][:3500]) + notify_all(f"JarvisQwen briefing {date}", state["briefing_md"][:3500]) return state diff --git a/server/app/core/memory/memory.py b/server/app/core/memory/memory.py index 31f6f38..7426990 100644 --- a/server/app/core/memory/memory.py +++ b/server/app/core/memory/memory.py @@ -10,6 +10,7 @@ from sqlalchemy import select from sqlalchemy.orm import Session +from ...db import session from ...models import Memory from ..router import llm, policy @@ -59,50 +60,62 @@ def _find_conflict(db: Session, owner_id: str, new_fact: str) -> Memory | None: return None -def _arbitrate(db: Session, old: Memory, new_fact: str) -> str: - """时序仲裁：生成保留历史连续性的调和摘要，而非硬覆盖。""" +def _arbitrate(old_content: str, new_fact: str) -> str: + """时序仲裁：生成保留历史连续性的调和摘要，而非硬覆盖。纯网络，无会话。""" prompt = ( "Below are two records on the same topic from different points in time; they may conflict. " "Write ONE sentence that reconciles them with explicit time sense " "(e.g. 'X was A before T1, updated to B since'):" - f"\nOld record: {old.content}\nNew record: {new_fact}" + f"\nOld record: {old_content}\nNew record: {new_fact}" ) - result = llm.complete(db, prompt, tier=policy.TIER_LIGHT, step="memory_arbitrate", max_tokens=200) - return result.text.strip() or f"{old.content} (updated: {new_fact})" + result = llm.complete(prompt, tier=policy.TIER_LIGHT, step="memory_arbitrate", max_tokens=200) + return result.text.strip() or f"{old_content} (updated: {new_fact})" -def consolidate(db: Session, owner_id: str) -> int: - """夜间整合：把近 24h 情节记忆交给轻量层提炼为语义记忆；冲突时时序仲裁而非硬覆盖。""" +def consolidate(owner_id: str) -> int: + """夜间整合：把近 24h 情节记忆交给轻量层提炼为语义记忆；冲突时时序仲裁而非硬覆盖。 + + 自管短会话：读情节 → LLM 提炼（无会话）→ 逐条短会话查冲突/写入， + 仲裁的 LLM 调用同样在会话外。 + """ cutoff = time.time() - 86400 - episodes = db.execute( - select(Memory).where( - Memory.kind == "episodic", Memory.owner_id == owner_id, - Memory.ts >= cutoff, Memory.archived == 0, - ) - ).scalars().all() + with session() as db: + episodes = [e.content[:500] for e in db.execute( + select(Memory).where( + Memory.kind == "episodic", Memory.owner_id == owner_id, + Memory.ts >= cutoff, Memory.archived == 0, + ) + ).scalars().all()] if len(episodes) < 3: return 0 - joined = "\n---\n".join(e.content[:500] for e in episodes[:40]) + joined = "\n---\n".join(episodes[:40]) prompt = ( "Below are work-log fragments from a research-assistant system over the past 24 hours. " "Extract 1-5 facts or patterns worth remembering long-term (topics the user cares " "about, recurring themes, explicit preferences). One per line, conclusions only:\n\n" + joined ) - result = llm.complete(db, prompt, tier=policy.TIER_LIGHT, step="memory_consolidate", max_tokens=512) + result = llm.complete(prompt, tier=policy.TIER_LIGHT, step="memory_consolidate", max_tokens=512) count = 0 for line in result.text.splitlines(): fact = line.strip().lstrip("0123456789.、- ") if len(fact) < 8: continue - conflict = _find_conflict(db, owner_id, fact) - if conflict is not None: - reconciled = _arbitrate(db, conflict, fact) - conflict.content = reconciled - conflict.confidence = min(1.0, conflict.confidence + 0.1) - conflict.ts = time.time() - conflict.tags = (conflict.tags + ",reconciled").strip(",") + with session() as db: + conflict = _find_conflict(db, owner_id, fact) + conflict_id = conflict.id if conflict is not None else None + conflict_content = conflict.content if conflict is not None else "" + if conflict_id is not None: + reconciled = _arbitrate(conflict_content, fact) # 网络，无会话 + with session() as db: + row = db.execute(select(Memory).where(Memory.id == conflict_id)).scalar_one_or_none() + if row is not None: + row.content = reconciled + row.confidence = min(1.0, row.confidence + 0.1) + row.ts = time.time() + row.tags = (row.tags + ",reconciled").strip(",") else: - db.add(Memory(kind="semantic", content=fact, tags="consolidated", owner_id=owner_id)) + with session() as db: + db.add(Memory(kind="semantic", content=fact, tags="consolidated", owner_id=owner_id)) count += 1 return count diff --git a/server/app/core/router/cascade.py b/server/app/core/router/cascade.py index c168ae3..d69c3dc 100644 --- a/server/app/core/router/cascade.py +++ b/server/app/core/router/cascade.py @@ -8,9 +8,6 @@ import re -from sqlalchemy.orm import Session - -from ...models import Task from . import llm, policy HEDGE_PATTERNS = re.compile( @@ -35,16 +32,15 @@ def _confidence_heuristic(text: str, min_len: int = 20) -> float: def complete_cascade( - db: Session, prompt: str, *, - task: Task | None = None, + task_id: str = "", step: str = "", max_tokens: int = 2048, confidence_threshold: float = 0.6, ) -> tuple[llm.LlmResult, bool]: """先用轻量层回答；置信度不足则升级前沿层重做。返回 (结果, 是否发生了升级)。""" - light_result = llm.complete(db, prompt, tier=policy.TIER_LIGHT, task=task, + light_result = llm.complete(prompt, tier=policy.TIER_LIGHT, task_id=task_id, step=f"{step}_light", max_tokens=max_tokens) if light_result.cached or light_result.simulated: return light_result, False # 缓存命中/dry-run 不参与级联判断 @@ -53,6 +49,6 @@ def complete_cascade( if confidence >= confidence_threshold: return light_result, False - frontier_result = llm.complete(db, prompt, tier=policy.TIER_FRONTIER, task=task, + frontier_result = llm.complete(prompt, tier=policy.TIER_FRONTIER, task_id=task_id, step=f"{step}_escalated", max_tokens=max_tokens) return frontier_result, True diff --git a/server/app/core/router/llm.py b/server/app/core/router/llm.py index 4d613fc..7c283e7 100644 --- a/server/app/core/router/llm.py +++ b/server/app/core/router/llm.py @@ -3,6 +3,13 @@ 六道工序：脱敏 → 缓存 → 预算 → 路由 → 弹性调用（退避/断路器/fallback）→ 审计记账。 任何 prompt 都不可能绕过安全与记账——这是本项目的核心工程约束。 +会话纪律（本模块是全项目的样板）：DB 会话只为一次交互而活，绝不跨网络调用。 +执行分三段： + ① 短会话：脱敏/缓存/预算/路由计划（连 API Key 都在此取好），提交即关闭 + ② 纯网络：litellm 调用（最长 120s），期间不持有任何会话或 SQLite 锁 + ③ 短会话：审计 + 缓存回填 + 任务花费累计 +SQLite 单写者模型下，任何"开着事务做网络"的写法都会锁死全站写操作。 + 未配置任何 API Key 时进入 dry-run 模式：返回模拟响应并在审计中标记 simulated， 保证系统无 Key 也能完整跑通流程（开发/演示用）。 """ @@ -10,8 +17,9 @@ from dataclasses import dataclass -from sqlalchemy.orm import Session +from sqlalchemy import select +from ...db import session from ...models import Task from ..budget import guard as budget_guard from ..cache import semantic as cache @@ -34,95 +42,90 @@ class RedactionBlocked(Exception): pass -def _call_litellm(db: Session, model: str, prompt: str, max_tokens: int) -> tuple[str, float, int, int]: - import litellm - - provider = providers.provider_of_model(model) - key_row = providers.pick_key(db, provider) - if key_row is None: - raise resilience.AllModelsFailed(f"No usable API key for provider {provider}") - kwargs: dict = { - "api_key": providers.decrypt_key(key_row.encrypted_key), - "max_tokens": max_tokens, - "timeout": 120, - } - call_model, api_base = providers.litellm_route(model, key_row) - if api_base: - kwargs["api_base"] = api_base - resp = litellm.completion(model=call_model, messages=[{"role": "user", "content": prompt}], **kwargs) - text = resp.choices[0].message.content or "" - usage = getattr(resp, "usage", None) - tin = getattr(usage, "prompt_tokens", 0) or 0 - tout = getattr(usage, "completion_tokens", 0) or 0 - cost = policy.exact_cost(model, tin, tout) # Qwen 官方价目表优先 - if cost is None: - try: - cost = float(litellm.completion_cost(completion_response=resp)) - except Exception: # noqa: BLE001 - cost = policy.estimate_cost(policy.TIER_FRONTIER, len(prompt)) - return text, cost, tin, tout - - -def _has_any_key(db: Session) -> bool: - from sqlalchemy import select - +def _has_any_key(db) -> bool: from ...models import ApiKey return db.execute(select(ApiKey).where(ApiKey.status == "active")).first() is not None def complete( - db: Session, prompt: str, *, tier: str = policy.TIER_LIGHT, - task: Task | None = None, + task_id: str = "", step: str = "", max_tokens: int = 2048, ) -> LlmResult: - task_id = task.id if task else "" - - # ① 脱敏（占位符替换，返回后还原） - level = str(get_setting(db, "redact_level")) - red = redact(prompt, level) - if red.blocked: - raise RedactionBlocked(red.reason) - safe_prompt = red.text - - # ② 语义缓存 - hit = cache.lookup(db, tier, safe_prompt) - if hit is not None: - audit_log.record(db, task_id=task_id, step=step, model="(cache)", input_text=safe_prompt, - output_text=hit, cached=True) - return LlmResult(text=red.restore(hit), model="(cache)", cost_usd=0, cached=True) - - # ③ 预算检查（超限抛 BudgetExceeded → 引擎将任务挂起） - budget_guard.check(db, task, upcoming_estimate=policy.estimate_cost(tier, len(safe_prompt))) - - # dry-run：没有任何 Key 时返回模拟响应，流程照走 - if not _has_any_key(db): - fake = f"[simulated/dry-run] tier={tier}, no API key configured. Prompt digest: {safe_prompt[:120]}" - audit_log.record(db, task_id=task_id, step=step, model="(simulated)", input_text=safe_prompt, - output_text=fake, simulated=True) - return LlmResult(text=fake, model="(simulated)", cost_usd=0, simulated=True) - - # ④⑤ 路由 + 弹性调用（退避/断路器/fallback 链） - # 先 commit：把进度/缓存计数等 autoflush 产生的脏写落库并释放 SQLite 写锁， - # 否则写锁会被持有横跨整个（最长 120s 的）网络调用，堵死全站 POST。 - db.commit() - chain = policy.models_for_tier(db, tier) - retries = int(get_setting(db, "max_retries")) - + # ---- ① 短会话：全部前置 DB 工作，返回前提交关闭 ---- + with session() as db: + level = str(get_setting(db, "redact_level")) + red = redact(prompt, level) + if red.blocked: + raise RedactionBlocked(red.reason) + safe_prompt = red.text + + hit = cache.lookup(db, tier, safe_prompt) + if hit is not None: + audit_log.record(db, task_id=task_id, step=step, model="(cache)", input_text=safe_prompt, + output_text=hit, cached=True) + return LlmResult(text=red.restore(hit), model="(cache)", cost_usd=0, cached=True) + + # 预算检查（超限抛 BudgetExceeded → 引擎将任务挂起）；任务行现查现用，不依赖外部 ORM 对象 + task_row = db.execute(select(Task).where(Task.id == task_id)).scalar_one_or_none() if task_id else None + budget_guard.check(db, task_row, upcoming_estimate=policy.estimate_cost(tier, len(safe_prompt))) + + # dry-run：没有任何 Key 时返回模拟响应，流程照走 + if not _has_any_key(db): + fake = f"[simulated/dry-run] tier={tier}, no API key configured. Prompt digest: {safe_prompt[:120]}" + audit_log.record(db, task_id=task_id, step=step, model="(simulated)", input_text=safe_prompt, + output_text=fake, simulated=True) + return LlmResult(text=fake, model="(simulated)", cost_usd=0, simulated=True) + + # 路由计划：每个候选模型的调用名/Key/endpoint 都在会话内取好，网络阶段零 DB 依赖 + retries = int(get_setting(db, "max_retries")) + plan: dict[str, tuple[str, str, str]] = {} # model -> (call_model, api_key, api_base) + for model in policy.models_for_tier(db, tier): + key_row = providers.pick_key(db, providers.provider_of_model(model)) + if key_row is None: + continue + call_model, api_base = providers.litellm_route(model, key_row) + plan[model] = (call_model, providers.decrypt_key(key_row.encrypted_key), api_base) + + chain = list(plan) + if not chain: + raise resilience.AllModelsFailed(f"No usable API key for any model in tier {tier}") + + # ---- ② 纯网络：不持有任何会话/写锁 ---- def _do(model: str) -> tuple[str, float, int, int]: - return _call_litellm(db, model, safe_prompt, max_tokens) + import litellm + + call_model, api_key, api_base = plan[model] + kwargs: dict = {"api_key": api_key, "max_tokens": max_tokens, "timeout": 120} + if api_base: + kwargs["api_base"] = api_base + resp = litellm.completion(model=call_model, messages=[{"role": "user", "content": safe_prompt}], **kwargs) + text = resp.choices[0].message.content or "" + usage = getattr(resp, "usage", None) + tin = getattr(usage, "prompt_tokens", 0) or 0 + tout = getattr(usage, "completion_tokens", 0) or 0 + cost = policy.exact_cost(model, tin, tout) # Qwen 官方价目表优先 + if cost is None: + try: + cost = float(litellm.completion_cost(completion_response=resp)) + except Exception: # noqa: BLE001 + cost = policy.estimate_cost(policy.TIER_FRONTIER, len(safe_prompt)) + return text, cost, tin, tout model, (text, cost, tin, tout) = resilience.call_with_fallbacks(chain, _do, retries_per_model=retries) - # ⑥ 审计记账 + 缓存回填 + 任务累计花费 - audit_log.record(db, task_id=task_id, step=step, model=model, tokens_in=tin, tokens_out=tout, - cost_usd=cost, input_text=safe_prompt, output_text=text) - cache.store(db, tier, safe_prompt, text, model) - if task is not None: - task.cost_usd += cost + # ---- ③ 短会话：审计记账 + 缓存回填 + 任务累计花费 ---- + with session() as db: + audit_log.record(db, task_id=task_id, step=step, model=model, tokens_in=tin, tokens_out=tout, + cost_usd=cost, input_text=safe_prompt, output_text=text) + cache.store(db, tier, safe_prompt, text, model) + if task_id: + task_row = db.execute(select(Task).where(Task.id == task_id)).scalar_one_or_none() + if task_row is not None: + task_row.cost_usd += cost return LlmResult(text=red.restore(text), model=model, cost_usd=cost) diff --git a/server/app/core/scheduler/runner.py b/server/app/core/scheduler/runner.py index b545bd9..08df7ab 100644 --- a/server/app/core/scheduler/runner.py +++ b/server/app/core/scheduler/runner.py @@ -46,9 +46,7 @@ def worker_loop() -> None: _stop.wait(2) continue try: - with session() as db: - task = db.execute(select(Task).where(Task.id == task_id)).scalar_one() - run_task(db, task) + run_task(task_id) # 引擎自管短会话；执行期间（含网络调用）不持有 DB 锁 except Exception: # noqa: BLE001 引擎内部已兜底，这里防调度器线程死亡 traceback.print_exc() with session() as db: @@ -91,11 +89,13 @@ def nightly_consolidate() -> None: from ..memory.memory import consolidate, decay_heat with session() as db: - for user in db.execute(select(User)).scalars().all(): - try: - consolidate(db, user.id) - except Exception: # noqa: BLE001 - traceback.print_exc() + user_ids = [u.id for u in db.execute(select(User)).scalars().all()] + for user_id in user_ids: # consolidate 内含 LLM 调用，自管短会话，不在此持有会话 + try: + consolidate(user_id) + except Exception: # noqa: BLE001 + traceback.print_exc() + with session() as db: decay_heat(db) evict_expired(db) diff --git a/server/tests/test_core.py b/server/tests/test_core.py index ad172b5..3d99023 100644 --- a/server/tests/test_core.py +++ b/server/tests/test_core.py @@ -114,17 +114,60 @@ def s2(ctx, state): task = Task(type="_test", owner_id=user.id, params_json="{}") db.add(task) db.flush() - run_task(db, task) - assert task.status == "FAILED" - cp = latest_checkpoint(db, task.id) + task_id = task.id + + run_task(task_id) # 引擎自管会话，只需 task_id + with session() as db: + assert db.get(Task, task_id).status == "FAILED" + cp = latest_checkpoint(db, task_id) assert cp is not None and cp.step_name == "s1" # s1 的成果被保住 + db.get(Task, task_id).status = "QUEUED" - task.status = "QUEUED" - run_task(db, task) - assert task.status == "DONE" + run_task(task_id) + with session() as db: + assert db.get(Task, task_id).status == "DONE" assert executed == ["s1", "s2", "s2"] # 重跑没有重复执行 s1 +# ---------- 会话纪律：步骤网络等待期间不得阻塞其他写者 ---------- +def test_writes_not_blocked_while_step_waits_on_network(): + """本项目曾因 worker 跨网络调用持有 SQLite 写锁而全站 POST 瘫痪。 + 此测试守卫会话纪律：步骤在"网络等待"（会话外耗时操作）期间， + 其他线程的写操作必须立即完成，而不是等到任务结束。""" + import threading + import time as _t + + from app.core.engine.engine import StepDef, register, run_task + from app.db import init_db, session + from app.models import Task, User + + init_db() + + def slow_network(ctx, state): + _t.sleep(2.0) # 模拟 LLM/PDF 下载等网络等待（正确写法：不持有会话） + return state + + register("_locktest", [StepDef("net", slow_network)]) + with session() as db: + user = User(name="lock_user", password_hash="x") + db.add(user) + db.flush() + task = Task(type="_locktest", owner_id=user.id, params_json="{}") + db.add(task) + db.flush() + task_id, owner_id = task.id, user.id + + worker = threading.Thread(target=run_task, args=(task_id,)) + worker.start() + _t.sleep(0.5) # 等 worker 进入 slow_network 的等待段 + t0 = _t.time() + with session() as db: # 模拟 API 线程的 POST 写入 + db.add(Task(type="_locktest", owner_id=owner_id, params_json="{}", status="CANCELLED")) + blocked_for = _t.time() - t0 + worker.join() + assert blocked_for < 1.0, f"并发写被阻塞 {blocked_for:.2f}s——步骤执行期间不应持有写锁" + + # ---------- 预算熔断 ---------- def test_budget_cutoff(): from app.core.budget.guard import BudgetExceeded, check diff --git a/server/tests/test_tier_features.py b/server/tests/test_tier_features.py index 0fb1a20..59e9238 100644 --- a/server/tests/test_tier_features.py +++ b/server/tests/test_tier_features.py @@ -70,6 +70,5 @@ def test_memory_arbitration_reconciles_conflict(): uid = user.id # consolidate 依赖 LLM 调用（dry-run 无 Key 时返回模拟响应），只验证不抛异常、流程可跑通 - with session() as db: - count = consolidate(db, uid) - assert count >= 0 # dry-run 下情节记忆不足 3 条会提前返回 0，属预期 + count = consolidate(uid) # 自管短会话，LLM 调用在会话外 + assert count >= 0 # dry-run 下情节记忆不足 3 条会提前返回 0，属预期