llm-to-agent/packages/agent-core/src/use-cases/agent-run-service.ts

import type { AgentRun, RunEvent } from "../domain/agent-run";
import type { UserInteraction } from "../domain/user-interaction";
import { CoreError } from "../errors/core-error";
import type { ModelPort, ModelRequest } from "../ports/model-port";
import type { RunRepository } from "../ports/run-repository";
import type { ClockPort, IdPort } from "../ports/system-ports";
import type { ConversationService } from "./conversation-service";
import type { InteractionService } from "./interaction-service";
import type { ProjectService } from "./project-service";
import type { WorkspaceService } from "./workspace-service";
import { isWorkspaceTool, workspaceTools } from "./workspace-tools";
import { interactionTool } from "./interaction-tool";
import {
  ensureNoActiveRun,
  lastSequence,
  messagesThroughTrigger,
  persistRunEvent,
  prepareRetryRun,
  requireRun,
  recoverInterruptedRun,
  safeErrorMessage,
} from "./run-service-helpers";
import type { StartedRun, StartRunInput } from "./run-types";
import { executeWorkspaceTool, prepareRunConversation } from "./run-workspace";

export class AgentRunService {
  private readonly controllers = new Map<string, AbortController>();
  private startQueue: Promise<void> = Promise.resolve();

  constructor(
    private readonly conversations: ConversationService,
    private readonly model: ModelPort,
    private readonly runs: RunRepository,
    private readonly clock: ClockPort,
    private readonly ids: IdPort,
    private readonly projects: ProjectService,
    private readonly interactions: InteractionService,
    private readonly workspace?: WorkspaceService,
  ) {}

  async start(input: StartRunInput, signal: AbortSignal): Promise<StartedRun> {
    return this.withStartLock(async () => {
      await ensureNoActiveRun(this.runs);
      const runId = this.ids.create();
      const conversation = await prepareRunConversation(
        input,
        runId,
        this.conversations,
        this.projects,
        this.workspace,
      );
      const timestamp = this.clock.now().toISOString();
      const triggerMessage = conversation.messages.at(-1);
      if (triggerMessage?.role !== "user")
        throw new CoreError(
          "RUN_TRIGGER_INVALID",
          "无法确定本次运行的用户消息",
        );
      const run: AgentRun = {
        id: runId,
        conversationId: conversation.id,
        triggerMessageId: triggerMessage.id,
        status: "running",
        createdAt: timestamp,
        updatedAt: timestamp,
      };
      await this.runs.create(run);
      return this.started(run, signal, 0);
    });
  }

  async resume(
    interaction: UserInteraction,
    signal: AbortSignal,
  ): Promise<StartedRun> {
    const current = await requireRun(this.runs, interaction.runId);
    if (current.status !== "waiting_user")
      throw new CoreError("RUN_NOT_WAITING", "该任务当前没有等待用户回答");
    if (!interaction.answer)
      throw new CoreError("INTERACTION_INVALID", "交互答案不存在");
    const run = {
      ...current,
      status: "running" as const,
      updatedAt: this.clock.now().toISOString(),
    };
    await this.runs.update(run);
    const sequence = lastSequence(await this.runs.listEvents(run.id));
    const continuation: NonNullable<ModelRequest["continuations"]>[number] = {
      toolCallId: interaction.toolCallId,
      toolName: "request_user_interaction",
      toolArguments: interaction.toolArguments,
      result: JSON.stringify(interaction.answer),
    };
    return this.started(run, signal, sequence, [continuation], interaction);
  }

