Commit 310a2b2e5a52143ebcfe19dbb0b0b2360aff285a

Authored by yibo
1 parent d91d352f

fix(agent-chat): 会话池按次回收,堵住心跳每 5 分钟漏一个 claude CLI 进程

sessions 里存的是**常驻**的 claude CLI 子进程(stream-json 模式,不是一次性的
`claude -p`),此前只有四条回收路径:协议 error / 子进程 disconnected / 手动删对话 /
手动 stop —— 「正常跑完的对话」不在其中。

而两路心跳都用一次性 convId 打进 /api/agent/:id/chat:
  - heartbeat-service:`heartbeat-${Date.now()}`,cto / coo 各 5 分钟一次
  - company-heartbeat-service:_sendAgentChat 的五个调用点(CEO 巡检 / Pipeline 步骤 /
    日程 / 告警 / 日报)convId 也全是一次性的,且无差别带 isHeartbeat:true

于是每次心跳新起一个进程且永不释放。实测单次运行 2h43m 攒到 82 个进程 / 17.5 GB RSS。

改法:
  - 心跳会话跑完立刻 closeSession(一次性 convId,会话复用对它零价值)
  - 普通对话上空闲 TTL(默认 30 分钟,AGENT_SESSION_IDLE_MS 可覆盖),复用 engine/idle-timer
  - 计时器只在「没有在途请求」的间隙走:getOrCreateProtocol 命中复用时 cancel,请求结束时
    重新上弦。所以再长的单次生成也不会被 TTL 打断,比在流式回调里 touch() 少一个调用点
  - 删对话 / stop 两处手抄的 disconnect+delete 收敛到 closeSession,否则它们不停表

守卫:agent-chat-session-reclaim.test.js 用现成的 /chat/:convId/stop 当黑盒探针
(查得到会话才返回 success:true),不为测试加导出口。对 HEAD 版本跑过:2 红 1 绿 ——
两条回收断言修前失败,「普通对话保留会话」修前修后都绿,证明复用能力没被顺手砍掉。

