宜博 / lemonboss

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