统一运行生命周期入口

CHANGELOG.md

@@ -5,6 +5,10 @@
 > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。
 > 工程口径的完整记录见 `PROGRESS.md` / git log。
 
+## 0.60.3 — 2026-07-28
+
+- 网页、微信、企业微信和定时任务现在采用一致的消息接收机制：消息会在开始处理前可靠保存，服务恰好在后台任务启动前重启时也不会静默丢失输入。
+
 ## 0.60.2 — 2026-07-27
 
 - 提升发送消息的可靠性：服务在接收消息后即使遇到重启或后台任务调度异常，也不会静默丢失用户刚发送的内容；非法的生图、视频模型选项也不再导致任务卡在运行中。

DESIGN.md

@@ -45,7 +45,8 @@ zcbot/
 ├── config/{agent.yaml, models/*.yaml, media/*.yaml}
 ├── workspace/users/<user_id>/{.memory/, .skills/, <working_dir>/}
 ├── 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 回收 |
 

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` 按安全门控跳过。

core/__init__.py

@@ -1,3 +1,3 @@
 # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。
 # 改版本只动这一行。
-__version__ = "0.60.2"
+__version__ = "0.60.3"

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)

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()

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",

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:

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

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:

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,