qwen-code/packages/core/src/agents/runtime/agent-interactive.ts

/**
 * @license
 * Copyright 2025 Qwen
 * SPDX-License-Identifier: Apache-2.0
 */

/**
 * @fileoverview AgentInteractive — persistent interactive agent.
 *
 * Composes AgentCore with on-demand message processing. Builds conversation
 * state (messages, pending approvals, live outputs) that the UI reads.
 */

import {
  createAbortController,
  createChildAbortController,
} from '../../utils/abortController.js';
import { createDebugLogger } from '../../utils/debugLogger.js';
import { type AgentEventEmitter, AgentEventType } from './agent-events.js';
import type {
  AgentApprovalRequestEvent,
  AgentToolOutputUpdateEvent,
  AgentToolResultEvent,
} from './agent-events.js';
import type { AgentStatsSummary } from './agent-statistics.js';
import type { AgentCore } from './agent-core.js';
import type { ContextState } from './agent-headless.js';
import type { GeminiChat } from '../../core/geminiChat.js';
import type { FunctionDeclaration } from '@google/genai';
import {
  ToolConfirmationOutcome,
  type ToolCallConfirmationDetails,
  type ToolResultDisplay,
} from '../../tools/tools.js';
import { AsyncMessageQueue } from '../../utils/asyncMessageQueue.js';
import {
  AgentTerminateMode,
  AgentStatus,
  isTerminalStatus,
  type AgentInteractiveConfig,
  type AgentMessage,
} from './agent-types.js';

const debugLogger = createDebugLogger('AGENT_INTERACTIVE');

/**
 * AgentInteractive — persistent interactive agent that processes
 * messages on demand.
 *
 * Three-level cancellation:
 * - `cancelCurrentRound()` — abort the current reasoning loop only
 * - `shutdown()` — graceful: stop accepting messages, wait for cycle
 * - `abort()` — immediate: master abort, set cancelled
 */
export class AgentInteractive {
  readonly config: AgentInteractiveConfig;
  private readonly core: AgentCore;
  private readonly queue = new AsyncMessageQueue<string>();

  private status: AgentStatus = AgentStatus.INITIALIZING;
  private error: string | undefined;
  private lastRoundError: string | undefined;
  private executionPromise: Promise<void> | undefined;
  private masterAbortController = createAbortController();
  private roundAbortController: AbortController | undefined;
  private chat: GeminiChat | undefined;
  private toolsList: FunctionDeclaration[] = [];
  private processing = false;
  private roundCancelledByUser = false;

  // Wall-clock timestamp when each currently-executing tool transitioned into
  // the scheduler's `executing` state. Keyed by callId. First TOOL_OUTPUT_UPDATE
  // carrying executionStartTime wins; later events that re-carry it are ignored
  // so the timer is stable. Lives on InteractiveAgent (not AgentCore) because
  // it's only consumed by the interactive UI's elapsed-time indicator.
  private readonly executionStartTimes = new Map<string, number>();

  constructor(config: AgentInteractiveConfig, core: AgentCore) {
    this.config = config;
    this.core = core;
    this.setupEventListeners();
  }

  // ─── Lifecycle ──────────────────────────────────────────────

  /**
   * Start the agent. Initializes the chat session, then kicks off
   * processing if an initialTask is configured.
   */
  async start(context: ContextState): Promise<void> {
    this.setStatus(AgentStatus.INITIALIZING);

    this.chat = await this.core.createChat(context, {
      interactive: true,
      extraHistory: this.config.chatHistory,
    });
    if (!this.chat) {
      this.error = 'Failed to create chat session';
      this.setStatus(AgentStatus.FAILED);
      return;
    }

    this.toolsList = await this.core.prepareTools();
    this.core.stats.start(Date.now());

    if (this.config.chatHistory?.length) {
      this.addMessage(
        'info',
        `History context from parent session included (${this.config.chatHistory.length} messages)`,
      );
    }

    if (this.config.initialTask) {
      this.queue.enqueue(this.config.initialTask);
      this.executionPromise = this.runLoop();
    }
  }

  /**
   * Run loop: process all pending messages, then settle status.
   * Exits when the queue is empty or the agent is aborted.
   */
  private async runLoop(): Promise<void> {
    this.processing = true;
    try {
      let message = this.queue.dequeue();
      while (message !== null && !this.masterAbortController.signal.aborted) {
        this.addMessage('user', message);
        await this.runOneRound(message);
        message = this.queue.dequeue();
      }

      if (this.masterAbortController.signal.aborted) {
        this.setStatus(AgentStatus.CANCELLED);
      } else {
        this.settleRoundStatus();
      }
    } catch (err) {
      this.error = err instanceof Error ? err.message : String(err);
      this.setStatus(AgentStatus.FAILED);
      debugLogger.error('AgentInteractive processing failed:', err);
    } finally {
      this.processing = false;
    }
  }

