/** * 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(); /** 追踪每个 session 的当前委派深度 */ private sessionDepth = new Map(); constructor( private engines: EngineProvider, private toolRegistry?: ToolRegistry, private preToolHooks: PreToolHook[] = [], private postToolHooks: PostToolHook[] = [], private defaultConfig?: Partial, ) { super(); } /** * L-18 修复: 热更新 SubAgent 的默认配置 * * 主 Agent 的配置变更(thinkingEnabled/thinkingEffort/contextLength 等)通过 * engine.updateConfig() 即时生效;但 SubAgent 在 delegate() 时从 defaultConfig * 复制配置,若 defaultConfig 不同步,新创建的 SubAgent 仍使用旧配置。 * * 此方法供 IPC 层在 config:set 时同步调用,确保后续 SubAgent 使用最新配置。 */ updateDefaultConfig(partial: Partial): void { this.defaultConfig = { ...this.defaultConfig, ...partial }; } /** * 委派子任务 * * 创建一个独立的 AgentLoopEngine 实例执行子任务。 * SubAgent 不共享主 Agent 的引擎状态,安全隔离。 */ async delegate(params: { taskId?: string; description: string; parentSessionId: string; maxIterations?: number; tools?: string[]; /** * v0.8.0 P1-2: 引擎级取消信号 —— delegate_task 工具透传 context.signal, * 用户中断父会话时 SubEngine 联动终止(此前工具被 registry 兜底超时杀掉, * SubEngine 仍在后台跑满剩余轮次)。 */ signal?: AbortSignal; }): Promise { 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, // v0.7.3 P3-1: SubAgent 与主引擎同源消费 enableReflection(REFLECTING 状态开关) enableReflection: this.defaultConfig?.enableReflection ?? false, // v0.7.4 P3-2 修正: SubAgent 继承主引擎的 temperature/maxTokens —— // 旧实现不读这两个键,新 SubAgent 恒用引擎 DEFAULT_CONFIG(0.0/63488), // 导致"热生效"对子任务不完整 temperature: this.defaultConfig?.temperature ?? 0.0, maxTokens: this.defaultConfig?.maxTokens ?? 63488, }, this.engines.createAdapter(), this.toolRegistry, this.preToolHooks, this.postToolHooks, ); subEngine.setFallbackAdapter(this.engines.getFallbackAdapter()); subEngine.setWorkspacePath(this.engines.getWorkspacePath()); // v0.8.0 P1-2: 外部取消信号联动 —— signal 已中止或运行中触发 abort 时 // 终止 SubEngine;finally 中解除监听防泄漏 let onExternalAbort: (() => void) | null = null; if (params.signal) { if (params.signal.aborted) { subEngine.abort(); } else { onExternalAbort = () => subEngine.abort(); params.signal.addEventListener('abort', onExternalAbort, { once: true }); } } // 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 { // v0.8.0 P1-2: 解除外部取消信号监听(防事件循环泄漏) if (params.signal && onExternalAbort) { params.signal.removeEventListener('abort', onExternalAbort); } // #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 []; // v0.6.4 修正注释与实现漂移:delegate_task 在所有深度下无条件排除 // (resolveTools 不感知 depth,旧注释中的"除非深度为 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) => { this.emit('subStreamEvent', { taskId, event }); }); engine.on('stateChange', (data: Record) => { this.emit('subStateChange', { taskId, data }); }); } }