增强消息运行可靠性

CHANGELOG.md

@@ -5,6 +5,10 @@
 > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。
 > 工程口径的完整记录见 `PROGRESS.md` / git log。
 
+## 0.60.2 — 2026-07-27
+
+- 提升发送消息的可靠性：服务在接收消息后即使遇到重启或后台任务调度异常，也不会静默丢失用户刚发送的内容；非法的生图、视频模型选项也不再导致任务卡在运行中。
+
 ## 0.60.1 — 2026-07-27
 
 - 联网搜索现在会以当前北京时间判断时效：未指定时间时不再擅自给搜索词添加旧年份，查询“最新、当前、近期”内容时会采用合适的近期结果范围；用户明确指定的年份或时间窗保持优先。

DESIGN.md

@@ -160,7 +160,8 @@ 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 锁 task 行防 idx race)
+            先校验请求，再用 SELECT FOR UPDATE 将 user 消息与 running 同事务提交；
+            BG worker 消费已持久化轮次，不重复追加 user，防 202 后崩溃丢输入/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
@@ -246,7 +247,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 | 单活 run gate(FOR UPDATE + 409)+ lifespan reaper;multi-worker 再换 lease |
+| 并发 POST 撞 messages.idx / 202 后进程退出丢输入 | 单活 run gate(FOR UPDATE + 409)下原子提交 user 消息与 running；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(联网搜索动态时效约束,bump 0.60.1)
+最后更新:2026-07-27(Web 消息接收事务化,bump 0.60.2)
 
 ---
 
@@ -23,6 +23,7 @@
 
 ### 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` 按安全门控跳过。
 - **07-27 / 0.60.1 / 联网搜索动态时效约束**:真实 task `24d6b609` 在 2026-07-27、用户未指定年份时，`deepseek_v4.flash` 两次自行给 `web_search` query 加 `2025`；工具原样转发且默认 `freshness=noLimit`，根因是 system prompt 虽注入 today 却仅标作宪法文件命名用途，工具 schema 也无年份策略。修为 system prompt 首段按 `Asia/Shanghai` 动态注入当前时间/日期/年份，并给唯一规则：未指定时间则 query 无年份，“最新/当前/近期”按当前日期配合 freshness，明确时间窗则服从用户；`web_search` 的 query/freshness schema 再局部强化。5 项定向测试通过。
 - **07-27 / 0.60.0 / 目录优先的新对话草稿页 + 首条消息自动命名**:登录未选任务与左栏「+ 新对话」统一进入不落 DB 的草稿页，选择/新建 working_dir 后首发才创建 task，避免空任务；原完整表单保留为「自定义」。新增 0023 `tasks.auto_title_pending` 一次性闸与 `auto_title` 可选创建字段，首条消息并行生成短标题（`usage_events.kind=task_title`），只改 task 显示名不动目录，人工 PATCH name 清闸且条件 UPDATE 防在途覆盖；失败保留「新对话」不阻塞主 run。
 - **07-27 / 0.59.6 / 对话内安全重命名 working_dir**:新增 `rename_working_dir` 受控工具，agent run 内只登记目标 leaf 名，正常回复结束并把当前 task 退出 running 后再落地，避免 executor/system prompt/宿主工具仍握旧 cwd；取消或失败不执行。文件面板与对话路径共用 `core/working_dirs.py` DB-aware 原语：锁定并同步更新共享目录的全部 task、活跃邻居与 no-subtask 冲突安全拒绝、DB UPDATE 后再做 FS rename。system prompt 明确当前 task_dir 禁走 shell/run_python 直接改名，定时 run 不挂该工具；无新增 schema/migration。全量 357 测试通过（17 项按环境跳过），DB 路由测试未设 `ZCBOT_TEST_DB_URL` 时按安全门控跳过。

core/__init__.py

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

core/loop.py

@@ -245,8 +245,19 @@ class AgentLoop:
             })
 
     def run(self, user_message: str) -> str:
+        """运行一个尚未落库的用户轮次（CLI、渠道与调度任务的兼容入口）。"""
+        return self._run(user_message)
+
+    def run_persisted_turn(self) -> str:
+        """运行已由调用方原子持久化的用户轮次（Web POST 入口）。"""
+        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)
+
+    def _run(self, user_message: Optional[str]) -> str:
         self._maybe_fold_context()
-        self.session.append({"role": "user", "content": user_message})
+        if user_message is not None:
+            self.session.append({"role": "user", "content": user_message})
 
         for _ in range(self.max_iterations):
             if self._is_cancelled():

tests/test_loop_persisted_turn.py

