diff --git a/server/src/core/agent-manager.js b/server/src/core/agent-manager.js index 45daa75..5536bf4 100644 --- a/server/src/core/agent-manager.js +++ b/server/src/core/agent-manager.js @@ -438,6 +438,9 @@ class AgentManager extends EventEmitter { }); this.sessionPool.activeKey = null; + // 调用结束 = 会话进入空闲,上弦。不上的话它就永远留在池子里: + // getKey 带 taskId,下一个图节点是新 key,这个会话再没人查得到,也再没人杀得掉。 + this.sessionPool.armIdle(session.key); if (logSession) await logSession.logAgentEnd(agentName, stdout, 'success'); return { agent: agentName, output: stdout, status: 'success' }; @@ -463,6 +466,9 @@ class AgentManager extends EventEmitter { } if (logSession) await logSession.logAgentError(agentName, err); + // 报错退出同样要上弦 —— 「跑挂的会话」正是 agent-chat 侧漏得最凶的那一类: + // 子进程活得好好的,错的是 API 调用,没人回收就是纯泄漏。 + this.sessionPool.armIdle(session.key); throw new Error(`Agent ${agentName} (${protocolName}) failed: ${err.message}`); } } diff --git a/server/src/core/session-pool.js b/server/src/core/session-pool.js index f076b0a..86286e3 100644 --- a/server/src/core/session-pool.js +++ b/server/src/core/session-pool.js @@ -15,6 +15,22 @@ const { EventEmitter } = require('events'); const { randomUUID } = require('crypto'); const { createProtocol } = require('./protocol-factory'); +const { IdleTimer } = require('../engine/idle-timer'); + +/** + * 会话空闲上限。与 routes/agent-chat.js 的 `sessions` 池共用同一个旋钮和同一个含义 + * —— 两边存的都是**常驻**的 claude CLI 子进程,回收判据也一样。 + * + * 在此之前这个池子只有三条回收路径:换新会话 / reset() / killActive()。 + * 「正常跑完」和「不可重试的报错」都不在其中,而 getKey 里带了 taskId + * —— 每个图节点一个新 key,于是一节点一进程、永不释放,且不报错。 + * 实测 23 小时漏出 7 个图执行进程(另有 38 个来自 agent-chat 侧的另一个洞)。 + * + * 每次读 env 而不是模块加载时读一次:单测要能改这个值,不然只能等 30 分钟。 + */ +function idleMs() { + return Number(process.env.AGENT_SESSION_IDLE_MS) || 30 * 60 * 1000; +} class SessionPool extends EventEmitter { constructor() { @@ -101,6 +117,7 @@ class SessionPool extends EventEmitter { launcher, sessionId, callQueue: Promise.resolve(), + idleTimer: null, // 由 armIdle 上弦;新建会话在首次调用期间故意不计时 }; this.sessions.set(sessionKey, session); @@ -115,6 +132,7 @@ class SessionPool extends EventEmitter { const existing = this.sessions.get(key); if (existing?.protocol?.isConnected) { + existing.idleTimer?.cancel(); // 请求在途期间不计空闲 this._emit(agentName, protocolName, 'reused', existing.sessionId, key); return existing; } @@ -132,10 +150,37 @@ class SessionPool extends EventEmitter { } /** + * 调用结束后给会话上弦;调用开始时由 getOrCreate 停表。 + * + * 计时器只在「没有在途请求」的间隙走,所以再长的单次生成也不会被 TTL 打断 + * —— 与 agent-chat 侧同一套写法,比在流式回调里 touch() 少一处要维护的调用点。 + * + * ponytail: 这是**兜底**不是精算 —— 一次性 taskId(图节点跑完就不会再用)也要 + * 白占一个 TTL 周期的内存,天花板是「并发会话数 × 单进程 RSS(实测约 300MB)× TTL」。 + * 要更紧的话,调用方在明知 taskId 一次性时直接 reset(key) 立刻释放;这里不替它判断, + * 因为 devloop 的 retry-with-reflection 复用同一个 taskId,一刀切关掉会砍掉复用能力。 + */ + armIdle(sessionKey) { + const session = this.sessions.get(sessionKey); + if (!session) return; + session.idleTimer?.cancel(); + const ms = idleMs(); + session.idleTimer = new IdleTimer( + ms, + () => { + console.log(`[SessionPool] 🧹 会话空闲超过 ${Math.round(ms / 60000)} 分钟,回收 ${sessionKey}`); + this.reset(sessionKey).catch(() => {}); + }, + { label: `pool-session:${sessionKey}` } + ).start(); + } + + /** * 重置(断开 + 清理)单个会话 */ async reset(sessionKey) { const session = this.sessions.get(sessionKey); + session?.idleTimer?.cancel(); if (session?.protocol) { this._emit( session.agentName, session.protocolName, 'closed', @@ -197,6 +242,7 @@ class SessionPool extends EventEmitter { if (!this.activeKey) return null; const session = this.sessions.get(this.activeKey); if (!session?.protocol) return null; + session.idleTimer?.cancel(); this._emit( session.agentName, session.protocolName, 'closed', diff --git a/server/tests/unit/session-pool-idle-reclaim.test.js b/server/tests/unit/session-pool-idle-reclaim.test.js new file mode 100644 index 0000000..aff9cff --- /dev/null +++ b/server/tests/unit/session-pool-idle-reclaim.test.js @@ -0,0 +1,101 @@ +/** + * SessionPool 空闲回收守卫 + * + * 背景:这个池子存的是**常驻**的 claude CLI 子进程,而 getKey 里带了 taskId + * —— 每个图节点一个新 key,跑完之后再没人查得到那个会话,也再没人杀得掉它。 + * 修之前池子只有三条回收路径(换新会话 / reset / killActive),「正常跑完」和 + * 「不可重试的报错」都不在其中,实测 23 小时漏出 7 个进程(约 2GB RSS)。 + * + * 症状是**静默的**:不报错、不写日志、git status 干净,只有内存一路涨。所以这里盯三条: + * 1. 上弦之后到点必须真的杀子进程 + 摘出池子 + * 2. 会话被复用时必须停表(否则长任务跑到一半被 TTL 打断,比泄漏更难查) + * 3. reset 之后计时器不许还活着(否则回收完还会再摸一次已经不存在的会话) + * + * 不起真 CLI:直接把假会话塞进 pool.sessions,只验回收机制本身。 + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { createRequire } from 'node:module'; + +const require_ = createRequire(import.meta.url); +const { SessionPool } = require_('../../src/core/session-pool'); + +const TTL = 30; // ms —— idleMs() 每次读 env,所以能在测试里调小 +const AGENT = 'graph-node-agent'; +const CONFIG = { client: 'claude', deploy: 'local', companyId: 'e2e-demo', taskId: 'run_dead_beef' }; +const WORKDIR = '/tmp/does-not-need-to-exist'; + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + +/** 一个只记录 disconnect 有没有被调过的假 protocol */ +function fakeSession(pool, { connected = true } = {}) { + const key = pool.getKey(AGENT, CONFIG, WORKDIR); + const session = { + key, + agentName: AGENT, + protocolName: 'claude-cli', + protocol: { isConnected: connected, sessionId: 'sess-1', disconnected: false, + disconnect() { this.disconnected = true; this.isConnected = false; } }, + launcher: null, + sessionId: 'sess-1', + callQueue: Promise.resolve(), + idleTimer: null, + }; + pool.sessions.set(key, session); + return { key, session }; +} + +describe('SessionPool 空闲回收', () => { + let prevEnv; + let pool; + + beforeEach(() => { + prevEnv = process.env.AGENT_SESSION_IDLE_MS; + process.env.AGENT_SESSION_IDLE_MS = String(TTL); + pool = new SessionPool(); + }); + + afterEach(() => { + if (prevEnv === undefined) delete process.env.AGENT_SESSION_IDLE_MS; + else process.env.AGENT_SESSION_IDLE_MS = prevEnv; + }); + + it('上弦后到点:子进程被杀,会话从池子里摘掉', async () => { + const { key, session } = fakeSession(pool); + + pool.armIdle(key); + expect(session.protocol.disconnected).toBe(false); // 还没到点,不许提前动手 + + await sleep(TTL * 3); + + expect(session.protocol.disconnected).toBe(true); + expect(pool.sessions.has(key)).toBe(false); + }); + + it('会话被复用时停表 —— 在途请求不会被 TTL 打断', async () => { + const { key, session } = fakeSession(pool); + + pool.armIdle(key); + const reused = await pool.getOrCreate(AGENT, CONFIG, {}, 'claude-cli', WORKDIR); + expect(reused).toBe(session); // 命中复用,没有新建 + + await sleep(TTL * 3); + + expect(session.protocol.disconnected).toBe(false); + expect(pool.sessions.has(key)).toBe(true); + }); + + it('reset 之后计时器不再触发', async () => { + const { key, session } = fakeSession(pool); + + pool.armIdle(key); + await pool.reset(key); + expect(session.protocol.disconnected).toBe(true); + expect(pool.sessions.has(key)).toBe(false); + + // 看 IdleTimer.fired 而不是「有没有再杀一次子进程」:会话已经不在池子里了, + // 超时回调本来就空转,拿 disconnected 断言等于什么都没测(写第一版时就是这么假绿的)。 + // fired 直接答的是「这只表还在不在走」—— 没取消的话它会留一个吊着的 setTimeout。 + await sleep(TTL * 3); + expect(session.idleTimer.fired).toBe(false); + }); +});