/** * Agent Manager — 统一 Agent 调用管理器(薄门面) * * 职责: * - Agent 配置查询(Registry + 静态配置) * - callAgent 调用入口 + Hook 编排 * - 协议执行 + 重试策略 * - 日志管理 * * 委托: * - 会话池管理 → SessionPool * - Protocol 实例化 → protocol-factory * - Prompt 执行 → stream-executor * - 运行时推断 → runtime-resolver */ const { EventEmitter } = require('events'); const fs = require('fs'); const path = require('path'); const { PromptBuilder } = require('./prompt-builder'); const { getGlobalLogManager, waitForGlobalLogManager } = require('./logging'); const { HookRunner } = require('./hook-runner'); const { resolve: resolveRuntime } = require('./runtime-resolver'); const { SessionPool } = require('./session-pool'); const locator = require('../../services/agent-locator'); const { runPromptInSession } = require('./stream-executor'); const { RunRegistry: EngineRunRegistry, FailoverError, } = require('../engine'); /** * 无流式数据的空闲上限,与 stream-executor.runPromptInSession 的 JSDoc 契约一致(30 分钟)。 * 真正的超时管理由上层心跳监控负责,这里只是兜底 —— 一个 agent 读几千行源码再动笔, * 中间静默十几分钟是正常的,不该被当成卡死。 */ const DEFAULT_IDLE_TIMEOUT_MS = 1_800_000; class AgentManager extends EventEmitter { constructor(config, workDir, options = {}) { super(); this.workDir = workDir; this.config = typeof config === 'string' ? JSON.parse(fs.readFileSync(config, 'utf-8')) : config; this.baseDir = path.join(__dirname, '../..'); this.promptBuilder = new PromptBuilder(this.config, this.baseDir); this.hookRunner = new HookRunner(this.config, this.baseDir); // 会话池 this.sessionPool = new SessionPool(); this.sessionPool.on('session_event', (payload) => this._emitSessionEvent(payload)); // Registry 集成 this.registry = options.registry || null; if (this.registry) this._bindRegistryEvents(); // Engine RunRegistry this.runRegistry = options.runRegistry || new EngineRunRegistry(); // 日志 this.sessionId = options.sessionId || `agent-${Date.now()}`; this.logManager = options.logManager || getGlobalLogManager(); this.logSession = options.logSession || null; this._logInitPromise = null; } // ═══════════════════════════════════════════════════ // 公共接口 // ═══════════════════════════════════════════════════ /** * 调用 Agent(主入口) */ async callAgent(agentName, task, options = {}) { const { onData, onError, onAcpEvent, sessionId, sessionScope } = options; // 日志会话 const logSession = await this._resolveLogSession(sessionId); // options 一路带进去:运行态目录的第三层键 taskId 就藏在 options.metadata 里(28 §5.1) const { agent, client } = this.getAgentConfig(agentName, options); if (!client) throw new Error(`Agent "${agentName}" 无法执行:缺少 client 配置`); const resolved = resolveRuntime(agent, client); const protocolName = resolved.protocol; console.log(`\n[${agentName}] 启动 (Client: ${agent.client}, Deploy: ${agent.deploy}, Protocol: ${protocolName})...`); if (logSession) { await logSession.logAgentStart(agentName, task, { client: agent.client, deploy: agent.deploy, protocol: protocolName, ...options.metadata, }); } // Hook 上下文 const hookCtx = { agentName, agentType: agent.type, sessionId: sessionId || this.sessionId, taskId: options.metadata?.taskId || `task-${Date.now()}`, workDir: this.workDir, taskPrompt: task.substring(0, 1000), }; try { await this.hookRunner.run('pre-task', agent, hookCtx); // agent 对象显式串进去,不让 PromptBuilder 再按名字回查共享 config: // 回查拿到的是「最后一次解析」的那份运行态坐标,并发时会是别人的 cwd。 const fullPrompt = this.buildFullPrompt(agentName, task, agent); const startTime = Date.now(); const result = await this._runViaProtocol( agentName, agent, resolved, fullPrompt, { ...options, logSession } ); hookCtx.duration = Date.now() - startTime; hookCtx.exitCode = 0; hookCtx.outputSummary = (result?.output || '').substring(0, 500); await this.hookRunner.run('post-task-success', agent, hookCtx); return result; } catch (err) { hookCtx.exitCode = 1; hookCtx.errorMsg = err.message; await this.hookRunner.run('post-task-failure', agent, hookCtx).catch(console.error); throw err; } } /** * 获取 Agent 配置(Registry > 静态配置) * * @param {string} agentName 寻址名,恒为 `cid:aid`(无公司时是裸 `aid`) * @param {object} [options] callAgent 的 options —— 这里只取 taskId,见下 */ getAgentConfig(agentName, options = {}) { // 运行态目录的第三层键(28 §5.1)。调用方把它塞在 metadata 里 // (graph-engine.defaultAgentExecutor 传的是 ctx.runId),所以两种位置都收。 // 取不到就交给 locator 落 `_no-task`:**绝不能拼出 `.../undefined/`** —— // 那种目录建得出来、CLI 也起得来,只是所有无 taskId 的调用叠在一个坑里且不报错。 const taskId = options.taskId || options.metadata?.taskId || null; if (this.registry) { const registered = this.registry.getAgent(agentName); if (registered) { // registry 命中也必须过 _applyRuntimeScope。这条分支原来直接 return, // 于是 companyId / taskId / companyDir / workDir 一个都不挂,三件事同时失守且不报错: // ① session key 的 runtime 段恒为 `_global/_no-task` → 两家公司复用同一个 CLI 会话; // ② _runViaProtocol 的 `agent.workDir || this.workDir` 退到**仓库根**当 cwd // → agent 直接在源码树里跑,边界守卫必炸; // ③ companyDir 为 null → 公司语境 / user.md / memory 一行不进提示词。 // 生产里眼下没人传 options.registry,所以是「接上就全线失守」的埋雷,不是活故障。 return { agent: this._scopeRegisteredAgent(agentName, registered.agent, taskId), client: registered.client, }; } } let agentConfig = this.config.agents?.[agentName]; // Dynamic Hot-Reload Fallback (Local File-System First) if (!agentConfig) { // 寻址收编到 agent-locator(28 号设计 §2.5)。地址恒为 `cid:aid` // (graph-engine.resolveAgentName 不再挑地址形态),`aid` 恒指向 system/agents/{aid} // 那一份本体 —— 批次 4 清空公司私有目录后,「这家公司有没有自己那一份」不再是个问题。 const { dir: agentDir, agentsRoot, agentId, companyId } = locator.resolveAgentDir(agentName); // 1. Check if it's an independent Agent if (fs.existsSync(agentDir)) { agentConfig = { type: 'orchestrator', protocol: 'claude-cli', transport: 'pipe', launcher: 'claude-local', client: 'claude-local', // 本体只有一份,prompt 恒指向它;公司语境(company-context.md)由 // _applyRuntimeScope 按 (agentId, companyId) 算,不再是 config.json 里的 // systemAgentRef 说了算(该字段随公司私有 config.json 一起废弃)。 prompt: path.join(agentDir, 'agent.md'), companyContextPath: null, description: `Dynamic Agent ${agentName}`, // 运行态坐标(G2)。cwd / 公司态目录由 _applyRuntimeScope 每次调用重算, // 这里只记「拿什么算」:cwd 下沉到 system/agents/{aid}/{cid}/{taskId}/。 runtimeMode: 'body', agentId, companyId, bodyDir: agentDir, companyDir: null, workDir: agentDir, }; } else { // 2. Check if it's a scoped skill inside any agent let foundPath = null; if (fs.existsSync(agentsRoot)) { for (const f of fs.readdirSync(agentsRoot, { withFileTypes: true })) { if (f.isDirectory() && !f.name.startsWith('.')) { const skillPath = path.join(agentsRoot, f.name, '.claude', 'skills', agentName, 'SKILL.md'); if (fs.existsSync(skillPath)) { foundPath = skillPath; break; } } } } if (foundPath) { // 找到 skill 所在的 Agent 目录,作为 CLI 工作目录 // {agentDir}/.claude/skills/{skillName}/SKILL.md → {agentDir}:要剥掉 4 层。 // 原来只剥 3 层,落在 `{agentDir}/.claude` 上 —— CLI 在那个 cwd 下找的是 // `.claude/.claude/skills/`,**永远发现不了这个 skill**,而进程照样起得来、 // 照样有输出,只是降级成了没有 skill 的通用会话。批次 3 之后这个错值还会 // 顺着新的 bodyDir 字段进提示词的「你的本体文件」坐标,把 agent 也带偏。 const ownerAgentDir = path.dirname(path.dirname(path.dirname(path.dirname(foundPath)))); agentConfig = { type: 'executor', protocol: 'claude-cli', transport: 'pipe', launcher: 'claude-local', client: 'claude-local', prompt: foundPath, workDir: ownerAgentDir, // CLI 进程在拥有该 skill 的 Agent 目录下运行 description: `Dynamic Skill ${agentName}`, // scoped skill 的 cwd **故意不下沉**:它靠 cwd/.claude/skills/ 被 CLI 发现, // 挪到 {taskId}/ 就找不到自己了。但 (companyId, taskId) 仍要挂上 —— // 会话隔离靠的是 key 里这两维,不是靠 cwd 恰好不同(见 session-pool.getKey)。 runtimeMode: 'skill', agentId: path.basename(ownerAgentDir), companyId, bodyDir: ownerAgentDir, companyDir: null, }; } } // Cache dynamic configuration to prevent PromptBuilder from failing if (agentConfig) { this.config.agents = this.config.agents || {}; this.config.agents[agentName] = agentConfig; } } if (!agentConfig) throw new Error(`Agent not found: ${agentName}`); // 缓存命中也要重算:agentConfig 被缓存进 this.config.agents[agentName],而 taskId // 是**每次调用**才有的。只在首次寻址时算一次的话,第二个任务会跑进第一个任务的目录, // 而且目录存在、CLI 起得来、日志也照写 —— 全无报错,只是所有任务的痕迹叠在一起。 const scopedAgent = this._applyRuntimeScope(agentConfig, taskId); // 缓存槽里**换成**本次的快照,而不是就地改写原对象:PromptBuilder 只按名字回查 // config.agents,buildFullPrompt(agentName) 这条没有 agent 对象的老路径得有个地方 // 读到运行态坐标。换对象的关键在于:已经返回给上一次调用的那份不会被这次动到。 if (this.config.agents?.[agentName]) this.config.agents[agentName] = scopedAgent; // Fallback valid client let clientConfig = scopedAgent.client ? (this.config.clients?.[scopedAgent.client] || null) : null; if (!clientConfig) clientConfig = { url: 'local' }; // provide a default dummy client info if missing return { agent: scopedAgent, client: clientConfig }; } /** * registry 里查出来的 agent 补上运行态坐标。 * * registry 的 `_agents[name]` 同样是**共享缓存对象**(整个进程一份,热重载才换), * 所以这里也只产出副本、绝不回写。 * 自带 workDir 的配置(静态 config 搬进 DB 的那类)保留自己的 cwd 不下沉 —— * 但 (companyId, taskId) 照挂:会话隔离读的是这两个字段,不是 cwd 恰好不同。 * * @private */ _scopeRegisteredAgent(agentName, registeredAgent, taskId) { const { agentId, companyId } = locator.parseAddress(agentName); const scoped = { agentId, companyId, ...registeredAgent }; if (!scoped.runtimeMode) scoped.runtimeMode = registeredAgent.workDir ? 'pinned' : 'body'; if (scoped.runtimeMode === 'body' && !scoped.bodyDir) scoped.bodyDir = locator.bodyDir(agentId); return this._applyRuntimeScope(scoped, taskId); } /** * 把本次调用的运行态坐标(cwd / 公司态目录 / 会话隔离键)算到 agentConfig 上。 * * 只对动态寻址出来的 agent 生效(静态 config.json 里的 agent 自带 workDir,不下沉)。 * 三种模式: * - `body` 本体共享后的目标形态:cwd = system/agents/{aid}/{cid}/{taskId}/ * - `skill` scoped skill,cwd 必须留在拥有它的本体目录里 * - `pinned` registry 里自带 workDir 的配置,cwd 由配置说了算 * 三种模式都挂 (companyId, taskId):会话隔离读的是这两个字段,不是 cwd。 * * **返回副本,绝不就地改写入参**:入参是缓存在 this.config.agents[name](或 registry * 内存缓存)里的共享对象。就地改写的话,同一寻址名两次时间上重叠的调用中,后一次会把 * 前一次的 taskId / workDir 覆盖掉 —— 而前一次此刻还卡在 callAgent 的 await * (logAgentStart / hookRunner.run)上,还没走到 _runViaProtocol 去读 agent.workDir。 * 醒来后两次调用塌进同一个 cwd + 同一个 sessionKey:复用同一个长驻 CLI 子进程、 * 上下文互串,而且两边都成功返回。跨公司因为缓存键含 cid 侥幸躲过,**同公司跨任务**必中。 * * @private */ _applyRuntimeScope(agentConfig, taskId) { if (!agentConfig?.runtimeMode) return agentConfig; const scoped = { ...agentConfig, taskId }; if (scoped.runtimeMode !== 'body') return scoped; scoped.companyDir = locator.companyDir(scoped.agentId, scoped.companyId); scoped.companyContextPath = path.join(scoped.companyDir, 'company-context.md'); scoped.workDir = locator.taskDir(scoped.agentId, scoped.companyId, taskId); // cwd 不存在 CLI 直接起不来(recursive 顺带把公司态目录一并建出来) fs.mkdirSync(scoped.workDir, { recursive: true }); return scoped; } /** * 构建完整 prompt * * @param {object} [agent] 本次调用解析出来的 agent 对象。传了就用它,不再让 PromptBuilder * 按名字回查共享 config —— 回查到的是「最后一次解析」的坐标,并发时属于别的任务。 * PromptBuilder 只认名字,所以做法是给它一份**只覆盖这一个名字**的 config 视图。 */ buildFullPrompt(agentName, task, agent = null) { if (!agent) return this.promptBuilder.build(agentName, task); const cfg = this.promptBuilder.config; const view = new PromptBuilder( { ...cfg, agents: { ...cfg.agents, [agentName]: agent } }, this.baseDir ); return view.build(agentName, task); } /** * 终止当前运行的进程/会话 */ killCurrentProcess() { const killed = this.sessionPool.killActive(); return !!killed; } /** * 获取可用 Agent 列表 */ listAgents() { if (this.registry) { return this.registry.listAgents().map(a => ({ key: a.name, type: a.type, client: a.client, deploy: a.deploy, description: a.description, })); } return Object.entries(this.config.agents || {}).map(([key, config]) => ({ key, type: config.type, client: config.client, deploy: config.deploy, description: config.description, })); } /** * 预热会话池 */ async prewarmReusableSessions(agentNames = null) { let names; if (Array.isArray(agentNames)) { names = agentNames; } else if (this.config.agents && Object.keys(this.config.agents).length > 0) { // 优先用当前配置中的 agents(已按组过滤) names = Object.keys(this.config.agents); } else if (this.registry) { names = this.registry.listAgents().map(a => a.name); } else { names = []; } const warmed = []; for (const name of names) { try { const { agent, client } = this.getAgentConfig(name); if (!client) continue; const resolved = resolveRuntime(agent, client); await this.sessionPool.getOrCreate( name, agent, resolved.launcherConfig, resolved.protocol, this.workDir ); warmed.push(name); } catch (err) { console.warn(`[AgentManager] prewarm 跳过 ${name}: ${err.message}`); } } if (warmed.length > 0) console.log(`[AgentManager] 预热完成: ${warmed.join(', ')}`); return warmed; } // ── 委托方法 ── closeAllSessions() { return this.sessionPool.closeAll(); } getSessionStats() { return this.sessionPool.getStats(); } // ═══════════════════════════════════════════════════ // 内部方法 // ═══════════════════════════════════════════════════ /** * 协议执行 + 失败重试 * @private */ async _runViaProtocol(agentName, agent, resolved, fullPrompt, options) { const logSession = options.logSession || this._getLogSession(); const protocolName = resolved.protocol; // 优先使用 Agent 自带的 workDir —— body 模式下它已经是任务态目录 // system/agents/{aid}/{cid}/{taskId}/(getAgentConfig 已 mkdir,cwd 不存在 CLI 起不来)。 // 会话隔离**主要**靠 agent.companyId / agent.taskId 进 sessionPool.getKey; // 但 (companyId, taskId) 只到「一次 run」的粒度——同一 run 内两个并发节点引用同一个 // system agent(fan-out 撞同 agentId,如 executor-fanout.json 的 impl-a/impl-b)时 // taskId 相同、会话键相同,会互相顶掉对方的 CLI 子进程。调用方(graph-engine)按 // 节点运行实例 id 传 options.sessionScope 来加一维,这里原样透传给 getKey 的第四个参数。 const effectiveWorkDir = agent.workDir || this.workDir; const sessionScope = options.sessionScope; let session = await this.sessionPool.getOrCreate( agentName, agent, resolved.launcherConfig, protocolName, effectiveWorkDir, sessionScope ); for (let attempt = 0; attempt < 2; attempt++) { try { this.sessionPool.activeKey = session.key; // 优先级:调用方显式传入 > 该 agent 的配置 > 默认值。 // // 修之前这里是 `this.config?...?.idleTimeout || 120_000`,且 `...options` 摊在前面、 // 被下一行的 idleTimeoutMs 覆盖 —— 两个后果叠在一起: // ① 默认 120 秒,比 stream-executor 自己 JSDoc 写的 30 分钟(1_800_000)小一个数量级, // 也和同进程 ClaudeCliProtocol 早已放到 24h 的 idle 上限完全对不上; // ② 调用方**无法覆写** —— 传了 options.idleTimeoutMs 也会被这一行盖掉。 // 真 CLI 下的实际后果:任何「先读几千行源码再动笔」的节点都在动笔前被杀 // (readme-writer 就是这么死的:idle timeout: no stream data for 120000ms), // 这是 27-P3B 第三次真跑「work 节点 → success」跑不通的唯一技术原因。 // // 「按静默时长杀 agent」在本仓库已经被判过一次死刑(ClaudeCliProtocol 那两个 // IDLE_TIMEOUT 常量的注释写着「已取消」),这里是漏网的一处。默认值对齐 // stream-executor 的契约,真正的超时管理交给上层心跳监控。 const idleTimeoutMs = options.idleTimeoutMs || this.config?.agents?.[agentName]?.idleTimeout || DEFAULT_IDLE_TIMEOUT_MS; const stdout = await runPromptInSession(session, fullPrompt, { ...options, logSession, idleTimeoutMs, }); 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' }; } catch (err) { this.sessionPool.activeKey = null; if (attempt === 0 && FailoverError.from(err).retryable) { this._emitSessionEvent({ agentName, protocol: protocolName, status: 'reconnecting', sessionId: session.protocol?.sessionId, sessionKey: session.key, reason: err.message, }); await this.sessionPool.reset(session.key); session = await this.sessionPool.getOrCreate( agentName, agent, resolved.launcherConfig, protocolName, effectiveWorkDir, sessionScope ); this._emitSessionEvent({ agentName, protocol: protocolName, status: 'reconnected', sessionId: session.protocol?.sessionId, sessionKey: session.key, }); continue; } if (logSession) await logSession.logAgentError(agentName, err); // 报错退出同样要上弦 —— 「跑挂的会话」正是 agent-chat 侧漏得最凶的那一类: // 子进程活得好好的,错的是 API 调用,没人回收就是纯泄漏。 this.sessionPool.armIdle(session.key); throw new Error(`Agent ${agentName} (${protocolName}) failed: ${err.message}`); } } } // ── 日志 ── /** @private */ async _resolveLogSession(sessionId) { if (sessionId) { await this._ensureLogManagerReady(); return this.logManager.getOrCreateSession(sessionId, { workDir: this.workDir }); } await this._ensureLogReady(); return this._getLogSession(); } /** @private */ async _ensureLogReady() { if (this.logSession) return this.logSession; if (!this._logInitPromise) { this._logInitPromise = waitForGlobalLogManager().then(() => { if (this.logManager.initialized && !this.logSession) { this.logSession = this.logManager.getOrCreateSession(this.sessionId, { workDir: this.workDir, }); } return this.logSession; }); } return this._logInitPromise; } /** @private */ async _ensureLogManagerReady() { if (this.logManager.initialized) return; if (this._logInitPromise) { await this._logInitPromise; } else if (typeof this.logManager.init === 'function') { await this.logManager.init(); } } /** @private */ _getLogSession() { return this.logSession; } // ── 事件 ── /** @private */ _emitSessionEvent(payload) { const agentType = payload.agentType || this.config?.agents?.[payload.agentName]?.type || null; const event = { type: 'agent_session', timestamp: new Date().toISOString(), agentType, ...payload, }; this.emit('session_event', event); console.log( `[Session] agent=${event.agentName} type=${event.agentType || '-'} ` + `protocol=${event.protocol} status=${event.status} ` + `sessionId=${event.sessionId || '-'} key=${event.sessionKey || '-'}` ); } /** @private */ _bindRegistryEvents() { this.registry.on('changed', ({ added, removed, modified }) => { console.log(`[AgentManager] Registry 热重载: +${added.length} -${removed.length} ~${modified.length}`); const fullConfig = this.registry.getFullConfig(); this.promptBuilder = new PromptBuilder(fullConfig, this.baseDir); this.hookRunner = new HookRunner(fullConfig, this.baseDir); this._emitSessionEvent({ agentName: null, status: 'config_reloaded', added, removed, modified, }); }); this.registry.on('registered', ({ name, config }) => { console.log(`[AgentManager] Agent 已动态注册: ${name}`); this._emitSessionEvent({ agentName: name, agentType: config.type, status: 'registered' }); }); this.registry.on('unregistered', ({ name }) => { console.log(`[AgentManager] Agent 已动态注销: ${name}`); this._emitSessionEvent({ agentName: name, status: 'unregistered' }); }); } } module.exports = AgentManager;