feat(agent-chat): 排队等待期间发 SSE queued/queue_timeout 信号，前端精确显示排队提示

之前排队等待（上条改动）期间前端只有通用的"思考中"，最长可能等到 60s 都看不出是在排队还是在正常生成，容易让人以为卡死了。 - agent-session-lock.js 新增 isLockBusy(agentId)：判断此刻是不是真被占着（僵尸锁不算），供路由层在真正排队前 peek 一眼。 - agent-chat.js 的 /:agentId/chat：锁被占用时提前建立 SSE 连接、先发一个 queued 事件（activeConvId + waitBudgetMs），前端能精确显示"排队等待中"；没被占用的常见路径完全跳过这段，不引入多余的建连时机差异。头一旦按 200 SSE 发出去就不能再改状态码，排队超时也改用 queue_timeout 事件收尾，不再是独立的 429 JSON 响应（没被占用、一开始就抢到锁的路径不受影响，仍是普通 SSE）。 - agentChatEvents.js 的 handleEvent 认识这两个新事件类型：queued 把 thinkingText 换成带等待预算的提示；queue_timeout 跟 error 分支一样落一条可见的助手消息，不会像原来的静默 429 那样挂在界面上没反应。 - 相应更新 agent-chat-interrupt.test.js（断言 SSE 里真的有 queued/queue_timeout 事件，超时分支状态码从 429 改为 200）+ agentChatEvents.test.js 新增两条用例。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CwGcvMfUcgCcweN5vjhTkn

