宜博 / lemonboss

Commit 04db77178ff914a5d2383a493b2bd0155711da04

Authored by yibo
1 parent 92bebc5d

fix(agent-chat): 429 排队等待替代立刻拒绝 + 前端不再静默吞掉非 200 响应 

- agent-session-lock.js 新增 acquireLockWithWait：锁被占用时按 pollMs 轮询排队等待
  AGENT_CHAT_QUEUE_WAIT_MS 预算（默认 60s），check-and-set 之间无 await，天然原子；
  等到锁释放就正常执行，等满预算仍抢不到才 429，不再"撞上就立刻拒绝"。
  内部自唤醒（TaskCallback/TaskDispatch 等 selfChat 调用）是 fire-and-forget，之前
  429 等于这次唤醒直接消失，只能靠别处巡检兜底；排队等待让大多数几秒到几十秒的
  短暂碰撞能自然衔接上。
- AgentChat.vue 的 handleSend() 之前拿到 fetch 响应后不检查 r.ok 就直接当 SSE 流读，
  非 200 响应体是普通 JSON，逐行解析全部匹配不上 event/data 前缀，被静默当空流吞掉——
  用户发的消息挂在界面上，没有任何回复也没有报错。现在先判 r.ok，非 200 时读 JSON
  错误体，429 显示"Agent 正忙，请稍后重试"，其他状态显示带状态码的错误提示。
- agent-chat-interrupt.test.js 相应更新：原来断言"并发第二个请求立刻 429"的用例
  改为断言"预算内排队等待、等到就正常执行、两次调用不重叠"，另加一条"预算耗尽仍
  抢不到锁时返回 429"的回归。

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CwGcvMfUcgCcweN5vjhTkn
Inline Side-by-side
Showing 4 changed files with 101 additions and 26 deletions
server/routes/agent-chat.js
  View file @04db771
