From 04db77178ff914a5d2383a493b2bd0155711da04 Mon Sep 17 00:00:00 2001 From: yibo Date: Mon, 31 Aug 2026 11:19:32 +0800 Subject: [PATCH] fix(agent-chat): 429 排队等待替代立刻拒绝 + 前端不再静默吞掉非 200 响应 --- server/routes/agent-chat.js | 43 +++++++++++++++++++++++++------------------ server/src/core/agent-session-lock.js | 26 +++++++++++++++++++++++++- server/tests/unit/agent-chat-interrupt.test.js | 48 +++++++++++++++++++++++++++++++++++++++++------- web/src/views/AgentChat.vue | 10 ++++++++++ 4 files changed, 101 insertions(+), 26 deletions(-) diff --git a/server/routes/agent-chat.js b/server/routes/agent-chat.js index 86867f3..bf40496 100644 --- a/server/routes/agent-chat.js +++ b/server/routes/agent-chat.js @@ -2565,7 +2565,12 @@ router.get('/:agentId/uploads/:convId/:filename', (req, res) => { // 同一 convId 的多轮对话不受限制（ACP 进程复用） // 32 号文档方案 B：这把锁挪到了共享模块 agent-session-lock.js，company.js 的招聘装配 // 重试逻辑也要用同一把——两条路径各自维护一份的话，各自都以为自己没有并发。 -const { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease } = require('../src/core/agent-session-lock'); +const { agentChatLocks, waitForLockRelease, acquireLockWithWait } = require('../src/core/agent-session-lock'); + +// 排队等待的预算：master/ceo 这类 agent 大部分并发碰撞是 TaskCallback 心跳唤醒跟人类 +// 消息撞在一起，占用窗口通常几秒到几十秒——等一等就能自然衔接上，不需要立刻 429。 +// 等满这个预算仍抢不到锁才真正拒绝（前端 429 会显示"Agent 正忙"提示，不是静默丢失）。 +const AGENT_CHAT_QUEUE_WAIT_MS = Number(process.env.AGENT_CHAT_QUEUE_WAIT_MS) || 60 * 1000; // 40 号文档方案一：Claude CLI 一次 agent_message_chunk 可能带几十到几百字符，SSE 原样 // 转发就是"一坨一坨蹦出来"，没有打字机效果。这里按固定步长切成小片再各发一条 SSE 事件—— @@ -2587,28 +2592,30 @@ 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) { - 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, - sameConversation: existingLock.convId === convId, - }); - } - // 超时释放 - console.warn(`[AgentChat] ⚠️ 强制释放超时锁: ${agentId} / ${existingLock.convId}`); - agentChatLocks.delete(agentId); + // 后来发现「立刻 429」对内部自唤醒（TaskCallback/TaskDispatch 等 selfChat 调用）代价很大： + // 那几条路径是 fire-and-forget，429 就等于这次唤醒直接消失、只能靠别处巡检兜底 + // （见 docs 里 selfChat 429 漏单的教训）。改成排队等待：锁被占用先原地等一段预算， + // 大部分碰撞是几秒到几十秒的短占用，等一等就能自然衔接上；等满预算仍抢不到才真正 429。 + 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({ + error: `Agent ${agentId} 正在执行另一个会话 (${lock.convId})，请稍后重试`, + activeConvId: lock.convId, + elapsedMs: elapsed, + sameConversation: lock.convId === convId, + }); + } + if (acquireResult.staleLock) { + console.warn(`[AgentChat] ⚠️ 强制释放超时锁: ${agentId} / ${acquireResult.staleLock.convId}`); } - agentChatLocks.set(agentId, { convId, startedAt: Date.now() }); const msgId = `msg-${Date.now()}-${randomUUID().substring(0, 8)}`; const sendSSE = setupSSE(res); diff --git a/server/src/core/agent-session-lock.js b/server/src/core/agent-session-lock.js index adcbcc7..7ecc608 100644 --- a/server/src/core/agent-session-lock.js +++ b/server/src/core/agent-session-lock.js @@ -31,4 +31,28 @@ function waitForLockRelease(agentId, budgetMs = 3000, pollMs = 100) { }); } -module.exports = { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease }; +/** + * 排队等待版加锁：锁被占用时不立刻让调用方判定失败，而是按 pollMs 轮询， + * 在 budgetMs 窗口内锁一空出来就同步 check-and-set（中间没有 await， + * 多个等待者不会同时都以为自己抢到），谁先摸到谁进；budget 耗尽仍抢不到才 + * 报"没抢到"，调用方按原来的 429 逻辑处理。 + * 没人占、或占用已超过 AGENT_LOCK_TIMEOUT（视为僵尸锁）都算"没被占"，直接抢。 + */ +function acquireLockWithWait(agentId, convId, budgetMs, pollMs = 200) { + return new Promise((resolve) => { + 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) { + agentChatLocks.set(agentId, { convId, startedAt: Date.now() }); + return resolve({ acquired: true, staleLock: stale ? existing : null }); + } + if (Date.now() >= deadline) return resolve({ acquired: false, lock: existing }); + setTimeout(tryAcquire, pollMs); + }; + tryAcquire(); + }); +} + +module.exports = { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease, acquireLockWithWait }; diff --git a/server/tests/unit/agent-chat-interrupt.test.js b/server/tests/unit/agent-chat-interrupt.test.js index 055dddd..2b51cf0 100644 --- a/server/tests/unit/agent-chat-interrupt.test.js +++ b/server/tests/unit/agent-chat-interrupt.test.js @@ -1,14 +1,18 @@ /** - * 40 号文档方案二的两条后端回归： + * 40 号文档方案二 + 429 排队等待改造的后端回归： * * 1. 风险①：同一 agentId + 同一 convId 的并发请求过去完全没有闸门（`agentChatLocks` 的判断 * 只挡"不同 convId"），一直靠前端 textarea 的 disabled 属性挡住。"打断并发送"要解锁输入框， * 这条路径就变得可达——两个循环同时喂同一个 ACP 会话是未定义行为。 * 2. `stop` 端点现在要等锁真正释放（或 3 秒预算到期）才返回，`cleaned` 字段供前端判断 * 能不能安全发第二条。 + * 3. 后续改造：并发碰撞不再"立刻 429"，改成排队等待 `AGENT_CHAT_QUEUE_WAIT_MS` 预算—— + * 等到锁释放就正常执行；等满预算仍抢不到才 429。这里用一个较短的测试专用预算 + * （500ms）验证两条分支：等得到 → 200，等不到 → 429。 */ process.env.LEMONBOSS_CLAUDE_MODE = 'mock'; process.env.AGENT_SESSION_IDLE_MS = '60000'; +process.env.AGENT_CHAT_QUEUE_WAIT_MS = '500'; import { describe, it, expect, beforeAll, afterAll, afterEach, vi } from 'vitest'; import fs from 'fs'; @@ -52,20 +56,51 @@ describe('agent-chat · 打断-排队（40 号文档）', () => { const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); - it('同一 convId 的第二个请求在第一个还没跑完时会被 429 拒绝，不会两个循环同时跑', async () => { + it('同一 convId 的第二个请求在锁释放预算内排队等待，等第一个跑完再执行，不会两个循环同时跑', async () => { const conv = 'race-same-conv'; const { ClaudeCliProtocol } = require_('@mas/acp-core'); const original = ClaudeCliProtocol.prototype.sendPrompt; - // 拖住第一个请求，制造一个真实的并发窗口 + const calls = []; // 记录每次调用的起止时刻，用来证明两次调用没有重叠 + // 拖住第一个请求，制造一个真实的并发窗口——延迟(100ms) < 测试预算(500ms)， + // 排队等待应该能在预算内等到锁释放，第二个请求正常执行而不是 429。 vi.spyOn(ClaudeCliProtocol.prototype, 'sendPrompt').mockImplementation(function (...args) { - return sleep(300).then(() => original.apply(this, args)); + const call = { start: Date.now(), end: null }; + calls.push(call); + return sleep(100).then(() => { call.end = Date.now(); return 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 }), - }); + }).then(r => r.text()); await sleep(50); // 确保 first 已经进了 handler、拿到了锁 + const secondRes = await fetch(`${base}/api/agent/${AGENT}/chat`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ message: 'second', conversationId: conv }), + }); + + expect(secondRes.status).toBe(200); // 排队等到了，不是瞬间 429 + await secondRes.text(); + await first; + + expect(calls.length).toBe(2); + expect(calls[1].start).toBeGreaterThanOrEqual(calls[0].end); // 严格先后，没有重叠 + }); + + it('排队等待预算耗尽仍抢不到锁时返回 429', async () => { + const conv = 'race-timeout'; + const { ClaudeCliProtocol } = require_('@mas/acp-core'); + const original = ClaudeCliProtocol.prototype.sendPrompt; + // 延迟(900ms) > 测试预算(500ms)：排队等满预算还是抢不到，只能 429。 + vi.spyOn(ClaudeCliProtocol.prototype, 'sendPrompt').mockImplementation(function (...args) { + return sleep(900).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 }), + }).then(r => r.text()); + await sleep(50); const second = await fetch(`${base}/api/agent/${AGENT}/chat`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ message: 'second', conversationId: conv }), @@ -75,8 +110,7 @@ describe('agent-chat · 打断-排队（40 号文档）', () => { const body = await second.json(); expect(body.sameConversation).toBe(true); - const firstRes = await first; - await firstRes.text(); // 读完流，等 finally 跑完 + await first; }); it('stop 之后：锁在预算内释放，返回 cleaned:true', async () => { diff --git a/web/src/views/AgentChat.vue b/web/src/views/AgentChat.vue index dbc6074..1a3425e 100644 --- a/web/src/views/AgentChat.vue +++ b/web/src/views/AgentChat.vue @@ -953,6 +953,16 @@ async function handleSend() { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ message: finalText, conversationId: st.activeConvId }), }) + if (!r.ok) { + // 非 200（比如 429 并发锁冲突）的响应体是普通 JSON，不是 SSE——不能当流读， + // 否则逐行解析全部匹配不上 event/data 前缀，会静默吞掉、界面上像没回复一样。 + const errBody = await r.json().catch(() => ({})) + const errText = r.status === 429 + ? `⏳ ${errBody.error || 'Agent 正忙，请稍后重试'}` + : `❌ 错误 (${r.status}): ${errBody.error || r.statusText}` + st.messages.push({ role: 'assistant', content: errText, timestamp: new Date().toISOString() }) + return + } const reader = r.body.getReader() const decoder = new TextDecoder() let buffer = '', lastEventType = '' -- libgit2 0.21.4