From 7cfeebfa89144400abe1de8f8ab475006c21ffbb Mon Sep 17 00:00:00 2001 From: caoqianming Date: Mon, 27 Jul 2026 17:50:31 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=BC=BA=E6=B6=88=E6=81=AF=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E5=8F=AF=E9=9D=A0=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 4 ++ DESIGN.md | 5 +- PROGRESS.md | 3 +- core/__init__.py | 2 +- core/loop.py | 13 ++++- tests/test_loop_persisted_turn.py | 65 +++++++++++++++++++++++ tests/test_rename_working_dir.py | 34 +++++++++++++ tests/test_web_routes_db.py | 85 ++++++++++++++++++++++++++++++- web/routers/messages.py | 43 +++++++++++++--- web/runs.py | 9 +++- 10 files changed, 250 insertions(+), 13 deletions(-) create mode 100644 tests/test_loop_persisted_turn.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b0c8e85..9433d37 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。 > 工程口径的完整记录见 `PROGRESS.md` / git log。 +## 0.60.2 — 2026-07-27 + +- 提升发送消息的可靠性：服务在接收消息后即使遇到重启或后台任务调度异常，也不会静默丢失用户刚发送的内容；非法的生图、视频模型选项也不再导致任务卡在运行中。 + ## 0.60.1 — 2026-07-27 - 联网搜索现在会以当前北京时间判断时效：未指定时间时不再擅自给搜索词添加旧年份，查询“最新、当前、近期”内容时会采用合适的近期结果范围；用户明确指定的年份或时间窗保持优先。 diff --git a/DESIGN.md b/DESIGN.md index 40339cd..d80f058 100644 --- a/DESIGN.md +++ b/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 回收 | diff --git a/PROGRESS.md b/PROGRESS.md index 2a1616a..6091f4f 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(联网搜索动态时效约束,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` 时按安全门控跳过。 diff --git a/core/__init__.py b/core/__init__.py index 8278627..23eea16 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.60.1" +__version__ = "0.60.2" diff --git a/core/loop.py b/core/loop.py index 870f29f..75360e7 100644 --- a/core/loop.py +++ b/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(): diff --git a/tests/test_loop_persisted_turn.py b/tests/test_loop_persisted_turn.py new file mode 100644 index 0000000..4686464 --- /dev/null +++ b/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() diff --git a/tests/test_rename_working_dir.py b/tests/test_rename_working_dir.py index 46e9b41..033edf5 100644 --- a/tests/test_rename_working_dir.py +++ b/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 diff --git a/tests/test_web_routes_db.py b/tests/test_web_routes_db.py index c1853bc..ee0da9b 100644 --- a/tests/test_web_routes_db.py +++ b/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 入口)。""" diff --git a/web/routers/messages.py b/web/routers/messages.py index a288ab7..4259794 100644 --- a/web/routers/messages.py +++ b/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 并行，不阻塞 diff --git a/web/runs.py b/web/runs.py index 189ed60..271d10a 100644 --- a/web/runs.py +++ b/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 据此补「已停止」卡,