metona-ai-desktop/electron/harness/agent-loop/engine.ts

/**
 * Agent Loop — ReAct 状态机引擎
 *
 * 生产级 ReAct Agent Loop，负责：
 * 1. 状态机管理循环生命周期
 * 2. 调用 Provider Adapter 获取 LLM 响应（支持流式）
 * 3. 解析输出、执行工具、注入观察
 * 4. 流式事件推送 UI
 * 5. 超时、重试、上下文压缩
 *
 * @see docs/生产级通用 AI Agent 智能体桌面应用：完整设计与构建指南.html — 第四章
 */

import { EventEmitter } from 'events';
import { nanoid } from 'nanoid';
import {
  AgentLoopState,
  TerminationReason,
  type IterationStep,
  type Thought,
  type AgentLoopConfig,
  type AgentLoopOutput,
  type TokenUsage,
} from './types';
import type {
  MetonaRequest,
  MetonaResponse,
  MetonaMessage,
  MetonaSystemPrompt,
  MetonaToolCall,
  MetonaToolResult,
  MetonaStreamEvent,
  IMetonaProviderAdapter,
  MetonaToolDef,
} from '../types';
import { MetonaStreamEventType, MetonaFinishReason } from '../types';

const DEFAULT_CONFIG: AgentLoopConfig = {
  maxIterations: 20,
  timeoutMs: 120_000,
  totalTimeoutMs: 600_000,
  enableReflection: false,
  compressionThreshold: 0.8,
  contextWindow: 128_000,
  retryCount: 3,
  temperature: 0.0,
  maxTokens: 63488,
  thinkingEnabled: true,
  thinkingEffort: 'high',
};

/**
 * ReAct Agent Loop 引擎
 */