feat(agent-chat): 排队等待期间发 SSE queued/queue_timeout 信号，前端精确显示排队提示
之前排队等待（上条改动）期间前端只有通用的"思考中"，最长可能等到 60s 都看不出是在排队还是在正常生成，容易让人以为卡死了。 - agent-session-lock.js 新增 isLockBusy(agentId)：判断此刻是不是真被占着（僵尸锁不算），供路由层在真正排队前 peek 一眼。 - agent-chat.js 的 /:agentId/chat：锁被占用时提前建立 SSE 连接、先发一个 queued 事件（activeConvId + waitBudgetMs），前端能精确显示"排队等待中"；没被占用的常见路径完全跳过这段，不引入多余的建连时机差异。头一旦按 200 SSE 发出去就不能再改状态码，排队超时也改用 queue_timeout 事件收尾，不再是独立的 429 JSON 响应（没被占用、一开始就抢到锁的路径不受影响，仍是普通 SSE）。 - agentChatEvents.js 的 handleEvent 认识这两个新事件类型：queued 把 thinkingText 换成带等待预算的提示；queue_timeout 跟 error 分支一样落一条可见的助手消息，不会像原来的静默 429 那样挂在界面上没反应。 - 相应更新 agent-chat-interrupt.test.js（断言 SSE 里真的有 queued/queue_timeout 事件，超时分支状态码从 429 改为 200）+ agentChatEvents.test.js 新增两条用例。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CwGcvMfUcgCcweN5vjhTkn
yibo
1 parent e9a21412
Showing 5 changed files with 120 additions and 17 deletions
server/routes/agent-chat.js
server/src/core/agent-session-lock.js
server/tests/unit/agent-chat-interrupt.test.js
web/src/composables/agentChatEvents.js
web/src/composables/agentChatEvents.test.js
@@ -2565,7 +2565,7 @@ router.get(&#39;/:agentId/uploads/:convId/:filename&#39;, (req, res) =&gt; {
 // 同一 convId 的多轮对话不受限制（ACP 进程复用）
 // 32 号文档方案 B：这把锁挪到了共享模块 agent-session-lock.js，company.js 的招聘装配
 // 重试逻辑也要用同一把——两条路径各自维护一份的话，各自都以为自己没有并发。
-const { agentChatLocks, waitForLockRelease, acquireLockWithWait } = require('../src/core/agent-session-lock');
+const { agentChatLocks, waitForLockRelease, acquireLockWithWait, isLockBusy } = require('../src/core/agent-session-lock');
 // 排队等待的预算：master/ceo 这类 agent 大部分并发碰撞是 TaskCallback 心跳唤醒跟人类
 // 消息撞在一起，占用窗口通常几秒到几十秒——等一等就能自然衔接上，不需要立刻 429。
@@ -2601,24 +2601,45 @@ router.post(&#39;/:agentId/chat&#39;, async (req, res) =&gt; {
   // 那几条路径是 fire-and-forget，429 就等于这次唤醒直接消失、只能靠别处巡检兜底
   // （见 docs 里 selfChat 429 漏单的教训）。改成排队等待：锁被占用先原地等一段预算，
   // 大部分碰撞是几秒到几十秒的短占用，等一等就能自然衔接上；等满预算仍抢不到才真正 429。
+  //
+  // 排队期间前端只看得到通用的"思考中"提示，跟正常生成没有区别，等上几十秒容易让人
+  // 以为卡死了。这里被占用时提前把 SSE 连接建起来发一个 queued 事件（带 activeConvId +
+  // 等待预算），前端能精确显示"排队等待中"；没被占用的常见路径完全跳过这段，不引入
+  // 多余的建连时机差异。headers 一旦按 200 SSE 发出去就不能再改状态码，所以这条路径下
+  // 排队超时也只能用专门的 queue_timeout 事件通知前端，不能再退回 429。
+  const busyLock = isLockBusy(agentId);
+  let sendSSE = null;
+  if (busyLock) {
+    sendSSE = setupSSE(res);
+    sendSSE('queued', {
+      activeConvId: busyLock.convId,
+      waitBudgetMs: AGENT_CHAT_QUEUE_WAIT_MS,
+    });
+  }
+
   const acquireResult = await acquireLockWithWait(agentId, convId, AGENT_CHAT_QUEUE_WAIT_MS);
   if (!acquireResult.acquired) {
     const { lock } = acquireResult;
     const elapsed = Date.now() - lock.startedAt;
     console.log(`[AgentChat] 🔒 拒绝 ${agentId} 并发请求 — 排队等待 ${(AGENT_CHAT_QUEUE_WAIT_MS / 1000).toFixed(0)}s 后仍被占用 (会话 ${lock.convId}，已占 ${(elapsed / 1000).toFixed(0)}s)`);
-    return res.status(429).json({
+    const payload = {
       error: `Agent ${agentId} 正在执行另一个会话 (${lock.convId})，请稍后重试`,
       activeConvId: lock.convId,
       elapsedMs: elapsed,
       sameConversation: lock.convId === convId,
-    });
+    };
+    if (sendSSE) {
+      sendSSE('queue_timeout', payload);
+      return res.end();
+    }
+    return res.status(429).json(payload);
   }
   if (acquireResult.staleLock) {
     console.warn(`[AgentChat] ⚠️ 强制释放超时锁: ${agentId} / ${acquireResult.staleLock.convId}`);
   }
   const msgId = `msg-${Date.now()}-${randomUUID().substring(0, 8)}`;
-  const sendSSE = setupSSE(res);
+  if (!sendSSE) sendSSE = setupSSE(res);
   const identity = loadIdentity(agentDir);
   const kind = resolveKind(req.body);
@@ -32,6 +32,18 @@ function waitForLockRelease(agentId, budgetMs = 3000, pollMs = 100) {
 }
 /**
+ * 判断某个 agent 此刻是不是真被占着（僵尸锁——占用超过 AGENT_LOCK_TIMEOUT——不算占用）。
+ * 占用中返回锁本身（供调用方取 convId/startedAt 用于提示），否则返回 null。
+ * 供路由层在真正排队前先 peek 一眼，决定要不要提前给前端发"排队中"信号。
+ */
+function isLockBusy(agentId) {
+  const existing = agentChatLocks.get(agentId);
+  if (!existing) return null;
+  if (Date.now() - existing.startedAt >= AGENT_LOCK_TIMEOUT) return null;
+  return existing;
+}
+
+/**
  * 排队等待版加锁：锁被占用时不立刻让调用方判定失败，而是按 pollMs 轮询，
  * 在 budgetMs 窗口内锁一空出来就同步 check-and-set（中间没有 await，
  * 多个等待者不会同时都以为自己抢到），谁先摸到谁进；budget 耗尽仍抢不到才
@@ -43,16 +55,16 @@ function acquireLockWithWait(agentId, convId, budgetMs, pollMs = 200) {
     const deadline = Date.now() + budgetMs;
     const tryAcquire = () => {
       const existing = agentChatLocks.get(agentId);
-      const stale = !!existing && Date.now() - existing.startedAt >= AGENT_LOCK_TIMEOUT;
-      if (!existing || stale) {
+      const busy = isLockBusy(agentId);
+      if (!busy) {
         agentChatLocks.set(agentId, { convId, startedAt: Date.now() });
-        return resolve({ acquired: true, staleLock: stale ? existing : null });
+        return resolve({ acquired: true, staleLock: existing && !busy ? existing : null });
       }
-      if (Date.now() >= deadline) return resolve({ acquired: false, lock: existing });
+      if (Date.now() >= deadline) return resolve({ acquired: false, lock: busy });
       setTimeout(tryAcquire, pollMs);
     };
     tryAcquire();
   });
 }
-module.exports = { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease, acquireLockWithWait };
+module.exports = { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease, acquireLockWithWait, isLockBusy };
@@ -7,8 +7,11 @@
  * 2. `stop` 端点现在要等锁真正释放（或 3 秒预算到期）才返回，`cleaned` 字段供前端判断
  *    能不能安全发第二条。
  * 3. 后续改造：并发碰撞不再"立刻 429"，改成排队等待 `AGENT_CHAT_QUEUE_WAIT_MS` 预算——
- *    等到锁释放就正常执行；等满预算仍抢不到才 429。这里用一个较短的测试专用预算
- *    （500ms）验证两条分支：等得到 → 200，等不到 → 429。
+ *    等到锁释放就正常执行；等满预算仍抢不到才拒绝。这里用一个较短的测试专用预算
+ *    （500ms）验证两条分支：等得到 → 正常执行，等不到 → 拒绝。
+ * 4. 排队期间前端不能只看到通用的"思考中"：锁被占用时会先把 SSE 连接建起来发一个
+ *    `queued` 事件；等满预算仍抢不到，因为响应头已经按 200 SSE 发出去、不能再改状态码，
+ *    改用 `queue_timeout` 事件收尾（不再是独立的 429 JSON 响应）。
  */
 process.env.LEMONBOSS_CLAUDE_MODE = 'mock';
 process.env.AGENT_SESSION_IDLE_MS = '60000';
@@ -56,6 +59,19 @@ describe(&#39;agent-chat · 打断-排队（40 号文档）&#39;, () =&gt; {
   const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+  /** 把 SSE 响应体解析成 [{event, data}]，供断言排队/超时事件是否真的发出来了 */
+  function parseSSE(text) {
+    const events = [];
+    let lastEvent = null;
+    for (const line of text.split('\n')) {
+      if (line.startsWith('event: ')) lastEvent = line.slice(7).trim();
+      else if (line.startsWith('data: ')) {
+        try { events.push({ event: lastEvent, data: JSON.parse(line.slice(6)) }); } catch { /* 忽略解析不了的行 */ }
+      }
+    }
+    return events;
+  }
+
   it('同一 convId 的第二个请求在锁释放预算内排队等待，等第一个跑完再执行，不会两个循环同时跑', async () => {
     const conv = 'race-same-conv';
     const { ClaudeCliProtocol } = require_('@mas/acp-core');
@@ -80,18 +96,24 @@ describe(&#39;agent-chat · 打断-排队（40 号文档）&#39;, () =&gt; {
     });
     expect(secondRes.status).toBe(200); // 排队等到了，不是瞬间 429
-    await secondRes.text();
+    const secondEvents = parseSSE(await secondRes.text());
+    // 排队期间要先发一条 queued，前端才能显示"排队等待中"而不是通用的"思考中"
+    const queuedEvent = secondEvents.find(e => e.event === 'queued');
+    expect(queuedEvent).toBeTruthy();
+    expect(queuedEvent.data.activeConvId).toBe(conv);
+    // 等到锁之后要继续正常跑完这轮对话，不是发完 queued 就结束
+    expect(secondEvents.some(e => e.event === 'session')).toBe(true);
     await first;
     expect(calls.length).toBe(2);
     expect(calls[1].start).toBeGreaterThanOrEqual(calls[0].end); // 严格先后，没有重叠
   });
-  it('排队等待预算耗尽仍抢不到锁时返回 429', async () => {
+  it('排队等待预算耗尽仍抢不到锁时以 queue_timeout 事件收尾（响应头已按 200 SSE 发出，改不了状态码）', async () => {
     const conv = 'race-timeout';
     const { ClaudeCliProtocol } = require_('@mas/acp-core');
     const original = ClaudeCliProtocol.prototype.sendPrompt;
-    // 延迟(900ms) > 测试预算(500ms)：排队等满预算还是抢不到，只能 429。
+    // 延迟(900ms) > 测试预算(500ms)：排队等满预算还是抢不到。
     vi.spyOn(ClaudeCliProtocol.prototype, 'sendPrompt').mockImplementation(function (...args) {
       return sleep(900).then(() => original.apply(this, args));
     });
@@ -106,9 +128,13 @@ describe(&#39;agent-chat · 打断-排队（40 号文档）&#39;, () =&gt; {
       body: JSON.stringify({ message: 'second', conversationId: conv }),
     });
-    expect(second.status).toBe(429);
-    const body = await second.json();
-    expect(body.sameConversation).toBe(true);
+    expect(second.status).toBe(200); // 头已经按 SSE 发出去了，不再是独立的 429 JSON 响应
+    const events = parseSSE(await second.text());
+    expect(events.some(e => e.event === 'queued')).toBe(true);
+    const timeoutEvent = events.find(e => e.event === 'queue_timeout');
+    expect(timeoutEvent).toBeTruthy();
+    expect(timeoutEvent.data.sameConversation).toBe(true);
+    expect(timeoutEvent.data.activeConvId).toBe(conv);
     await first;
   });
@@ -80,6 +80,22 @@ export function createAgentChatEventHandler({
           st.thinkingText = data.message || '思考中...';
         }
         break;
+      // 429 排队等待改造：锁被占用时后端会先发这条，让"思考中"从通用文案变成
+      // 能看出真实原因的排队提示，而不是干等到超时/开始输出都看不出区别。
+      case 'queued': {
+        const waitSec = Math.round((data.waitBudgetMs || 0) / 1000);
+        st.thinkingText = `⏳ 排队等待中 — Agent 正在处理另一个会话，最多等待 ${waitSec} 秒`;
+        if (isCurrent) scrollToBottom();
+        break;
+      }
+      // 排队等到预算耗尽：这条路径的响应头已经按 200 SSE 发出去了，后端没法再改状态码，
+      // 只能用这个专门的事件类型收尾——跟 'error' 分支一样直接落一条可见的助手消息。
+      case 'queue_timeout':
+        st.messages.push({ role: 'assistant', content: `⏳ ${data.error || 'Agent 排队等待超时，请稍后重试'}`, timestamp: new Date().toISOString() });
+        st.thinkingText = '';
+        st.thinkingContent = '';
+        if (isCurrent) scrollToBottom();
+        break;
       case 'agent_output':
         st.streaming = true;
         pushChunk(st, data.message || '');
@@ -87,6 +87,34 @@ describe(&#39;agentChatEvents · 串台加固&#39;, () =&gt; {
   });
 });
+describe('agentChatEvents · 排队等待提示（429 排队等待改造）', () => {
+  const handleEvent = createAgentChatEventHandler({
+    getCurrentAgentId: () => 'agent-A',
+    onNewConversation: vi.fn(),
+    scrollToBottom: vi.fn(),
+    scrollAcpLog: vi.fn(),
+    scrollSubLogs: vi.fn(),
+  });
+
+  it('queued 事件把思考文案换成带等待预算的排队提示，而不是通用的"思考中"', () => {
+    const st = makeState();
+    handleEvent(st, 'agent-A', 'queued', { activeConvId: 'conv-busy', waitBudgetMs: 60000 });
+    expect(st.thinkingText).toContain('排队等待中');
+    expect(st.thinkingText).toContain('60');
+  });
+
+  it('queue_timeout 事件落一条可见的助手消息，而不是像原来 429 那样静默无回复', () => {
+    const st = makeState();
+    handleEvent(st, 'agent-A', 'queued', { activeConvId: 'conv-busy', waitBudgetMs: 500 });
+    handleEvent(st, 'agent-A', 'queue_timeout', { error: 'Agent 正在执行另一个会话 (conv-busy)，请稍后重试' });
+
+    expect(st.messages).toHaveLength(1);
+    expect(st.messages[0].role).toBe('assistant');
+    expect(st.messages[0].content).toContain('conv-busy');
+    expect(st.thinkingText).toBe(''); // 收尾清空，不留一个卡住的"排队等待中"
+  });
+});
+
 describe('agentChatEvents · 打字机渲染缓冲', () => {
   it('文本按固定节奏匀速吐出，而不是一次性整段落地', () => {
     vi.useFakeTimers();