... ... @@ -2565,7 +2565,12 @@ router.get(&#39;/:agentId/uploads/:convId/:filename&#39;, (req, res) =&gt; {
2565 2565 // 同一 convId 的多轮对话不受限制（ACP 进程复用）
2566 2566 // 32 号文档方案 B：这把锁挪到了共享模块 agent-session-lock.js，company.js 的招聘装配
2567 2567 // 重试逻辑也要用同一把——两条路径各自维护一份的话，各自都以为自己没有并发。
2568   -const { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease } = require('../src/core/agent-session-lock');
  2568 +const { agentChatLocks, waitForLockRelease, acquireLockWithWait } = require('../src/core/agent-session-lock');
  2569 +
  2570 +// 排队等待的预算：master/ceo 这类 agent 大部分并发碰撞是 TaskCallback 心跳唤醒跟人类
  2571 +// 消息撞在一起，占用窗口通常几秒到几十秒——等一等就能自然衔接上，不需要立刻 429。
  2572 +// 等满这个预算仍抢不到锁才真正拒绝（前端 429 会显示"Agent 正忙"提示，不是静默丢失）。
  2573 +const AGENT_CHAT_QUEUE_WAIT_MS = Number(process.env.AGENT_CHAT_QUEUE_WAIT_MS) || 60 * 1000;
2569 2574  
2570 2575 // 40 号文档方案一：Claude CLI 一次 agent_message_chunk 可能带几十到几百字符，SSE 原样
2571 2576 // 转发就是"一坨一坨蹦出来"，没有打字机效果。这里按固定步长切成小片再各发一条 SSE 事件——
... ... @@ -2587,28 +2592,30 @@ router.post(&#39;/:agentId/chat&#39;, async (req, res) =&gt; {
2587 2592  
2588 2593 const convId = reqConvId || `conv-${Date.now().toString(36)}`;
2589 2594  
2590   - // 🔒 并发互斥检查：同一 Agent 的并发请求被拒绝——无论 convId 是否相同。
  2595 + // 🔒 并发互斥检查：同一 Agent 的并发请求排队等待，等不到才拒绝——无论 convId 是否相同。
2591 2596 // 40 号文档风险①：这里原来只挡"不同 convId"，同一 convId 的第二个请求会直接
2592 2597 // 覆盖锁往下跑，等于两个循环同时喂同一个 ACP 会话（未定义行为）。此前一直靠前端
2593 2598 // textarea 的 disabled 属性挡住，从没被服务端真正堵过；"打断并发送"要解锁输入框，
2594 2599 // 这条路径就变得可达了，必须在这里把闸门补上。
2595   - const existingLock = agentChatLocks.get(agentId);
2596   - if (existingLock) {
2597   - const elapsed = Date.now() - existingLock.startedAt;
2598   - if (elapsed < AGENT_LOCK_TIMEOUT) {
2599   - console.log(`[AgentChat] 🔒 拒绝 ${agentId} 并发请求 — 已有活跃会话 ${existingLock.convId} (${(elapsed / 1000).toFixed(0)}s)`);
2600   - return res.status(429).json({
2601   - error: `Agent ${agentId} 正在执行另一个会话 (${existingLock.convId})，请等待完成`,
2602   - activeConvId: existingLock.convId,
2603   - elapsedMs: elapsed,
2604   - sameConversation: existingLock.convId === convId,
2605   - });
2606   - }
2607   - // 超时释放
2608   - console.warn(`[AgentChat] ⚠️ 强制释放超时锁: ${agentId} / ${existingLock.convId}`);
2609   - agentChatLocks.delete(agentId);
  2600 + // 后来发现「立刻 429」对内部自唤醒（TaskCallback/TaskDispatch 等 selfChat 调用）代价很大：
  2601 + // 那几条路径是 fire-and-forget，429 就等于这次唤醒直接消失、只能靠别处巡检兜底
  2602 + // （见 docs 里 selfChat 429 漏单的教训）。改成排队等待：锁被占用先原地等一段预算，
  2603 + // 大部分碰撞是几秒到几十秒的短占用，等一等就能自然衔接上；等满预算仍抢不到才真正 429。
  2604 + const acquireResult = await acquireLockWithWait(agentId, convId, AGENT_CHAT_QUEUE_WAIT_MS);
  2605 + if (!acquireResult.acquired) {
  2606 + const { lock } = acquireResult;
  2607 + const elapsed = Date.now() - lock.startedAt;
  2608 + console.log(`[AgentChat] 🔒 拒绝 ${agentId} 并发请求 — 排队等待 ${(AGENT_CHAT_QUEUE_WAIT_MS / 1000).toFixed(0)}s 后仍被占用 (会话 ${lock.convId}，已占 ${(elapsed / 1000).toFixed(0)}s)`);
  2609 + return res.status(429).json({
  2610 + error: `Agent ${agentId} 正在执行另一个会话 (${lock.convId})，请稍后重试`,
  2611 + activeConvId: lock.convId,
  2612 + elapsedMs: elapsed,
  2613 + sameConversation: lock.convId === convId,
  2614 + });
  2615 + }
  2616 + if (acquireResult.staleLock) {
  2617 + console.warn(`[AgentChat] ⚠️ 强制释放超时锁: ${agentId} / ${acquireResult.staleLock.convId}`);
2610 2618 }
2611   - agentChatLocks.set(agentId, { convId, startedAt: Date.now() });
2612 2619  
2613 2620 const msgId = `msg-${Date.now()}-${randomUUID().substring(0, 8)}`;
2614 2621 const sendSSE = setupSSE(res);
... ...
server/src/core/agent-session-lock.js
  View file @04db771
... ... @@ -31,4 +31,28 @@ function waitForLockRelease(agentId, budgetMs = 3000, pollMs = 100) {
31 31 });
32 32 }
33 33  
34   -module.exports = { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease };
  34 +/**
  35 + * 排队等待版加锁：锁被占用时不立刻让调用方判定失败，而是按 pollMs 轮询，
  36 + * 在 budgetMs 窗口内锁一空出来就同步 check-and-set（中间没有 await，
  37 + * 多个等待者不会同时都以为自己抢到），谁先摸到谁进；budget 耗尽仍抢不到才
  38 + * 报"没抢到"，调用方按原来的 429 逻辑处理。
  39 + * 没人占、或占用已超过 AGENT_LOCK_TIMEOUT（视为僵尸锁）都算"没被占"，直接抢。
  40 + */
  41 +function acquireLockWithWait(agentId, convId, budgetMs, pollMs = 200) {
  42 + return new Promise((resolve) => {
  43 + const deadline = Date.now() + budgetMs;
  44 + const tryAcquire = () => {
  45 + const existing = agentChatLocks.get(agentId);
  46 + const stale = !!existing && Date.now() - existing.startedAt >= AGENT_LOCK_TIMEOUT;
  47 + if (!existing || stale) {
  48 + agentChatLocks.set(agentId, { convId, startedAt: Date.now() });
  49 + return resolve({ acquired: true, staleLock: stale ? existing : null });
  50 + }
  51 + if (Date.now() >= deadline) return resolve({ acquired: false, lock: existing });
  52 + setTimeout(tryAcquire, pollMs);
  53 + };
  54 + tryAcquire();
  55 + });
  56 +}
  57 +
  58 +module.exports = { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease, acquireLockWithWait };
... ...
server/tests/unit/agent-chat-interrupt.test.js
  View file @04db771
1 1 /**
2   - * 40 号文档方案二的两条后端回归：
  2 + * 40 号文档方案二 + 429 排队等待改造的后端回归：
3 3 *
4 4 * 1. 风险①：同一 agentId + 同一 convId 的并发请求过去完全没有闸门（`agentChatLocks` 的判断
5 5 * 只挡"不同 convId"），一直靠前端 textarea 的 disabled 属性挡住。"打断并发送"要解锁输入框，
6 6 * 这条路径就变得可达——两个循环同时喂同一个 ACP 会话是未定义行为。
7 7 * 2. `stop` 端点现在要等锁真正释放（或 3 秒预算到期）才返回，`cleaned` 字段供前端判断
8 8 * 能不能安全发第二条。
  9 + * 3. 后续改造：并发碰撞不再"立刻 429"，改成排队等待 `AGENT_CHAT_QUEUE_WAIT_MS` 预算——
  10 + * 等到锁释放就正常执行；等满预算仍抢不到才 429。这里用一个较短的测试专用预算
  11 + * （500ms）验证两条分支：等得到 → 200，等不到 → 429。
9 12 */
10 13 process.env.LEMONBOSS_CLAUDE_MODE = 'mock';
11 14 process.env.AGENT_SESSION_IDLE_MS = '60000';
  15 +process.env.AGENT_CHAT_QUEUE_WAIT_MS = '500';
12 16  
13 17 import { describe, it, expect, beforeAll, afterAll, afterEach, vi } from 'vitest';
14 18 import fs from 'fs';
... ... @@ -52,20 +56,51 @@ describe(&#39;agent-chat · 打断-排队（40 号文档）&#39;, () =&gt; {
52 56  
53 57 const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
54 58  
55   - it('同一 convId 的第二个请求在第一个还没跑完时会被 429 拒绝，不会两个循环同时跑', async () => {
  59 + it('同一 convId 的第二个请求在锁释放预算内排队等待，等第一个跑完再执行，不会两个循环同时跑', async () => {
56 60 const conv = 'race-same-conv';
57 61 const { ClaudeCliProtocol } = require_('@mas/acp-core');
58 62 const original = ClaudeCliProtocol.prototype.sendPrompt;
59   - // 拖住第一个请求，制造一个真实的并发窗口
  63 + const calls = []; // 记录每次调用的起止时刻，用来证明两次调用没有重叠
  64 + // 拖住第一个请求，制造一个真实的并发窗口——延迟(100ms) < 测试预算(500ms)，
  65 + // 排队等待应该能在预算内等到锁释放，第二个请求正常执行而不是 429。
60 66 vi.spyOn(ClaudeCliProtocol.prototype, 'sendPrompt').mockImplementation(function (...args) {
61   - return sleep(300).then(() => original.apply(this, args));
  67 + const call = { start: Date.now(), end: null };
  68 + calls.push(call);
  69 + return sleep(100).then(() => { call.end = Date.now(); return original.apply(this, args); });
62 70 });
63 71  
64 72 const first = fetch(`${base}/api/agent/${AGENT}/chat`, {
65 73 method: 'POST', headers: { 'Content-Type': 'application/json' },
66 74 body: JSON.stringify({ message: 'first', conversationId: conv }),
67   - });
  75 + }).then(r => r.text());
68 76 await sleep(50); // 确保 first 已经进了 handler、拿到了锁
  77 + const secondRes = await fetch(`${base}/api/agent/${AGENT}/chat`, {
  78 + method: 'POST', headers: { 'Content-Type': 'application/json' },
  79 + body: JSON.stringify({ message: 'second', conversationId: conv }),
  80 + });
  81 +
  82 + expect(secondRes.status).toBe(200); // 排队等到了，不是瞬间 429
  83 + await secondRes.text();
  84 + await first;
  85 +
  86 + expect(calls.length).toBe(2);
  87 + expect(calls[1].start).toBeGreaterThanOrEqual(calls[0].end); // 严格先后，没有重叠
  88 + });
  89 +
  90 + it('排队等待预算耗尽仍抢不到锁时返回 429', async () => {
  91 + const conv = 'race-timeout';
  92 + const { ClaudeCliProtocol } = require_('@mas/acp-core');
  93 + const original = ClaudeCliProtocol.prototype.sendPrompt;
  94 + // 延迟(900ms) > 测试预算(500ms)：排队等满预算还是抢不到，只能 429。
  95 + vi.spyOn(ClaudeCliProtocol.prototype, 'sendPrompt').mockImplementation(function (...args) {
  96 + return sleep(900).then(() => original.apply(this, args));
  97 + });
  98 +
  99 + const first = fetch(`${base}/api/agent/${AGENT}/chat`, {
  100 + method: 'POST', headers: { 'Content-Type': 'application/json' },
  101 + body: JSON.stringify({ message: 'first', conversationId: conv }),
  102 + }).then(r => r.text());
  103 + await sleep(50);
69 104 const second = await fetch(`${base}/api/agent/${AGENT}/chat`, {
70 105 method: 'POST', headers: { 'Content-Type': 'application/json' },
71 106 body: JSON.stringify({ message: 'second', conversationId: conv }),
... ... @@ -75,8 +110,7 @@ describe(&#39;agent-chat · 打断-排队（40 号文档）&#39;, () =&gt; {
75 110 const body = await second.json();
76 111 expect(body.sameConversation).toBe(true);
77 112  
78   - const firstRes = await first;
79   - await firstRes.text(); // 读完流，等 finally 跑完
  113 + await first;
80 114 });
81 115  
82 116 it('stop 之后：锁在预算内释放，返回 cleaned:true', async () => {
... ...
web/src/views/AgentChat.vue
  View file @04db771
... ... @@ -953,6 +953,16 @@ async function handleSend() {
953 953 method: 'POST', headers: { 'Content-Type': 'application/json' },
954 954 body: JSON.stringify({ message: finalText, conversationId: st.activeConvId }),
955 955 })
  956 + if (!r.ok) {
  957 + // 非 200（比如 429 并发锁冲突）的响应体是普通 JSON，不是 SSE——不能当流读，
  958 + // 否则逐行解析全部匹配不上 event/data 前缀，会静默吞掉、界面上像没回复一样。
  959 + const errBody = await r.json().catch(() => ({}))
  960 + const errText = r.status === 429
  961 + ? `⏳ ${errBody.error || 'Agent 正忙，请稍后重试'}`
  962 + : `❌ 错误 (${r.status}): ${errBody.error || r.statusText}`
  963 + st.messages.push({ role: 'assistant', content: errText, timestamp: new Date().toISOString() })
  964 + return
  965 + }
956 966 const reader = r.body.getReader()
957 967 const decoder = new TextDecoder()
958 968 let buffer = '', lastEventType = ''
... ...