diff --git a/CHANGELOG.md b/CHANGELOG.md index 9433d37..b227d70 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。 > 工程口径的完整记录见 `PROGRESS.md` / git log。 +## 0.60.3 — 2026-07-28 + +- 网页、微信、企业微信和定时任务现在采用一致的消息接收机制：消息会在开始处理前可靠保存，服务恰好在后台任务启动前重启时也不会静默丢失输入。 + ## 0.60.2 — 2026-07-27 - 提升发送消息的可靠性：服务在接收消息后即使遇到重启或后台任务调度异常，也不会静默丢失用户刚发送的内容；非法的生图、视频模型选项也不再导致任务卡在运行中。 diff --git a/DESIGN.md b/DESIGN.md index d80f058..45cb893 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -45,7 +45,8 @@ zcbot/ ├── config/{agent.yaml, models/*.yaml, media/*.yaml} ├── workspace/users//{.memory/, .skills/, /} ├── web/ # app.py=工厂+lifespan 编排;routers/*(11 路由模块,register 范式); -│ # background/scheduler_runner/wechat_runner(后台协程);runs(BG worker) +│ # background/scheduler_runner/wechat_runner(后台协程); +│ # run_lifecycle(统一抢占/落消息/调度);runs(BG worker) │ # + auth/admin/broker/sinks/common/schemas/model_gate/userfiles/static/ ├── db/migrations/ # alembic └── main.py # 入口:web / db / probe / user @@ -160,8 +161,9 @@ Tasks POST/GET/PATCH/DELETE /v1/tasks*(POST 可选 auto_title;分页+筛选+ DELETE=软删,FS 不动) GET /v1/folders(working_dir + task 计数) GET/POST /v1/tasks/{id}/messages(POST 起 run;单活 run:running/cancelling→409, - 先校验请求，再用 SELECT FOR UPDATE 将 user 消息与 running 同事务提交； - BG worker 消费已持久化轮次，不重复追加 user，防 202 后崩溃丢输入/idx race) + 先校验请求；Web/渠道/定时共用 run_lifecycle，以 SELECT FOR UPDATE + 将 user 消息与 running 同事务提交并统一登记 broker/inflight； + BG worker 消费已持久化轮次，不重复追加 user，防调度前崩溃丢输入/idx race) GET /v1/tasks/{id}/events(SSE) POST /v1/tasks/{id}/cancel(协作式,202) Auth POST /v1/auth/login(platform_key)/ login_password / change_password;GET /v1/me Files GET /v1/files?path= / upload / download / delete / rename @@ -247,7 +249,7 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB) | running task 被 rename/delete | 后端校验 + UI 禁按钮 | | DB-then-FS 中断孤儿 | rename DB 先行可回滚;delete 后台 GC 扫"FS 有 DB 无" | | 同 wd 多 task 并发写同名 | known limitation,频率近 0;软警告 banner;宪法文件已按 short_id 命名隔离 | -| 并发 POST 撞 messages.idx / 202 后进程退出丢输入 | 单活 run gate(FOR UPDATE + 409)下原子提交 user 消息与 running；worker 只消费已持久化轮次；lifespan reaper 收敛残留 running，multi-worker 再换 lease | +| 各入口并发撞 messages.idx / 接收后进程退出丢输入 | Web/渠道/定时共用 run_lifecycle：单活 gate(FOR UPDATE)下原子提交 user 消息与 running、统一调度 worker；worker 只消费已持久化轮次；lifespan reaper 收敛残留 running，multi-worker 再换 lease | | shell/run_python 无沙箱开放外部 = 主机沦陷 | **Stage C 是 hard prereq**;`BLOCKED_PATTERNS` 是 trivial-bypass 装饰品,不再加规则(黑名单 fundamentally broken),防线在 OS 层 | | sandbox 出站越权 / 资源滥用 | default-deny + 受控 proxy;硬限制 + 软配额 + idle 回收 | diff --git a/PROGRESS.md b/PROGRESS.md index 6091f4f..f99fb2e 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -2,7 +2,7 @@ > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。 -最后更新:2026-07-27(Web 消息接收事务化,bump 0.60.2) +最后更新:2026-07-28(全入口 run 生命周期收口,bump 0.60.3) --- @@ -21,6 +21,10 @@ ## 已完成关键能力 +### 2026-07-28 + +- **07-28 / 0.60.3 / Web、渠道、定时任务统一 run 生命周期**:新增 `web/run_lifecycle.py`，收口 task 行锁/忙碌判断、user 消息与 running 原子提交、broker/inflight 登记及调度失败转 error；网页、微信/企微、定时任务全部消费已持久化轮次，消除渠道与 scheduler 原有的 worker 启动前丢输入窗口。Web 模型降级/自动标题、渠道回复、定时结果统计继续留在各自模块，CLI 保留旧 `run(message)`，无新表、migration、队列或 run 实体。全量 379 测试通过（17 项按环境跳过）。 + ### 2026-07 - **07-27 / 0.60.2 / Web 消息接收事务化（消除 202 后丢输入窗口）**:`POST /messages` 先完成媒体 variant 校验，再在 task 行锁保护下把用户消息与 `run_status=running` 同事务提交；Web worker 新增已持久化轮次入口，只消费 Session 末尾 user 而不重复 append，CLI/渠道/定时入口继续走旧入口保持兼容。后台 coroutine 调度失败会把 task 收敛为 error，不再留下假 running；不新增 runs 表、队列或 migration。全量 375 测试通过（17 项按环境跳过），新增 DB 路由测试因未设置 `ZCBOT_TEST_DB_URL` 按安全门控跳过。 diff --git a/core/__init__.py b/core/__init__.py index 23eea16..8307c2f 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.60.2" +__version__ = "0.60.3" diff --git a/core/loop.py b/core/loop.py index 75360e7..c2ca343 100644 --- a/core/loop.py +++ b/core/loop.py @@ -249,7 +249,7 @@ class AgentLoop: return self._run(user_message) def run_persisted_turn(self) -> str: - """运行已由调用方原子持久化的用户轮次（Web POST 入口）。""" + """运行已由调用方原子持久化的用户轮次（Web 生命周期入口）。""" if not self.session.messages or self.session.messages[-1].get("role") != "user": raise RuntimeError("persisted turn requires the latest message to be user") return self._run(None) diff --git a/tests/test_run_lifecycle.py b/tests/test_run_lifecycle.py new file mode 100644 index 0000000..5a70c9a --- /dev/null +++ b/tests/test_run_lifecycle.py @@ -0,0 +1,141 @@ +"""统一 run 生命周期服务的无 DB 回归测试。""" +from __future__ import annotations + +import asyncio +import unittest +from contextlib import contextmanager +from types import SimpleNamespace +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +from web import run_lifecycle + + +class ClaimRunTests(unittest.TestCase): + def test_claim_atomically_adds_message_and_marks_running(self) -> None: + tid, uid = uuid4(), uuid4() + task = SimpleNamespace( + task_id=tid, + user_id=uid, + run_status="idle", + model_profile="old", + ) + session = MagicMock() + session.execute.side_effect = [ + SimpleNamespace(scalar_one_or_none=lambda: task), + SimpleNamespace(scalar_one=lambda: 7), + MagicMock(), + ] + + @contextmanager + def fake_scope(): + yield session + + def prepare(_session, locked_task): + self.assertIs(locked_task, task) + return {"model_profile": "new"}, {"marker": "ok"} + + with patch.object(run_lifecycle, "session_scope", fake_scope): + claim = run_lifecycle.claim_run_with_message( + tid, uid, "持久化消息", prepare=prepare, + ) + + self.assertEqual(claim.metadata, {"marker": "ok"}) + added = session.add.call_args.args[0] + self.assertEqual(added.task_id, tid) + self.assertEqual(added.idx, 7) + self.assertEqual( + added.payload, + {"role": "user", "content": "持久化消息"}, + ) + update_stmt = session.execute.call_args_list[-1].args[0] + params = update_stmt.compile().params + self.assertEqual(params["run_status"], "running") + self.assertIsNone(params["run_error"]) + self.assertEqual(params["model_profile"], "new") + + def test_busy_task_is_rejected_before_message_insert(self) -> None: + session = MagicMock() + session.execute.return_value = SimpleNamespace( + scalar_one_or_none=lambda: SimpleNamespace(run_status="running") + ) + + @contextmanager + def fake_scope(): + yield session + + with ( + patch.object(run_lifecycle, "session_scope", fake_scope), + self.assertRaises(run_lifecycle.RunTaskBusy), + ): + run_lifecycle.claim_run_with_message(uuid4(), uuid4(), "消息") + session.add.assert_not_called() + + +class ScheduleRunTests(unittest.IsolatedAsyncioTestCase): + async def test_schedule_uses_persisted_entry_and_tracks_inflight(self) -> None: + tid, uid = uuid4(), uuid4() + app = SimpleNamespace(state=SimpleNamespace(inflight={})) + worker = MagicMock() + broker = MagicMock() + + with ( + patch.object(run_lifecycle, "run_agent_bg", worker), + patch.object(run_lifecycle, "broker", broker), + ): + task = run_lifecycle.schedule_claimed_run( + app, + tid, + uid, + "已落库", + image_variant="image-x", + video_variant="video-y", + scheduled=True, + ) + self.assertEqual(app.state.inflight[task], tid) + await task + await asyncio.sleep(0) + + worker.assert_called_once_with( + tid, + uid, + "已落库", + "image-x", + "video-y", + True, + user_message_persisted=True, + ) + self.assertEqual(app.state.inflight, {}) + broker.start.assert_called_once_with(tid) + + async def test_schedule_failure_marks_error_and_closes_broker(self) -> None: + tid, uid = uuid4(), uuid4() + app = SimpleNamespace(state=SimpleNamespace(inflight={})) + session = MagicMock() + broker = MagicMock() + + @contextmanager + def fake_scope(): + yield session + + with ( + patch.object(run_lifecycle, "session_scope", fake_scope), + patch.object(run_lifecycle, "broker", broker), + patch.object( + run_lifecycle.asyncio, + "create_task", + side_effect=RuntimeError("loop closed"), + ), + self.assertRaises(run_lifecycle.RunScheduleError), + ): + run_lifecycle.schedule_claimed_run(app, tid, uid, "已落库") + + update_stmt = session.execute.call_args.args[0] + params = update_stmt.compile().params + self.assertEqual(params["run_status"], "error") + self.assertIn("loop closed", params["run_error"]) + broker.close.assert_called_once_with(tid) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_web_routes_db.py b/tests/test_web_routes_db.py index ee0da9b..c272a46 100644 --- a/tests/test_web_routes_db.py +++ b/tests/test_web_routes_db.py @@ -244,7 +244,7 @@ class MessageRunDurabilityTests(unittest.TestCase): with ( patch("web.routers.messages.resolve_image_model", return_value=""), patch("web.routers.messages.resolve_video_model", return_value=""), - patch("web.routers.messages.run_agent_bg", side_effect=fake_worker), + patch("web.run_lifecycle.run_agent_bg", side_effect=fake_worker), ): r = _client.post( f"/v1/tasks/{tid}/messages", diff --git a/web/routers/messages.py b/web/routers/messages.py index 4259794..0c3bcac 100644 --- a/web/routers/messages.py +++ b/web/routers/messages.py @@ -1,7 +1,7 @@ """Messages 路由:历史/目录/发消息起 run/取消/清空/润色/bg proc/SSE。 -run 的单活闸(FOR UPDATE + 409)与 drain 背压都在这里;BG worker 本体在 -web/runs.py::run_agent_bg。 +drain 背压与 Web 特有模型门控在这里；run 抢占/调度收口在 +web/run_lifecycle.py，BG worker 本体在 web/runs.py。 """ from __future__ import annotations @@ -11,14 +11,13 @@ from uuid import UUID from fastapi import Depends, HTTPException from fastapi.responses import StreamingResponse -from sqlalchemy import func, select, update +from sqlalchemy import select, update from core.storage import session_scope from core.storage.models import Message, Task, User from ..broker import broker from ..common import ( - INSTANCE, assert_owns_task, iso, outline_snippet, @@ -34,7 +33,13 @@ from ..model_gate import ( resolve_video_model, skill_pinned_profiles, ) -from ..runs import run_agent_bg +from ..run_lifecycle import ( + RunScheduleError, + RunTaskBusy, + RunTaskNotFound, + claim_run_with_message, + schedule_claimed_run, +) from ..schemas import MessageRequest, OptimizePromptRequest from ..userfiles import load_user_root @@ -211,33 +216,13 @@ def register_message_routes(app, *, require_user) -> None: # running 后才发现参数非法，会留下一个实际上没有 worker 的假活跃任务。 image_variant = resolve_image_model(body.image_model, user_id=user_id) video_variant = resolve_video_model(body.video_model, user_id=user_id) - with session_scope() as s: - row = s.execute( - select( - Task.run_status, - Task.model_profile, - Task.auto_title_pending, - ) - .where(Task.task_id == tid, Task.user_id == user_id) - .with_for_update() - ).first() - if row is None: - raise HTTPException(404, f"task not found: {tid}") - if row.run_status in ("running", "cancelling"): - raise HTTPException( - 409, - f"task already has an active run (status={row.run_status}); " - f"wait for it to finish or cancel", - ) - values: dict = { - "run_status": "running", "run_error": None, - "run_owner": INSTANCE or None, - } + def _prepare_claim(s, task): + values: dict = {} # 档位门控:存量 task 的模型已不在用户档位内(如管理员下调了档位)→ 本次起 # 持久落回 flash(基线必含),UI 下拉随之显示 flash。不报错、不打断会话历史, # 符合"老 task 下次发消息直接切 flash"。当前 task 模型仍在档内则原样不动。 # plan/role 用同一 session 读(避免在 FOR UPDATE 事务里再开嵌套 session)。 - cur_profile = row.model_profile or "" + cur_profile = task.model_profile or "" if cur_profile: from core.model_access import is_allowed urow = s.execute( @@ -252,51 +237,40 @@ def register_message_routes(app, *, require_user) -> None: fb_profile, fb_model_id = resolve_model_profile(FALLBACK_MODEL_PROFILE) values["model_profile"] = fb_profile values["model"] = fb_model_id - # task 行锁串行化同一会话的 idx 分配。用户消息与 running 状态同事务提交： - # 只要客户端拿到 202，这一轮输入就已可从 DB 恢复；worker 不再承担首条 - # user 消息的持久化责任。 - next_idx = s.execute( - select(func.coalesce(func.max(Message.idx), -1) + 1) - .where(Message.task_id == tid) - ).scalar_one() - s.add(Message( - task_id=tid, - idx=int(next_idx), - payload={"role": "user", "content": content}, - )) - s.execute( - update(Task).where(Task.task_id == tid).values(**values) - ) - title_profile = values.get("model_profile", cur_profile) - should_auto_title = bool(row.auto_title_pending) - broker.start(tid) # 清上一轮 done 标记,新订阅者才能看到流式 - # commit 后 lock 释放;BG 线程接管(sink 通过 broker 把 event 桥回 asyncio loop)。 - # 登记到 app.state.inflight:① 关停 drain 时 await 它收尾 ② 持强引用防 task 被 GC - # 中途回收(asyncio.create_task 不留引用是已知坑)。done 回调自摘除。 - run_coro = asyncio.to_thread( - run_agent_bg, tid, user_id, content, image_variant, video_variant, - user_message_persisted=True, - ) + return values, { + "title_profile": values.get("model_profile", cur_profile), + "should_auto_title": bool(task.auto_title_pending), + } + try: - run_task = asyncio.create_task(run_coro) - except Exception as e: - # 极少见的 event-loop 调度失败也不能让 task 永久留在 running。消息已经 - # 持久化，明确标 error，后续可直接续跑；关闭未调度 coroutine 避免告警。 - run_coro.close() - err = f"background scheduling failed: {type(e).__name__}: {e}" - with session_scope() as s: - s.execute( - update(Task).where(Task.task_id == tid).values( - run_status="error", run_error=err, - ) - ) - broker.close(tid) + claim = claim_run_with_message( + tid, user_id, content, prepare=_prepare_claim, + ) + except RunTaskNotFound: + raise HTTPException(404, f"task not found: {tid}") + except RunTaskBusy as e: + raise HTTPException( + 409, + f"task already has an active run (status={e.status}); " + f"wait for it to finish or cancel", + ) + + try: + schedule_claimed_run( + app, + tid, + user_id, + content, + image_variant=image_variant, + video_variant=video_variant, + ) + except RunScheduleError: raise HTTPException( 500, "message persisted, but background scheduling failed; retry this task", ) - app.state.inflight[run_task] = tid - run_task.add_done_callback(lambda t: app.state.inflight.pop(t, None)) + title_profile = claim.metadata["title_profile"] + should_auto_title = claim.metadata["should_auto_title"] # 快速入口只在首条消息时 pending=true。辅助标题与主 run 并行，不阻塞 # TTFT/SSE；generate_task_title 内再次查闸并用条件 UPDATE 防人工改名竞态。 if should_auto_title: diff --git a/web/run_lifecycle.py b/web/run_lifecycle.py new file mode 100644 index 0000000..c655260 --- /dev/null +++ b/web/run_lifecycle.py @@ -0,0 +1,141 @@ +"""Web 运行生命周期的统一入口：事务抢占、消息落库与后台 worker 调度。 + +只收口所有 Web 形态入口共有的正确性边界；模型降级、自动标题、渠道回复和定时 +结果统计仍由各自调用方负责。不引入持久化 run 实体或队列。 +""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from typing import Any, Callable, Optional +from uuid import UUID + +from sqlalchemy import func, select, update + +from core.storage import session_scope +from core.storage.models import Message, Task + +from .broker import broker +from .common import INSTANCE +from .runs import run_agent_bg + + +class RunTaskNotFound(Exception): + """目标 task 不存在或不属于指定用户。""" + + +class RunTaskBusy(Exception): + """目标 task 已有活跃 run。""" + + def __init__(self, status: str) -> None: + self.status = status + super().__init__(f"task already has an active run (status={status})") + + +class RunScheduleError(Exception): + """消息已持久化，但后台 coroutine 未能登记到 event loop。""" + + +PrepareClaim = Callable[[Any, Task], tuple[dict[str, Any], dict[str, Any]]] + + +@dataclass(frozen=True) +class RunClaim: + """抢占成功后返回给调用方的领域元数据。""" + + metadata: dict[str, Any] = field(default_factory=dict) + + +def claim_run_with_message( + task_id: UUID, + user_id: UUID, + user_message: str, + *, + prepare: Optional[PrepareClaim] = None, +) -> RunClaim: + """在 task 行锁下原子提交 user 消息和 `run_status=running`。 + + `prepare` 在持锁事务内执行，可读取 task/用户状态并返回： + `(额外 task 更新字段, 调用方元数据)`。它用于 Web 模型降级和自动标题快照， + 不把这些领域规则塞进通用生命周期层。回调抛错会使整个事务回滚。 + """ + with session_scope() as s: + task = s.execute( + select(Task) + .where(Task.task_id == task_id, Task.user_id == user_id) + .with_for_update() + ).scalar_one_or_none() + if task is None: + raise RunTaskNotFound(str(task_id)) + if task.run_status in ("running", "cancelling"): + raise RunTaskBusy(task.run_status) + + extra_values: dict[str, Any] = {} + metadata: dict[str, Any] = {} + if prepare is not None: + extra_values, metadata = prepare(s, task) + + # task 行锁串行化同一 task 的 idx 分配，无需额外序列表或 advisory lock。 + next_idx = s.execute( + select(func.coalesce(func.max(Message.idx), -1) + 1) + .where(Message.task_id == task_id) + ).scalar_one() + s.add(Message( + task_id=task_id, + idx=int(next_idx), + payload={"role": "user", "content": user_message}, + )) + values = { + "run_status": "running", + "run_error": None, + "run_owner": INSTANCE or None, + **extra_values, + } + s.execute(update(Task).where(Task.task_id == task_id).values(**values)) + return RunClaim(metadata=dict(metadata)) + + +def schedule_claimed_run( + app, + task_id: UUID, + user_id: UUID, + user_message: str, + *, + image_variant: str = "", + video_variant: str = "", + scheduled: bool = False, +) -> asyncio.Task: + """调度已完成事务抢占的 run，并统一登记 broker / inflight。 + + 调度失败时消息不能回滚（事务已经提交），因此把 task 收敛为 error 并抛出 + `RunScheduleError`，调用方可转成 HTTP 500、渠道错误回复或定时失败记录。 + """ + broker.start(task_id) + run_coro = asyncio.to_thread( + run_agent_bg, + task_id, + user_id, + user_message, + image_variant, + video_variant, + scheduled, + user_message_persisted=True, + ) + try: + run_task = asyncio.create_task(run_coro) + except Exception as e: + run_coro.close() + err = f"background scheduling failed: {type(e).__name__}: {e}" + with session_scope() as s: + s.execute( + update(Task).where(Task.task_id == task_id).values( + run_status="error", + run_error=err, + ) + ) + broker.close(task_id) + raise RunScheduleError(err) from e + + app.state.inflight[run_task] = task_id + run_task.add_done_callback(lambda t: app.state.inflight.pop(t, None)) + return run_task diff --git a/web/runs.py b/web/runs.py index 271d10a..b029edd 100644 --- a/web/runs.py +++ b/web/runs.py @@ -18,7 +18,6 @@ from core.storage.telemetry import record_run_error from core.toolfail import alert_provider_critical from .broker import broker -from .common import INSTANCE from .sinks import WebEventSink @@ -41,8 +40,9 @@ def run_agent_bg( image_variant / video_variant:本 run 用哪个 image/video variant 装 tool(空 → yaml 第一个)。 随消息 POST 传进来,不入 DB —— UI 下拉的选择就跟在这一条消息上生效。 - user_message_persisted=True 仅供 Web POST：用户消息已和 tasks.running 同事务提交， - worker 从 Session 恢复后直接处理最后一条 user，避免重复落库。其余入口保持旧行为。 + user_message_persisted=True 供 Web / 渠道 / 定时生命周期入口：用户消息已和 + tasks.running 同事务提交，worker 从 Session 恢复后直接处理最后一条 user， + 避免重复落库。CLI 等非 Web 入口保持旧行为。 """ from core.agent_builder import build_agent, sync_task_tokens cancel_check = lambda tid=task_id: broker.is_cancelled(tid) @@ -148,6 +148,13 @@ async def run_channel_conversation(app, uid, text, attachments, *, channel): from core.wechat import service as _wx from core.wechat.ilink import attachment_basename from core.wechat.inbound import extract_last_assistant_text + from .run_lifecycle import ( + RunScheduleError, + RunTaskBusy, + RunTaskNotFound, + claim_run_with_message, + schedule_claimed_run, + ) # 解析/建该渠道常驻 chat task(不存在自动建)—— 与 push 记录(send_to_user)共用 # ensure_channel_chat_task,避免两条建 task 路径漂移。wechat 无 binding → 返回 None。 @@ -206,24 +213,17 @@ async def run_channel_conversation(app, uid, text, attachments, *, channel): extra = "\n".join(lines) text = f"{text}\n\n{extra}" if text.strip() else extra - # 抢 run 锁:正忙 → 提示稍候(同用户串行;ClawBot loop 本就串行,wecom 回调靠此挡并发) - with session_scope() as s: - row = s.execute( - select(Task.run_status).where(Task.task_id == tid).with_for_update() - ).first() - if row is None: - return "[出错] 对话 task 不存在" - if row.run_status in ("running", "cancelling"): - return "上一条还在处理中,请稍候再发。" - s.execute(update(Task).where(Task.task_id == tid).values( - run_status="running", run_error=None, run_owner=INSTANCE or None)) - - broker.start(tid) - runner = asyncio.create_task(asyncio.to_thread( - run_agent_bg, tid, uid, text, "", "", False, - )) - app.state.inflight[runner] = tid - runner.add_done_callback(lambda t: app.state.inflight.pop(t, None)) + # 渠道消息也与 running 原子提交，消除进程在 worker 启动前退出时的丢输入窗口。 + try: + claim_run_with_message(tid, uid, text) + except RunTaskNotFound: + return "[出错] 对话 task 不存在" + except RunTaskBusy: + return "上一条还在处理中,请稍候再发。" + try: + runner = schedule_claimed_run(app, tid, uid, text) + except RunScheduleError as e: + return f"[出错] {e}" await runner with session_scope() as s: diff --git a/web/scheduler_runner.py b/web/scheduler_runner.py index 93216fa..d57c61b 100644 --- a/web/scheduler_runner.py +++ b/web/scheduler_runner.py @@ -24,9 +24,14 @@ from core.storage import session_scope from core.storage.models import ScheduledJob, Task from .broker import broker -from .common import INSTANCE from .model_gate import resolve_model_profile -from .runs import run_agent_bg +from .run_lifecycle import ( + RunScheduleError, + RunTaskBusy, + RunTaskNotFound, + claim_run_with_message, + schedule_claimed_run, +) def start_scheduler(app, cfg: dict) -> Optional[asyncio.Task]: @@ -87,30 +92,34 @@ def start_scheduler(app, cfg: dict) -> Optional[asyncio.Task]: ScheduledJob.job_id == job_id ).values(bound_task_id=tid)) - # 抢 run 锁(同 post_message):busy → 本次跳过,下个 cron 点再来 - with session_scope() as s: - row = s.execute( - select(Task.run_status).where(Task.task_id == tid).with_for_update() - ).first() - if row is None: - record_result(job_id, status="error", task_id=tid, error="目标 task 不存在") - return - if row.run_status in ("running", "cancelling"): - record_result(job_id, status="skipped", task_id=tid, - error="目标 task 正忙,本次跳过") - print(f"[scheduler] job {str(job_id)[:8]} skipped (task busy)") - return - s.execute(update(Task).where(Task.task_id == tid).values( - run_status="running", run_error=None, - run_owner=INSTANCE or None)) - message = build_run_message(snap) - broker.start(tid) - runner = asyncio.create_task(asyncio.to_thread( - run_agent_bg, tid, uid, message, "", "", True, - )) - app.state.inflight[runner] = tid - runner.add_done_callback(lambda t: app.state.inflight.pop(t, None)) + try: + claim_run_with_message(tid, uid, message) + except RunTaskNotFound: + record_result( + job_id, status="error", task_id=tid, + error="目标 task 不存在", + ) + return + except RunTaskBusy: + record_result( + job_id, status="skipped", task_id=tid, + error="目标 task 正忙,本次跳过", + ) + print(f"[scheduler] job {str(job_id)[:8]} skipped (task busy)") + return + try: + runner = schedule_claimed_run( + app, tid, uid, message, scheduled=True, + ) + except RunScheduleError as e: + record_result( + job_id, status="error", task_id=tid, error=str(e), + ) + print( + f"[scheduler] job {str(job_id)[:8]} scheduling failed: {e}" + ) + return timeout = int(snap.get("timeout_seconds") or 0) timed_out = False @@ -124,9 +133,9 @@ def start_scheduler(app, cfg: dict) -> Optional[asyncio.Task]: else: await runner - # 超时被掐断:run_agent_bg 对 ok/cancelled 都把 run_status 收回 idle - # (二者在 DB 里不可区分),只有这里知道本次是 timeout 中断的。必须记为 - # error —— 否则会误判成 ok(掩盖"跑到一半没推送"),且不计入连续失败/不触发 + # 超时被掐断:run_agent_bg 把 task 标 cancelled，但只有这里知道它是 + # scheduler timeout 而非用户停止。job 必须另记 error + # —— 否则会误判成 ok(掩盖"跑到一半没推送"),且不计入连续失败/不触发 # 兜底。半成品不投递 notify,直接收尾返回。 if timed_out: record_result(job_id, status="error", task_id=tid,