线上验证:重启后第一轮心跳 2 触发 / 2 回收,server 名下 claude 子进程数 0。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
server/routes/agent-chat.js
@@ -19,6 +19,7 @@ const { randomUUID } = require(&#39;crypto&#39;); @@ -19,6 +19,7 @@ const { randomUUID } = require(&#39;crypto&#39;);
19 const multer = require('multer'); 19 const multer = require('multer');
20 const { ClaudeCliProtocol } = require('@mas/acp-core'); 20 const { ClaudeCliProtocol } = require('@mas/acp-core');
21 const { Parser } = require('../src/core/parser'); 21 const { Parser } = require('../src/core/parser');
  22 +const { IdleTimer } = require('../src/engine/idle-timer');
22 const AgentManager = require('../src/core/agent-manager'); 23 const AgentManager = require('../src/core/agent-manager');
23 const locator = require('../services/agent-locator'); 24 const locator = require('../services/agent-locator');
24 const { OkrService } = require('../services/okr-service'); 25 const { OkrService } = require('../services/okr-service');
@@ -742,14 +743,56 @@ function listConversations(agentId) { @@ -742,14 +743,56 @@ function listConversations(agentId) {
742 // ACP Session 池 — 支持多轮 + 上下文恢复 743 // ACP Session 池 — 支持多轮 + 上下文恢复
743 // ═══════════════════════════════════════ 744 // ═══════════════════════════════════════
744 745
745 -// key: `${agentId}:${conversationId}` → { protocol, messageCount } 746 +// key: `${agentId}:${conversationId}` → { protocol, messageCount, idleTimer }
746 const sessions = new Map(); 747 const sessions = new Map();
747 748
  749 +// ── 会话回收 ──
  750 +// sessions 里存的是**常驻**的 claude CLI 子进程(stream-json 模式,不是一次性的 `claude -p`)。
  751 +// 在此之前只有四条回收路径:协议 error / 子进程 disconnected / 手动删对话 / 手动 stop ——
  752 +// 也就是说「正常跑完的对话」永远不回收。心跳每 5 分钟用一个**全新** convId 打进来
  753 +// (heartbeat-service.js:206),于是每次心跳都新起一个进程且永不释放,实测 2h40m 攒了 82 个
  754 +// 进程 / 17.5 GB。所以:心跳跑完立刻回收,普通对话按空闲 TTL 回收。
  755 +const SESSION_IDLE_MS = Number(process.env.AGENT_SESSION_IDLE_MS) || 30 * 60 * 1000;
  756 +
  757 +/** 只从池子里摘掉(子进程已经自己死了的场景),顺手停表 */
  758 +function dropSession(key) {
  759 + const entry = sessions.get(key);
  760 + entry?.idleTimer?.cancel();
  761 + sessions.delete(key);
  762 + return entry;
  763 +}
  764 +
  765 +/** 摘掉 + 杀子进程 */
  766 +function closeSession(key, reason) {
  767 + const entry = dropSession(key);
  768 + if (!entry) return false;
  769 + try { entry.protocol.disconnect(); } catch { }
  770 + console.log(`[AgentChat] 🧹 会话已回收 ${key} (${reason})`);
  771 + return true;
  772 +}
  773 +
  774 +/**
  775 + * 请求结束后给会话上弦;请求开始时由 getOrCreateProtocol 停表。
  776 + * 计时器只在「没有在途请求」的间隙走,所以再长的单次生成也不会被 TTL 打断
  777 + * —— 这比在流式回调里 touch() 少一处需要维护的调用点。
  778 + */
  779 +function armSessionIdle(key) {
  780 + const entry = sessions.get(key);
  781 + if (!entry) return;
  782 + entry.idleTimer?.cancel();
  783 + entry.idleTimer = new IdleTimer(
  784 + SESSION_IDLE_MS,
  785 + () => closeSession(key, `空闲超过 ${Math.round(SESSION_IDLE_MS / 60000)} 分钟`),
  786 + { label: `agent-session:${key}` }
  787 + ).start();
  788 +}
  789 +
748 async function getOrCreateProtocol(agentId, convId) { 790 async function getOrCreateProtocol(agentId, convId) {
749 const key = `${agentId}:${convId}`; 791 const key = `${agentId}:${convId}`;
750 let entry = sessions.get(key); 792 let entry = sessions.get(key);
751 793
752 if (entry && entry.protocol.isConnected) { 794 if (entry && entry.protocol.isConnected) {
  795 + entry.idleTimer?.cancel(); // 请求在途期间不计空闲
753 return entry; 796 return entry;
754 } 797 }
755 798
@@ -801,12 +844,12 @@ async function getOrCreateProtocol(agentId, convId) { @@ -801,12 +844,12 @@ async function getOrCreateProtocol(agentId, convId) {
801 // 绑定错误监听器,防止 EventEmitter 抛出未捕获异常导致整个 Node 进程奔溃 (例如 spawn claude ENOENT) 844 // 绑定错误监听器,防止 EventEmitter 抛出未捕获异常导致整个 Node 进程奔溃 (例如 spawn claude ENOENT)
802 protocol.on('error', (err) => { 845 protocol.on('error', (err) => {
803 console.error(`[AgentChat] 🚨 ${identity.name} 协议级异常:`, err.message); 846 console.error(`[AgentChat] 🚨 ${identity.name} 协议级异常:`, err.message);
804 - sessions.delete(key); // 下次提问将重建进程 847 + dropSession(key); // 下次提问将重建进程
805 releaseLock(); 848 releaseLock();
806 }); 849 });
807 850
808 protocol.on('disconnected', ({ code, signal }) => { 851 protocol.on('disconnected', ({ code, signal }) => {
809 - sessions.delete(key); 852 + dropSession(key);
810 releaseLock(); 853 releaseLock();
811 console.log(`[AgentChat] 💤 ${identity.name} 会话断开 (code=${code}, signal=${signal})`); 854 console.log(`[AgentChat] 💤 ${identity.name} 会话断开 (code=${code}, signal=${signal})`);
812 }); 855 });
@@ -820,7 +863,7 @@ async function getOrCreateProtocol(agentId, convId) { @@ -820,7 +863,7 @@ async function getOrCreateProtocol(agentId, convId) {
820 await protocol.initialize(); 863 await protocol.initialize();
821 await protocol.newSession({ cwd }); 864 await protocol.newSession({ cwd });
822 865
823 - const newEntry = { protocol, messageCount: 0, agentId, convId }; 866 + const newEntry = { protocol, messageCount: 0, agentId, convId, idleTimer: null };
824 sessions.set(key, newEntry); 867 sessions.set(key, newEntry);
825 console.log(`[AgentChat] 🧬 ${identity.name} ACP 会话已创建 (conv: ${convId})`); 868 console.log(`[AgentChat] 🧬 ${identity.name} ACP 会话已创建 (conv: ${convId})`);
826 869
@@ -1337,10 +1380,7 @@ router.delete(&#39;/:agentId/conversations/:convId&#39;, (req, res) =&gt; { @@ -1337,10 +1380,7 @@ router.delete(&#39;/:agentId/conversations/:convId&#39;, (req, res) =&gt; {
1337 const uploadDir = getUploadDir(agentId, convId); 1380 const uploadDir = getUploadDir(agentId, convId);
1338 if (fs.existsSync(uploadDir)) fs.rmSync(uploadDir, { recursive: true, force: true }); 1381 if (fs.existsSync(uploadDir)) fs.rmSync(uploadDir, { recursive: true, force: true });
1339 // 4. 断开会话 1382 // 4. 断开会话
1340 - const key = `${agentId}:${convId}`;  
1341 - const entry = sessions.get(key);  
1342 - if (entry && entry.protocol) entry.protocol.disconnect();  
1343 - sessions.delete(key); 1383 + closeSession(`${agentId}:${convId}`, '对话被删除');
1344 console.log(`[AgentChat] 🗑️ 已删除对话 ${agentId}/${convId} (含uploads)`); 1384 console.log(`[AgentChat] 🗑️ 已删除对话 ${agentId}/${convId} (含uploads)`);
1345 res.json({ success: true }); 1385 res.json({ success: true });
1346 } catch (err) { res.status(500).json({ error: err.message }); } 1386 } catch (err) { res.status(500).json({ error: err.message }); }
@@ -1350,10 +1390,7 @@ router.delete(&#39;/:agentId/conversations/:convId&#39;, (req, res) =&gt; { @@ -1350,10 +1390,7 @@ router.delete(&#39;/:agentId/conversations/:convId&#39;, (req, res) =&gt; {
1350 router.post('/:agentId/chat/:convId/stop', (req, res) => { 1390 router.post('/:agentId/chat/:convId/stop', (req, res) => {
1351 const { agentId, convId } = req.params; 1391 const { agentId, convId } = req.params;
1352 const key = `${agentId}:${convId}`; 1392 const key = `${agentId}:${convId}`;
1353 - const entry = sessions.get(key);  
1354 - if (entry && entry.protocol) {  
1355 - entry.protocol.disconnect(); // 关闭底层 claude cli 进程  
1356 - sessions.delete(key); // 下次从冷启动重新加载 1393 + if (closeSession(key, '用户手动中断')) {
1357 return res.json({ success: true, message: 'Agent 已中断' }); 1394 return res.json({ success: true, message: 'Agent 已中断' });
1358 } 1395 }
1359 res.json({ success: false, message: '未找到运行中的 Agent 进程' }); 1396 res.json({ success: false, message: '未找到运行中的 Agent 进程' });
@@ -2709,14 +2746,18 @@ ${resultsText.join(&#39;\n\n&#39;)} @@ -2709,14 +2746,18 @@ ${resultsText.join(&#39;\n\n&#39;)}
2709 saveMeta(agentId, convId, meta); 2746 saveMeta(agentId, convId, meta);
2710 2747
2711 // 清理断开的 session 2748 // 清理断开的 session
2712 - const key = `${agentId}:${convId}`;  
2713 - sessions.delete(key); 2749 + dropSession(`${agentId}:${convId}`);
2714 2750
2715 sendSSE('error', { message: err.message }); 2751 sendSSE('error', { message: err.message });
2716 console.log(`[SSE-DEBUG] 🚨 sendSSE('error') called:`, err.message.substring(0, 200)); 2752 console.log(`[SSE-DEBUG] 🚨 sendSSE('error') called:`, err.message.substring(0, 200));
2717 } finally { 2753 } finally {
2718 agentChatLocks.delete(agentId); 2754 agentChatLocks.delete(agentId);
2719 res.end(); 2755 res.end();
  2756 + // 心跳会话是一次性的(heartbeat-service 每次都生成新 convId),留在池子里就是纯泄漏。
  2757 + // 普通对话保留会话复用能力,只上空闲 TTL。
  2758 + const sessionKey = `${agentId}:${convId}`;
  2759 + if (req.body.isHeartbeat) closeSession(sessionKey, '心跳结束');
  2760 + else armSessionIdle(sessionKey);
2720 } 2761 }
2721 }); 2762 });
2722 2763
server/tests/unit/agent-chat-session-reclaim.test.js 0 → 100644
  1 +/**
  2 + * 锁的是 claude CLI 子进程泄漏 —— 实测 2h40m 攒了 82 个进程 / 17.5 GB。
  3 + *
  4 + * agent-chat 的 `sessions` Map 存的是**常驻**的 claude CLI 子进程(stream-json 模式,
  5 + * 不是一次性的 `claude -p`)。修之前只有四条回收路径:协议 error / 子进程 disconnected /
  6 + * 手动删对话 / 手动 stop —— 「正常跑完的对话」不在其中。而 heartbeat-service 每 5 分钟
  7 + * 用一个**全新** convId(`heartbeat-${Date.now()}`)打进来,于是每次心跳都新起一个进程且
  8 + * 永不释放。cto / coo 两个本体的 intervalMinutes 都是 5,正好对上观测到的「每 5 分钟涨 2-3 个」。
  9 + *
  10 + * 探针用现成的 `/chat/:convId/stop`:它只在 sessions 里查得到条目时才返回 success:true,
  11 + * 所以不用为测试往模块里加导出口。断言全是「会话还在不在」,不看 chat 自己的响应内容。
  12 + */
  13 +process.env.LEMONBOSS_CLAUDE_MODE = 'mock'; // 必须早于 require:acp-core 的 mock 开关是模块级常量
  14 +process.env.AGENT_SESSION_IDLE_MS = '400'; // 同理,SESSION_IDLE_MS 是模块级常量
  15 +
  16 +import { describe, it, expect, beforeAll, afterAll } from 'vitest';
  17 +import fs from 'fs';
  18 +import path from 'path';
  19 +import express from 'express';
  20 +import { createRequire } from 'node:module';
  21 +
  22 +const require_ = createRequire(import.meta.url);
  23 +const ROOT = path.resolve(__dirname, '../../..');
  24 +const AGENT = 'zz-session-reclaim';
  25 +const BODY = path.join(ROOT, 'system', 'agents', AGENT);
  26 +
  27 +describe('agent-chat · 会话回收(心跳即时回收 + 空闲 TTL)', () => {
  28 + let base, srv;
  29 +
  30 + beforeAll(async () => {
  31 + fs.mkdirSync(BODY, { recursive: true });
  32 + fs.writeFileSync(path.join(BODY, 'config.json'), JSON.stringify({ id: AGENT, type: 'skill', name: AGENT, status: 'active' }, null, 2));
  33 + fs.writeFileSync(path.join(BODY, 'identity.md'), `# ${AGENT}\n`);
  34 + fs.writeFileSync(path.join(BODY, 'agent.md'), '# 会话回收单测用的常驻本体\n');
  35 +
  36 + const app = express();
  37 + app.use(express.json());
  38 + app.use('/api/agent', require_('../../routes/agent-chat'));
  39 + await new Promise((r) => { srv = app.listen(0, '127.0.0.1', () => { base = `http://127.0.0.1:${srv.address().port}`; r(); }); });
  40 + });
  41 +
  42 + afterAll(async () => {
  43 + if (srv) await new Promise((r) => srv.close(r));
  44 + // 本体定义(config/identity/agent.md)留下,与 zz-tenant-* 同一约定:测试建出来的 agent 也是
  45 + // 注册表条目,建了又删会让并行用例撞上「列到了名字、进去时已经没了」的 flaky。只清运行态。
  46 + for (const d of ['conversations', '_global', 'memory']) fs.rmSync(path.join(BODY, d), { recursive: true, force: true });
  47 + });
  48 +
  49 + /** 跑完一轮对话(SSE 要读到流结束,否则 finally 还没执行) */
  50 + const chat = async (convId, extra = {}) => {
  51 + const res = await fetch(`${base}/api/agent/${AGENT}/chat`, {
  52 + method: 'POST',
  53 + headers: { 'Content-Type': 'application/json' },
  54 + body: JSON.stringify({ message: 'ping', conversationId: convId, ...extra }),
  55 + });
  56 + await res.text();
  57 + };
  58 +
  59 + /** 探针:会话是否还留在池子里 */
  60 + const alive = async (convId) => {
  61 + const res = await fetch(`${base}/api/agent/${AGENT}/chat/${convId}/stop`, { method: 'POST' });
  62 + return (await res.json()).success === true;
  63 + };
  64 +
  65 + const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
  66 +
  67 + it('心跳会话跑完立刻回收(这是泄漏的正主)', async () => {
  68 + const conv = `heartbeat-${Date.now().toString(36)}`;
  69 + await chat(conv, { isHeartbeat: true });
  70 + expect(await alive(conv)).toBe(false);
  71 + });
  72 +
  73 + it('普通对话保留会话(复用能力不能被顺手砍掉)', async () => {
  74 + const conv = `conv-keep-${Date.now().toString(36)}`;
  75 + await chat(conv);
  76 + expect(await alive(conv)).toBe(true);
  77 + });
  78 +
  79 + it('普通对话空闲超过 TTL 后被回收', async () => {
  80 + const conv = `conv-ttl-${Date.now().toString(36)}`;
  81 + await chat(conv);
  82 + await sleep(700); // > AGENT_SESSION_IDLE_MS
  83 + expect(await alive(conv)).toBe(false);
  84 + });
  85 +});
system/agents/zz-session-reclaim/agent.md 0 → 100644
  1 +# 会话回收单测用的常驻本体
system/agents/zz-session-reclaim/config.json 0 → 100644
  1 +{
  2 + "id": "zz-session-reclaim",
  3 + "type": "skill",
  4 + "name": "zz-session-reclaim",
  5 + "status": "active"
  6 +}
0 \ No newline at end of file 7 \ No newline at end of file
system/agents/zz-session-reclaim/identity.md 0 → 100644
  1 +# zz-session-reclaim