/** * 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 { runPromptInSession } = require('./stream-executor'); const { RunRegistry: EngineRunRegistry, FailoverError, } = require('../engine'); 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); const { agent, client } = this.getAgentConfig(agentName); 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); const fullPrompt = this.buildFullPrompt(agentName, task); 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 > 静态配置) */ getAgentConfig(agentName) { if (this.registry) { const registered = this.registry.getAgent(agentName); if (registered) return { agent: registered.agent, client: registered.client }; } let agentConfig = this.config.agents?.[agentName]; // Dynamic Hot-Reload Fallback (Local File-System First) if (!agentConfig) { let agentDir; let agentsRoot; if (agentName.includes(':')) { const [compId, subId] = agentName.split(':'); agentsRoot = path.resolve(__dirname, `../../../data/companies/${compId}/agents`); agentDir = path.join(agentsRoot, subId); } else { agentsRoot = path.resolve(__dirname, '../../../system/agents'); agentDir = path.join(agentsRoot, 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: path.join(agentDir, 'agent.md'), workDir: agentDir, // CLI 进程在 Agent 自己的目录下运行,自动读取 .claude/skills/ description: `Dynamic Agent ${agentName}` }; } 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 工作目录 const ownerAgentDir = path.dirname(path.dirname(path.dirname(foundPath))); // .claude/skills/xxx/SKILL.md → agent dir 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}` }; } } // 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}`); // Fallback valid client let clientConfig = agentConfig.client ? (this.config.clients?.[agentConfig.client] || null) : null; if (!clientConfig) clientConfig = { url: 'local' }; // provide a default dummy client info if missing return { agent: agentConfig, client: clientConfig }; } /** * 构建完整 prompt */ buildFullPrompt(agentName, task) { return this.promptBuilder.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(确保 CLI 进程运行在 Agent 目录下,能读取 .claude/skills/) const effectiveWorkDir = agent.workDir || this.workDir; let session = await this.sessionPool.getOrCreate( agentName, agent, resolved.launcherConfig, protocolName, effectiveWorkDir ); for (let attempt = 0; attempt < 2; attempt++) { try { this.sessionPool.activeKey = session.key; const idleTimeoutMs = this.config?.agents?.[agentName]?.idleTimeout || 120_000; const stdout = await runPromptInSession(session, fullPrompt, { ...options, logSession, idleTimeoutMs, }); this.sessionPool.activeKey = null; 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 ); this._emitSessionEvent({ agentName, protocol: protocolName, status: 'reconnected', sessionId: session.protocol?.sessionId, sessionKey: session.key, }); continue; } if (logSession) await logSession.logAgentError(agentName, err); 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;