grpc Cancelled Error in Vercel SDK streamText

[AshishAvesta]

It's working fine while working with generateText, but when we use it with streamText, it gives an error.

const { text } = await generateText({
model,
prompt: 'What is Laminar flow?',
experimental_telemetry: {
isEnabled: true,
tracer: getTracer(),
},
});

const { textStream } = await streamText({
model,
messages: promptMessages,
abortSignal: signal,
maxSteps: 10,
experimental_telemetry: {
isEnabled: true,
tracer: getTracer(),
},
onChunk: ({ chunk }) => { ... }
});

it gives error like :

Error: 1 CANCELLED: Call cancelled
at callErrorFromStatus (/app/node_modules/@grpc/grpc-js/src/call.ts:82:17)
at Object.onReceiveStatus (/app/node_modules/@grpc/grpc-js/src/client.ts:360:55)
at Object.onReceiveStatus (/app/node_modules/@grpc/grpc-js/src/client-interceptors.ts:458:34)
at Object.onReceiveStatus (/app/node_modules/@grpc/grpc-js/src/client-interceptors.ts:419:48)
at /app/node_modules/@grpc/grpc-js/src/resolving-call.ts:169:24
at processTicksAndRejections (node:internal/process/task_queues:85:11)

[skull8888888]

@AshishAvesta Thank you for the issue. We're working on the new version of SDK right now, and will include fix for this in the next release. We will release it today or tomorrow.

[skull8888888]

@AshishAvesta can you please provide more context? I can't reproduce it on my side.

I'm running this and everything works and trace

import { getTracer, Laminar } from "@lmnr-ai/lmnr";
import { streamText } from "ai"
import { openai } from "@ai-sdk/openai";
import 'dotenv/config'

Laminar.initialize();


async function main() {
  const { textStream } = await streamText({
    model: openai("gpt-4.1-mini"),
    messages: [
      {
        role: "user",
        content: "What is the capital of France?"
      }
    ],
    experimental_telemetry: {
      isEnabled: true,
      tracer: getTracer(),
    },
  });

  for await (const textPart of textStream) {
    process.stdout.write(textPart);
  }

  Laminar.flush();
}

main()
  

And I run it with ts-node test.ts

[AshishAvesta]

Here’s a clear and meaningful version of your explanation you can post in the GitHub issue:

---

Hi, thank you for looking into this.

To provide more context:

I initialize Laminar when my app starts using the following code:

Laminar.initialize({
  projectApiKey: apiKey,
});

Later, I have a service that runs when a user starts a thread or chat. Here's a simplified version of the code:

const { textStream } = await streamText({
  model,
  tools: preparedTools,
  messages: promptMessages,
  abortSignal: signal,
  maxSteps: 10,
  temperature: modelConfig.temperature,
  experimental_telemetry: telemetryOptions,
  onChunk: ({ chunk }) => {
    const typedChunk = this.convertChunk(chunk);

    if (typedChunk.type === "text") {
      fullResponse += (typedChunk as TextContent).text;
    }

    onChunk(typedChunk, signal);
  },
  onError({ error }) {
    onChunk(
      {
        type: "text",
        text: "Error: " + error,
        timestamp: new Date().toISOString()
      },
      signal
    );
  }
});

However, I do not currently call Laminar.flush() or Laminar.shutdown() after this. I wasn’t sure if those are required in this flow, and whether omitting them might cause issues such as missing telemetry or incomplete data being sent.

Please let me know if I should explicitly call flush() or shutdown() after the interaction ends, or if there's a recommended pattern for cleaning up or finalising Laminar usage in such streaming scenarios.

Thanks again!

---

Let me know if you'd like to tailor this more to your specific use case or style.

[skull8888888]

flush and shutdown are primarily required if you run standalone script. The purpose of those functions is to flush collected span data before the script ends.

For your issue, does it happen consistently? Also, I'm curious why do you use onChunk instead of iterating with textStream?

[AshishAvesta]

I'm not running this in a standalone script and using onChunk because I'm streaming data back to the front end without delay, and also using MCP servers (for passing or giving context to LLM).

