Method startStream

src/node/services/streamManager.ts:3625–3775

* Starts a new stream for a workspace, automatically cancelling any existing stream * * Uses per-workspace mutex to prevent concurrent streams. The mutex ensures: * 1. Only one startStream can execute at a time per workspace * 2. Old stream fully exits before new stream starts * 3. No

(
    workspaceId: string,
    messages: ModelMessage[],
    model: LanguageModel,
    modelString: string,
    historySequence: number,
    system: string,
    runtime: Runtime,
    messageId: string,
    abortSignal?: AbortSignal,
    tools?: Record<string, Tool>,
    initialMetadata?: Partial<MuxMetadata>,
    providerOptions?: Record<string, unknown>,
    maxOutputTokens?: number,
    toolPolicy?: ToolPolicy,
    providedStreamToken?: StreamToken,
    hasQueuedMessages?: (dispatchMode?: "tool-end" | "turn-end") => boolean,
    workspaceName?: string,
    thinkingLevel?: string,
    headers?: Record<string, string | undefined>,
    anthropicCacheTtlOverride?: AnthropicCacheTtl,
    callSettingsOverrides?: ResolvedCallSettingsOverrides,
    onChunk?: StreamTextOnChunk,
    onStepMessages?: (messages: ModelMessage[]) => void,
    providedRuntimeTempDir?: string,
    modelFallback?: ModelFallbackOptions
  )

   * 3. No race conditions in stream registration or cleanup
   */
  async startStream(
    workspaceId: string,
    messages: ModelMessage[],
    model: LanguageModel,
    modelString: string,
    historySequence: number,
    system: string,
    runtime: Runtime,
    messageId: string,
    abortSignal?: AbortSignal,
    tools?: Record<string, Tool>,
    initialMetadata?: Partial<MuxMetadata>,
    providerOptions?: Record<string, unknown>,
    maxOutputTokens?: number,
    toolPolicy?: ToolPolicy,
    providedStreamToken?: StreamToken,
    hasQueuedMessages?: (dispatchMode?: "tool-end" | "turn-end") => boolean,
    workspaceName?: string,
    thinkingLevel?: string,
    headers?: Record<string, string | undefined>,
    anthropicCacheTtlOverride?: AnthropicCacheTtl,
    callSettingsOverrides?: ResolvedCallSettingsOverrides,
    onChunk?: StreamTextOnChunk,
    onStepMessages?: (messages: ModelMessage[]) => void,
    providedRuntimeTempDir?: string,
    modelFallback?: ModelFallbackOptions
  ): Promise<Result<StreamToken, SendMessageError>> {
    const typedWorkspaceId = workspaceId as WorkspaceId;

    if (messages.length === 0) {
      return Err({
        type: "unknown",
        raw: "Invalid prompt: messages must not be empty",
      });
    }

    // Get or create mutex for this workspace
    if (!this.streamLocks.has(typedWorkspaceId)) {
      this.streamLocks.set(typedWorkspaceId, new AsyncMutex());
    }
    const mutex = this.streamLocks.get(typedWorkspaceId)!;

    try {
      // Acquire lock - guarantees only one startStream per workspace
      // Lock is automatically released when scope exits via Symbol.asyncDispose
      await using _lock = await mutex.acquire();

      // DEBUG: Log stream start
      log.debug(
        `[STREAM START] workspaceId=${workspaceId} historySequence=${historySequence} model=${modelString}`
      );

      const streamAbortController = new AbortController();
      const unlinkAbortSignal = linkAbortSignal(abortSignal, streamAbortController);

      let runtimeTempDir: string | undefined;
      let streamRegistered = false;