metona-ai-desktop/electron/harness/orchestration/orchestrator.ts

/**
 * Task Orchestrator — 任务编排器
 *
 * 支持父子委派模式：主 Agent 委派子任务给 SubAgent。
 * 每个 SubAgent 运行在**独立的 AgentLoopEngine 实例**中，避免状态污染。
 *
 * 安全保障：
 * 1. 独立引擎实例 — SubAgent 不共享主 Agent 的引擎状态
 * 2. 递归深度限制 — 默认最大 3 层，防止无限递归
 * 3. 工具白名单隔离 — SubAgent 默认不继承 delegate_task（防止递归）
 * 4. 真正的 abort — 通过引擎引用调用 engine.abort()
 * 5. 事件隔离 — SubAgent 的流式事件不直接转发到前端，仅通过 orchestrator 事件通知
 *
 * P2-10 改造：
 * - 依赖 EngineProvider（AgentEngineManager）而非单个 mainEngine：
 *   SubAgent 通过工厂获取独立 adapter 实例，彻底消除 abort 信号互踩问题
 *   （原实现 SubEngine 共享主引擎 adapter，setAbortSignal 单槽位会互相覆盖）
 * - 新增 abortByParent(parentSessionId)：用户中断会话时联动中断其派生的 SubAgent
 *
 * @see docs/生产级通用 AI Agent 智能体桌面应用：完整设计与构建指南.html — 第五章
 */

import { EventEmitter } from 'events';
import { nanoid } from 'nanoid';
import { AgentLoopEngine } from '../agent-loop/engine';
import type { AgentLoopConfig } from '../agent-loop/types';
import type {
  MetonaMessage,
  MetonaSystemPrompt,
  MetonaToolDef,
  IMetonaProviderAdapter,
} from '../types';
import type { ToolRegistry } from '../tools/registry';
import type { PreToolHook } from '../hooks/pre-tool';
import type { PostToolHook } from '../hooks/post-tool';
import log from 'electron-log';

export interface SubAgentResult {
  taskId: string;
  /** v0.5.0: 委派方（父会话）ID — 供事件消费者按会话过滤 */
  parentSessionId: string;
  result: string;
  success: boolean;
  durationMs: number;
  iterations: number;
}

/**
 * v0.5.0: SubAgent 生命周期事件载荷（统一广播给前端 AgentMonitor 展示）
 */
export interface SubAgentEvent {
  taskId: string;
  parentSessionId: string;
  description: string;
  status: 'delegated' | 'running' | 'completed' | 'error';
  depth: number;
  durationMs?: number;
  iterations?: number;
  error?: string;
}

/**
 * P2-10: 引擎供给接口（由 AgentEngineManager 实现）
 * orchestrator 不再持有单个引擎引用，而是按需创建独立实例。
 */
export interface EngineProvider {
  /** 主 adapter（读取 contextWindow 等元信息） */
  getAdapter(): IMetonaProviderAdapter;
  /** 创建独立 adapter 实例（SubAgent 专用，隔离 abort 信号） */
  createAdapter(): IMetonaProviderAdapter;
  /** 故障转移 Provider（可为 null） */
  getFallbackAdapter(): IMetonaProviderAdapter | null;
  /** 工作空间路径 */
  getWorkspacePath(): string;
}

interface SubAgentHandle {
  taskId: string;
  parentSessionId: string;
  description: string;
  status: 'pending' | 'running' | 'completed' | 'error';
  depth: number;
  engine?: AgentLoopEngine;
  result?: SubAgentResult;
  abort: () => void;
  getStatus: () => { taskId: string; status: string; description: string; depth: number };
}

/** 默认递归深度限制 */
const MAX_DELEGATION_DEPTH = 3;

export class TaskOrchestrator extends EventEmitter {
  private activeSubAgents = new Map<string, SubAgentHandle>();
  /** 追踪每个 session 的当前委派深度 */
  private sessionDepth = new Map<string, number>();

  constructor(
    private engines: EngineProvider,
    private toolRegistry?: ToolRegistry,
    private preToolHooks: PreToolHook[] = [],
    private postToolHooks: PostToolHook[] = [],
    private defaultConfig?: Partial<AgentLoopConfig>,
  ) {
    super();
  }