  /**
   * Run a single reasoning round for one message.
   * Creates a per-round AbortController so cancellation is scoped.
   */
  private async runOneRound(message: string): Promise<void> {
    if (!this.chat) return;

    this.setStatus(AgentStatus.RUNNING);
    this.lastRoundError = undefined;
    this.roundCancelledByUser = false;
    this.roundAbortController = createChildAbortController(
      this.masterAbortController,
    );

    try {
      const initialMessages = [
        { role: 'user' as const, parts: [{ text: message }] },
      ];

      const result = await this.core.runReasoningLoop(
        this.chat,
        initialMessages,
        this.toolsList,
        this.roundAbortController,
        {
          maxTurns: this.config.maxTurnsPerMessage,
          maxTimeMinutes: this.config.maxTimeMinutesPerMessage,
        },
      );

      // Surface non-normal termination as a visible info message and as
      // lastRoundError so Arena can distinguish limit stops from successes.
      if (
        result.terminateMode &&
        result.terminateMode !== AgentTerminateMode.GOAL
      ) {
        const msg = terminateModeMessage(result.terminateMode);
        if (msg) {
          this.addMessage('info', msg.text, { metadata: { level: msg.level } });
        }
        this.lastRoundError = `Terminated: ${result.terminateMode}`;
      }
    } catch (err) {
      // User-initiated cancellation already logged by cancelCurrentRound().
      if (this.roundCancelledByUser) return;
      // Agent survives round errors — log and settle status in runLoop.
      const errorMessage = err instanceof Error ? err.message : String(err);
      this.lastRoundError = errorMessage;
      debugLogger.error('AgentInteractive round error:', err);
      this.addMessage('info', errorMessage, { metadata: { level: 'error' } });
    } finally {
      // Helper's reverse-cleanup detaches the parent listener automatically
      // when the round controller aborts; abort here so cleanup fires whether
      // or not the round was already cancelled.
      this.roundAbortController?.abort();
      this.roundAbortController = undefined;
    }
  }

  // ─── Cancellation ──────────────────────────────────────────

  /**
   * Cancel only the current reasoning round.
   * Adds a visible "cancelled" info message and clears pending approvals.
   */
  cancelCurrentRound(): void {
    this.roundCancelledByUser = true;
    this.roundAbortController?.abort();
    this.core.clearPendingApprovals();
    this.addMessage('info', 'Agent round cancelled.', {
      metadata: { level: 'warning' },
    });
  }

  /**
   * Graceful shutdown: stop accepting messages and wait for current
   * processing to finish.
   */
  async shutdown(): Promise<void> {
    this.queue.drain();
    if (this.executionPromise) {
      await this.executionPromise;
    }
    // If no processing cycle ever ran (no initialTask, no messages),
    // ensure the agent reaches a terminal status.
    if (!isTerminalStatus(this.status)) {
      this.setStatus(AgentStatus.COMPLETED);
    }
  }

  /**
   * Immediate abort: cancel everything and set status to cancelled.
   */
  abort(): void {
    this.masterAbortController.abort();
    this.queue.drain();
    this.core.clearPendingApprovals();
  }

  // ─── Message Queue ─────────────────────────────────────────

  /**
   * Enqueue a message for the agent to process.
   */
  enqueueMessage(message: string): void {
    this.queue.enqueue(message);
    if (!this.processing) {
      this.executionPromise = this.runLoop();
    }
  }

  // ─── State Accessors (delegates to AgentCore) ──────────────

  getMessages(): readonly AgentMessage[] {
    return this.core.getMessages();
  }

  getStatus(): AgentStatus {
    return this.status;
  }

  getError(): string | undefined {
    return this.error;
  }

  getLastRoundError(): string | undefined {
    return this.lastRoundError;
  }

  getStats(): AgentStatsSummary {
    return this.core.getExecutionSummary();
  }

  /** The prompt token count from the most recent model call. */
  getLastPromptTokenCount(): number {
    return this.core.lastPromptTokenCount;
  }

  getCore(): AgentCore {
    return this.core;
  }

  getEventEmitter(): AgentEventEmitter {
    return this.core.getEventEmitter();
  }

  /**
   * Returns tool calls currently awaiting user approval.
   * Keyed by callId → full ToolCallConfirmationDetails (with onConfirm).
   * The UI reads this to render confirmation dialogs inside ToolGroupMessage.
   */
  getPendingApprovals(): ReadonlyMap<string, ToolCallConfirmationDetails> {
    return this.core.getPendingApprovals();
  }

  /**
   * Returns live output for currently-executing tools.
   * Keyed by callId → latest ToolResultDisplay (replaces on each update).
   * Entries are cleared when TOOL_RESULT arrives for the call.
   */
  getLiveOutputs(): ReadonlyMap<string, ToolResultDisplay> {
    return this.core.getLiveOutputs();
  }

  /**
   * Returns PTY PIDs for currently-executing interactive shell tools.
   * Keyed by callId → PID. Populated from TOOL_OUTPUT_UPDATE when pid is
   * present; cleared when TOOL_RESULT arrives. The UI uses this to enable
   * interactive shell input via HistoryItemDisplay's activeShellPtyId prop.
   */
  getShellPids(): ReadonlyMap<string, number> {
    return this.core.getShellPids();
  }

