Resumable Streams

When building chat interfaces, it’s common to encounter network interruptions, page refreshes, or serverless function timeouts that can break the connection to an in-progress agent. Where a standard chat implementation would require the user to resend their message and wait for the entire response again, workflow runs are durable, and so are the streams attached to them. This means a stream can be resumed at any point, optionally only syncing the data that was missed since the last connection. Resumable streams come out of the box with Workflow DevKit. The WorkflowChatTransport helper provides a drop-in transport for the AI SDK that handles client-side resumption logic automatically.

Implementing Stream Resumption

Let’s add stream resumption to a basic durable agent:

Return the Run ID from Your API

Modify your chat endpoint to include the workflow run ID in a response header:

app/api/chat/route.ts

import { createUIMessageStreamResponse, type UIMessage } from "ai";
import { convertToModelMessages } from "ai";
import { start } from "workflow/api";
import { chatWorkflow } from "@/workflows/chat/workflow";

export async function POST(req: Request) {
  const { messages }: { messages: UIMessage[] } = await req.json();
  const modelMessages = convertToModelMessages(messages);

  const run = await start(chatWorkflow, [modelMessages]);

  return createUIMessageStreamResponse({
    stream: run.readable,
    headers: {
      "x-workflow-run-id": run.runId, // For reconnection
    },
  });
}

Add a Stream Reconnection Endpoint

Create a new API route that returns the stream for an existing run:

app/api/chat/[id]/stream/route.ts

import { createUIMessageStreamResponse } from "ai";
import { getRun } from "workflow/api";

export async function GET(
  request: Request,
  { params }: { params: Promise<{ id: string }> }
) {
  const { id } = await params;
  const { searchParams } = new URL(request.url);

  // Client provides the last chunk index they received
  const startIndexParam = searchParams.get("startIndex");
  const startIndex = startIndexParam
    ? parseInt(startIndexParam, 10)
    : undefined;

  // Fetch the existing run
  const run = getRun(id);
  const stream = run.getReadable({ startIndex });

  return createUIMessageStreamResponse({ stream });
}

The startIndex parameter ensures the client can resume from exactly where it left off.

Use `WorkflowChatTransport` in the Client

Replace the default transport in useChat with WorkflowChatTransport:

app/page.tsx

"use client";

import { useChat } from "@ai-sdk/react";
import { WorkflowChatTransport } from "@workflow/ai";
import { useMemo } from "react";

export default function ChatPage() {
  // Check for an active workflow run on mount
  const activeRunId = useMemo(() => {
    if (typeof window === "undefined") return;
    return localStorage.getItem("active-workflow-run-id") ?? undefined;
  }, []);

  const { messages, sendMessage, status } = useChat({
    resume: Boolean(activeRunId),
    transport: new WorkflowChatTransport({
      api: "/api/chat",

      // Store the run ID when a new chat starts
      onChatSendMessage: (response) => {
        const workflowRunId = response.headers.get("x-workflow-run-id");
        if (workflowRunId) {
          localStorage.setItem("active-workflow-run-id", workflowRunId);
        }
      },

      // Clear the run ID when the chat completes
      onChatEnd: () => {
        localStorage.removeItem("active-workflow-run-id");
      },

      // Use the stored run ID for reconnection
      prepareReconnectToStreamRequest: ({ api, ...rest }) => {
        const runId = localStorage.getItem("active-workflow-run-id");
        if (!runId) throw new Error("No active workflow run ID found");
        return {
          ...rest,
          api: `/api/chat/${encodeURIComponent(runId)}/stream`,
        };
      },
    }),
  });

  return (
    <div>
      {/* ... render your chat UI ... */}
    </div>
  );
}

Now try refreshing the page during a response, or simulate a network interruption. The client will automatically reconnect to the same stream and continue from where it left off.

How It Works

Initial request: When the user sends a message, WorkflowChatTransport makes a POST to /api/chat
Run ID storage: The API starts a workflow and returns the run ID in the x-workflow-run-id header
Callback execution: onChatSendMessage stores this run ID in localStorage
Automatic reconnection: If the stream is interrupted before receiving a “finish” chunk, the transport automatically reconnects
URL construction: prepareReconnectToStreamRequest builds the reconnection URL using the stored run ID
Resume from index: The reconnection endpoint returns the stream from the last known position
Cleanup: When the stream completes, onChatEnd clears the stored run ID

This approach also handles page refreshes. The client will automatically reconnect to the stream from the last known position when the UI loads with a stored run ID.

Advanced Configuration

Custom Run ID Storage

Instead of localStorage, store the run ID in a database:

lineNumbers

