contrib: add @temporalio/workflow-streams

.github/CODEOWNERS

@@ -4,4 +4,5 @@
 # someone opens a pull request.
 * @temporalio/sdk
 
-/contrib/ai-sdk/ @temporalio/sdk @temporalio/ai-sdk
+/contrib/ai-sdk/           @temporalio/sdk @temporalio/ai-sdk
+/contrib/workflow-streams/ @temporalio/sdk @temporalio/ai-sdk

.github/workflows/conventions.yml

@@ -53,5 +53,7 @@ jobs:
 
       - run: pnpm run lint:check
         env:
-          NODE_OPTIONS: '--max-old-space-size=8192'
+          NODE_OPTIONS: --max-old-space-size=8192
       - run: pnpm run lint:prune
+        env:
+          NODE_OPTIONS: --max-old-space-size=8192

.github/workflows/docs.yml

@@ -51,6 +51,7 @@ jobs:
         run: pnpm run docs
         env:
           ALGOLIA_API_KEY: ${{ secrets.ALGOLIA_API_KEY }}
+          NODE_OPTIONS: --max-old-space-size=8192
 
       - name: Publish production docs
         if: ${{ inputs.publish_target == 'prod' }}

contrib/ai-sdk/package.json

@@ -24,9 +24,12 @@
     "workerThreads": false
   },
   "dependencies": {
+    "@temporalio/activity": "workspace:*",
+    "@temporalio/client": "workspace:*",
     "@temporalio/common": "workspace:*",
     "@temporalio/plugin": "workspace:*",
     "@temporalio/workflow": "workspace:*",
+    "@temporalio/workflow-streams": "workspace:*",
     "@ungap/structured-clone": "^1.3.0",
     "headers-polyfill": "^4.0.3",
     "web-streams-polyfill": "^4.2.0"

contrib/ai-sdk/src/__tests__/test-ai-sdk.ts

@@ -41,7 +41,13 @@ import { workflowInterceptorModules } from '@temporalio/testing';
 import { bundleWorkflowCode, DefaultLogger, Runtime } from '@temporalio/worker';
 import type { InjectedSinks } from '@temporalio/worker';
 import type { BaseContext } from '@temporalio/test-helpers';
-import { test as anyTest, helpers, Worker, TestWorkflowEnvironment } from '@temporalio/test-helpers';
+import {
+  test as anyTest,
+  createBaseBundlerOptions,
+  helpers,
+  Worker,
+  TestWorkflowEnvironment,
+} from '@temporalio/test-helpers';
 import { AiSdkPlugin, createActivities } from '..';
 import {
   embeddingWorkflow,
@@ -234,6 +240,7 @@ const test = anyTest as TestFn<BaseContext>;
 test.before(async (t) => {
   const env = await TestWorkflowEnvironment.createLocal();
   const workflowBundle = await bundleWorkflowCode({
+    ...createBaseBundlerOptions(),
     workflowsPath: require.resolve('./workflows/ai-sdk'),
     workflowInterceptorModules,
     logger: new DefaultLogger('WARN'),
@@ -465,6 +472,7 @@ test('Telemetry', async (t) => {
       ],
       taskQueue: 'test-ai-telemetry',
       workflowsPath: require.resolve('./workflows/ai-sdk'),
+      bundlerOptions: createBaseBundlerOptions(),
 
       interceptors: {
         client: {

contrib/ai-sdk/src/activities.ts

@@ -1,15 +1,24 @@
 import type {
   LanguageModelV3CallOptions,
+  LanguageModelV3Content,
+  LanguageModelV3FinishReason,
   LanguageModelV3GenerateResult,
+  LanguageModelV3Usage,
   EmbeddingModelV3Result,
   SharedV3ProviderOptions,
   SharedV3Headers,
+  SharedV3Warning,
   ProviderV3,
 } from '@ai-sdk/provider';
 import { asSchema, type Schema, type ToolExecutionOptions } from 'ai';
 import { ApplicationFailure } from '@temporalio/common';
+import { Context } from '@temporalio/activity';
+import { WorkflowStreamClient } from '@temporalio/workflow-streams/client';
+import type { Duration } from '@temporalio/common/lib/time';
 import type { McpClientFactories, McpClientFactory } from './mcp';
 
+const encoder = new TextEncoder();
+
 /**
  * Arguments for invoking a language model activity.
  */
@@ -18,6 +27,16 @@ export interface InvokeModelArgs {
   options: LanguageModelV3CallOptions;
 }
 
+/**
+ * Arguments for invoking a streaming language model activity.
+ */
+export interface InvokeModelStreamingArgs extends InvokeModelArgs {
+  modelId: string;
+  options: LanguageModelV3CallOptions;
+  streamingTopic: string;
+  streamingBatchInterval?: Duration;
+}
+
 /**
  * Result from a language model invocation.
  * This is an alias to the AI SDK's LanguageModelV3GenerateResult for type safety.
@@ -75,6 +94,118 @@ export function createActivities(provider: ProviderV3, mcpClientFactories?: McpC
       const model = provider.languageModel(args.modelId);
       return await model.doGenerate(args.options);
     },
+
+    /**
+     * Streaming-aware model activity.
+     *
+     * Calls `model.doStream()`, publishes each yielded AI SDK stream part
+     * as JSON to the stream side channel, and returns the assembled
+     * `LanguageModelV3GenerateResult`. Consumers receive native AI SDK
+     * stream-part types (text-delta, reasoning-delta, tool-input-delta,
+     * response-metadata, finish, ...); no normalization happens here.
+     */
+    async invokeModelStreaming(args: InvokeModelStreamingArgs): Promise<InvokeModelResult> {
+      await using stream = WorkflowStreamClient.fromWithinActivity({
+        batchInterval: args.streamingBatchInterval ?? '100 milliseconds',
+      });
+      const events = stream.topic(args.streamingTopic);
+
+      const model = provider.languageModel(args.modelId);
+      const streamResult = await model.doStream(args.options);
+
+      const content: LanguageModelV3Content[] = [];
+      let finishReason: LanguageModelV3FinishReason = { unified: 'other', raw: undefined };
+      let usage: LanguageModelV3Usage = {
+        inputTokens: { total: undefined, noCache: undefined, cacheRead: undefined, cacheWrite: undefined },
+        outputTokens: { total: undefined, text: undefined, reasoning: undefined },
+      };
+      const warnings: SharedV3Warning[] = [];
+      let responseMetadata: Record<string, unknown> | undefined;
+
+      const textBlocks = new Map<string, string>();
+      const reasoningBlocks = new Map<string, string>();
+
+      const reader = streamResult.stream.getReader();
+
+      while (true) {
+        const { done, value: part } = await reader.read();
+        if (done) break;
+
+        Context.current().heartbeat();
+
+        // Publish the raw stream part as JSON so consumers can switch on
+        // the native AI SDK type. Accumulation below is for the final
+        // assembled result this activity returns.
+        events.publish(encoder.encode(JSON.stringify(part)));
+
+        switch (part.type) {
+          case 'stream-start':
+            warnings.push(...part.warnings);
+            break;
+          case 'text-start':
+            textBlocks.set(part.id, '');
+            break;
+          case 'text-delta':
+            textBlocks.set(part.id, (textBlocks.get(part.id) ?? '') + part.delta);
+            break;
+          case 'text-end':
+            content.push({
+              type: 'text',
+              text: textBlocks.get(part.id) ?? '',
+              providerMetadata: part.providerMetadata,
+            });
+            textBlocks.delete(part.id);
+            break;
+          case 'reasoning-start':
+            reasoningBlocks.set(part.id, '');
+            break;
+          case 'reasoning-delta':
+            reasoningBlocks.set(part.id, (reasoningBlocks.get(part.id) ?? '') + part.delta);
+            break;
+          case 'reasoning-end':
+            content.push({
+              type: 'reasoning',
+              text: reasoningBlocks.get(part.id) ?? '',
+              providerMetadata: part.providerMetadata,
+            });
+            reasoningBlocks.delete(part.id);
+            break;
+          case 'response-metadata':
+            responseMetadata = {
+              id: part.id,
+              timestamp: part.timestamp,
+              modelId: part.modelId,
+            };
+            break;
+          case 'finish':
+            finishReason = part.finishReason;
+            usage = part.usage;
+            break;
+          default:
+            // tool-call, tool-result, file, source — collect as content
+            if (
+              'type' in part &&
+              (part.type === 'tool-call' ||
+                part.type === 'tool-result' ||
+                part.type === 'file' ||
+                part.type === 'source')
+            ) {
+              content.push(part);
+            }
+            break;
+        }
+      }
+
+      return {
+        content,
+        finishReason,
+        usage,
+        warnings,
+        request: streamResult.request,
+        response: responseMetadata ? { ...responseMetadata, ...streamResult.response } : streamResult.response,
+      };
+    },
+
     async invokeEmbeddingModel(args: InvokeEmbeddingModelArgs): Promise<InvokeEmbeddingModelResult> {
       const model = provider.embeddingModel(args.modelId);
       return await model.doEmbed({

contrib/ai-sdk/src/mcp.ts

@@ -44,7 +44,7 @@ export class TemporalMCPClient {
         toolName,
         {
           description: toolResult.description,
-          execute: async (input, options) => {
+          execute: async (input: unknown, options: unknown) => {
             const activities = workflow.proxyActivities({
               summary: toolName,
               startToCloseTimeout: '10 minutes',

contrib/ai-sdk/src/provider.ts

@@ -1,3 +1,4 @@
+import type { ReadableStreamDefaultController } from 'node:stream/web';
 import type {
   EmbeddingModelV3,
   EmbeddingModelV3CallOptions,
@@ -13,6 +14,11 @@ import type {
 import * as workflow from '@temporalio/workflow';
 import type { ActivityOptions } from '@temporalio/workflow';
 import { ApplicationFailure } from '@temporalio/common';
+import type { Duration } from '@temporalio/common/lib/time';
+
+// `ReadableStream` is a sandbox global; type-only import keeps `node:stream/web`
+// out of the workflow bundle (es2023 lib has no DOM types).
+declare const ReadableStream: typeof import('node:stream/web').ReadableStream;
 
 /**
  * Options for configuring the TemporalProvider with per-model activity settings.
@@ -28,7 +34,22 @@ export interface TemporalProviderOptions {
    * Activity options specific to language model calls.
    * Merged with default options, with these taking precedence.
    */
-  languageModel?: ActivityOptions;
+  languageModel?: ActivityOptions & {
+    /**
+     * Topic name on the workflow's stream that streaming model calls publish
+     * raw stream parts to. When set, `doStream` is enabled and routes through
+     * the streaming activity; when unset, `doStream` throws. Pick a unique
+     * name per concurrent streaming call to keep event streams separable.
+     */
+    streamingTopic?: string;
+
+    /**
+     * Batch interval for the per-activity `WorkflowStreamClient` that
+     * publishes stream parts back to the workflow. Lower values reduce
+     * latency at the cost of more signal traffic. Defaults to 100ms.
+     */
+    streamingBatchInterval?: Duration;
+  };
 
   /**
    * Activity options specific to embedding model calls.
@@ -46,11 +67,16 @@ export interface TemporalProviderOptions {
 export class TemporalLanguageModel implements LanguageModelV3 {
   readonly specificationVersion = 'v3';
   readonly provider = 'temporal';
+  private readonly streamingTopic: string | undefined;
+  private readonly streamingBatchInterval: Duration | undefined;
 
   constructor(
     readonly modelId: string,
-    readonly options?: ActivityOptions
-  ) {}
+    readonly options?: ActivityOptions & { streamingTopic?: string; streamingBatchInterval?: Duration }
+  ) {
+    this.streamingTopic = options?.streamingTopic;
+    this.streamingBatchInterval = options?.streamingBatchInterval;
+  }
 
   get supportedUrls(): Record<string, RegExp[]> {
     return {};
@@ -75,8 +101,59 @@ export class TemporalLanguageModel implements LanguageModelV3 {
     return result;
   }
 
-  doStream(_options: LanguageModelV3CallOptions): PromiseLike<LanguageModelV3StreamResult> {
-    throw ApplicationFailure.nonRetryable('Streaming not supported.');
+  async doStream(options: LanguageModelV3CallOptions): Promise<LanguageModelV3StreamResult> {
+    if (this.streamingTopic === undefined) {
+      throw ApplicationFailure.nonRetryable(
+        'Streaming not enabled. Set streamingTopic in languageModel provider options.'
+      );
+    }
+
+    // Call the streaming activity, which publishes tokens via stream
+    // and returns the accumulated result.
+    const activities = workflow.proxyActivities({
+      startToCloseTimeout: '10 minutes',
+      ...this.options,
+    });
+    const result = await activities.invokeModelStreaming!({
+      modelId: this.modelId,
+      options,
+      streamingTopic: this.streamingTopic,
+      streamingBatchInterval: this.streamingBatchInterval,
+    });
+    if (result === undefined) {
+      throw ApplicationFailure.nonRetryable('Received undefined response from streaming model activity.');
+    }
+
+    // Wrap the accumulated result as a ReadableStream that replays the content.
+    // Real-time token streaming already happened via stream in the activity.
+    const stream = new ReadableStream({
+      start(controller: ReadableStreamDefaultController) {
+        controller.enqueue({ type: 'stream-start', warnings: result.warnings ?? [] });
+        let partIndex = 0;
+        for (const item of result.content ?? []) {
+          const id = `part-${partIndex++}`;
+          if (item.type === 'text') {
+            controller.enqueue({ type: 'text-start', id });
+            controller.enqueue({ type: 'text-delta', id, delta: item.text });
+            controller.enqueue({ type: 'text-end', id });
+          } else if (item.type === 'reasoning') {
+            controller.enqueue({ type: 'reasoning-start', id });
+            controller.enqueue({ type: 'reasoning-delta', id, delta: item.text });
+            controller.enqueue({ type: 'reasoning-end', id });
+          } else {
+            controller.enqueue(item);
+          }
+        }
+        controller.enqueue({
+          type: 'finish',
+          finishReason: result.finishReason,
+          usage: result.usage,
+        });
+        controller.close();
+      },
+    });
+
+    return { stream, request: result.request, response: result.response };
   }
 }