diff --git a/CHANGELOG.md b/CHANGELOG.md index 0e3849c..5ea9723 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ > 开发中的用户文案可先写入 `## Unreleased`；该区不会被前端解析，正式发布时再替换为数字版本和日期。 > 工程口径的完整记录见 `PROGRESS.md` / git log。 +## Unreleased + +- 修复深度思考或事件流重连期间点击“停止”偶尔没有反应的问题；停止信号会在状态落库前立即送达，重复停止也会可靠补发。 + ## 0.71.0 — 2026-09-03 - 提高工具参数异常时的自动恢复率，产物发布和工具健康分类也更加稳定准确。 diff --git a/DESIGN.md b/DESIGN.md index ab46ef1..0ac90f7 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -174,7 +174,7 @@ Eval 与生产 core 解耦，通过现有 `/v1` API 创建专用任务、监听 **无感部署(蓝绿,0.41)**:生产双 systemd 实例(`ZCBOT_INSTANCE=blue/green`)+ nginx upstream 切流(对外仍 8765),取代单实例 restart 的 503 窗口。双实例并存的互踩由三件事消解:`tasks.run_owner`(0020)让 reaper 只收自己色、sandbox 容器名/label 带实例色各管各的、微信长轮询 PG advisory lock 选主;定时任务 claim 本就 SKIP LOCKED 天然安全。部署编排在 `deploy/update_bluegreen.sh`(流程/兜底见 RUN)。 -**broker 外置(Redis pub/sub,✅ 0.42)**:0.41 落地时 event/cancel broker 留在进程内,代价是切换窗口内"刷新看不到旧实例 run 直播 / 停止送达不到"两个边缘。2026-07-06 重评(用户量已非个位数)决定实施 —— ①部署时总有 in-flight run,窗口边缘从偶发变常态;②单进程逼近天花板后,最近的扩容手段是稳态双实例同时接流量,broker 外置是它的硬前提(否则 POST 与 SSE 落不同实例直播全瞎)。选 Redis 不选 PG LISTEN/NOTIFY(token 级 delta 全过主库太吵:NOTIFY 全局队列 + 8KB payload 限制,把实时路径耦到 PG)、不选 nginx sticky hash(upstream 变更时 hash 重排,恰在部署窗口失效,治标)。实现(`web/broker.py`,LocalRunBroker/RedisRunBroker 同接口鸭子类型):`ZCBOT_REDIS_URL` env 开关,不设即进程内(dev 零影响);event 走 `PUBLISH zcbot:ev:` + 单条 async pubsub reader 路由到本地订阅 queue(Redis 只管跨进程一跳,fan-out 最后一公里仍在进程内);done = SETEX key(60s)+ publish 双通道,订阅先挂 channel 再查 key 不漏;cancel = SETEX/GET/DEL key,loop 在 chunk 间 poll(localhost ~0.1ms)。取消另以共享 PG 中已经提交的 `tasks.run_status='cancelling'` 作可靠兜底：运行线程每秒低频读取一次，未配置 Redis 的蓝绿窗口或 Redis 运行中故障时仍可跨实例停止；PG 不承载 token 直播事件。容错纪律:启动 ping 不通 fail-fast(同 sandbox init),运行中失败 10s 节流 log + 降级丢帧,reader 1s 退避重连。Redis 不解决的:run 本体仍绑进程(进程死 run 死,drain/reaper 不变)、线程池上限(调 `ZCBOT_RUN_MAX_WORKERS` 的事)。 +**broker 外置(Redis pub/sub,✅ 0.42)**:0.41 落地时 event/cancel broker 留在进程内,代价是切换窗口内"刷新看不到旧实例 run 直播 / 停止送达不到"两个边缘。2026-07-06 重评(用户量已非个位数)决定实施 —— ①部署时总有 in-flight run,窗口边缘从偶发变常态;②单进程逼近天花板后,最近的扩容手段是稳态双实例同时接流量,broker 外置是它的硬前提(否则 POST 与 SSE 落不同实例直播全瞎)。选 Redis 不选 PG LISTEN/NOTIFY(token 级 delta 全过主库太吵:NOTIFY 全局队列 + 8KB payload 限制,把实时路径耦到 PG)、不选 nginx sticky hash(upstream 变更时 hash 重排,恰在部署窗口失效,治标)。实现(`web/broker.py`,LocalRunBroker/RedisRunBroker 同接口鸭子类型):`ZCBOT_REDIS_URL` env 开关,不设即进程内(dev 零影响);event 走 `PUBLISH zcbot:ev:` + 单条 async pubsub reader 路由到本地订阅 queue(Redis 只管跨进程一跳,fan-out 最后一公里仍在进程内);done = SETEX key(60s)+ publish 双通道,订阅先挂 channel 再查 key 不漏;cancel 在鉴权后先置本实例 Event，再 SETEX/GET/DEL key 跨实例传递，loop 在 chunk 间优先 poll 本地信号(通常约 100ms 内退出，不等待 Redis 或 DB 往返)。取消另以共享 PG 中已经提交的 `tasks.run_status='cancelling'` 作可靠兜底：运行线程每秒低频读取一次，未配置 Redis 的蓝绿窗口或 Redis 运行中故障时仍可跨实例停止；PG 不承载 token 直播事件。容错纪律:启动 ping 不通 fail-fast(同 sandbox init),运行中失败 10s 节流 log + 降级丢帧,reader 1s 退避重连。Redis 不解决的:run 本体仍绑进程(进程死 run 死,drain/reaper 不变)、线程池上限(调 `ZCBOT_RUN_MAX_WORKERS` 的事)。 ### 7.1 心智模型:Task 一等公民 + Dir 文件副视图 diff --git a/tests/frontend_chat_queue.test.mjs b/tests/frontend_chat_queue.test.mjs index 74266f7..f06449a 100644 --- a/tests/frontend_chat_queue.test.mjs +++ b/tests/frontend_chat_queue.test.mjs @@ -55,12 +55,15 @@ test("composer uses one Codex-style action button for stop and queued send", () assert.match(html, /\.msg\.queued/); assert.match(chat, /card\.className = "msg user queued"/); assert.match(chat, /function composerHasPayload\(\)/); + assert.match(chat, /function isCurrentTaskRunActive\(\)/); assert.match(chat, /btn\.textContent = "停止"/); assert.match(chat, /if \(composerHasPayload\(\)\)[\s\S]*queueCurrentMessage\(\)/); assert.match(chat, /_actionMode === "loading" \|\| _actionMode === "submitting"[\s\S]*btn\.disabled = true/); assert.match(chat, /setActionMode\("submitting"\)/); assert.match(chat, /if \(state\.taskId === taskId\) \$\("chat-input"\)\.value = ""/); assert.match(chat, /if \(state\.taskId === taskId\) setActionMode\("streaming"\)/); + assert.match(chat, /const activeRun = isCurrentTaskRunActive\(\)/); + assert.match(chat, /else if \(activeRun\)[\s\S]*cancelCurrentTask\(\)/); assert.match(chat, /function queueCurrentMessage\(\)/); assert.match(chat, /async function dispatchNextQueuedMessage\(taskId\)/); assert.match(chat, /void dispatchNextQueuedMessage\(ctx\.taskId\)/); diff --git a/tests/test_broker_redis.py b/tests/test_broker_redis.py index 6f3aa34..2c3dc97 100644 --- a/tests/test_broker_redis.py +++ b/tests/test_broker_redis.py @@ -10,6 +10,7 @@ 用两个 broker 实例共享同一 FakeServer 模拟蓝绿两进程。 """ import asyncio +import threading import unittest from uuid import uuid4 @@ -134,8 +135,36 @@ class TestRedisBroker(unittest.TestCase): b.emit(tid, {"type": "text", "delta": "x"}) # 不抛 self.assertFalse(b.is_cancelled(tid)) # 降级 False b.request_cancel(tid) # 不抛 + self.assertFalse(b.is_cancelled(tid)) b.close(tid) # 不抛 + def test_cancel_uses_local_signal_while_redis_write_is_blocked(self): + """Redis 卡顿时，同实例 run 无需等待网络超时即可看到停止。""" + class SlowBroken: + def __init__(self): + self.writing = threading.Event() + self.release = threading.Event() + + def set(self, *args, **kwargs): + self.writing.set() + self.release.wait(2) + raise ConnectionError("redis down") + + def get(self, *args, **kwargs): + raise ConnectionError("redis down") + + redis = SlowBroken() + b = RedisRunBroker("redis://fake:6379/0", sync_client=redis, + async_client=fakeredis.aioredis.FakeRedis(decode_responses=True)) + tid = uuid4() + worker = threading.Thread(target=b.request_cancel, args=(tid,), daemon=True) + worker.start() + self.assertTrue(redis.writing.wait(1)) + self.assertTrue(b.is_cancelled(tid)) + redis.release.set() + worker.join(1) + self.assertFalse(b.is_cancelled(tid)) + def test_bind_loop_fail_fast_when_unreachable(self): class Boom: def ping(self): diff --git a/tests/test_web_routes_db.py b/tests/test_web_routes_db.py index 6ee8591..cff03d5 100644 --- a/tests/test_web_routes_db.py +++ b/tests/test_web_routes_db.py @@ -194,8 +194,16 @@ class TasksCrudTests(unittest.TestCase): tid = _client.post("/v1/tasks", json={"name": "状态闸任务"}, headers=_AUTH).json()["task_id"] # idle 时 cancel → 409 self.assertEqual(_client.post(f"/v1/tasks/{tid}/cancel", headers=_AUTH).status_code, 409) - # running 时 clear → 409;回 idle 后 clear → 200 且归零 + # running 时 cancel 可重试：第一次切 cancelling，第二次幂等补发信号。 _set_run_status(tid, "running") + with patch("web.routers.messages.broker.request_cancel") as request_cancel: + first = _client.post(f"/v1/tasks/{tid}/cancel", headers=_AUTH) + second = _client.post(f"/v1/tasks/{tid}/cancel", headers=_AUTH) + self.assertEqual(first.status_code, 202, first.text) + self.assertEqual(second.status_code, 202, second.text) + self.assertEqual(request_cancel.call_count, 2) + + # cancelling 时 clear → 409;回 idle 后 clear → 200 且归零 self.assertEqual(_client.post(f"/v1/tasks/{tid}/clear", headers=_AUTH).status_code, 409) _set_run_status(tid, "idle") d = _client.post(f"/v1/tasks/{tid}/clear", headers=_AUTH).json() diff --git a/web/broker.py b/web/broker.py index afa3329..148503d 100644 --- a/web/broker.py +++ b/web/broker.py @@ -173,6 +173,9 @@ class RedisRunBroker: self._pubsub: Any = None self._reader_task: Optional[asyncio.Task] = None self._subs: dict[UUID, set[asyncio.Queue]] = defaultdict(set) + # Redis 写入期间的同实例瞬时旁路：HTTP 线程先置 Event，run 线程下一次 + # 轮询立即命中；写入结束即清，Redis key 负责跨实例传递和持久兜底。 + self._cancel_flags: dict[UUID, threading.Event] = {} self._loop: Optional[asyncio.AbstractEventLoop] = None self._last_err_log = 0.0 @@ -345,14 +348,23 @@ class RedisRunBroker: # ─────────────── cancel signaling ─────────────── def request_cancel(self, task_id: UUID) -> None: + self._cancel_flags.setdefault(task_id, threading.Event()).set() try: self._redis.set(f"{self._CANCEL_PREFIX}{task_id}", "1", ex=self._CANCEL_TTL) except Exception as e: self._log_err("request_cancel", e) + finally: + self._cancel_flags.pop(task_id, None) def is_cancelled(self, task_id: UUID) -> bool: - """AgentLoop 在 stream chunk 间 poll(工作线程);redis 挂了返 False 降级 - (cancel 暂不可达,等价于 redis 前的跨实例行为,run 不受影响)。""" + """AgentLoop 在 stream chunk 间 poll；同实例先读 Event，跨实例再读 Redis。 + + Redis 运行中故障时跨实例快信号不可达，但同实例 Event 与共享 DB fallback + 仍可取消，不影响正常 run。 + """ + local = self._cancel_flags.get(task_id) + if local is not None and local.is_set(): + return True try: return self._redis.get(f"{self._CANCEL_PREFIX}{task_id}") is not None except Exception as e: @@ -360,6 +372,7 @@ class RedisRunBroker: return False def clear_cancel(self, task_id: UUID) -> None: + self._cancel_flags.pop(task_id, None) try: self._redis.delete(f"{self._CANCEL_PREFIX}{task_id}") except Exception as e: diff --git a/web/routers/messages.py b/web/routers/messages.py index b5631d0..d0abe3b 100644 --- a/web/routers/messages.py +++ b/web/routers/messages.py @@ -349,8 +349,11 @@ def register_message_routes(app, *, require_user) -> None: """向当前 task 的活跃 run 发协作式 cancel 信号。 - 单活 run 形态下"取消当前活动"语义无歧义;客户端只需 task_id - 校验 task 归属 user;否则 404 - - tasks.run_status 不是 `running` → 409(idle / cancelling / error 都不能 cancel) - - 标 `cancelling`(过渡态),BG 线程 loop 在 stream chunk 间 + 工具调用之间 poll 看见即退; + - `running` → 标 `cancelling`;已经是 `cancelling` 时幂等重发 cancel 信号 + - 其他终态 → 409 + - 鉴权和状态校验后先发 broker 快信号，再持久化 `cancelling` 作为跨实例兜底； + 不让事务提交延迟模型停止 + - BG 线程 loop 在 stream chunk 间 + 工具调用之间 poll 看见即退; 退出后 finally 写终态(正常→idle,异常→error) - LLM 走 streaming,cancel 延迟 ~ 单 chunk 间隔(100ms 级) """ @@ -362,19 +365,23 @@ def register_message_routes(app, *, require_user) -> None: row = s.execute( select(Task.run_status, Task.title_source) .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 != "running": + if row.run_status not in ("running", "cancelling"): raise HTTPException( 409, f"task not running (run_status={row.run_status}); cannot cancel", ) - s.execute( - update(Task).where(Task.task_id == tid).values(run_status="cancelling") - ) - broker.request_cancel(tid) + # 先走低延迟 broker。DB UPDATE/commit 即使被短事务阻塞，当前模型读取线程 + # 也能先在约 100ms 的轮询周期内退出；cancelling 列继续承担跨实例可靠兜底。 + broker.request_cancel(tid) + if row.run_status == "running": + s.execute( + update(Task) + .where(Task.task_id == tid, Task.run_status == "running") + .values(run_status="cancelling") + ) return {"ok": True, "task_id": str(tid), "run_status": "cancelling"} # ───────────── Background procs(bg proc,DESIGN §8.12)───────────── diff --git a/web/static/js/chat.js b/web/static/js/chat.js index bacc762..00e246a 100644 --- a/web/static/js/chat.js +++ b/web/static/js/chat.js @@ -1366,6 +1366,16 @@ function isCurrentTaskStreaming() { return !!getLiveRun(state.taskId); } +function isCurrentTaskRunActive() { + if (isCurrentTaskStreaming()) return true; + const meta = state.taskMeta; + return !!( + meta + && meta.task_id === state.taskId + && (meta.run_status === "running" || meta.run_status === "cancelling") + ); +} + // 直播卡片内文字按「轮次」分段:每段一个 .body,工具调用会关闭当前段,之后的新文字 // 在卡片底部另起一段 —— 使流式文字与工具卡按时序穿插、最新文字始终贴在底部可见。 // 历史渲染天然按消息分段,直播这样分段后两态结构一致,run 结束 reload 无跳变。 @@ -2079,11 +2089,14 @@ export async function openSoftwareJobResults(taskId, outputDir) { } function chatAction() { - const busy = isCurrentTaskStreaming() || hasRunningProc(state.taskId); + // SSE 重连 / 收尾交错时 liveRuns 可能短暂缺席，但 task 的持久状态仍是 running。 + // 按钮既然处于运行态，点击空 composer 必须继续走 cancel，不能退化成“发送空消息”。 + const activeRun = isCurrentTaskRunActive(); + const busy = activeRun || hasRunningProc(state.taskId); if (busy) { if (composerHasPayload()) { queueCurrentMessage(); - } else if (isCurrentTaskStreaming()) { + } else if (activeRun) { cancelCurrentTask(); } else { killTaskProcs(state.taskId); @@ -3160,21 +3173,44 @@ async function sendMessage(overrideText) { async function cancelCurrentTask() { const run = getLiveRun(state.taskId); - if (!state.taskId || !run) return; - run.cancelling = true; + const taskId = state.taskId; + if (!taskId || !isCurrentTaskRunActive()) return; + if (run) run.cancelling = true; + if (state.taskMeta && state.taskMeta.task_id === taskId) { + state.taskMeta.run_status = "cancelling"; + } setActionMode("cancelling"); - syncTaskRowRunIndicator(state.taskId); + syncTaskRowRunIndicator(taskId); $("chat-hint").textContent = "停止中…"; try { - await api("POST", `/v1/tasks/${state.taskId}/cancel`); + await api("POST", `/v1/tasks/${taskId}/cancel`); // 不重置 streaming / 按钮 — 等 SSE 的 cancelled / done 走完一并清 } catch (e) { if (e.status === 401) { logout(); return; } - // 409 = 已结束 / 已 cancelling,不算错;其他贴 toast - if (e.status !== 409) appendErrorCard("cancel: " + e.message); - run.cancelling = false; + // 409 只可能来自滚动升级中的旧后端：回读真实状态，仍在取消就保持停止中， + // 已终止则让现有 SSE/任务刷新接管。网络错误才恢复为可重试的“停止”。 + if (e.status === 409) { + try { + const meta = await api("GET", `/v1/tasks/${taskId}`); + if (state.taskId !== taskId) return; + state.taskMeta = meta; + if (meta.run_status === "cancelling") return; + if (meta.run_status !== "running") { + if (run) discardInactiveLiveRun(taskId); + else setActionMode("idle"); + $("chat-hint").textContent = ""; + return; + } + } catch (_) { /* 下面恢复按钮，允许用户再次尝试 */ } + } else { + appendErrorCard("cancel: " + e.message); + } + if (run) run.cancelling = false; + if (state.taskMeta && state.taskMeta.task_id === taskId) { + state.taskMeta.run_status = "running"; + } setActionMode("streaming"); - syncTaskRowRunIndicator(run.taskId); + syncTaskRowRunIndicator(taskId); $("chat-hint").textContent = "接收中…"; } }