  /**
   * Returns wall-clock start timestamps (ms since epoch) for currently-
   * executing tools, from the scheduler's `→ executing` transition.
   * Keyed by callId; entries are cleared when TOOL_RESULT arrives. The UI
   * uses this to render an elapsed-time indicator that excludes approval
   * and scheduling wait.
   */
  getExecutionStartTimes(): ReadonlyMap<string, number> {
    return this.executionStartTimes;
  }

  /**
   * Wait for the run loop to finish (used by InProcessBackend).
   */
  async waitForCompletion(): Promise<void> {
    if (this.executionPromise) {
      await this.executionPromise;
    }
  }

  // ─── Private Helpers ───────────────────────────────────────

  /**
   * Settle status after the run loop empties.
   * On success → IDLE (agent stays alive for follow-up messages).
   * On error → FAILED (terminal).
   */
  private settleRoundStatus(): void {
    if (this.lastRoundError && !this.roundCancelledByUser) {
      this.setStatus(AgentStatus.FAILED);
    } else {
      this.setStatus(AgentStatus.IDLE);
    }
  }

  private setStatus(newStatus: AgentStatus): void {
    const previousStatus = this.status;
    if (previousStatus === newStatus) return;

    this.status = newStatus;

    this.core.eventEmitter?.emit(AgentEventType.STATUS_CHANGE, {
      agentId: this.config.agentId,
      previousStatus,
      newStatus,
      roundCancelledByUser: this.roundCancelledByUser || undefined,
      timestamp: Date.now(),
    });
  }

  private addMessage(
    role: AgentMessage['role'],
    content: string,
    options?: { thought?: boolean; metadata?: Record<string, unknown> },
  ): void {
    this.core.pushMessage(role, content, options);
  }

  /**
   * Wraps TOOL_WAITING_APPROVAL's onConfirm so a Cancel outcome aborts
   * the current round (headless agents bypass this path entirely).
   * Core already owns the message / live-output / shell-PID listeners.
   */
  private setupEventListeners(): void {
    const emitter = this.core.eventEmitter;

    emitter.on(
      AgentEventType.TOOL_OUTPUT_UPDATE,
      (event: AgentToolOutputUpdateEvent) => {
        if (
          event.executionStartTime !== undefined &&
          !this.executionStartTimes.has(event.callId)
        ) {
          this.executionStartTimes.set(event.callId, event.executionStartTime);
        }
      },
    );

    emitter.on(AgentEventType.TOOL_RESULT, (event: AgentToolResultEvent) => {
      this.executionStartTimes.delete(event.callId);
    });

    emitter.on(
      AgentEventType.TOOL_WAITING_APPROVAL,
      (event: AgentApprovalRequestEvent) => {
        const fullDetails = {
          ...event.confirmationDetails,
          onConfirm: async (
            outcome: Parameters<ToolCallConfirmationDetails['onConfirm']>[0],
            payload?: Parameters<ToolCallConfirmationDetails['onConfirm']>[1],
          ) => {
            this.core.deletePendingApproval(event.callId);
            // Nudge the UI to re-render so the tool transitions visually
            // from Confirming → Executing without waiting for the first
            // real TOOL_OUTPUT_UPDATE from the tool's execution.
            this.core.eventEmitter.emit(AgentEventType.TOOL_OUTPUT_UPDATE, {
              subagentId: this.core.subagentId,
              round: event.round,
              callId: event.callId,
              outputChunk: '',
              timestamp: Date.now(),
            });
            await event.respond(outcome, payload);
            // When the user denies a tool, cancel the round immediately
            // so the agent doesn't waste a turn "acknowledging" the denial.
            if (outcome === ToolConfirmationOutcome.Cancel) {
              this.cancelCurrentRound();
            }
          },
        } as ToolCallConfirmationDetails;

        this.core.setPendingApproval(event.callId, fullDetails);
      },
    );
  }
}

/**
 * Map a non-GOAL terminate mode to a visible status message for the UI,
 * or return null to suppress the message entirely.
 *
 * CANCELLED is suppressed here because cancelCurrentRound() already emits
 * its own warning. SHUTDOWN is suppressed as a normal lifecycle end.
 */
function terminateModeMessage(
  mode: AgentTerminateMode,
): { text: string; level: 'info' | 'warning' | 'error' } | null {
  switch (mode) {
    case AgentTerminateMode.MAX_TURNS:
      return {
        text: 'Agent stopped: maximum turns reached.',
        level: 'warning',
      };
    case AgentTerminateMode.TIMEOUT:
      return { text: 'Agent stopped: time limit reached.', level: 'warning' };
    case AgentTerminateMode.ERROR:
      return { text: 'Agent stopped due to an error.', level: 'error' };
    case AgentTerminateMode.CANCELLED:
    case AgentTerminateMode.SHUTDOWN:
      return null;
    default:
      return null;
  }
}