  /**
   * L-18 修复: 热更新 SubAgent 的默认配置
   *
   * 主 Agent 的配置变更（thinkingEnabled/thinkingEffort/contextLength 等）通过
   * engine.updateConfig() 即时生效；但 SubAgent 在 delegate() 时从 defaultConfig
   * 复制配置，若 defaultConfig 不同步，新创建的 SubAgent 仍使用旧配置。
   *
   * 此方法供 IPC 层在 config:set 时同步调用，确保后续 SubAgent 使用最新配置。
   */
  updateDefaultConfig(partial: Partial<AgentLoopConfig>): void {
    this.defaultConfig = { ...this.defaultConfig, ...partial };
  }

  /**
   * 委派子任务
   *
   * 创建一个独立的 AgentLoopEngine 实例执行子任务。
   * SubAgent 不共享主 Agent 的引擎状态，安全隔离。
   */
  async delegate(params: {
    taskId?: string;
    description: string;
    parentSessionId: string;
    maxIterations?: number;
    tools?: string[];
  }): Promise<SubAgentResult> {
    const taskId = params.taskId ?? `sub_${nanoid(8)}`;
    const startMs = Date.now();

    // ===== 递归深度检查 =====
    const currentDepth = this.sessionDepth.get(params.parentSessionId) ?? 0;
    if (currentDepth >= MAX_DELEGATION_DEPTH) {
      log.warn(
        `[Orchestrator] Delegation depth limit reached (${currentDepth}) for session ${params.parentSessionId}`,
      );
      return {
        taskId,
        parentSessionId: params.parentSessionId,
        result: `SubAgent delegation depth limit reached (${MAX_DELEGATION_DEPTH}). Cannot delegate further.`,
        success: false,
        durationMs: 0,
        iterations: 0,
      };
    }
    const depth = currentDepth + 1;
    this.sessionDepth.set(params.parentSessionId, depth);

    this.emit('taskDelegated', {
      taskId,
      description: params.description,
      parentSessionId: params.parentSessionId,
      depth,
    });

    // ===== 创建独立的引擎实例（P2-10: 独立 adapter，隔离 abort 信号） =====
    const subEngine = new AgentLoopEngine(
      {
        maxIterations: params.maxIterations ?? 10,
        totalTimeoutMs: 300_000, // 子任务总超时 5 分钟
        thinkingEnabled: this.defaultConfig?.thinkingEnabled ?? true,
        thinkingEffort: this.defaultConfig?.thinkingEffort ?? 'medium',
        contextLength: this.defaultConfig?.contextLength,
        contextWindow: this.defaultConfig?.contextWindow ?? 128_000,
      },
      this.engines.createAdapter(),
      this.toolRegistry,
      this.preToolHooks,
      this.postToolHooks,
    );
    subEngine.setFallbackAdapter(this.engines.getFallbackAdapter());
    subEngine.setWorkspacePath(this.engines.getWorkspacePath());

    // v0.5.0: 转发 SubEngine 的流式/状态事件（供 TRACE 录制与前端可观测）
    // 事件以 taskId 为 sessionId，录制到独立 JSONL 文件，不污染父会话的流
    this.forwardSubEngineEvents(subEngine, taskId);

    // ===== 工具白名单设置 =====
    const allowedTools = this.resolveTools(params.tools);
    subEngine.setTools(allowedTools);

    const handle: SubAgentHandle = {
      taskId,
      parentSessionId: params.parentSessionId,
      description: params.description,
      status: 'running',
      depth,
      engine: subEngine,
      abort: () => {
        subEngine.abort();
        handle.status = 'error';
        this.activeSubAgents.delete(taskId);
      },
      getStatus: () => ({
        taskId,
        status: handle.status,
        description: handle.description,
        depth: handle.depth,
      }),
    };

    this.activeSubAgents.set(taskId, handle);
    this.emit('taskStarted', {
      taskId,
      description: params.description,
      parentSessionId: params.parentSessionId,
      depth,
    });

    try {
      // 构建用户消息
      const userMessage: MetonaMessage = {
        role: 'user',
        content: params.description,
        timestamp: Date.now(),
      };

      // 构建 System Prompt（子 Agent 专用）
      const systemPrompt = this.buildSubAgentPrompt(params.description, depth);

      // 运行 Agent Loop（同步等待完成）
      const output = await subEngine.runStream(
        userMessage,
        taskId,
        [], // 子 Agent 无历史
        systemPrompt,
      );

      const durationMs = Date.now() - startMs;
      const success = output.terminationReason === 'completed';
      const result: SubAgentResult = {
        taskId,
        parentSessionId: params.parentSessionId,
        result: output.finalAnswer,
        success,
        durationMs,
        iterations: output.iterations.length,
      };

      handle.status = success ? 'completed' : 'error';
      handle.result = result;
      this.emit('taskCompleted', result);

      log.info(
        `[Orchestrator] SubAgent "${taskId}" (depth=${depth}) ${success ? 'completed' : 'failed'} in ${durationMs}ms, ${output.iterations.length} iterations`,
      );

      return result;
    } catch (error) {
      const durationMs = Date.now() - startMs;
      const errMsg = (error as Error).message;
      const result: SubAgentResult = {
        taskId,
        parentSessionId: params.parentSessionId,
        result: errMsg,
        success: false,
        durationMs,
        iterations: 0,
      };

      handle.status = 'error';
      handle.result = result;
      this.emit('taskError', {
        taskId,
        parentSessionId: params.parentSessionId,
        description: params.description,
        error: errMsg,
      });

      log.error(`[Orchestrator] SubAgent "${taskId}" (depth=${depth}) error: ${errMsg}`);

      return result;
    } finally {
      // #5 修复: 统一在 finally 块恢复 sessionDepth，覆盖正常完成/异常/abort 所有路径
      // 审查修复: 如果 abortAll 已 clear sessionDepth，不再恢复（避免覆盖紧急清理）。
      // 场景：用户紧急中断时 abortAll 先 clear，若 SubEngine 随后才返回执行 finally，
      // 不应把已清空的 sessionDepth 又 set 回 currentDepth。
      if (this.sessionDepth.has(params.parentSessionId)) {
        if (currentDepth === 0) {
          this.sessionDepth.delete(params.parentSessionId);
        } else {
          this.sessionDepth.set(params.parentSessionId, currentDepth);
        }
      }
      this.activeSubAgents.delete(taskId);
      // P2-10: SubEngine 使用独立 adapter 实例，无需恢复主引擎的 abort signal
    }
  }

