/** * 40 号文档方案二 + 429 排队等待改造的后端回归: * * 1. 风险①:同一 agentId + 同一 convId 的并发请求过去完全没有闸门(`agentChatLocks` 的判断 * 只挡"不同 convId"),一直靠前端 textarea 的 disabled 属性挡住。"打断并发送"要解锁输入框, * 这条路径就变得可达——两个循环同时喂同一个 ACP 会话是未定义行为。 * 2. `stop` 端点现在要等锁真正释放(或 3 秒预算到期)才返回,`cleaned` 字段供前端判断 * 能不能安全发第二条。 * 3. 后续改造:并发碰撞不再"立刻 429",改成排队等待 `AGENT_CHAT_QUEUE_WAIT_MS` 预算—— * 等到锁释放就正常执行;等满预算仍抢不到才拒绝。这里用一个较短的测试专用预算 * (500ms)验证两条分支:等得到 → 正常执行,等不到 → 拒绝。 * 4. 排队期间前端不能只看到通用的"思考中":锁被占用时会先把 SSE 连接建起来发一个 * `queued` 事件;等满预算仍抢不到,因为响应头已经按 200 SSE 发出去、不能再改状态码, * 改用 `queue_timeout` 事件收尾(不再是独立的 429 JSON 响应)。 */ 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'; 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)); /** 把 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'); const original = ClaudeCliProtocol.prototype.sendPrompt; const calls = []; // 记录每次调用的起止时刻,用来证明两次调用没有重叠 // 拖住第一个请求,制造一个真实的并发窗口——延迟(100ms) < 测试预算(500ms), // 排队等待应该能在预算内等到锁释放,第二个请求正常执行而不是 429。 vi.spyOn(ClaudeCliProtocol.prototype, 'sendPrompt').mockImplementation(function (...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 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('排队等待预算耗尽仍抢不到锁时以 queue_timeout 事件收尾(响应头已按 200 SSE 发出,改不了状态码)', async () => { const conv = 'race-timeout'; const { ClaudeCliProtocol } = require_('@mas/acp-core'); const original = ClaudeCliProtocol.prototype.sendPrompt; // 延迟(900ms) > 测试预算(500ms):排队等满预算还是抢不到。 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 }), }); 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; }); 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; }); });