diff --git a/docs/todo/system-design/40-agent-chat-streaming-interrupt.md b/docs/todo/system-design/40-agent-chat-streaming-interrupt.md new file mode 100644 index 0000000..b9572c3 --- /dev/null +++ b/docs/todo/system-design/40-agent-chat-streaming-interrupt.md @@ -0,0 +1,206 @@ +# 40-agent-chat-streaming-interrupt. Agent Chat 打字机平滑 + 打断-排队 + 串台加固 + +> **日期**: 2026-08-30 +> **状态**: 📝 需求文档,讨论中,尚未实现 +> **触发**: 用户反馈"前端刷新不及时、没有打字机效果",怀疑要不要把传输层从 HTTP 换成 WebSocket +> 来解决打字机效果、多 Agent/多聊天切换、执行中无法打断这几个问题。经排查现状代码 + +> 对照 [hermes-webui](file:///Users/yibo/Documents/lemonAI/hermes-webui)、 +> [hermes-agent](file:///Users/yibo/Documents/lemonAI/hermes-agent)、 +> [deepseek-harness](file:///Users/yibo/Documents/lemonAI/deepseek-harness) 三个参考实现, +> 结论是**传输层不是瓶颈,不需要引入 WebSocket**——三个参考项目全部走 SSE/HTTP 流式, +> 没有一个用 WebSocket。真正的差距是另外三件独立的事,本文档逐一给方案。 + +--- + +## 起因 + +现状排查发现:LemonBoss 的 Agent Chat **已经是 SSE 流式**,不是普通 HTTP 一问一答—— +`web/src/views/AgentChat.vue:941` 用 `fetch().body.getReader()` 逐行解析 +`event:`/`data:`,后端 `server/routes/agent-chat.js:2542`(`POST /:agentId/chat`)用 +`setupSSE(res)` 把 Claude CLI 子进程吐出的 ACP `agent_message_chunk` 事件实时转发给前端。 +多聊天"串台"(切换 Agent 时旧流写串到新 Agent 状态里)已经在 `9b1c3f1` 修复:每个 Agent +一份独立的 `chatStates` reactive 状态 + `requestAgentId` 快照,事件处理函数(`handleEvent`, +`AgentChat.vue:991`)全程用 `isCurrent()` 校验"这条流是不是还对应当前选中的 Agent"。 + +对照 hermes-webui(`api/gateway_chat.py` 的 `_stream_writeback_is_current`)和 +deepseek-harness(`interruptByParent`:中止当前 turn → 新消息进 FIFO → 排队执行), +确认这套"per-stream id + is-current 校验"就是业界通用解法,LemonBoss 已经用对了模式, +不需要换传输层来重新解决一遍。 + +真正剩下的三个独立问题: + +1. **打字机不流畅**——不是传输问题,是 ACP chunk 粒度太粗(Claude CLI 一次吐一大段,不是 + 逐字/逐 token),SSE 本身能把这段文本秒级送达前端,但前端拿到手就是一大坨,没有"逐字吐出" + 的观感。 +2. **执行中无法打断并接上新消息**——现状是"先 Stop 再发"两步操作,且发送框在 + `chatState.sending` 为真时整个被 `disabled`(`AgentChat.vue:192`),物理上打不了字。 + deepseek-harness 有专门的打断-排队状态机,LemonBoss 没有。 +3. **串台已修,但没有回归测试兜底**——`9b1c3f1` 是一次性修复,`isCurrent()` 校验散落在 + `handleEvent` 十几个分支里,没有测试盯着,将来改动容易在某个分支漏掉校验又退回去。 + +--- + +## 一、现状代码地图(后续设计都基于这份) + +| 组件 | 位置 | 现状行为 | +|------|------|----------| +| 前端发送/接收 | `AgentChat.vue:908-969`(`handleSend`) | `fetch()` 开流,`st.sending=true` 期间 textarea `disabled`(`:192`),流结束才 `sending=false` | +| 前端事件分发 | `AgentChat.vue:991-1042`(`handleEvent`) | 按 `requestAgentId === agentId.value` 做 `isCurrent()` 校验,非当前 Agent 的流只落状态不刷 UI | +| 后端 SSE 入口 | `agent-chat.js:2542`(`POST /:agentId/chat`) | 每次请求 `while(loopCount<10)` 循环喂 prompt 给 ACP 协议,`onUpdate` 把 `agent_message_chunk` 原样透传成一条 SSE `agent_output` 事件 | +| 后端并发闸门 | `agent-chat.js:2552-2568`(`agentChatLocks`,键是 `agentId`) | 同一 Agent、**不同** convId 并发请求直接 `429` 拒绝;同一 Agent、**相同** convId 的并发请求**不会**被这段拒绝——第二个请求会直接覆盖锁往下跑,等于两个循环同时喂同一个 ACP 会话。目前靠前端 `disabled` 挡住,服务端没有闸门(见下文风险①) | +| 停止端点 | `agent-chat.js:1505`(`POST /:agentId/chat/:convId/stop`) | 调用 `closeSession(key, reason)`(`agent-chat.js:840`),即 `entry.protocol.disconnect()` 杀掉常驻 CLI 子进程,是"硬中止",不接排队 | +| 锁释放 | `agent-chat.js:2880`(`finally` 块) | 请求 handler 退出时统一 `agentChatLocks.delete(agentId)` + `res.end()`;中止后原请求是否会立刻走到这里,取决于 `protocol.disconnect()` 是否让 handler 内部 `await` 的 ACP 调用抛错——**需要在实现阶段用真实 CLI 验证一次**(见风险②) | + +--- + +## 二、方案一:打字机平滑(两端都做) + +**根因**:ACP `agent_message_chunk` 的粒度由 Claude CLI 决定,一次可能带几十到几百字符, +SSE 转发是逐 chunk 转发(`agent-chat.js:2628` `sendSSE('agent_output', ...)`),前端收到 +就是这一整段直接拼进 `st.streamContent`,没有二次拆分,所以观感是"一坨一坨蹦出来"而不是 +打字机。 + +### 2.1 后端:转发前按字符数切片 + +在 `agent-chat.js:2624` 的 `onUpdate` 里,把 `update.content.text` 按固定步长(比如 8-12 +字符一片)拆成多条 `sendSSE('agent_output', ...)`,而不是一次性整段发送。这是唯一需要碰 +后端的点,改动局限在这一个回调函数内,不影响日志落盘(`logMessageChunk` 仍然整段写,只是 +发给前端的部分被拆片)。 + +### 2.2 前端:渲染节流缓冲 + +在 `handleEvent` 收到 `agent_output` 时(`AgentChat.vue:1015`),不直接把整段 `data.message` +拼进 `st.streamContent`,改成推进一个逐字符渲染队列,用 `requestAnimationFrame` 或 +`setInterval` 按固定速率(比如 30-50 字符/秒)从队列取字符追加显示。这样即使后端某次真的 +一次性吐了一大段(比如网络抖动攒批),前端也能把它"抹平"成匀速的打字机效果,不依赖后端切片 +的粒度是否稳定。 + +两端都做的原因:后端切片解决"数据到达的最小单位太粗",前端缓冲解决"到达节奏不均匀"(网络 +延迟、CLI 输出突发)——只做一端在极端情况下(比如后端切了片但网络一次性把好几个 SSE 事件 +一起送到)还是会看起来一卡一卡。 + +--- + +## 三、方案二:打断-排队状态机 + +### 3.1 交互设计 + +保留现有 **Stop 按钮**(`AgentChat.vue:193`,⏹)不变——纯中止,点了就是杀会话,不接后续。 + +新增:**执行中解锁输入框**,允许在 `chatState.sending === true` 时继续打字。此时发送按钮 +的语义从"发送"变成"**打断并发送**",两个操作在视觉上要能一眼区分开——用户分不清点了哪个, +后果是"以为消息接上了,其实这轮直接被丢掉": + +- **Stop 按钮**(`AgentChat.vue:193`,⏹):图标/文案不变,补一句 tooltip「仅停止,不发送 + 新消息」。 +- **发送按钮在 sending 态下**:换一个视觉上明显不同的图标/配色(暖色调,区别于平时发送用的 + 主色),tooltip 改为「发送并打断当前回复」。 +- **打断成功后**,在对话记录里插入一条不打断阅读的**分隔标记**(细线 + 小字,类似 + "— 已打断上一条回复 —"),让用户回看聊天记录时能看懂"上一条为什么没说完"——这是 + ChatGPT / Claude.ai 自己在"stop and continue"场景下的通用做法。 + +点击"打断并发送"后的具体流程: + +1. 前端把发送按钮切到"打断中…"状态,锁住重复点击(防止手速快连点两次触发两轮打断); +2. 调用现有 `stop` 端点(复用 `agent-chat.js:1505`,不新增中止路由,但要改内部实现,见 + 3.1.1); +3. 根据 `stop` 返回的 `cleaned` 字段决定下一步(见 3.1.1),确认可以安全发送后,把这条新 + 消息按**正常发送流程**投出去,带**同一个 convId**,接续同一个对话历史,并在发送成功后 + 插入上面说的分隔标记。 + +这本质是"打断并立即顶替",但工程实现上做成一个**深度为 1 的排队**(先确认上一个 turn 真的 +已经退出,再发新的),不是前端拿到点击事件就立刻并发发第二个请求——原因见风险①: +服务端目前对"同一 convId 并发"没有闸门,真并发会导致两个循环同时写同一个 ACP 会话。 + +#### 3.1.1 `stop` 端点改造:后端确认清理完成,而不是前端猜时间 + +现状 `stop` 端点(`agent-chat.js:1505-1512`)调用 `closeSession` 后立刻同步返回,不代表 +原来那个 `/chat` 请求的 handler 已经真正退出(风险②)。改造成: + +- `stop` 端点内部在调用 `closeSession` 之后,轮询等待 `agentChatLocks` 里对应 `agentId` + 的锁被清掉(即原请求走到 `agent-chat.js:2880` 的 `finally`),**预算 3 秒**(`closeSession` + 杀子进程本身是毫秒级,慢的是原请求 `await` 被异常唤醒到走完 `finally` 这段,3 秒是给日志 + flush 这类边缘情况留的余量,正常路径应该用不到)。 +- 3 秒内确认清理完成 → 返回 `{ success: true, cleaned: true }`,前端立刻发新消息,用户基本 + 感觉不到延迟。 +- 3 秒仍未确认(极端情况) → 返回 `{ success: true, cleaned: false }`(尽力而为,不算失败), + 前端**不自动强发**(会撞回风险①的并发洞),而是在输入框附近给一条不遮挡输入的提示 + 「上一条还在收尾…」+ 一个「强制发送」按钮,把"要不要冒这个小概率风险"的决定权交还给用户, + 而不是系统默默赌一把。 + +### 3.2 需要新增/改动的点 + +| 位置 | 改动 | +|------|------| +| `AgentChat.vue:192` | textarea 的 `:disabled="chatState.sending"` 改成允许 sending 态下继续输入;`autoResize`/`handleKeydown` 不受影响 | +| `AgentChat.vue:194` | 发送按钮增加"sending 态下点击 = 打断并发送"分支,UI 上给一个过渡态(如按钮文案/图标切换) | +| `AgentChat.vue:handleSend`(`:908`) | 抽出"等待上一个 turn 收尾"的等待逻辑,供打断分支复用 | +| `agent-chat.js:2552-2568`(`agentChatLocks`) | **补上同 convId 并发闸门**(风险①,见下)——无论是否做打断功能,这都是一个已经存在、只是暂时没被前端触发到的并发 bug,建议顺带修 | +| `agent-chat.js:1505`(stop 端点) | 改成内部轮询确认清理完成再返回,加 `cleaned: true/false` 字段,见 3.1.1 | + +### 3.3 风险 / 需要在实现阶段验证的点 + +1. **同 convId 并发无闸门(现存 bug,非本次引入)**:`agent-chat.js:2554` 的判断是 + `existingLock.convId !== convId` 才拒绝,同 convId 的第二个请求会直接覆盖锁往下跑。 + 目前靠前端 `disabled` 属性挡住,从没被服务端真正堵过。一旦解锁输入框,这条路径变得 + 可达,必须先补服务端闸门,否则用户手速快一点就能让两个循环同时喂同一个 ACP 会话(未定义 + 行为,轻则消息乱序,重则 CLI 子进程状态损坏)。 +2. **中止后原请求是否会及时退出,还未验证**:`closeSession` 只是 `protocol.disconnect()`, + 原来那个还挂着 `await` 的 `POST /chat` 请求 handler 会不会因此立刻抛错走到 + `finally`(`:2880`)释放锁、结束响应,取决于 `ClaudeCliProtocol` 底层对子进程异常退出的 + 处理方式。**已拍板的应对方案见 3.1.1**(`stop` 端点内部轮询 + 3 秒预算 + `cleaned` 字段 + + 前端超时兜底按钮);实现阶段仍需要用一次真实 Claude CLI 会话验证「中止后原请求平均/最坏 + 多久会真正退出」,如果实测远超 3 秒(比如需要等底层网络超时),要回来调整这个预算数字, + 而不是改设计本身。 + +--- + +## 四、方案三:串台加固(回归测试) + +`9b1c3f1` 的修复方式是在十几个 `handleEvent` 分支里散落地加 `if (isCurrent) ...`,容易在 +未来新增分支时漏掉。建议补一个**前端组件级测试**(而不是走完整 E2E),思路: + +- 模拟两个 Agent 的 `chatStates`,对 Agent A 发起一次"慢"流(用假的 SSE 事件序列,中间插 + 延时),在流还没结束时把 `agentId.value` 切到 Agent B; +- 断言 Agent A 后续到达的事件(`thinking` / `agent_output` / `tool_call` 等每一种 SSE + 事件类型都要覆盖到)只更新了 `chatStates.get('A')`,不影响当前显示(Agent B 的视图); +- 反向断言:切回 Agent A 时,之前"憋住"没刷 UI 的内容应该已经正确落在 `chatStates.get('A')` + 里,切回来就能看到完整历史,不丢消息。 + +这条测试的价值不是证明现在没 bug(`9b1c3f1` 已经手工验证过),而是**卡住以后的回归**—— +后续如果给 `handleEvent` 加新的 SSE 事件类型,忘了包 `isCurrent()` 判断,测试会红。 + +--- + +## 五、不做的事(明确排除) + +- **不引入 WebSocket**。三个参考项目和 LemonBoss 现状都证明 SSE 足够,问题不在传输层。 + 真要引入 WebSocket 的场景是"没有聊天窗口打开时也要收到 Agent 的后台推送"(比如心跳异常 + 主动弹通知),这是一个不同的需求,不属于本文档范围,需要的话应该另开文档讨论。 +- **不做"多条消息同时排队"的完整 FIFO**。deepseek-harness 那套是为多 Agent + parent-child 场景设计的,LemonBoss 这里是单用户单会话打断,深度 1("打断当前的,接上 + 新的")已经覆盖需求,做完整队列是过度设计。 + +--- + +## 六、实施顺序建议 + +三个方案彼此独立,互不阻塞,建议顺序(从风险低/收益快到风险高): + +1. **方案三**(串台回归测试)——纯新增测试,零风险,先把安全网铺好。 +2. **方案一**(打字机平滑)——只碰渲染逻辑和一个转发回调,不改并发/状态机,风险最低。 +3. **方案二**(打断-排队)——涉及并发闸门(风险①必须先修)和状态机改造,且风险②需要真实 + CLI 验证,建议放最后,且实现前先用一次真实 Claude CLI 会话验证"中止后原请求多久会退出"。 + +## 七、已拍板的设计决策 + +以下两点原本是未决问题,已确认,写在这里避免以后重新讨论一遍: + +1. **超时策略**:由后端 `stop` 端点内部确认清理完成(轮询 `agentChatLocks`),预算 **3 秒**; + 超时不代表失败,返回 `cleaned: false`,前端不自动强发,展示「上一条还在收尾…」+ 手动 + 「强制发送」按钮,风险由用户显式承担,不做静默重试。详见 3.1.1。 +2. **UI 区分**:Stop 按钮和"sending 态下的发送按钮"必须视觉上可区分(图标/配色 + 不同 + tooltip),打断成功后在对话记录里插入一条分隔标记,便于事后回看。详见 3.1。 + +本文档目前没有遗留的未决问题;进入实现阶段后如果冒出新的(比如风险②实测的具体数字要调整 +预算),回来更新这一节。 diff --git a/server/routes/agent-chat.js b/server/routes/agent-chat.js index 7ea113e..d93b486 100644 --- a/server/routes/agent-chat.js +++ b/server/routes/agent-chat.js @@ -1502,13 +1502,18 @@ router.delete('/:agentId/conversations/:convId', (req, res) => { }); /** 停止当前正在生成的任务 */ -router.post('/:agentId/chat/:convId/stop', (req, res) => { +// 40 号文档方案二(3.1.1):中止后不能立刻同步返回——原来那个 /chat 请求的 handler +// 是否已经走到 finally(释放 agentChatLocks、res.end())还不确定,"打断并发送"要用这个 +// 信号判断能不能安全发第二条。这里轮询等最多 3 秒,超时不算失败,只是 cleaned:false, +// 交给前端决定要不要"强制发送"。 +router.post('/:agentId/chat/:convId/stop', async (req, res) => { const { agentId, convId } = req.params; const key = `${agentId}:${convId}`; - if (closeSession(key, '用户手动中断')) { - return res.json({ success: true, message: 'Agent 已中断' }); + if (!closeSession(key, '用户手动中断')) { + return res.json({ success: false, message: '未找到运行中的 Agent 进程' }); } - res.json({ success: false, message: '未找到运行中的 Agent 进程' }); + const cleaned = await waitForLockRelease(agentId); + res.json({ success: true, message: 'Agent 已中断', cleaned }); }); /** 读取 bootstrap file */ @@ -2537,7 +2542,17 @@ router.get('/:agentId/uploads/:convId/:filename', (req, res) => { // 同一 convId 的多轮对话不受限制(ACP 进程复用) // 32 号文档方案 B:这把锁挪到了共享模块 agent-session-lock.js,company.js 的招聘装配 // 重试逻辑也要用同一把——两条路径各自维护一份的话,各自都以为自己没有并发。 -const { agentChatLocks, AGENT_LOCK_TIMEOUT } = require('../src/core/agent-session-lock'); +const { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease } = require('../src/core/agent-session-lock'); + +// 40 号文档方案一:Claude CLI 一次 agent_message_chunk 可能带几十到几百字符,SSE 原样 +// 转发就是"一坨一坨蹦出来",没有打字机效果。这里按固定步长切成小片再各发一条 SSE 事件—— +// 只影响推给前端的粒度,`logMessageChunk` 落盘的仍是整段,不受影响。 +const TYPEWRITER_CHUNK_SIZE = 10; +function sliceForTypewriter(text, size = TYPEWRITER_CHUNK_SIZE) { + const pieces = []; + for (let i = 0; i < text.length; i += size) pieces.push(text.slice(i, i + size)); + return pieces; +} router.post('/:agentId/chat', async (req, res) => { const { message, conversationId: reqConvId } = req.body; @@ -2549,16 +2564,21 @@ router.post('/:agentId/chat', async (req, res) => { const convId = reqConvId || `conv-${Date.now().toString(36)}`; - // 🔒 并发互斥检查:同一 Agent 不同 convId 的并发请求被拒绝 + // 🔒 并发互斥检查:同一 Agent 的并发请求被拒绝——无论 convId 是否相同。 + // 40 号文档风险①:这里原来只挡"不同 convId",同一 convId 的第二个请求会直接 + // 覆盖锁往下跑,等于两个循环同时喂同一个 ACP 会话(未定义行为)。此前一直靠前端 + // textarea 的 disabled 属性挡住,从没被服务端真正堵过;"打断并发送"要解锁输入框, + // 这条路径就变得可达了,必须在这里把闸门补上。 const existingLock = agentChatLocks.get(agentId); - if (existingLock && existingLock.convId !== convId) { + if (existingLock) { const elapsed = Date.now() - existingLock.startedAt; if (elapsed < AGENT_LOCK_TIMEOUT) { console.log(`[AgentChat] 🔒 拒绝 ${agentId} 并发请求 — 已有活跃会话 ${existingLock.convId} (${(elapsed / 1000).toFixed(0)}s)`); return res.status(429).json({ error: `Agent ${agentId} 正在执行另一个会话 (${existingLock.convId}),请等待完成`, activeConvId: existingLock.convId, - elapsedMs: elapsed + elapsedMs: elapsed, + sameConversation: existingLock.convId === convId, }); } // 超时释放 @@ -2625,7 +2645,9 @@ router.post('/:agentId/chat', async (req, res) => { console.log(`[SSE-DEBUG] 📤 agent_output (${update.content.text.length} chars)`); fullOutput += update.content.text; logMessageChunk(agentId, convId, update.content.text); - sendSSE('agent_output', { agent: agentId, message: update.content.text }); + for (const piece of sliceForTypewriter(update.content.text)) { + sendSSE('agent_output', { agent: agentId, message: piece }); + } } else if (update?.sessionUpdate === 'agent_thought_chunk' && update.content?.text) { logMessageChunk(agentId, convId, update.content.text); sendSSE('thinking', { message: update.content.text }); diff --git a/server/src/core/agent-session-lock.js b/server/src/core/agent-session-lock.js index e42d66c..adcbcc7 100644 --- a/server/src/core/agent-session-lock.js +++ b/server/src/core/agent-session-lock.js @@ -14,4 +14,21 @@ const agentChatLocks = new Map(); // agentId → { convId, startedAt } const AGENT_LOCK_TIMEOUT = 30 * 60 * 1000; // 30 分钟安全阀,与原实现一致 -module.exports = { agentChatLocks, AGENT_LOCK_TIMEOUT }; +/** + * 40 号文档「打断-排队」用:中止一个 turn 之后,轮询等这把锁真正被清掉 + * (即原请求的 handler 走到 finally),而不是前端凭感觉睡一个固定时长。 + * 查不到锁本身也算"已清空"——用于 stop 端点在锁已经因为别的原因释放时仍能返回 cleaned:true。 + */ +function waitForLockRelease(agentId, budgetMs = 3000, pollMs = 100) { + return new Promise((resolve) => { + const deadline = Date.now() + budgetMs; + const check = () => { + if (!agentChatLocks.has(agentId)) return resolve(true); + if (Date.now() >= deadline) return resolve(false); + setTimeout(check, pollMs); + }; + check(); + }); +} + +module.exports = { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease }; diff --git a/server/tests/unit/agent-chat-interrupt.test.js b/server/tests/unit/agent-chat-interrupt.test.js new file mode 100644 index 0000000..055dddd --- /dev/null +++ b/server/tests/unit/agent-chat-interrupt.test.js @@ -0,0 +1,118 @@ +/** + * 40 号文档方案二的两条后端回归: + * + * 1. 风险①:同一 agentId + 同一 convId 的并发请求过去完全没有闸门(`agentChatLocks` 的判断 + * 只挡"不同 convId"),一直靠前端 textarea 的 disabled 属性挡住。"打断并发送"要解锁输入框, + * 这条路径就变得可达——两个循环同时喂同一个 ACP 会话是未定义行为。 + * 2. `stop` 端点现在要等锁真正释放(或 3 秒预算到期)才返回,`cleaned` 字段供前端判断 + * 能不能安全发第二条。 + */ +process.env.LEMONBOSS_CLAUDE_MODE = 'mock'; +process.env.AGENT_SESSION_IDLE_MS = '60000'; + +import { describe, it, expect, beforeAll, afterAll, afterEach, vi } from 'vitest'; +import fs from 'fs'; +import path from 'path'; +import express from 'express'; +import { createRequire } from 'node:module'; + +const require_ = createRequire(import.meta.url); +const ROOT = path.resolve(__dirname, '../../..'); +const AGENT = 'zz-chat-interrupt'; +const BODY = path.join(ROOT, 'system', 'agents', AGENT); + +function clearRuntimeState() { + for (const d of ['conversations', '_global', 'memory']) { + fs.rmSync(path.join(BODY, d), { recursive: true, force: true }); + } +} + +describe('agent-chat · 打断-排队(40 号文档)', () => { + let base, srv; + + beforeAll(async () => { + clearRuntimeState(); + fs.mkdirSync(BODY, { recursive: true }); + fs.writeFileSync(path.join(BODY, 'config.json'), JSON.stringify({ id: AGENT, type: 'skill', name: AGENT, status: 'active' }, null, 2)); + fs.writeFileSync(path.join(BODY, 'identity.md'), `# ${AGENT}\n`); + fs.writeFileSync(path.join(BODY, 'agent.md'), '# 打断-排队单测用的常驻本体\n'); + + const app = express(); + app.use(express.json()); + app.use('/api/agent', require_('../../routes/agent-chat')); + await new Promise((r) => { srv = app.listen(0, '127.0.0.1', () => { base = `http://127.0.0.1:${srv.address().port}`; r(); }); }); + }); + + afterAll(async () => { + if (srv) await new Promise((r) => srv.close(r)); + clearRuntimeState(); + }); + + afterEach(() => vi.restoreAllMocks()); + + const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + + it('同一 convId 的第二个请求在第一个还没跑完时会被 429 拒绝,不会两个循环同时跑', async () => { + const conv = 'race-same-conv'; + const { ClaudeCliProtocol } = require_('@mas/acp-core'); + const original = ClaudeCliProtocol.prototype.sendPrompt; + // 拖住第一个请求,制造一个真实的并发窗口 + vi.spyOn(ClaudeCliProtocol.prototype, 'sendPrompt').mockImplementation(function (...args) { + return sleep(300).then(() => original.apply(this, args)); + }); + + const first = fetch(`${base}/api/agent/${AGENT}/chat`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ message: 'first', conversationId: conv }), + }); + await sleep(50); // 确保 first 已经进了 handler、拿到了锁 + const second = await fetch(`${base}/api/agent/${AGENT}/chat`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ message: 'second', conversationId: conv }), + }); + + expect(second.status).toBe(429); + const body = await second.json(); + expect(body.sameConversation).toBe(true); + + const firstRes = await first; + await firstRes.text(); // 读完流,等 finally 跑完 + }); + + it('stop 之后:锁在预算内释放,返回 cleaned:true', async () => { + const conv = 'stop-cleaned'; + const chatDone = fetch(`${base}/api/agent/${AGENT}/chat`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ message: 'ping', conversationId: conv }), + }).then(r => r.text()); + + await sleep(50); // 等它先跑起来、拿到锁 + const stopRes = await fetch(`${base}/api/agent/${AGENT}/chat/${conv}/stop`, { method: 'POST' }); + const stopBody = await stopRes.json(); + expect(stopBody.success).toBe(true); + expect(stopBody.cleaned).toBe(true); + + await chatDone; + }); + + it('stop 之后立刻发同一个 convId 的新消息不会被 429 拒绝(cleaned 确认过锁已释放)', async () => { + const conv = 'stop-then-resend'; + const chatDone = fetch(`${base}/api/agent/${AGENT}/chat`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ message: 'ping', conversationId: conv }), + }).then(r => r.text()); + + await sleep(50); + const stopRes = await fetch(`${base}/api/agent/${AGENT}/chat/${conv}/stop`, { method: 'POST' }); + const { cleaned } = await stopRes.json(); + expect(cleaned).toBe(true); + + const resendRes = await fetch(`${base}/api/agent/${AGENT}/chat`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ message: 'resend', conversationId: conv }), + }); + expect(resendRes.status).toBe(200); + await resendRes.text(); + await chatDone; + }); +}); diff --git a/system/agents/zz-chat-interrupt/agent.md b/system/agents/zz-chat-interrupt/agent.md new file mode 100644 index 0000000..da566e4 --- /dev/null +++ b/system/agents/zz-chat-interrupt/agent.md @@ -0,0 +1 @@ +# 打断-排队单测用的常驻本体 diff --git a/system/agents/zz-chat-interrupt/config.json b/system/agents/zz-chat-interrupt/config.json new file mode 100644 index 0000000..6cc3644 --- /dev/null +++ b/system/agents/zz-chat-interrupt/config.json @@ -0,0 +1,6 @@ +{ + "id": "zz-chat-interrupt", + "type": "skill", + "name": "zz-chat-interrupt", + "status": "active" +} \ No newline at end of file diff --git a/system/agents/zz-chat-interrupt/identity.md b/system/agents/zz-chat-interrupt/identity.md new file mode 100644 index 0000000..a205872 --- /dev/null +++ b/system/agents/zz-chat-interrupt/identity.md @@ -0,0 +1 @@ +# zz-chat-interrupt diff --git a/vitest.config.js b/vitest.config.js index 7da9f66..5f9d9f0 100644 --- a/vitest.config.js +++ b/vitest.config.js @@ -4,7 +4,7 @@ export default defineConfig({ test: { globals: true, environment: 'node', - include: ['src/**/*.test.js', 'server/src/tests/**/*.test.js', 'server/tests/**/*.test.js', 'packages/**/*.test.js'], + include: ['src/**/*.test.js', 'server/src/tests/**/*.test.js', 'server/tests/**/*.test.js', 'packages/**/*.test.js', 'web/src/**/*.test.js'], coverage: { provider: 'v8', reporter: ['text', 'html'], diff --git a/web/src/composables/agentChatEvents.js b/web/src/composables/agentChatEvents.js new file mode 100644 index 0000000..dc51085 --- /dev/null +++ b/web/src/composables/agentChatEvents.js @@ -0,0 +1,131 @@ +/** + * 从 AgentChat.vue 抽出来的两块纯逻辑: + * + * 1. `createAgentChatEventHandler` —— SSE 事件分发器。只认 requestAgentId 是否等于 + * 调用时刻"当前选中的 agent",据此决定要不要触碰全局态(路由/对话列表/DOM 滚动); + * 传入的 `st`(每个 agent 一份的状态)永远无条件更新,不受这个判断影响——这样切走 + * 了不会打断/污染,切回来能看到完整的后台进度(9b1c3f1 修的就是这条不变式)。 + * 抽成不依赖组件实例的纯函数,是为了能在不挂载整个 Vue 组件的前提下对它做回归测试 + * (40 号文档方案三)。 + * + * 2. `createTypewriterQueue` / `flushTypewriter` —— 打字机渲染缓冲(40 号文档方案一)。 + * 后端已经把大段文本切成小片再发 SSE,这里再加一层匀速吐出,抹平网络到达节奏不均匀 + * (突发多个 SSE 事件挤在一起)的情况。 + */ + +const TYPEWRITER_TICK_MS = 20; +const TYPEWRITER_CHARS_PER_TICK = 2; + +/** 把新到的文本接到 st 的打字机队列尾部,并确保有一个定时器在匀速吐出 */ +export function createTypewriterQueue({ tickMs = TYPEWRITER_TICK_MS, charsPerTick = TYPEWRITER_CHARS_PER_TICK } = {}) { + return function pushChunk(st, text) { + if (!text) return; + st._typeQueue = (st._typeQueue || '') + text; + if (st._typeTimer) return; // 已经在吐了,接上队尾即可 + st._typeTimer = setInterval(() => { + if (!st._typeQueue) { + clearInterval(st._typeTimer); + st._typeTimer = null; + return; + } + st.streamContent += st._typeQueue.slice(0, charsPerTick); + st._typeQueue = st._typeQueue.slice(charsPerTick); + }, tickMs); + }; +} + +/** 立刻把队列里剩下的字符吐光,用在"这轮已经结束/被打断,不用再演匀速"的时刻 */ +export function flushTypewriter(st) { + if (st._typeTimer) { + clearInterval(st._typeTimer); + st._typeTimer = null; + } + if (st._typeQueue) { + st.streamContent += st._typeQueue; + st._typeQueue = ''; + } +} + +/** + * @param {() => string} getCurrentAgentId 读取"用户当前选中的 agent id" + * @param {(requestAgentId: string, data: object, st: object) => void} onNewConversation + * 仅在 isCurrent 时调用:处理路由跳转 + 对话列表插入这类全局副作用 + * @param {() => void} scrollToBottom + * @param {() => void} scrollAcpLog + * @param {() => void} scrollSubLogs + * @param {(st: object, text: string) => void} [pushChunk] 打字机队列的入队函数,默认新建一个 + */ +export function createAgentChatEventHandler({ + getCurrentAgentId, + onNewConversation, + scrollToBottom, + scrollAcpLog, + scrollSubLogs, + pushChunk = createTypewriterQueue(), +}) { + return function handleEvent(st, requestAgentId, type, data) { + const isCurrent = getCurrentAgentId() === requestAgentId; + switch (type) { + case 'session': + if (data.conversationId && !st.activeConvId) { + st.activeConvId = data.conversationId; + if (isCurrent) onNewConversation(requestAgentId, data, st); + } + break; + case 'thinking': + if (data.message && !data.message.startsWith('🏗️')) { + st.thinkingContent += data.message; + if (isCurrent) scrollToBottom(); + } else { + st.thinkingText = data.message || '思考中...'; + } + break; + case 'agent_output': + st.streaming = true; + pushChunk(st, data.message || ''); + if (isCurrent) scrollToBottom(); + break; + case 'tool_call': + st.thinkingText = data.message || `🔧 ${data.tool}`; + st.acpLog.push({ type: 'tool', icon: '🔧', text: `${data.tool}`, ts: Date.now() }); + if (isCurrent) scrollAcpLog(); + break; + case 'tool_input': + st.acpLog.push({ type: 'input', icon: '📥', text: data.input?.slice(0, 300) || '', ts: Date.now() }); + if (isCurrent) scrollAcpLog(); + break; + case 'tool_result': + st.acpLog.push({ type: 'result', icon: '📤', text: data.result?.slice(0, 300) || '', ts: Date.now() }); + if (isCurrent) scrollAcpLog(); + break; + case 'session_cost': + st.acpLog.push({ type: 'cost', icon: '💰', text: `$${data.cost?.toFixed(4)} (${((data.duration || 0) / 1000).toFixed(1)}s)`, ts: Date.now() }); + break; + case 'acp_raw': + if (data.method && !data.method.includes('content_block_delta') && !data.method.includes('message_start')) { + st.acpLog.push({ type: 'raw', icon: '⚡', text: data.method, ts: Date.now() }); + if (isCurrent) scrollAcpLog(); + } + break; + case 'sub_agent': + if (!st.subAgents[data.callId]) st.subAgents[data.callId] = { agent: data.agent, status: 'running', content: '', collapsed: false }; + if (data.event === 'chunk' && data.chunk) { + st.subAgents[data.callId].content += data.chunk; + if (isCurrent) scrollSubLogs(); + } else if (data.event === 'done') st.subAgents[data.callId].status = data.status; + else if (data.event === 'start') st.subAgents[data.callId].status = 'running'; + if (isCurrent) scrollToBottom(); + break; + case 'error': + st.messages.push({ role: 'assistant', content: `❌ ${data.message}`, timestamp: new Date().toISOString() }); + st.acpLog.push({ type: 'error', icon: '🚨', text: data.message, ts: Date.now() }); + st.thinkingText = ''; + st.thinkingContent = ''; + if (isCurrent) { + scrollToBottom(); + scrollAcpLog(); + } + break; + } + }; +} diff --git a/web/src/composables/agentChatEvents.test.js b/web/src/composables/agentChatEvents.test.js new file mode 100644 index 0000000..9f99b99 --- /dev/null +++ b/web/src/composables/agentChatEvents.test.js @@ -0,0 +1,129 @@ +/** + * 回归测试:9b1c3f1 修的"多聊天串台"不变式(40 号文档方案三)。 + * + * 不变式:handleEvent 收到的每一条事件都要无条件写进它自己的 st(每个 agent 一份), + * 但只有在事件所属的 requestAgentId 仍等于"用户当前选中的 agent"时,才允许触碰 + * 全局副作用(路由跳转 / 对话列表 / DOM 滚动)。这条测试不需要挂载整个 Vue 组件—— + * handleEvent 已经被抽成不依赖组件实例的纯函数(agentChatEvents.js),直接喂事件、 + * 断言 st 内容和副作用回调的调用次数即可。 + */ +import { describe, it, expect, vi } from 'vitest'; +import { createAgentChatEventHandler, createTypewriterQueue, flushTypewriter } from './agentChatEvents.js'; + +function makeState() { + return { + messages: [], activeConvId: null, + sending: false, streaming: false, streamContent: '', + subAgents: {}, subLogList: [], acpLog: [], + thinkingText: '', thinkingContent: '', showThinking: true, + }; +} + +describe('agentChatEvents · 串台加固', () => { + it('切走之后,旧 agent 的事件继续更新它自己的 st,但不再触发全局副作用', () => { + let current = 'agent-A'; + const scrollToBottom = vi.fn(); + const scrollAcpLog = vi.fn(); + const onNewConversation = vi.fn(); + const handleEvent = createAgentChatEventHandler({ + getCurrentAgentId: () => current, + onNewConversation, + scrollToBottom, + scrollAcpLog, + scrollSubLogs: vi.fn(), + pushChunk: (st, text) => { st.streamContent += text; }, // 跳过打字机定时器,测同步内容 + }); + + const stA = makeState(); + handleEvent(stA, 'agent-A', 'session', { conversationId: 'conv-A' }); + handleEvent(stA, 'agent-A', 'agent_output', { message: '你好' }); + expect(scrollToBottom).toHaveBeenCalledTimes(1); // 只有 agent_output 会调 scrollToBottom + expect(onNewConversation).toHaveBeenCalledTimes(1); // session 走的是 onNewConversation + + // 用户切到另一个 agent + current = 'agent-B'; + scrollToBottom.mockClear(); + scrollAcpLog.mockClear(); + + handleEvent(stA, 'agent-A', 'agent_output', { message: ',世界' }); + handleEvent(stA, 'agent-A', 'tool_call', { tool: 'bash' }); + + // st 照常被写:内容没丢 + expect(stA.streamContent).toBe('你好,世界'); + expect(stA.acpLog).toHaveLength(1); + // 但不再是当前 agent,全局副作用一次都不该触发 + expect(scrollToBottom).not.toHaveBeenCalled(); + expect(scrollAcpLog).not.toHaveBeenCalled(); + + // 切回来之后,新事件恢复触发副作用,且历史内容完好 + current = 'agent-A'; + handleEvent(stA, 'agent-A', 'tool_call', { tool: 'grep' }); + expect(scrollAcpLog).toHaveBeenCalledTimes(1); + expect(stA.acpLog).toHaveLength(2); + }); + + it('两个 agent 同时在跑,事件不会互相串台到对方的 st 里', () => { + let current = 'agent-A'; + const handleEvent = createAgentChatEventHandler({ + getCurrentAgentId: () => current, + onNewConversation: vi.fn(), + scrollToBottom: vi.fn(), + scrollAcpLog: vi.fn(), + scrollSubLogs: vi.fn(), + pushChunk: (st, text) => { st.streamContent += text; }, + }); + + const stA = makeState(); + const stB = makeState(); + // 交替喂 A、B 两条流的事件,模拟后台并发 + handleEvent(stA, 'agent-A', 'agent_output', { message: 'A1' }); + handleEvent(stB, 'agent-B', 'agent_output', { message: 'B1' }); + current = 'agent-B'; // 用户切到 B + handleEvent(stA, 'agent-A', 'agent_output', { message: 'A2' }); + handleEvent(stB, 'agent-B', 'agent_output', { message: 'B2' }); + + expect(stA.streamContent).toBe('A1A2'); + expect(stB.streamContent).toBe('B1B2'); + }); +}); + +describe('agentChatEvents · 打字机渲染缓冲', () => { + it('文本按固定节奏匀速吐出,而不是一次性整段落地', () => { + vi.useFakeTimers(); + try { + const st = makeState(); + const pushChunk = createTypewriterQueue({ tickMs: 10, charsPerTick: 2 }); + pushChunk(st, 'abcdef'); + + expect(st.streamContent).toBe(''); // 入队即刻不落地,等定时器吐 + vi.advanceTimersByTime(10); + expect(st.streamContent).toBe('ab'); + vi.advanceTimersByTime(10); + expect(st.streamContent).toBe('abcd'); + vi.advanceTimersByTime(10); + expect(st.streamContent).toBe('abcdef'); + vi.advanceTimersByTime(10); // 队列空了,定时器该自己停 + expect(st._typeTimer).toBeNull(); + } finally { + vi.useRealTimers(); + } + }); + + it('flushTypewriter 立刻吐光剩余队列(打断/结束时用)', () => { + vi.useFakeTimers(); + try { + const st = makeState(); + const pushChunk = createTypewriterQueue({ tickMs: 10, charsPerTick: 1 }); + pushChunk(st, 'hello'); + vi.advanceTimersByTime(10); // 只吐了 1 个字符 + expect(st.streamContent).toBe('h'); + + flushTypewriter(st); + expect(st.streamContent).toBe('hello'); + expect(st._typeTimer).toBeNull(); + expect(st._typeQueue).toBe(''); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/web/src/views/AgentChat.vue b/web/src/views/AgentChat.vue index 961cd63..dbc6074 100644 --- a/web/src/views/AgentChat.vue +++ b/web/src/views/AgentChat.vue @@ -128,24 +128,27 @@