@@ -0,0 +1,65 @@
+"""AgentLoop 已持久化用户轮次入口的回归测试。"""
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+from uuid import uuid4
+
+from core.loop import AgentLoop
+
+
+class _Session:
+    def __init__(self, messages=None):
+        self.messages = list(messages or [])
+        self.appended = []
+
+    def append(self, message):
+        self.messages.append(message)
+        self.appended.append(message)
+        return uuid4()
+
+
+def _loop(session: _Session) -> AgentLoop:
+    loop = AgentLoop(
+        llm=MagicMock(),
+        executor=MagicMock(),
+        session=session,
+        capabilities=SimpleNamespace(max_iterations=1),
+        user_id=uuid4(),
+        working_dir=Path("."),
+        cancel_check=lambda: True,
+    )
+    loop._maybe_fold_context = MagicMock()
+    return loop
+
+
+class PersistedTurnTests(unittest.TestCase):
+    def test_persisted_turn_does_not_append_duplicate_user_message(self) -> None:
+        session = _Session([{"role": "user", "content": "已落库"}])
+        result = _loop(session).run_persisted_turn()
+
+        self.assertEqual(result, "[cancelled]")
+        self.assertEqual(session.appended, [])
+        self.assertEqual(session.messages, [{"role": "user", "content": "已落库"}])
+
+    def test_persisted_turn_requires_latest_user_message(self) -> None:
+        session = _Session([{"role": "assistant", "content": "旧回复"}])
+
+        with self.assertRaisesRegex(RuntimeError, "latest message"):
+            _loop(session).run_persisted_turn()
+
+    def test_legacy_run_still_appends_user_message(self) -> None:
+        session = _Session()
+        result = _loop(session).run("新消息")
+
+        self.assertEqual(result, "[cancelled]")
+        self.assertEqual(
+            session.appended,
+            [{"role": "user", "content": "新消息"}],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_rename_working_dir.py

@@ -58,6 +58,40 @@ class RenameWorkingDirToolTests(unittest.TestCase):
 
 
 class DeferredRenameWorkerTests(unittest.TestCase):
+    def test_persisted_web_turn_uses_non_appending_agent_entry(self) -> None:
+        from web import runs
+
+        tid = uuid4()
+        uid = uuid4()
+        agent = SimpleNamespace(
+            run=MagicMock(return_value="wrong entry"),
+            run_persisted_turn=MagicMock(return_value="ok"),
+            deferred_actions=DeferredTaskActions(),
+            sink=None,
+        )
+
+        @contextmanager
+        def fake_scope():
+            yield SimpleNamespace(execute=MagicMock())
+
+        with tempfile.TemporaryDirectory() as tmp:
+            task_dir = Path(tmp)
+            with (
+                patch("core.agent_builder.build_agent", return_value=(
+                    agent, MagicMock(), str(tid), MagicMock(), task_dir,
+                )),
+                patch("core.agent_builder.sync_task_tokens"),
+                patch.object(runs, "session_scope", fake_scope),
+                patch.object(runs, "broker", MagicMock()),
+            ):
+                runs.run_agent_bg(
+                    tid, uid, "已落库",
+                    user_message_persisted=True,
+                )
+
+        agent.run_persisted_turn.assert_called_once_with()
+        agent.run.assert_not_called()
+
     def test_normal_run_renames_after_status_commit_before_done(self) -> None:
         from web import runs
 

tests/test_web_routes_db.py

@@ -10,18 +10,21 @@
   move 被引用→409、递归删被引用→409、软删后放行
 - upload(磁盘配额 gate 路径)+ 根目录列表(system_wd_names DB 查询)
 - clear / cancel 的状态闸;schedules 404 路径;/v1/models 档位过滤
+- messages 参数失败不改状态；202 前用户消息已持久化且 worker 不重复写入
 
 隔离:测试专属 User 行 + 随机 uid 子树,teardown DB 行与 FS 整树删除;
-POST messages / optimize_prompt(会起真 LLM)不在此测。
+optimize_prompt(会起真 LLM)不在此测。
 """
 from __future__ import annotations
 
 import os
 import shutil
 import sys
+import threading
 import unittest
 import uuid
 from pathlib import Path
+from unittest.mock import patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
@@ -59,6 +62,8 @@ if _DB_OK:
     from web.auth import AuthConfig, mint_token
 
     _app = create_app()
+    # 不进入 lifespan，手工补消息路由登记后台任务所需的运行态容器。
+    _app.state.inflight = {}
     _client = TestClient(_app)   # 不进 with:不跑 lifespan
     _UID = uuid.uuid4()
     _TOKEN, _ = mint_token(AuthConfig.from_env(), _UID)
@@ -186,6 +191,84 @@ class TasksCrudTests(unittest.TestCase):
         self.assertEqual(_client.get(f"/v1/tasks/{tid}/outline", headers=_AUTH).json()["items"], [])
 
 
+class MessageRunDurabilityTests(unittest.TestCase):
+    def _mk_task(self, name: str) -> str:
+        r = _client.post("/v1/tasks", json={"name": name}, headers=_AUTH)
+        self.assertEqual(r.status_code, 201, r.text)
+        return r.json()["task_id"]
+
+    def test_validation_failure_keeps_task_idle_and_writes_no_message(self):
+        from fastapi import HTTPException
+        from sqlalchemy import select
+
+        tid = self._mk_task("非法媒体参数")
+        with patch(
+            "web.routers.messages.resolve_image_model",
+            side_effect=HTTPException(400, "invalid image model"),
+        ):
+            r = _client.post(
+                f"/v1/tasks/{tid}/messages",
+                json={"content": "不会落库", "image_model": "invalid"},
+                headers=_AUTH,
+            )
+
+        self.assertEqual(r.status_code, 400, r.text)
+        with session_scope() as s:
+            task = s.execute(
+                select(Task.run_status).where(Task.task_id == uuid.UUID(tid))
+            ).scalar_one()
+            messages = s.execute(
+                select(Message).where(Message.task_id == uuid.UUID(tid))
+            ).scalars().all()
+        self.assertEqual(task, "idle")
+        self.assertEqual(messages, [])
+
+    def test_message_is_committed_before_worker_and_not_duplicated(self):
+        from sqlalchemy import select
+
+        tid = self._mk_task("消息持久化")
+        seen = {}
+        worker_done = threading.Event()
+
+        def fake_worker(task_id, user_id, user_message, *args, **kwargs):
+            with session_scope() as s:
+                payloads = s.execute(
+                    select(Message.payload)
+                    .where(Message.task_id == task_id)
+                    .order_by(Message.idx)
+                ).scalars().all()
+            seen["payloads"] = payloads
+            seen["persisted"] = kwargs.get("user_message_persisted")
+            worker_done.set()
+
+        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),
+        ):
+            r = _client.post(
+                f"/v1/tasks/{tid}/messages",
+                json={"content": "必须先落库"},
+                headers=_AUTH,
+            )
+            self.assertTrue(worker_done.wait(5), "后台 worker 未启动")
+
+        self.assertEqual(r.status_code, 202, r.text)
+        self.assertTrue(seen["persisted"])
+        self.assertEqual(
+            seen["payloads"],
+            [{"role": "user", "content": "必须先落库"}],
+        )
+        with session_scope() as s:
+            payloads = s.execute(
+                select(Message.payload)
+                .where(Message.task_id == uuid.UUID(tid))
+                .order_by(Message.idx)
+            ).scalars().all()
+        self.assertEqual(payloads, [{"role": "user", "content": "必须先落库"}])
+        _set_run_status(tid, "idle")
+
+
 class FilesDbAwareTests(unittest.TestCase):
     """顶层目录 = task.working_dir 的 DB-aware 分支(§7.4 唯一 mutation 入口)。"""
 

web/routers/messages.py

@@ -207,6 +207,10 @@ def register_message_routes(app, *, require_user) -> None:
         content, _sc_hit = await asyncio.to_thread(_shortcuts.expand, _ws, user_id, content)
         if _sc_hit:
             print(f"[shortcut] {str(user_id)[:8]} '{_sc_hit}' expanded")
+        # 所有可能返回 4xx 的 variant 校验必须发生在事务写入之前。否则 task 已提交
+        # 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(
@@ -248,22 +252,49 @@ 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)
-        # image_model / video_model 在 POST 时校验,避免 BG 线程里抛在 sink 之外难追;空串透传不查 yaml。
-        # 显式选中非空 variant 时过档位门控(档外 → 403);空串不查(走 yaml 默认)。
-        image_variant = resolve_image_model(body.image_model, user_id=user_id)
-        video_variant = resolve_video_model(body.video_model, user_id=user_id)
         broker.start(tid)  # 清上一轮 done 标记,新订阅者才能看到流式
         # commit 后 lock 释放;BG 线程接管(sink 通过 broker 把 event 桥回 asyncio loop)。
         # 登记到 app.state.inflight:① 关停 drain 时 await 它收尾 ② 持强引用防 task 被 GC
         # 中途回收(asyncio.create_task 不留引用是已知坑)。done 回调自摘除。
-        run_task = asyncio.create_task(asyncio.to_thread(
+        run_coro = asyncio.to_thread(
             run_agent_bg, tid, user_id, content, image_variant, video_variant,
-        ))
+            user_message_persisted=True,
+        )
+        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)
+            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))
         # 快速入口只在首条消息时 pending=true。辅助标题与主 run 并行，不阻塞

web/runs.py

@@ -26,6 +26,7 @@ def run_agent_bg(
     task_id: UUID, user_id: UUID, user_message: str,
     image_variant: str = "", video_variant: str = "",
     scheduled: bool = False,
+    *, user_message_persisted: bool = False,
 ) -> None:
     """工作线程:`build_agent(resume=True)` → 装 WebEventSink + cancel_check → `agent.run` → 写 tasks.run_status。
 
@@ -40,6 +41,8 @@ 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，避免重复落库。其余入口保持旧行为。
     """
     from core.agent_builder import build_agent, sync_task_tokens
     cancel_check = lambda tid=task_id: broker.is_cancelled(tid)
@@ -53,7 +56,11 @@ def run_agent_bg(
             scheduled_run=scheduled,
         )
         agent.sink = WebEventSink(broker, task_id)
-        result = agent.run(user_message)
+        result = (
+            agent.run_persisted_turn()
+            if user_message_persisted
+            else agent.run(user_message)
+        )
         sync_task_tokens(task_state)
         # 收尾终态:agent.run 在任一取消路径都 return "[cancelled]"(loop.py)——
         # 用户停止 → 落持久 cancelled(前端 renderPersistedRunTerminal 据此补「已停止」卡,