agent-manager.js 17.4 KB

Edit Raw Blame History

/**
 * 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');

/**
 * 无流式数据的空闲上限，与 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);
    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)) {
          // P1 虚拟引用寻址：company-hired 的 agent 可能在 config.json 里带
          // systemAgentRef（company-service.hire() 走引用模式雇佣时写入），
          // 此时 agent.md 不再物理拷贝进私有目录，改为指向共享的
          // system_agents.legacy_path/agent.md（见 docs/todo/system-design/27-p1-virtual-reference.md §4.2）。
          let promptPath = path.join(agentDir, 'agent.md');
          let companyContextPath = null;
          const configJsonPath = path.join(agentDir, 'config.json');
          if (fs.existsSync(configJsonPath)) {
            let ref;
            try {
              const dirConfig = JSON.parse(fs.readFileSync(configJsonPath, 'utf-8'));
              ref = dirConfig?.systemAgentRef;
            } catch (e) {
              // config.json 损坏/不可解析：退回 legacy 行为，不让寻址崩溃
              ref = undefined;
            }
            if (ref) {
              const sharedAgentMdPath = require('../../services/graph-service').lookupSharedAgentMd(ref);
              if (sharedAgentMdPath) {
                promptPath = sharedAgentMdPath;
                companyContextPath = path.join(agentDir, 'company-context.md');
              }
              // sharedAgentMdPath 为 null（理论上不应发生，防御性处理）：
              // promptPath/companyContextPath 保持上面初始化的 legacy 默认值。
            }
          }

          agentConfig = {
             type: 'orchestrator',
             protocol: 'claude-cli',
             transport: 'pipe',
             launcher: 'claude-local',
             client: 'claude-local',
             prompt: promptPath,
             companyContextPath,
             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;

        // 优先级：调用方显式传入 > 该 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;

        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;