Method runFactory

packages/runtime/src/agent/index.ts:1796–1964

(
    input: RunAgentInput,
    config: BuiltInAgentFactoryConfig,
  )

  }

  private runFactory(
    input: RunAgentInput,
    config: BuiltInAgentFactoryConfig,
  ): Observable<BaseEvent> {
    if (this.abortController) {
      throw new Error(
        "Agent is already running. Call abortRun() first or create a new instance.",
      );
    }

    // Set synchronously before Observable creation to close TOCTOU window
    this.abortController = new AbortController();
    const controller = this.abortController;

    return new Observable<BaseEvent>((subscriber) => {
      const startEvent: RunStartedEvent = {
        type: EventType.RUN_STARTED,
        threadId: input.threadId,
        runId: input.runId,
      };
      subscriber.next(startEvent);

      const ctx: AgentFactoryContext = {
        input,
        abortController: controller,
        abortSignal: controller.signal,
        interrupt: async (interrupts: Interrupt[]) => {
          const resume = input.resume ?? [];
          const ids = new Set(interrupts.map((i) => i.id));
          const matching = resume.filter((r) => ids.has(r.interruptId));
          // All requested interrupts addressed → resume: return responses.
          if (interrupts.length > 0 && matching.length === interrupts.length) {
            return matching;
          }
          // Fresh run (or not yet addressed) → pause.
          throw new InterruptSignal(interrupts);
        },
      };

      // Resume injection (aisdk/tanstack): map each ResumeEntry to a tool-role
      // message keyed by interruptId (=== the paused tool call's id) and append
      // it to the messages the factory sees. Both SDK converters
      // (convertMessagesToVercelAISDKMessages / convertInputToTanStackAI) turn a
      // tool-role message into that SDK's native tool-result, so the model
      // continues — no SDK-specific approval-response wiring needed. The `custom`
      // factory reads input.resume itself via ctx.interrupt(), so leave it alone.
      // Idempotent: skip entries the client already recorded as a tool-result
      // message in the thread (useInterrupt persists resolutions so the
      // conversation stays well-formed across turns). Only synthesize results
      // for entries that aren't already answered, so we never double-answer a
      // tool call.
      const answeredToolCallIds = new Set(
        input.messages
          .filter((m) => m.role === "tool")
          .map((m) => (m as { toolCallId?: string }).toolCallId)
          .filter((id): id is string => typeof id === "string"),
      );
      const resumeToolMessages: Message[] = (input.resume ?? [])