  /**
   * 解析工具白名单
   *
   * - 如果指定了 tools，使用白名单（自动排除 delegate_task 防止递归）
   * - 如果未指定，使用所有已启用工具（同样排除 delegate_task）
   */
  private resolveTools(toolNames?: string[]): MetonaToolDef[] {
    if (!this.toolRegistry) return [];

    // 始终排除 delegate_task 防止递归（除非深度为 1 且显式要求）
    const EXCLUDE_TOOLS = new Set(['delegate_task']);

    if (toolNames && toolNames.length > 0) {
      // 使用白名单模式
      const resolved: MetonaToolDef[] = [];
      const notFound: string[] = [];

      for (const name of toolNames) {
        if (EXCLUDE_TOOLS.has(name)) continue; // 静默排除
        const tool = this.toolRegistry.get(name);
        if (tool) {
          resolved.push(tool.definition);
        } else {
          notFound.push(name);
        }
      }

      if (notFound.length > 0) {
        log.warn(`[Orchestrator] Tools not found: ${notFound.join(', ')}`);
      }

      return resolved;
    }

    // 未指定白名单 — 使用所有已启用工具（排除 delegate_task）
    return this.toolRegistry.listTools().filter((t) => !EXCLUDE_TOOLS.has(t.name));
  }

  /**
   * 构建 SubAgent 的 System Prompt
   */
  private buildSubAgentPrompt(description: string, depth: number): MetonaSystemPrompt {
    return {
      roleDefinition: `You are a SubAgent (delegation depth: ${depth}) executing a specific sub-task delegated by the parent Agent.\nYour goal is to complete the assigned task efficiently and return a clear, concise result.\nFocus only on the task at hand. Do not delegate further.`,
      outputConstraints: `Complete the task and provide a clear summary of your findings or actions.\nRespond in the same language as the task description.\nKeep your response focused and relevant — the parent Agent will use your result to continue its work.`,
      safetyGuidelines: `Do not access files outside the workspace.\nDo not execute dangerous commands.\nIf the task cannot be completed, explain why clearly.`,
    };
  }

