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

import type { AgentRun, RunEvent } from "../domain/agent-run";
import type { Conversation } from "../domain/conversation";
import { CoreError } from "../errors/core-error";
import type { ModelPort } 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 { ProjectService } from "./project-service";

export type StartRunInput =
  | Readonly<{ kind: "ordinary"; message: string }>
  | Readonly<{ kind: "project"; projectId: string; message: string }>
  | Readonly<{ kind: "existing"; conversationId: string; message: string }>;

export type StartedRun = Readonly<{
  run: AgentRun;
  events: AsyncIterable<RunEvent>;
}>;

export class AgentRunService {
  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,
  ) {}

  async start(input: StartRunInput, signal: AbortSignal): Promise<StartedRun> {
    const runId = this.ids.create();
    let conversation: Conversation;
    if (input.kind === "ordinary") {
      conversation = await this.conversations.createWithFirstMessage(
        input.message,
        runId,
      );
    } else if (input.kind === "project") {
      await this.projects.requireAvailable(input.projectId);
      conversation = await this.conversations.createProjectWithFirstMessage(
        input.projectId,
        input.message,
        runId,
      );
    } else {
      const existing = await this.conversations.getConversation(
        input.conversationId,
      );
      if (existing.projectId)
        await this.projects.requireAvailable(existing.projectId);
      conversation = await this.conversations.appendUserMessage(
        input.conversationId,
        input.message,
        runId,
      );
    }
    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 { run, events: this.execute(run, signal) };
  }

  private async *execute(
    run: AgentRun,
    signal: AbortSignal,
  ): AsyncIterable<RunEvent> {
    let sequence = 0;
    const event = async (
      type: RunEvent["type"],
      payload: Record<string, unknown>,
    ) => {
      const value: RunEvent = {
        runId: run.id,
        sequence: ++sequence,
        timestamp: this.clock.now().toISOString(),
        type,
        payload,
      };
      await this.runs.appendEvent(value);
      return value;
    };
    try {
      yield await event("run.started", { conversationId: run.conversationId });
      const messageId = this.ids.create();
      yield await event("message.started", { messageId });
      const conversation = await this.conversations.getConversation(
        run.conversationId,
      );
      let content = "";
      for await (const modelEvent of this.model.stream(
        { model: "default", messages: conversation.messages },
        signal,
      )) {
        if (modelEvent.type !== "text.delta") continue;
        content += modelEvent.delta;
        yield await event("message.delta", {
          messageId,
          delta: modelEvent.delta,
        });
      }
      if (!content)
        throw new CoreError("MODEL_RESPONSE_INVALID", "模型没有返回有效内容");
      await this.conversations.appendAssistantMessage(
        run.conversationId,
        messageId,
        run.id,
        content,
      );
      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) {
      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),
      });
    }
  }
}

function safeErrorMessage(cause: unknown): string {
  return cause instanceof CoreError ? cause.message : "模型服务暂时不可用";
}