宜博 / lemonboss

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

  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
    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 }),
    });

    expect(second.status).toBe(429);
    const body = await second.json();
    expect(body.sameConversation).toBe(true);

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