/** * SessionPool — 可复用会话池管理 * * 职责: * - 会话的创建 / 获取 / 重置 / 关闭 * - 生命周期事件通知 * - 会话统计 * * 不负责: * - Protocol 选择(由 protocol-factory 决定) * - Prompt 执行(由 stream-executor 决定) * - Agent 配置查询(由 AgentManager 决定) */ const { EventEmitter } = require('events'); const { randomUUID } = require('crypto'); const { createProtocol } = require('./protocol-factory'); class SessionPool extends EventEmitter { constructor() { super(); this.sessions = new Map(); this.activeKey = null; } /** * 生成会话唯一 Key * * key 里必须带 (companyId, taskId)(28 §6 批次 3-12):Agent 本体合并成一份之后 * `agentName` 不再唯一标识一次运行 —— 两家公司调同一个本体,key 相同就会**复用同一个 * CLI 会话**,A 公司的上下文直接漏进 B 公司的对话,而且两边都成功返回、不报错。 * 不能靠 workDir 顺带带上:body 模式的 cwd 确实含这两段,legacy / scoped skill 模式的 * 不含(cwd 分别是公司私有目录和本体目录),指望「路径里恰好有」就是又一份手抄的约定。 * 缺省值与 agent-locator 的保留值同名,含义一致:没有公司 / 没有任务。 */ getKey(agentName, agentConfig, workDir, scope) { const runtime = `${agentConfig.companyId || '_global'}/${agentConfig.taskId || '_no-task'}`; const base = `acp:${agentName}:${agentConfig.client}:${agentConfig.deploy}:${runtime}:${workDir}`; return scope ? `${base}:${scope}` : base; } /** * 创建新会话(Launcher + Transport + Protocol 全链路) * * @param {string} agentName * @param {object} agentConfig - { type, client, deploy, ... } * @param {object} launcherConfig - 来自 runtime-resolver * @param {string} protocolName - acp / claude-stream / codex-jsonl * @param {string} workDir * @param {string} [scope] - 会话细分维度(如 conversationId);必须与 getOrCreate 查询时用的一致 * @returns {Promise} session object */ async create(agentName, agentConfig, launcherConfig, protocolName, workDir, scope) { // scope 必须透传:不透传的话 getOrCreate 用「带 scope 的 key」查、create 用 // 「不带 scope 的 key」存,两个 key 永远对不上 —— 查不到就再建一个,会话池里 // 堆的全是查不到的死会话,且每次调用都新起一个 CLI 进程,同样不报错。 const sessionKey = this.getKey(agentName, agentConfig, workDir, scope); this._emit(agentName, protocolName, 'initializing', null, sessionKey); // 通过工厂创建 Protocol(无 if/else) const protocol = createProtocol( protocolName, { id: agentName, name: agentName }, { yolo: true } ); let launcher = null; if (protocolName === 'claude-cli') { // ClaudeCliProtocol 自行管理子进程(每次 sendPrompt 启动 claude -p) // 不需要 launcher/transport await protocol.connect(null, workDir); await protocol.initialize(); await protocol.newSession({ cwd: workDir }); } else { // 其他协议:组合 Launcher + Transport + JSON-RPC const { createConnection } = require('@mas/acp-core'); const { transport, launcher: lnch } = await createConnection( agentConfig, launcherConfig, workDir ); launcher = lnch; await protocol.connect(transport, workDir); await protocol.initialize(); await protocol.newSession({ cwd: workDir }); } const sessionId = protocol.sessionId || `${protocolName}-${agentName}-${randomUUID()}`; if (!protocol.sessionId) { protocol.sessionId = sessionId; } this._emit(agentName, protocolName, 'ready', sessionId, sessionKey); const session = { key: sessionKey, agentName, protocolName, protocol, launcher, sessionId, callQueue: Promise.resolve(), }; this.sessions.set(sessionKey, session); return session; } /** * 获取已有会话或创建新会话 */ async getOrCreate(agentName, agentConfig, launcherConfig, protocolName, workDir, scope) { const key = this.getKey(agentName, agentConfig, workDir, scope); const existing = this.sessions.get(key); if (existing?.protocol?.isConnected) { this._emit(agentName, protocolName, 'reused', existing.sessionId, key); return existing; } // 清理失效会话 if (existing?.protocol) { try { existing.protocol.disconnect(); } catch { } if (existing.launcher) { try { await existing.launcher.stop(); } catch { } } this.sessions.delete(key); } return this.create(agentName, agentConfig, launcherConfig, protocolName, workDir, scope); } /** * 重置(断开 + 清理)单个会话 */ async reset(sessionKey) { const session = this.sessions.get(sessionKey); if (session?.protocol) { this._emit( session.agentName, session.protocolName, 'closed', session.sessionId, sessionKey, 'reset' ); try { session.protocol.disconnect(); } catch { } if (session.launcher) { try { await session.launcher.stop(); } catch { } } } this.sessions.delete(sessionKey); if (this.activeKey === sessionKey) { this.activeKey = null; } } /** * 关闭所有会话(并行) */ async closeAll() { const keys = [...this.sessions.keys()]; if (keys.length === 0) return; console.log(`[AgentManager] 关闭 ${keys.length} 个会话...`); await Promise.allSettled(keys.map(key => this.reset(key))); this.sessions.clear(); this.activeKey = null; console.log(`[AgentManager] 所有会话已关闭`); } /** * 会话池统计 */ getStats() { const list = []; for (const [key, session] of this.sessions) { list.push({ key, agentName: session.agentName, protocol: session.protocolName, sessionId: session.sessionId, connected: !!session.protocol?.isConnected, active: this.activeKey === key, }); } return { total: list.length, connected: list.filter(s => s.connected).length, active: this.activeKey, sessions: list, }; } /** * 终止当前活跃会话 * @returns {object|null} 被终止的会话信息,或 null */ killActive() { if (!this.activeKey) return null; const session = this.sessions.get(this.activeKey); if (!session?.protocol) return null; this._emit( session.agentName, session.protocolName, 'closed', session.sessionId, session.key, 'killed' ); session.protocol.disconnect(); this.sessions.delete(this.activeKey); this.activeKey = null; return session; } /** @private */ _emit(agentName, protocol, status, sessionId, sessionKey, reason) { const payload = { agentName, protocol, status, sessionId, sessionKey }; if (reason) payload.reason = reason; this.emit('session_event', payload); } } module.exports = { SessionPool };