diff --git a/server/routes/agent-chat.js b/server/routes/agent-chat.js index bf40496..b5b8ca8 100644 --- a/server/routes/agent-chat.js +++ b/server/routes/agent-chat.js @@ -2565,7 +2565,7 @@ router.get('/:agentId/uploads/:convId/:filename', (req, res) => { // 同一 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('/:agentId/chat', async (req, res) => { // 那几条路径是 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); diff --git a/server/src/core/agent-session-lock.js b/server/src/core/agent-session-lock.js index 7ecc608..cf50415 100644 --- a/server/src/core/agent-session-lock.js +++ b/server/src/core/agent-session-lock.js @@ -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 }; diff --git a/server/tests/unit/agent-chat-interrupt.test.js b/server/tests/unit/agent-chat-interrupt.test.js index 2b51cf0..7344320 100644 --- a/server/tests/unit/agent-chat-interrupt.test.js +++ b/server/tests/unit/agent-chat-interrupt.test.js @@ -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('agent-chat · 打断-排队（40 号文档）', () => { 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('agent-chat · 打断-排队（40 号文档）', () => { }); 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('agent-chat · 打断-排队（40 号文档）', () => { 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; }); diff --git a/web/src/composables/agentChatEvents.js b/web/src/composables/agentChatEvents.js index dc51085..54e2f27 100644 --- a/web/src/composables/agentChatEvents.js +++ b/web/src/composables/agentChatEvents.js @@ -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 || ''); diff --git a/web/src/composables/agentChatEvents.test.js b/web/src/composables/agentChatEvents.test.js index 9f99b99..f4c1f2d 100644 --- a/web/src/composables/agentChatEvents.test.js +++ b/web/src/composables/agentChatEvents.test.js @@ -87,6 +87,34 @@ describe('agentChatEvents · 串台加固', () => { }); }); +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();