I'm calling this function :

/**

• Converts a raw chunk from Vercel AI SDK to MessageContent
  */
  private convertChunk(rawChunk: any): MessageContent {
  if (rawChunk.type === "text-delta") {
  return {
  type: "text",
  text: rawChunk.textDelta,
  timestamp: this.createTimestamp()
  };
  } else if (rawChunk.type === "tool-call") {
  return {
  type: "tool-call",
  timestamp: this.createTimestamp(),
  toolCallId: rawChunk.toolCallId,
  toolName: rawChunk.toolName,
  args: rawChunk.args || {}
  };
  } else if (rawChunk.type === "tool-result") {
  return {
  type: "tool-result",
  timestamp: this.createTimestamp(),
  toolCallId: rawChunk.toolCallId,
  toolName: rawChunk.toolName,
  result: rawChunk.result || {}
  };
  }
  // Default case, create empty text content
  return {
  type: "text",
  text: "",
  timestamp: this.createTimestamp()
  };
  }

[AshishAvesta]

And for the model, I used createProviderRegistry.

https://ai-sdk.dev/docs/reference/ai-sdk-core/provider-registry

Does this happen due to the model?

[skull8888888]

@AshishAvesta can you please provide full snippet of your code. I can't really reproduce an issue on my side

[AshishAvesta]

// ai-response.service.vercel.ts

// --- External Libraries ---
import { streamText, createProviderRegistry } from "ai";
import { createOpenAI } from "@ai-sdk/openai";
import { createAnthropic } from "@ai-sdk/anthropic";
import { v4 as uuidv4 } from "uuid";

// --- Domain Interfaces and Errors ---
import { BaseAIService } from "../base/base-ai-service";
import {
  IAIService,
  ModelConfig,
  MessageContent,
  TextContent,
  ResponseResult,
} from "../../../../domain/interfaces/services/ai.service.interface";
import { AIResponseError } from "../../../../domain/errors/chat.errors";
import { getObservabilityService } from "../../../../domain/factories/observability.factory";
import { IObservabilityService } from "../../../../domain/interfaces/services/observability.service.interface";

// --- Types ---
type Provider = "openai" | "anthropic";
type ProviderModelId = `${Provider}:${string}`;
type ApiKeys = Record<Provider, string>;

/**
 * Vercel AI SDK implementation of the AI Response Service
 */
export class VercelAIService extends BaseAIService implements IAIService {
  private createProviderRegistry(
    apiKeys: Partial<ApiKeys>
  ): ReturnType<typeof createProviderRegistry> {
    return createProviderRegistry({
      openai: createOpenAI({ apiKey: apiKeys.openai || "" }),
      anthropic: createAnthropic({ apiKey: apiKeys.anthropic || "" }),
    });
  }

  private createModelIdentifier(
    providerName: string,
    modelName: string
  ): ProviderModelId {
    const provider = providerName.toLowerCase() as Provider;

    if (provider !== "openai" && provider !== "anthropic") {
      throw new AIResponseError(`Unsupported AI provider: ${providerName}`);
    }

    return `${provider}:${modelName}`;
  }

  private buildMessages(params: {
    message: string;
    previousMessages: Array<{
      id: string;
      role: "user" | "assistant";
      content: MessageContent[];
    }>;
    systemMessage?: string;
  }): Message[] {
    const { previousMessages, message, systemMessage } = params;
    const messages: Message[] = [];

    const enhancedSystemMessage = this.createEnhancedSystemMessage(systemMessage);

    messages.push({
      id: uuidv4(),
      role: "system",
      content: enhancedSystemMessage,
    });

    messages.push(
      ...previousMessages.map((msg) => ({
        id: msg.id,
        role: msg.role,
        content: this.contentArrayToString(msg.content),
      })),
      { id: uuidv4(), role: "user", content: message }
    );

    return messages;
  }