const transport = new WorkflowChatTransport({
  api: "/api/chat",
  onChatSendMessage: async (response) => {
    const workflowRunId = response.headers.get("x-workflow-run-id");
    if (workflowRunId) {
      await fetch("/api/sessions", {
        method: "POST",
        body: JSON.stringify({ sessionId, workflowRunId }),
      });
    }
  },
  prepareReconnectToStreamRequest: async ({ api, ...rest }) => {
    const response = await fetch(`/api/sessions/${sessionId}`);
    const { workflowRunId } = await response.json();
    return {
      ...rest,
      api: `/api/chat/${encodeURIComponent(workflowRunId)}/stream`,
    };
  },
});

Error Handling

Handle reconnection errors gracefully:

lineNumbers

const [error, setError] = useState<string | null>(null);

const transport = new WorkflowChatTransport({
  api: "/api/chat",
  maxConsecutiveErrors: 5, // Default is 3
  prepareReconnectToStreamRequest: ({ api, ...rest }) => {
    try {
      const runId = localStorage.getItem("active-workflow-run-id");
      if (!runId) {
        setError("No active session found");
        throw new Error("No active workflow run ID found");
      }
      return {
        ...rest,
        api: `/api/chat/${encodeURIComponent(runId)}/stream`,
      };
    } catch (err) {
      setError(err instanceof Error ? err.message : "Failed to reconnect");
      throw err;
    }
  },
});

Authentication

Add authentication headers to reconnection requests:

lineNumbers

const transport = new WorkflowChatTransport({
  api: "/api/chat",
  prepareReconnectToStreamRequest: ({ api, ...rest }) => {
    const runId = localStorage.getItem("active-workflow-run-id");
    const token = localStorage.getItem("auth-token");
    
    if (!runId) throw new Error("No active workflow run ID found");
    
    return {
      ...rest,
      api: `/api/chat/${encodeURIComponent(runId)}/stream`,
      headers: {
        Authorization: `Bearer ${token}`,
      },
    };
  },
});

Manual Reconnection

Trigger reconnection manually:

lineNumbers

import { useChat } from "@ai-sdk/react";
import { WorkflowChatTransport } from "@workflow/ai";
import { useState } from "react";

export default function ChatPage() {
  const [transport] = useState(() => new WorkflowChatTransport({ api: "/api/chat" }));
  
  const { messages } = useChat({ transport });

  const handleReconnect = async () => {
    const runId = localStorage.getItem("active-workflow-run-id");
    if (!runId) return;

    await transport.reconnectToStream({
      chatId: runId,
    });
  };

  return (
    <div>
      <button onClick={handleReconnect}>Reconnect</button>
      {/* ... render messages ... */}
    </div>
  );
}

Testing Resumability

Simulate Network Interruption

Test stream resumption by artificially interrupting the connection:

lineNumbers

// In browser console:
const controller = new AbortController();
fetch("/api/chat/[runId]/stream", { signal: controller.signal });

// Abort after 2 seconds
setTimeout(() => controller.abort(), 2000);

Simulate Function Timeout

Test resumption after serverless function timeout:

app/api/chat/route.ts

export async function POST(req: Request) {
  const { messages } = await req.json();
  const modelMessages = convertToModelMessages(messages);

  const run = await start(chatWorkflow, [modelMessages]);

  // Simulate timeout by closing connection early
  if (process.env.NODE_ENV === "development") {
    setTimeout(() => {
      // Connection will be closed, triggering reconnection
    }, 5000);
  }

  return createUIMessageStreamResponse({
    stream: run.readable,
    headers: { "x-workflow-run-id": run.runId },
  });
}

WorkflowChatTransport API Reference - Full configuration options
Streaming - Understanding workflow streams
getRun() API Reference - Retrieving existing runs
Chat Session Modeling - Managing conversation state

Getting Started

Installation

Core Concepts

Building AI Agents

How It Works

Deployment

Observability

Testing

Error Reference

Implementing Stream Resumption

Return the Run ID from Your API

Add a Stream Reconnection Endpoint

Use `WorkflowChatTransport` in the Client

How It Works

Advanced Configuration

Custom Run ID Storage

Error Handling

Authentication

Manual Reconnection

Testing Resumability

Simulate Network Interruption

Simulate Function Timeout

​Implementing Stream Resumption

​Return the Run ID from Your API

​Add a Stream Reconnection Endpoint

​Use WorkflowChatTransport in the Client

​How It Works

​Advanced Configuration

​Custom Run ID Storage

​Error Handling

​Authentication

​Manual Reconnection

​Testing Resumability

​Simulate Network Interruption

​Simulate Function Timeout

​Related Documentation

Implementing Stream Resumption

Return the Run ID from Your API

Add a Stream Reconnection Endpoint

Use `WorkflowChatTransport` in the Client

How It Works

Advanced Configuration

Custom Run ID Storage

Error Handling

Authentication

Manual Reconnection

Testing Resumability

Simulate Network Interruption

Simulate Function Timeout

Related Documentation