宜博 / lemonboss

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