  async generateStreamingResponse(
    params: {
      modelConfig: ModelConfig;
      message: string;
      threadId: string;
      previousMessages: Array<{
        id: string;
        role: "user" | "assistant";
        content: MessageContent[];
      }>;
      userId: string;
      systemMessage?: string;
      tools?: Record<string, any>;
    },
    onChunk: (chunk: MessageContent, signal?: AbortSignal) => void,
    signal?: AbortSignal
  ): Promise<ResponseResult> {
    if (signal?.aborted) {
      return {
        fullResponse: "",
        responseContent: [],
        traceId: "aborted-before-start",
      };
    }

    let fullResponse = "";
    let aborted = false;
    const requestId = uuidv4();
    const observabilityService = getObservabilityService();

    const abortHandler = () => {
      aborted = true;
    };
    signal?.addEventListener("abort", abortHandler);

    try {
      const { modelConfig, message, previousMessages, systemMessage, userId } = params;
      const { modelName, provider } = modelConfig;

      const registry = this.createProviderRegistry({
        [provider.name]: provider.apiKey,
      });

      const promptMessages = this.buildMessages({
        message,
        previousMessages,
        systemMessage,
      });

      const fullModelId = this.createModelIdentifier(provider.name, modelName);
      const model = registry.languageModel(fullModelId);

      const telemetryOptions = this.getTelemetryOptions(
        observabilityService,
        userId,
        requestId
      );

      const { textStream } = await streamText({
        model,
        messages: promptMessages,
        abortSignal: signal,
        maxSteps: 10,
        temperature: modelConfig.temperature,
        experimental_telemetry: telemetryOptions,
        onChunk: ({ chunk }) => {
          const typedChunk = this.convertChunk(chunk);
          if (typedChunk.type === "text") {
            fullResponse += (typedChunk as TextContent).text;
          }
          onChunk(typedChunk, signal);
        },
        onError: ({ error }) => {
          onChunk(
            {
              type: "text",
              text: "Error: " + error,
              timestamp: new Date().toISOString(),
            },
            signal
          );
        },
      });

      try {
        for await (const _ of textStream) {
          if (aborted || signal?.aborted) break;
        }
      } catch (error) {
        if (aborted || signal?.aborted) {
          return {
            fullResponse,
            responseContent: [],
            traceId: "aborted-during-stream",
          };
        }
        throw error;
      }

      return {
        fullResponse,
        responseContent: [],
        traceId: requestId,
      };
    } catch (error) {
      throw new AIResponseError(
        error instanceof Error ? error.message : "Failed to generate AI response"
      );
    } finally {
      signal?.removeEventListener("abort", abortHandler);
    }
  }

  private getTelemetryOptions(
    observabilityService: IObservabilityService,
    userId?: string,
    requestId?: string
  ): any {
    if (observabilityService.isEnabled() && observabilityService.supportsTracing?.()) {
      const metadata = observabilityService.getRequestMetadata?.(userId, requestId) || {};
      const tracer = observabilityService.getTracer?.();

      if (tracer) {
        return {
          isEnabled: true,
          tracer,
          metadata,
        };
      }
    }

    return { isEnabled: false };
  }

  private convertChunk(rawChunk: any): MessageContent {
    if (rawChunk.type === "text-delta") {
      return {
        type: "text",
        text: rawChunk.textDelta,
        timestamp: this.createTimestamp(),
      };
    } else if (rawChunk.type === "tool-call") {
      return {
        type: "tool-call",
        timestamp: this.createTimestamp(),
        toolCallId: rawChunk.toolCallId,
        toolName: rawChunk.toolName,
        args: rawChunk.args || {},
      };
    } else if (rawChunk.type === "tool-result") {
      return {
        type: "tool-result",
        timestamp: this.createTimestamp(),
        toolCallId: rawChunk.toolCallId,
        toolName: rawChunk.toolName,
        result: rawChunk.result || {},
      };
    }

    return {
      type: "text",
      text: "",
      timestamp: this.createTimestamp(),
    };
  }
}

[AshishAvesta]

@AshishAvesta can you please provide full snippet of your code. I can't really reproduce an issue on my side

I discovered that the problem stemmed from the model. When I used the model as per your snippet, it worked for me as well. However, when I tried using my own method with createOpenAI(@ai-sdk/openai), I received an error.

Do you support models created using createOpenAI?