宜博 / lemonboss

/**
 * 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;
  });
});