  async cancel(runId: string): Promise<readonly RunEvent[]> {
    const run = await requireRun(this.runs, runId);
    if (
      run.status === "cancelled" ||
      run.status === "completed" ||
      run.status === "failed"
    )
      return [];
    if (run.status === "running") {
      const controller = this.controllers.get(runId);
      if (!controller)
        throw new CoreError(
          "RUN_NOT_CANCELLABLE",
          "任务正在恢复中，请稍后重试",
        );
      controller.abort();
      return [];
    }
    const interaction = await this.interactions.cancelPending(runId);
    if (!interaction)
      throw new CoreError("INTERACTION_NOT_FOUND", "等待中的交互请求不存在");
    let sequence = lastSequence(await this.runs.listEvents(run.id));
    const cancelled = await persistRunEvent(
      this.runs,
      this.clock,
      run.id,
      ++sequence,
      "interaction.cancelled",
      { interactionId: interaction.id },
    );
    await this.runs.update({
      ...run,
      status: "cancelled",
      updatedAt: this.clock.now().toISOString(),
    });
    const finished = await persistRunEvent(
      this.runs,
      this.clock,
      run.id,
      ++sequence,
      "run.cancelled",
      {},
    );
    return [cancelled, finished];
  }

  async retry(runId: string, signal: AbortSignal): Promise<StartedRun> {
    return this.withStartLock(async () => {
      const run = await prepareRetryRun(
        runId,
        this.conversations,
        this.runs,
        this.clock,
        this.ids,
      );
      return this.started(run, signal, 0);
    });
  }

  async lastSequence(runId: string): Promise<number> {
    await requireRun(this.runs, runId);
    return lastSequence(await this.runs.listEvents(runId));
  }

  getRun(runId: string): Promise<AgentRun> {
    return requireRun(this.runs, runId);
  }

  async listEvents(runId: string): Promise<readonly RunEvent[]> {
    await requireRun(this.runs, runId);
    return this.runs.listEvents(runId);
  }

  async recoverInterrupted(): Promise<readonly RunEvent[]> {
    const active = await this.runs.findActive();
    if (
      active?.status === "waiting_user" &&
      (await this.interactions.findPendingByRun(active.id))
    )
      return [];
    return recoverInterruptedRun(this.runs, this.clock);
  }

  private started(
    run: AgentRun,
    externalSignal: AbortSignal,
    sequence: number,
    continuations?: NonNullable<ModelRequest["continuations"]>,
    interaction?: UserInteraction,
  ): StartedRun {
    const controller = new AbortController();
    this.controllers.set(run.id, controller);
    const signal = AbortSignal.any([externalSignal, controller.signal]);
    return {
      run,
      events: this.execute(run, signal, sequence, continuations, interaction),
    };
  }

  private async withStartLock<T>(operation: () => Promise<T>): Promise<T> {
    const previous = this.startQueue;
    let release = () => {};
    this.startQueue = new Promise<void>((resolve) => {
      release = resolve;
    });
    await previous;
    try {
      return await operation();
    } finally {
      release();
    }
  }