  /**
   * 中断指定子任务
   */
  abortTask(taskId: string): boolean {
    const handle = this.activeSubAgents.get(taskId);
    if (handle && handle.status === 'running') {
      handle.abort();
      return true;
    }
    return false;
  }

  /**
   * P2-10: 中断指定父会话派生的所有 SubAgent
   * （用户中断会话时由 IPC abort handler 联动调用，消除"会话停了子任务还在跑"）
   *
   * v0.5.1: 返回值从数量改为被中止的 taskId 列表 — 调用方需据此清理这些
   * SubAgent 的 pending 工具确认（SubEngine 以 taskId 为 sessionId 写入
   * ConfirmationHook，父会话的 clearPending(sessionId) 清不到它们；不清理
   * 会导致中止后孤儿工具在用户补批时执行副作用）
   */
  abortByParent(parentSessionId: string): string[] {
    const abortedTaskIds: string[] = [];
    for (const handle of this.activeSubAgents.values()) {
      if (handle.parentSessionId === parentSessionId && handle.status === 'running') {
        handle.abort();
        abortedTaskIds.push(handle.taskId);
      }
    }
    if (abortedTaskIds.length > 0) {
      log.info(
        `[Orchestrator] Aborted ${abortedTaskIds.length} SubAgent(s) of session ${parentSessionId}`,
      );
    }
    return abortedTaskIds;
  }

  /**
   * 完成子任务（外部触发，保留接口兼容）
   */
  completeTask(taskId: string, result: string, success: boolean): void {
    const handle = this.activeSubAgents.get(taskId);
    if (handle && handle.status === 'running') {
      handle.status = success ? 'completed' : 'error';
      handle.result = {
        taskId,
        parentSessionId: handle.parentSessionId,
        result,
        success,
        durationMs: 0,
        iterations: 0,
      };
      this.activeSubAgents.delete(taskId);
      this.emit('taskCompleted', handle.result);
    }
  }

  getActiveAgentsStatus(): Array<{
    taskId: string;
    status: string;
    description: string;
    depth: number;
  }> {
    return Array.from(this.activeSubAgents.values()).map((a) => a.getStatus());
  }

  /**
   * 中断所有子任务
   */
  abortAll(): void {
    for (const agent of this.activeSubAgents.values()) {
      agent.abort();
    }
    this.activeSubAgents.clear();
    // 审查修复: 恢复 sessionDepth.clear()，保留紧急清理能力。
    // #5 修复曾移除此行，但若 SubEngine 卡死不返回，delegate 的 finally 永远不会执行，
    // sessionDepth 将永久残留。此处 clear 确保紧急路径能立即恢复状态。
    // 配合 delegate finally 块的 has() 检查：若已被 clear，finally 不再恢复（避免覆盖）。
    this.sessionDepth.clear();
  }

  /**
   * v0.5.0: 转发 SubEngine 的事件到 orchestrator 事件总线
   *
   * SubEngine 不经过 AgentEngineManager 的转发管道（刻意隔离，避免污染父会话流），
   * 此处将其流式/状态事件重新包装为 subStreamEvent / subStateChange 广播：
   * - IPC 层监听后录制到独立 TRACE 文件（sessionId = taskId）
   * - 生命周期事件（taskDelegated 等）另行广播给前端 AgentMonitor
   *
   * @param engine SubAgent 的独立引擎实例
   * @param taskId 子任务 ID（SubEngine runStream 的 sessionId）
   */
  private forwardSubEngineEvents(engine: AgentLoopEngine, taskId: string): void {
    engine.on('streamEvent', (event: Record<string, unknown>) => {
      this.emit('subStreamEvent', { taskId, event });
    });
    engine.on('stateChange', (data: Record<string, unknown>) => {
      this.emit('subStateChange', { taskId, data });
    });
  }
}