From 310a2b2e5a52143ebcfe19dbb0b0b2360aff285a Mon Sep 17 00:00:00 2001 From: yibo Date: Tue, 18 Aug 2026 17:23:34 +0800 Subject: [PATCH] fix(agent-chat): 会话池按次回收，堵住心跳每 5 分钟漏一个 claude CLI 进程 --- server/routes/agent-chat.js | 69 +++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------- server/tests/unit/agent-chat-session-reclaim.test.js | 85 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ system/agents/zz-session-reclaim/agent.md | 1 + system/agents/zz-session-reclaim/config.json | 6 ++++++ system/agents/zz-session-reclaim/identity.md | 1 + 5 files changed, 148 insertions(+), 14 deletions(-) create mode 100644 server/tests/unit/agent-chat-session-reclaim.test.js create mode 100644 system/agents/zz-session-reclaim/agent.md create mode 100644 system/agents/zz-session-reclaim/config.json create mode 100644 system/agents/zz-session-reclaim/identity.md diff --git a/server/routes/agent-chat.js b/server/routes/agent-chat.js index 58844f0..53d9713 100644 --- a/server/routes/agent-chat.js +++ b/server/routes/agent-chat.js @@ -19,6 +19,7 @@ const { randomUUID } = require('crypto'); const multer = require('multer'); const { ClaudeCliProtocol } = require('@mas/acp-core'); const { Parser } = require('../src/core/parser'); +const { IdleTimer } = require('../src/engine/idle-timer'); const AgentManager = require('../src/core/agent-manager'); const locator = require('../services/agent-locator'); const { OkrService } = require('../services/okr-service'); @@ -742,14 +743,56 @@ function listConversations(agentId) { // ACP Session 池 — 支持多轮 + 上下文恢复 // ═══════════════════════════════════════ -// key: `${agentId}:${conversationId}` → { protocol, messageCount } +// key: `${agentId}:${conversationId}` → { protocol, messageCount, idleTimer } const sessions = new Map(); +// ── 会话回收 ── +// sessions 里存的是**常驻**的 claude CLI 子进程（stream-json 模式，不是一次性的 `claude -p`）。 +// 在此之前只有四条回收路径：协议 error / 子进程 disconnected / 手动删对话 / 手动 stop —— +// 也就是说「正常跑完的对话」永远不回收。心跳每 5 分钟用一个**全新** convId 打进来 +// （heartbeat-service.js:206），于是每次心跳都新起一个进程且永不释放，实测 2h40m 攒了 82 个 +// 进程 / 17.5 GB。所以：心跳跑完立刻回收，普通对话按空闲 TTL 回收。 +const SESSION_IDLE_MS = Number(process.env.AGENT_SESSION_IDLE_MS) || 30 * 60 * 1000; + +/** 只从池子里摘掉（子进程已经自己死了的场景），顺手停表 */ +function dropSession(key) { + const entry = sessions.get(key); + entry?.idleTimer?.cancel(); + sessions.delete(key); + return entry; +} + +/** 摘掉 + 杀子进程 */ +function closeSession(key, reason) { + const entry = dropSession(key); + if (!entry) return false; + try { entry.protocol.disconnect(); } catch { } + console.log(`[AgentChat] 🧹 会话已回收 ${key} (${reason})`); + return true; +} + +/** + * 请求结束后给会话上弦；请求开始时由 getOrCreateProtocol 停表。 + * 计时器只在「没有在途请求」的间隙走，所以再长的单次生成也不会被 TTL 打断 + * —— 这比在流式回调里 touch() 少一处需要维护的调用点。 + */ +function armSessionIdle(key) { + const entry = sessions.get(key); + if (!entry) return; + entry.idleTimer?.cancel(); + entry.idleTimer = new IdleTimer( + SESSION_IDLE_MS, + () => closeSession(key, `空闲超过 ${Math.round(SESSION_IDLE_MS / 60000)} 分钟`), + { label: `agent-session:${key}` } + ).start(); +} + async function getOrCreateProtocol(agentId, convId) { const key = `${agentId}:${convId}`; let entry = sessions.get(key); if (entry && entry.protocol.isConnected) { + entry.idleTimer?.cancel(); // 请求在途期间不计空闲 return entry; } @@ -801,12 +844,12 @@ async function getOrCreateProtocol(agentId, convId) { // 绑定错误监听器，防止 EventEmitter 抛出未捕获异常导致整个 Node 进程奔溃 (例如 spawn claude ENOENT) protocol.on('error', (err) => { console.error(`[AgentChat] 🚨 ${identity.name} 协议级异常:`, err.message); - sessions.delete(key); // 下次提问将重建进程 + dropSession(key); // 下次提问将重建进程 releaseLock(); }); protocol.on('disconnected', ({ code, signal }) => { - sessions.delete(key); + dropSession(key); releaseLock(); console.log(`[AgentChat] 💤 ${identity.name} 会话断开 (code=${code}, signal=${signal})`); }); @@ -820,7 +863,7 @@ async function getOrCreateProtocol(agentId, convId) { await protocol.initialize(); await protocol.newSession({ cwd }); - const newEntry = { protocol, messageCount: 0, agentId, convId }; + const newEntry = { protocol, messageCount: 0, agentId, convId, idleTimer: null }; sessions.set(key, newEntry); console.log(`[AgentChat] 🧬 ${identity.name} ACP 会话已创建 (conv: ${convId})`); @@ -1337,10 +1380,7 @@ router.delete('/:agentId/conversations/:convId', (req, res) => { const uploadDir = getUploadDir(agentId, convId); if (fs.existsSync(uploadDir)) fs.rmSync(uploadDir, { recursive: true, force: true }); // 4. 断开会话 - const key = `${agentId}:${convId}`; - const entry = sessions.get(key); - if (entry && entry.protocol) entry.protocol.disconnect(); - sessions.delete(key); + closeSession(`${agentId}:${convId}`, '对话被删除'); console.log(`[AgentChat] 🗑️ 已删除对话 ${agentId}/${convId} (含uploads)`); res.json({ success: true }); } catch (err) { res.status(500).json({ error: err.message }); } @@ -1350,10 +1390,7 @@ router.delete('/:agentId/conversations/:convId', (req, res) => { router.post('/:agentId/chat/:convId/stop', (req, res) => { const { agentId, convId } = req.params; const key = `${agentId}:${convId}`; - const entry = sessions.get(key); - if (entry && entry.protocol) { - entry.protocol.disconnect(); // 关闭底层 claude cli 进程 - sessions.delete(key); // 下次从冷启动重新加载 + if (closeSession(key, '用户手动中断')) { return res.json({ success: true, message: 'Agent 已中断' }); } res.json({ success: false, message: '未找到运行中的 Agent 进程' }); @@ -2709,14 +2746,18 @@ ${resultsText.join('\n\n')} saveMeta(agentId, convId, meta); // 清理断开的 session - const key = `${agentId}:${convId}`; - sessions.delete(key); + dropSession(`${agentId}:${convId}`); sendSSE('error', { message: err.message }); console.log(`[SSE-DEBUG] 🚨 sendSSE('error') called:`, err.message.substring(0, 200)); } finally { agentChatLocks.delete(agentId); res.end(); + // 心跳会话是一次性的（heartbeat-service 每次都生成新 convId），留在池子里就是纯泄漏。 + // 普通对话保留会话复用能力，只上空闲 TTL。 + const sessionKey = `${agentId}:${convId}`; + if (req.body.isHeartbeat) closeSession(sessionKey, '心跳结束'); + else armSessionIdle(sessionKey); } }); diff --git a/server/tests/unit/agent-chat-session-reclaim.test.js b/server/tests/unit/agent-chat-session-reclaim.test.js new file mode 100644 index 0000000..1345b39 --- /dev/null +++ b/server/tests/unit/agent-chat-session-reclaim.test.js @@ -0,0 +1,85 @@ +/** + * 锁的是 claude CLI 子进程泄漏 —— 实测 2h40m 攒了 82 个进程 / 17.5 GB。 + * + * agent-chat 的 `sessions` Map 存的是**常驻**的 claude CLI 子进程（stream-json 模式， + * 不是一次性的 `claude -p`）。修之前只有四条回收路径：协议 error / 子进程 disconnected / + * 手动删对话 / 手动 stop —— 「正常跑完的对话」不在其中。而 heartbeat-service 每 5 分钟 + * 用一个**全新** convId（`heartbeat-${Date.now()}`）打进来，于是每次心跳都新起一个进程且 + * 永不释放。cto / coo 两个本体的 intervalMinutes 都是 5，正好对上观测到的「每 5 分钟涨 2-3 个」。 + * + * 探针用现成的 `/chat/:convId/stop`：它只在 sessions 里查得到条目时才返回 success:true， + * 所以不用为测试往模块里加导出口。断言全是「会话还在不在」，不看 chat 自己的响应内容。 + */ +process.env.LEMONBOSS_CLAUDE_MODE = 'mock'; // 必须早于 require：acp-core 的 mock 开关是模块级常量 +process.env.AGENT_SESSION_IDLE_MS = '400'; // 同理，SESSION_IDLE_MS 是模块级常量 + +import { describe, it, expect, beforeAll, afterAll } 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-session-reclaim'; +const BODY = path.join(ROOT, 'system', 'agents', AGENT); + +describe('agent-chat · 会话回收（心跳即时回收 + 空闲 TTL）', () => { + let base, srv; + + beforeAll(async () => { + 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)); + // 本体定义（config/identity/agent.md）留下，与 zz-tenant-* 同一约定：测试建出来的 agent 也是 + // 注册表条目，建了又删会让并行用例撞上「列到了名字、进去时已经没了」的 flaky。只清运行态。 + for (const d of ['conversations', '_global', 'memory']) fs.rmSync(path.join(BODY, d), { recursive: true, force: true }); + }); + + /** 跑完一轮对话（SSE 要读到流结束，否则 finally 还没执行） */ + const chat = async (convId, extra = {}) => { + const res = await fetch(`${base}/api/agent/${AGENT}/chat`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ message: 'ping', conversationId: convId, ...extra }), + }); + await res.text(); + }; + + /** 探针：会话是否还留在池子里 */ + const alive = async (convId) => { + const res = await fetch(`${base}/api/agent/${AGENT}/chat/${convId}/stop`, { method: 'POST' }); + return (await res.json()).success === true; + }; + + const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + + it('心跳会话跑完立刻回收（这是泄漏的正主）', async () => { + const conv = `heartbeat-${Date.now().toString(36)}`; + await chat(conv, { isHeartbeat: true }); + expect(await alive(conv)).toBe(false); + }); + + it('普通对话保留会话（复用能力不能被顺手砍掉）', async () => { + const conv = `conv-keep-${Date.now().toString(36)}`; + await chat(conv); + expect(await alive(conv)).toBe(true); + }); + + it('普通对话空闲超过 TTL 后被回收', async () => { + const conv = `conv-ttl-${Date.now().toString(36)}`; + await chat(conv); + await sleep(700); // > AGENT_SESSION_IDLE_MS + expect(await alive(conv)).toBe(false); + }); +}); diff --git a/system/agents/zz-session-reclaim/agent.md b/system/agents/zz-session-reclaim/agent.md new file mode 100644 index 0000000..8ffff49 --- /dev/null +++ b/system/agents/zz-session-reclaim/agent.md @@ -0,0 +1 @@ +# 会话回收单测用的常驻本体 diff --git a/system/agents/zz-session-reclaim/config.json b/system/agents/zz-session-reclaim/config.json new file mode 100644 index 0000000..6259e02 --- /dev/null +++ b/system/agents/zz-session-reclaim/config.json @@ -0,0 +1,6 @@ +{ + "id": "zz-session-reclaim", + "type": "skill", + "name": "zz-session-reclaim", + "status": "active" +} \ No newline at end of file diff --git a/system/agents/zz-session-reclaim/identity.md b/system/agents/zz-session-reclaim/identity.md new file mode 100644 index 0000000..16b5f4d --- /dev/null +++ b/system/agents/zz-session-reclaim/identity.md @@ -0,0 +1 @@ +# zz-session-reclaim -- libgit2 0.21.4