  private async *execute(
    run: AgentRun,
    signal: AbortSignal,
    initialSequence: number,
    initialContinuations: NonNullable<ModelRequest["continuations"]> = [],
    resolvedInteraction?: UserInteraction,
  ): AsyncIterable<RunEvent> {
    let sequence = initialSequence;
    let messageId: string | undefined;
    let content = "";
    let messagePersisted = false;
    const event = async (
      type: RunEvent["type"],
      payload: Record<string, unknown>,
    ) =>
      persistRunEvent(this.runs, this.clock, run.id, ++sequence, type, payload);
    try {
      if (initialContinuations.length === 0)
        yield await event("run.started", {
          conversationId: run.conversationId,
        });
      if (resolvedInteraction)
        yield await event("interaction.resolved", {
          interactionId: resolvedInteraction.id,
          answer: resolvedInteraction.answer,
        });
      messageId = this.ids.create();
      yield await event("message.started", { messageId });
      const conversation = await this.conversations.getConversation(
        run.conversationId,
      );
      const modelMessages = run.retryOfRunId
        ? messagesThroughTrigger(conversation, run.triggerMessageId)
        : conversation.messages;
      signal.throwIfAborted();
      const continuations = [...initialContinuations];
      let toolCount = 0;
      modelLoop: while (true) {
        let requestedTool = false;
        for await (const modelEvent of this.model.stream(
          {
            model: "default",
            messages: modelMessages,
            tools: [interactionTool, ...workspaceTools],
            ...(continuations.length ? { continuations } : {}),
          },
          signal,
        )) {
          signal.throwIfAborted();
          if (modelEvent.type === "text.delta") {
            content += modelEvent.delta;
            yield await event("message.delta", {
              messageId,
              delta: modelEvent.delta,
            });
            continue;
          }
          if (modelEvent.type === "tool.requested") {
            requestedTool = true;
            if (isWorkspaceTool(modelEvent.name)) {
              if (++toolCount > 12)
                throw new CoreError(
                  "TOOL_LIMIT_EXCEEDED",
                  "文件工具调用次数过多",
                );
              yield await event("tool.started", {
                messageId,
                toolCallId: modelEvent.toolCallId,
                name: modelEvent.name,
              });
              const tool = await executeWorkspaceTool(
                this.workspace,
                run.conversationId,
                modelEvent.name,
                modelEvent.arguments,
              );
              if (!tool.failed) {
                yield await event("tool.completed", {
                  messageId,
                  toolCallId: modelEvent.toolCallId,
                  name: modelEvent.name,
                  summary: tool.summary,
                });
              } else {
                yield await event("tool.failed", {
                  messageId,
                  toolCallId: modelEvent.toolCallId,
                  name: modelEvent.name,
                  code: tool.failed.code,
                  message: tool.failed.message,
                });
              }
              continuations.push({
                toolCallId: modelEvent.toolCallId,
                toolName: modelEvent.name,
                toolArguments: modelEvent.arguments,
                result: tool.result,
              });
              continue modelLoop;
            }
            if (modelEvent.name !== interactionTool.name)
              throw new CoreError(
                "TOOL_NOT_SUPPORTED",
                "模型请求了尚未支持的工具",
              );
            await this.conversations.appendAssistantMessage(
              run.conversationId,
              messageId,
              run.id,
              content,
            );
            messagePersisted = true;
            yield await event("message.completed", { messageId, content });
            const interaction = await this.interactions.create({
              runId: run.id,
              conversationId: run.conversationId,
              messageId,
              toolCallId: modelEvent.toolCallId,
              toolArguments: modelEvent.arguments,
            });
            await this.runs.update({
              ...run,
              status: "waiting_user",
              updatedAt: this.clock.now().toISOString(),
            });
            yield await event("interaction.requested", { interaction });
            return;
          }
        }
        if (!requestedTool) break;
      }
      signal.throwIfAborted();
      if (!content)
        throw new CoreError("MODEL_RESPONSE_INVALID", "模型没有返回有效内容");
      await this.conversations.appendAssistantMessage(
        run.conversationId,
        messageId,
        run.id,
        content,
      );
      messagePersisted = true;
      yield await event("message.completed", { messageId, content });
      await this.runs.update({
        ...run,
        status: "completed",
        updatedAt: this.clock.now().toISOString(),
      });
      yield await event("run.completed", {});
    } catch (cause) {
      if (signal.aborted) {
        if (content && messageId && !messagePersisted) {
          await this.conversations.appendAssistantMessage(
            run.conversationId,
            messageId,
            run.id,
            content,
          );
          yield await event("message.completed", {
            messageId,
            content,
            cancelled: true,
          });
        }
        await this.runs.update({
          ...run,
          status: "cancelled",
          updatedAt: this.clock.now().toISOString(),
        });
        yield await event("run.cancelled", {});
        return;
      }
      const code =
        cause instanceof CoreError ? cause.code : "MODEL_UNAVAILABLE";
      await this.runs.update({
        ...run,
        status: "failed",
        errorCode: code,
        updatedAt: this.clock.now().toISOString(),
      });
      yield await event("run.failed", {
        code,
        message: safeErrorMessage(cause),
      });
    } finally {
      this.controllers.delete(run.id);
    }
  }
}