Classify Workflow stream failures (#3850)

## Summary & Motivation

Stream infrastructure failures (HTTP/2 session wedges, transport timeouts, non-2xx stream responses) surfaced as plain `Error`, so terminal classification attributed them to customer code as `USER_ERROR`. They now carry a catchable `StreamError` with a `STREAM_ERROR` run error code, attributed to the SDK and retried when transport-level or 5xx.

The v4 events response body is wrapped so a post-header stream failure is classified and reported to the dispatcher recycler — a response header arriving is not yet a successful streamed request.

## Test Plan

Unit tests added across classification, serialization round-trip, the streamer, and the v4 transport; 331 `@workflow/core` and 123 `@workflow/world-vercel` focused tests pass.

Author: alangenfeld

.changeset/stream-error-classification.md

@@ -0,0 +1,10 @@
+---
+'@workflow/errors': minor
+'@workflow/core': patch
+'@workflow/cli': patch
+'@workflow/web-shared': patch
+'@workflow/world-vercel': patch
+'workflow': minor
+---
+
+Add a catchable `StreamError` and classify Workflow stream infrastructure failures as `STREAM_ERROR` instead of `USER_ERROR`.

packages/cli/src/lib/inspect/hydration.ts

@@ -157,6 +157,7 @@ const ERROR_REVIVER_KEYS = [
   'ReferenceError',
   'RetryableError',
   'RuntimeDecryptionError',
+  'StreamError',
   'SyntaxError',
   'TypeError',
   'URIError',

packages/core/src/classify-error.test.ts

@@ -5,6 +5,7 @@ import {
   ReplayDivergenceError,
   RUN_ERROR_CODES,
   RuntimeDecryptionError,
+  StreamError,
   ThrottleError,
   TooEarlyError,
   WorkflowDeploymentMismatchError,
@@ -153,6 +154,12 @@ describe('classifyRunError', () => {
     ).toBe(RUN_ERROR_CODES.WORLD_CONTRACT_ERROR);
   });
 
+  it('classifies StreamError as STREAM_ERROR', () => {
+    expect(classifyRunError(new StreamError('stream write failed'))).toBe(
+      RUN_ERROR_CODES.STREAM_ERROR
+    );
+  });
+
   it('classifies a TRANSPORT error as WORLD_CONTRACT_ERROR (backend fault, not USER_ERROR)', () => {
     // Transport blips are normally redelivered via the queue (see
     // isRetryableWorldError); if one ever reaches terminal classification it is
@@ -194,6 +201,22 @@ describe('isRetryableWorldError', () => {
     ).toBe(true);
   });
 
+  it('retries transport and server-side StreamErrors, but not terminal 4xx responses', () => {
+    expect(isRetryableWorldError(new StreamError('stream read failed'))).toBe(
+      true
+    );
+    expect(
+      isRetryableWorldError(
+        new StreamError('stream service failed', { status: 503 })
+      )
+    ).toBe(true);
+    expect(
+      isRetryableWorldError(
+        new StreamError('stream request was invalid', { status: 400 })
+      )
+    ).toBe(false);
+  });
+
   it('treats TRANSPORT and TIMEOUT codes as retryable', () => {
     expect(
       isRetryableWorldError(

packages/core/src/classify-error.ts

@@ -6,6 +6,7 @@ import {
   type RunErrorCode,
   RuntimeDecryptionError,
   StepNotRegisteredError,
+  StreamError,
   ThrottleError,
   WorkflowDeploymentMismatchError,
   WorkflowNotRegisteredError,
@@ -29,7 +30,11 @@ const WORLD_CONTRACT_ERROR_CODES = new Set([
  * from. Kept distinct from `WORLD_CONTRACT_ERROR_CODES` so a transport blip is
  * never misclassified as the server returning a malformed response.
  */
-const RETRYABLE_WORLD_ERROR_CODES = new Set(['TRANSPORT', 'TIMEOUT']);
+const RETRYABLE_WORLD_ERROR_CODES = new Set([
+  'TRANSPORT',
+  'TIMEOUT',
+  RUN_ERROR_CODES.STREAM_ERROR,
+]);
 
 /**
  * Set of error names that should classify as generic `RUNTIME_ERROR`. Each
@@ -100,6 +105,9 @@ export function isRetryableWorldError(err: unknown): boolean {
   if (ThrottleError.is(err)) {
     return true;
   }
+  if (StreamError.is(err)) {
+    return err.status === undefined || err.status >= 500;
+  }
   if (!WorkflowWorldError.is(err)) {
     return false;
   }
@@ -133,6 +141,10 @@ export function classifyRunError(err: unknown): RunErrorCode {
   // attribute an outage correctly. Note the retryable variants are normally
   // redelivered via the queue (see `isRetryableWorldError`) and only reach this
   // terminal classification if the run ultimately gives up.
+  if (StreamError.is(err)) {
+    return RUN_ERROR_CODES.STREAM_ERROR;
+  }
+
   if (isWorldContractError(err) || isRetryableWorldError(err)) {
     return RUN_ERROR_CODES.WORLD_CONTRACT_ERROR;
   }

packages/core/src/describe-error.ts

@@ -85,6 +85,8 @@ const MAX_DELIVERIES_HINT =
   'The workflow queue exceeded its max-delivery budget. This usually indicates a persistent runtime failure — check the most recent stack traces for the underlying cause.';
 const MAX_EVENTS_HINT =
   'The workflow exceeded the maximum number of events per run. This usually means unbounded work in the workflow function — e.g. a loop that keeps creating steps without terminating. Break long-running workflows into child workflows to stay under the limit.';
+const STREAM_ERROR_HINT =
+  'Workflow stream infrastructure failed while reading or writing data. This is an SDK or backend failure, not an error in your workflow code; please retry the run and report persistent failures with the runId.';
 const WORLD_CONTRACT_HINT =
   'The workflow backend returned data that violated the SDK contract. This is not retryable; please report it with the stack trace and runId.';
 const DEPLOYMENT_MISMATCH_HINT =
@@ -144,6 +146,13 @@ export function describeRunError(
       hint: CORRUPTED_EVENT_LOG_HINT,
     };
   }
+  if (errorCode === RUN_ERROR_CODES.STREAM_ERROR) {
+    return {
+      attribution: 'sdk',
+      errorCode,
+      hint: STREAM_ERROR_HINT,
+    };
+  }
   if (errorCode === RUN_ERROR_CODES.WORLD_CONTRACT_ERROR) {
     return {
       attribution: 'sdk',
@@ -264,6 +273,14 @@ export function describeError(
     };
   }
 
+  if (effectiveCode === RUN_ERROR_CODES.STREAM_ERROR) {
+    return {
+      attribution: 'sdk',
+      errorCode: effectiveCode,
+      hint: STREAM_ERROR_HINT,
+    };
+  }
+
   if (effectiveCode === RUN_ERROR_CODES.WORLD_CONTRACT_ERROR) {
     return {
       attribution: 'sdk',

packages/core/src/reconnecting-framed-stream.test.ts

@@ -1,4 +1,4 @@
-import { StreamExpiredError } from '@workflow/errors';
+import { StreamError, StreamExpiredError } from '@workflow/errors';
 import { SPEC_VERSION_CURRENT, type World } from '@workflow/world';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 
@@ -660,8 +660,11 @@ describe('createReconnectingFramedStream', () => {
     setWorld(world);
 
     const stream = createReconnectingFramedStream(RUN_ID, 's', 0);
-    await expect(readAll(stream)).rejects.toThrow(
-      /exceeded maximum reconnection attempts/
+    const error = await readAll(stream).catch((cause: unknown) => cause);
+    expect(StreamError.is(error)).toBe(true);
+    expect(error).toHaveProperty(
+      'message',
+      expect.stringMatching(/exceeded maximum reconnection attempts/)
     );
     // Initial connect + one connect per allowed reconnect; the following
     // reconnect throws before opening another stream.

packages/core/src/runtime/quickjs-serde.test.ts

@@ -19,6 +19,7 @@
 
 import fs from 'node:fs';
 import { createRequire } from 'node:module';
+import { StreamError } from '@workflow/errors';
 import { QuickJS } from 'quickjs-wasi';
 import { afterAll, beforeAll, describe, expect, it } from 'vitest';
 import {
@@ -162,6 +163,21 @@ describe('wire parity: guest serialize matches the reference codec', () => {
         return e;
       },
     ],
+    [
+      'StreamError',
+      '(() => { const cause = new Error("socket closed"); cause.stack = "cause-stack"; const e = new Error("stream failed", { cause }); e.name = "StreamError"; e.stack = "stream-stack"; e.status = 503; e.url = "https://workflow.example/stream"; return e; })()',
+      () => {
+        const cause = new Error('socket closed');
+        cause.stack = 'cause-stack';
+        const error = new StreamError('stream failed', {
+          cause,
+          status: 503,
+          url: 'https://workflow.example/stream',
+        });
+        error.stack = 'stream-stack';
+        return error;
+      },
+    ],
     [
       'Error with cause',
       '(() => { const c = new Error("cause"); c.stack = "cs"; const e = new Error("outer", { cause: c }); e.stack = "os"; return e; })()',

packages/core/src/runtime/quickjs-serde.ts

@@ -256,6 +256,7 @@ const SYMBOL_NAMES = [
   '@workflow/errors//HookConflictError',
   '@workflow/errors//RetryableError',
   '@workflow/errors//RuntimeDecryptionError',
+  '@workflow/errors//StreamError',
 ] as const;
 type SymbolName = (typeof SYMBOL_NAMES)[number];
 
@@ -1167,6 +1168,23 @@ export function createQuickJSSerde(
       if (Object.hasOwn(shape, 'cause')) reduced.cause = shape.cause;
       return reduced;
     },
+    StreamError: (value) => {
+      if (!isHandle(value) || !value.isError) return false;
+      if (chainedString(value, 'name') !== 'StreamError') return false;
+      const shape = reduceErrorShape(value) as Record<string, unknown>;
+      const reduced: Record<string, unknown> = {
+        message: shape.message,
+        stack: shape.stack,
+      };
+      if (Object.hasOwn(shape, 'cause')) reduced.cause = shape.cause;
+      const status = own(value, 'status');
+      if (status && !status.isUndefined) reduced.status = status;
+      else status?.dispose();
+      const url = own(value, 'url');
+      if (url && !url.isUndefined) reduced.url = url;
+      else url?.dispose();
+      return reduced;
+    },
     SyntaxError: namedErrorSubclassReducer('SyntaxError'),
     TypeError: namedErrorSubclassReducer('TypeError'),
     URIError: namedErrorSubclassReducer('URIError'),
@@ -1849,6 +1867,37 @@ export function createQuickJSSerde(
       }
       return error;
     },
+    StreamError: (value: JSValueHandle) => {
+      const cls = registeredErrorClass('@workflow/errors//StreamError');
+      let error: JSValueHandle;
+      if (cls) {
+        const options = vm.newObject();
+        for (const property of ['cause', 'status', 'url'] as const) {
+          if (!guestHasOwn(value, property)) continue;
+          const propertyValue = own(value, property) ?? vm.undefined;
+          options.setProp(property, propertyValue);
+          if (propertyValue !== vm.undefined) propertyValue.dispose();
+        }
+        const message = own(value, 'message') ?? vm.undefined;
+        error = vm.construct(cls, message, options);
+        if (message !== vm.undefined) message.dispose();
+        options.dispose();
+        const stack = own(value, 'stack');
+        if (stack && !stack.isUndefined) define(error, 'stack', stack);
+        stack?.dispose();
+        cls.dispose();
+      } else {
+        error = buildError(i.Error, value, { name: 'StreamError' });
+        for (const property of ['status', 'url'] as const) {
+          const propertyValue = own(value, property);
+          if (propertyValue && !propertyValue.isUndefined) {
+            define(error, property, propertyValue);
+          }
+          propertyValue?.dispose();
+        }
+      }
+      return error;
+    },
     SyntaxError: namedErrorSubclassReviver('SyntaxError'),
     TypeError: namedErrorSubclassReviver('TypeError'),
     URIError: namedErrorSubclassReviver('URIError'),

packages/core/src/runtime/runs.ts

@@ -1,4 +1,4 @@
-import { EntityConflictError } from '@workflow/errors';
+import { EntityConflictError, StreamError } from '@workflow/errors';
 import {
   BULK_CANCEL_MAX_RUN_IDS,
   type BulkCancelWorkflowRunResult,
@@ -409,7 +409,8 @@ export async function readStream(
   try {
     return await world.streams.get(runId, streamId, options?.startIndex);
   } catch (err) {
-    throw new Error(
+    if (StreamError.is(err)) throw err;
+    throw new StreamError(
       `Failed to read stream ${streamId}: ${err instanceof Error ? err.message : String(err)}`,
       { cause: err }
     );
@@ -426,7 +427,8 @@ export async function listStreams(
   try {
     return await world.streams.list(runId);
   } catch (err) {
-    throw new Error(
+    if (StreamError.is(err)) throw err;
+    throw new StreamError(
       `Failed to list streams for run ${runId}: ${err instanceof Error ? err.message : String(err)}`,
       { cause: err }
     );

packages/core/src/serialization.test.ts

@@ -4,7 +4,9 @@ import {
   FatalError,
   HookConflictError,
   RetryableError,
+  RUN_ERROR_CODES,
   RuntimeDecryptionError,
+  StreamError,
 } from '@workflow/errors';
 import { WORKFLOW_DESERIALIZE, WORKFLOW_SERIALIZE } from '@workflow/serde';
 import { beforeAll, describe, expect, it, vi } from 'vitest';
@@ -4714,6 +4716,34 @@ describe('dehydrate/hydrateRunError', () => {
     expect(cause?.name).toBe('OperationError');
   });
 
+  it('should round-trip StreamError as a catchable error type', async () => {
+    const original = new StreamError('stream write failed', {
+      cause: new Error('HTTP 500'),
+      status: 503,
+      url: 'https://workflow.example/stream',
+    });
+    const serialized = await dehydrateRunError(
+      original,
+      mockRunId,
+      noEncryptionKey
+    );
+    const hydrated = (await hydrateRunError(
+      serialized,
+      mockRunId,
+      noEncryptionKey
+    )) as StreamError;
+
+    expect(StreamError.is(hydrated)).toBe(true);
+    expect(hydrated).toBeInstanceOf(StreamError);
+    expect(hydrated).toMatchObject({
+      message: 'stream write failed',
+      status: 503,
+      url: 'https://workflow.example/stream',
+      code: RUN_ERROR_CODES.STREAM_ERROR,
+    });
+    expect((hydrated.cause as Error).message).toBe('HTTP 500');
+  });
+
   it('should produce DEVALUE_V1-prefixed binary output', async () => {
     const serialized = await dehydrateRunError(
       new Error('x'),