export class AgentLoopEngine extends EventEmitter {
  private currentState: AgentLoopState = AgentLoopState.INIT;
  private iterations: IterationStep[] = [];
  private currentIteration = 0;
  private startTime = 0;
  private totalTokens: TokenUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0 };
  private aborted = false;
  private config: AgentLoopConfig;
  private tools: MetonaToolDef[] = [];
  private currentSessionId: string = '';
  private currentRequestId: string = '';
  private workspacePath: string = '';

  constructor(
    config: Partial<AgentLoopConfig> = {},
    private adapter: IMetonaProviderAdapter,
    private toolRegistry?: import('../tools/registry').ToolRegistry,
    private preToolHooks: import('../hooks/pre-tool').PreToolHook[] = [],
    private postToolHooks: import('../hooks/post-tool').PostToolHook[] = [],
  ) {
    super();
    this.config = { ...DEFAULT_CONFIG, ...config };
  }

  /**
   * 设置工作空间路径（工具执行时传入 context）
   */
  setWorkspacePath(path: string): void {
    this.workspacePath = path;
  }

  /**
   * 设置可用工具列表
   */
  setTools(tools: MetonaToolDef[]): void {
    this.tools = tools;
  }

  /**
   * 获取当前工具列表（用于子任务恢复）
   */
  getTools(): MetonaToolDef[] {
    return [...this.tools];
  }

  /**
   * 热切换 Provider Adapter（设置变更时调用）
   */
  setAdapter(adapter: IMetonaProviderAdapter): void {
    this.adapter = adapter;
  }

  /**
   * 执行完整的 ReAct 循环（流式模式）
   *
   * @param userInput 用户输入文本
   * @param sessionId 会话 ID
   * @param history 历史消息
   * @param systemPrompt System Prompt
   */
  async runStream(
    userMessage: MetonaMessage,
    sessionId: string,
    history: MetonaMessage[],
    systemPrompt: MetonaSystemPrompt,
  ): Promise<AgentLoopOutput> {
    this.startTime = Date.now();
    this.aborted = false;
    this.iterations = [];
    this.currentIteration = 0;
    this.currentSessionId = sessionId;
    this.totalTokens = { promptTokens: 0, completionTokens: 0, totalTokens: 0 };

    try {
      await this.transitionTo(AgentLoopState.INIT);

      // 构建消息列表（保留 images 字段）
      const messages: MetonaMessage[] = [
        ...history,
        {
          role: 'user',
          content: userMessage.content,
          images: userMessage.images,
          timestamp: Date.now(),
        },
      ];

      // === 主循环 ===
      while (
        this.currentIteration < this.config.maxIterations &&
        !this.aborted &&
        Date.now() - this.startTime < this.config.totalTimeoutMs
      ) {
        this.currentIteration++;

        // 构建请求
        const request: MetonaRequest = {
          meta: {
            sessionId,
            iteration: this.currentIteration,
            requestId: `r_${nanoid(12)}`,
            timestamp: Date.now(),
            agentVersion: '1.0.0',
          },
          systemPrompt,
          messages,
          tools: this.tools.length > 0 ? this.tools : undefined,
          params: {
            maxTokens: this.config.maxTokens,
            temperature: this.config.temperature,
            stream: true,
            thinkingEnabled: this.config.thinkingEnabled,
            thinkingEffort: this.config.thinkingEffort,
            contextLength: this.config.contextLength,
          },
        };

        const step = await this.executeOneIterationStream(request, sessionId);
        this.iterations.push(step);

        // 将 assistant 回复加入消息历史
        if (step.thought) {
          const assistantMsg: MetonaMessage = {
            role: 'assistant',
            content: step.thought.content,
            reasoningContent: step.thought.reasoningContent,
            toolCalls: step.toolCalls,
            timestamp: Date.now(),
            iteration: this.currentIteration,
          };
          messages.push(assistantMsg);
        }

        // 如果没有工具调用，视为最终输出
        if (!step.toolCalls || step.toolCalls.length === 0) {
          return this.finish(TerminationReason.COMPLETED, step.thought?.content);
        }

        // 执行工具并将结果加入消息
        if (step.toolResults) {
          for (const result of step.toolResults) {
            messages.push({
              role: 'tool',
              content: typeof result.result === 'string' ? result.result : JSON.stringify(result.result),
              toolResult: result,
              timestamp: Date.now(),
              iteration: this.currentIteration,
            });
          }
        }
      }

      // 循环退出判断
      if (this.aborted) return this.finish(TerminationReason.USER_INTERRUPT);
      if (this.currentIteration >= this.config.maxIterations)
        return this.finish(TerminationReason.MAX_ITERATIONS);
      return this.finish(TerminationReason.TIMEOUT);
    } catch (error) {
      this.emit('error', { error: (error as Error).message });
      return this.finish(TerminationReason.ERROR, undefined, error as Error);
    }
  }

  /** 中断循环 */
  abort(): void {
    this.aborted = true;
    this.emit('aborted');
  }

  /** 获取当前状态 */
  getState(): AgentLoopState {
    return this.currentState;
  }

  // ========== 私有方法 ==========

  /**
   * 执行单次迭代（流式模式）
   */
  private async executeOneIterationStream(
    request: MetonaRequest,
    sessionId: string,
  ): Promise<IterationStep> {
    this.currentRequestId = request.meta.requestId;
    const step: IterationStep = {
      iteration: this.currentIteration,
      state: AgentLoopState.THINKING,
      startedAt: Date.now(),
    };

    try {
      // === THINKING: 流式调用 LLM ===
      await this.transitionTo(AgentLoopState.THINKING);
      this.emit('stateChange', {
        sessionId,
        iteration: this.currentIteration,
        state: 'THINKING',
      });

      let fullContent = '';
      let reasoningContent = '';
      const toolCallsBuffer = new Map<number, { name: string; argsBuffer: string }>();
      let tokenUsage: TokenUsage | undefined;

      // 流式接收响应
      for await (const event of this.chatStreamWithRetry(request)) {
        if (this.aborted) break;

        // 过滤掉每轮的 DONE 事件 — 只在全部迭代完成后发送一个最终 DONE
        if (event.type === MetonaStreamEventType.DONE) continue;

        // 转发流式事件到渲染进程
        this.emit('streamEvent', event);

        switch (event.type) {
          case MetonaStreamEventType.TEXT_DELTA:
            if (event.delta) fullContent += event.delta;
            break;

          case MetonaStreamEventType.REASONING_DELTA:
            if (event.delta) reasoningContent += event.delta;
            break;

          case MetonaStreamEventType.TOOL_CALL_DELTA:
            if (event.toolCallDelta) {
              const { index, name, argsDelta } = event.toolCallDelta;
              if (!toolCallsBuffer.has(index)) {
                toolCallsBuffer.set(index, { name: name ?? '', argsBuffer: '' });
              }
              const buf = toolCallsBuffer.get(index)!;
              if (name) buf.name = name;
              if (argsDelta) buf.argsBuffer += argsDelta;
            }
            break;

          case MetonaStreamEventType.TOOL_CALL_COMPLETE:
            if (event.toolCall) {
              // 工具调用完成，记录到 step
              if (!step.toolCalls) step.toolCalls = [];
              step.toolCalls.push(event.toolCall);
            }
            break;

          case MetonaStreamEventType.USAGE:
            if (event.usage) {
              tokenUsage = {
                promptTokens: event.usage.inputTokens ?? 0,
                completionTokens: event.usage.outputTokens ?? 0,
                totalTokens: event.usage.totalTokens ?? 0,
              };
            }
            break;

          case MetonaStreamEventType.ERROR:
            if (event.error) {
              throw new Error(event.error.message);
            }
            break;
        }
      }

      // 处理缓冲区中的工具调用（TOOL_CALL_DELTA 拼接）
      if (toolCallsBuffer.size > 0 && (!step.toolCalls || step.toolCalls.length === 0)) {
        step.toolCalls = [];
        for (const [, buf] of toolCallsBuffer) {
          let args: Record<string, unknown>;
          try {
            args = buf.argsBuffer ? JSON.parse(buf.argsBuffer) : {};
          } catch {
            // JSON 解析失败，使用空参数（LLM 仍请求了该工具调用）
            args = {};
          }
          step.toolCalls.push({
            id: `tc_${nanoid(8)}`,
            name: buf.name,
            args,
            iteration: this.currentIteration,
            timestamp: Date.now(),
          });
        }
      }

      // 记录 Thought
      if (fullContent || reasoningContent) {
        step.thought = {
          id: `thought-${this.currentIteration}`,
          content: fullContent,
          reasoningContent,
          timestamp: Date.now(),
          iteration: this.currentIteration,
        };
      }

      // 记录 Token 使用
      if (tokenUsage) {
        step.tokenUsage = tokenUsage;
        this.accumulateTokens(tokenUsage);
      }

      // === EXECUTING: 执行工具调用 ===
      if (step.toolCalls && step.toolCalls.length > 0) {
        await this.transitionTo(AgentLoopState.EXECUTING);
        this.emit('stateChange', {
          sessionId,
          iteration: this.currentIteration,
          state: 'EXECUTING',
        });

        step.toolResults = [];
        for (const tc of step.toolCalls) {
          const result = await this.executeToolSafely(tc);
          step.toolResults.push(result);

          // 转发工具结果到渲染进程
          this.emit('streamEvent', {
            type: MetonaStreamEventType.TOOL_RESULT,
            requestId: request.meta.requestId,
            sessionId,
            iteration: this.currentIteration,
            seq: 0,
            timestamp: Date.now(),
            toolResult: result,
          });
        }
      }

      // === OBSERVING ===
      await this.transitionTo(AgentLoopState.OBSERVING);

      // === 上下文压缩（每 5 轮检查一次） ===
      if (this.currentIteration % 5 === 0) {
        await this.transitionTo(AgentLoopState.COMPRESSING);
        this.emit('stateChange', {
          sessionId,
          iteration: this.currentIteration,
          state: 'COMPRESSING',
        });

        // 在 context 中标记压缩点
        const keepRecent = 3;
        if (this.iterations.length > keepRecent) {
          this.emit('compressed', { iteration: this.currentIteration, keptRounds: keepRecent });
        }
      }

      step.completedAt = Date.now();
      step.state = AgentLoopState.OBSERVING;
      return step;
    } catch (error) {
      step.completedAt = Date.now();
      step.state = AgentLoopState.TERMINATED;
      throw error;
    }
  }

  /**
   * 安全执行工具调用（经过 Hook 管道）
   */
  private async executeToolSafely(toolCall: MetonaToolCall): Promise<MetonaToolResult> {
    const startTs = Date.now();

    if (!this.toolRegistry) {
      return {
        toolCallId: toolCall.id, toolName: toolCall.name,
        result: null, success: false,
        error: `Tool '${toolCall.name}' not available: no ToolRegistry configured`,
        durationMs: Date.now() - startTs, timestamp: Date.now(),
      };
    }

    // 前置 Hook 管道
    for (const hook of this.preToolHooks) {
      const result = await hook.beforeExecute(toolCall, this.currentSessionId);
      if (result.blocked) {
        return {
          toolCallId: toolCall.id, toolName: toolCall.name,
          result: null, success: false, error: `Blocked: ${result.reason}`,
          durationMs: Date.now() - startTs, timestamp: Date.now(),
        };
      }
    }

    // 执行工具
    const toolResult = await this.toolRegistry.execute(toolCall, {
      sessionId: this.currentSessionId ?? '',
      workspacePath: this.workspacePath,
      iteration: this.currentIteration,
      requestId: this.currentRequestId,
    });

    // 后置 Hook 管道
    for (const hook of this.postToolHooks) {
      await hook.afterExecute(toolCall, toolResult, this.currentSessionId);
    }

    return toolResult;
  }

  /**
   * 带重试的流式调用
   *
   * 如果 adapter 抛出错误，在 retryCount 次数内重试，每次等待 1 秒。
   */
  private async *chatStreamWithRetry(request: MetonaRequest): AsyncIterable<MetonaStreamEvent> {
    let lastError: unknown;
    for (let attempt = 0; attempt <= this.config.retryCount; attempt++) {
      try {
        yield* this.adapter.chatStream(request);
        return;
      } catch (error) {
        lastError = error;
        if (attempt < this.config.retryCount) {
          await new Promise((resolve) => setTimeout(resolve, 1000));
        }
      }
    }
    throw lastError;
  }

  private async transitionTo(state: AgentLoopState): Promise<void> {
    const previous = this.currentState;
    this.currentState = state;
    this.emit('stateChange', {
      previous,
      current: state,
      state,
      sessionId: this.currentSessionId,
      iteration: this.currentIteration,
    });
  }

  private accumulateTokens(usage: TokenUsage): void {
    this.totalTokens.promptTokens += usage.promptTokens;
    this.totalTokens.completionTokens += usage.completionTokens;
    this.totalTokens.totalTokens += usage.totalTokens;
  }

  private finish(
    reason: TerminationReason,
    answer?: string,
    error?: Error,
  ): AgentLoopOutput {
    // 发送唯一的最终 DONE 事件 — UI 只在此处结束流式状态
    this.emit('streamEvent', {
      type: MetonaStreamEventType.DONE,
      requestId: this.currentRequestId,
      sessionId: this.currentSessionId,
      iteration: this.currentIteration,
      seq: 0,
      timestamp: Date.now(),
    });

    this.currentState = AgentLoopState.TERMINATED;
    return {
      finalAnswer: answer ?? (error ? error.message : 'No answer produced'),
      terminationReason: reason,
      iterations: this.iterations,
      totalTokenUsage: this.totalTokens,
      durationMs: Date.now() - this.startTime,
      metadata: { config: this.config, error: error?.message },
    };
  }
}