Core: require specifying runId when writing to stream

.changeset/purple-pianos-stare.md

@@ -0,0 +1,10 @@
+---
+"@workflow/world-postgres": patch
+"@workflow/world-vercel": patch
+"@workflow/world-local": patch
+"@workflow/web-shared": patch
+"@workflow/world": patch
+"@workflow/core": patch
+---
+
+Require specifying runId when writing to stream

packages/core/e2e/e2e.test.ts

@@ -1,3 +1,4 @@
+import { withResolvers } from '@workflow/utils';
 import { assert, describe, expect, test } from 'vitest';
 import { dehydrateWorkflowArguments } from '../src/serialization';
 import { cliInspectJson, isLocalDeployment } from './utils';
@@ -21,18 +22,29 @@
 
   url.searchParams.set('workflowFile', workflowFile);
   url.searchParams.set('workflowFn', workflowFn);
+
+  const ops: Promise<void>[] = [];
+  const { promise: runIdPromise, resolve: resolveRunId } =
+    withResolvers<string>();
+  const dehydratedArgs = dehydrateWorkflowArguments(args, ops, runIdPromise);
+
   const res = await fetch(url, {
     method: 'POST',
-    body: JSON.stringify(dehydrateWorkflowArguments(args, [], globalThis)),
+    body: JSON.stringify(dehydratedArgs),
   });
   if (!res.ok) {
     throw new Error(
       `Failed to trigger workflow: ${res.url} ${
         res.status
       }: ${await res.text()}`
     );
   }
   const run = await res.json();
+  resolveRunId(run.runId);
+
+  // Resolve and wait for any stream operations
+  await Promise.all(ops);
+
   return run;
 }
 
@@ -79,7 +91,7 @@
   ])('addTenWorkflow', { timeout: 60_000 }, async (workflow) => {
     const run = await triggerWorkflow(workflow, [123]);
     const returnValue = await getWorkflowReturnValue(run.runId);
     expect(returnValue).toBe(133);
 
     const { json } = await cliInspectJson(`runs ${run.runId} --withData`);
     expect(json).toMatchObject({
@@ -291,7 +303,7 @@
       method: 'POST',
       body: JSON.stringify({}),
     });
     expect(res.status).toBe(404);
     const body = await res.text();
     expect(body).toBe('');
   });

packages/core/src/observability.ts

@@ -44,15 +44,21 @@ const streamPrintRevivers: Record<string, (value: any) => any> = {
 };
 
 const hydrateStepIO = <
-  T extends { stepId?: string; input?: any; output?: any },
+  T extends { stepId?: string; input?: any; output?: any; runId?: string },
 >(
   step: T
 ): T => {
   return {
     ...step,
     input:
       step.input && Array.isArray(step.input) && step.input.length
-        ? hydrateStepArguments(step.input, [], globalThis, streamPrintRevivers)
+        ? hydrateStepArguments(
+            step.input,
+            [],
+            step.runId as string,
+            globalThis,
+            streamPrintRevivers
+          )
         : step.input,
     output: step.output
       ? hydrateStepReturnValue(step.output, globalThis, streamPrintRevivers)
@@ -79,14 +85,17 @@ const hydrateWorkflowIO = <
       ? hydrateWorkflowReturnValue(
           workflow.output,
           [],
+          workflow.runId as string,
           globalThis,
           streamPrintRevivers
         )
       : workflow.output,
   };
 };
 
-const hydrateEventData = <T extends { eventId?: string; eventData?: any }>(
+const hydrateEventData = <
+  T extends { eventId?: string; eventData?: any; runId?: string },
+>(
   event: T
 ): T => {
   if (!event.eventData) {
@@ -99,7 +108,13 @@ const hydrateEventData = <T extends { eventId?: string; eventData?: any }>(
     eventData: Object.fromEntries(
       Object.entries(event.eventData).map(([key, value]) => [
         key,
-        hydrateStepArguments(value as any, [], globalThis),
+        hydrateStepArguments(
+          value as any,
+          [],
+          event.runId as string,
+          globalThis,
+          streamPrintRevivers
+        ),
       ])
     ),
   };
@@ -110,9 +125,16 @@ const hydrateHookMetadata = <T extends { hookId?: string; metadata?: any }>(
 ): T => {
   return {
     ...hook,
-    metadata: hook.metadata
-      ? hydrateStepArguments(hook.metadata, [], globalThis)
-      : hook.metadata,
+    metadata:
+      hook.metadata && 'runId' in hook
+        ? hydrateStepArguments(
+            hook.metadata,
+            [],
+            hook.runId as string,
+            globalThis,
+            streamPrintRevivers
+          )
+        : hook.metadata,
   };
 };
 

packages/core/src/runtime.ts

@@ -180,7 +180,7 @@ export class Run<TResult> {
   ): ReadableStream<R> {
     const { ops = [], global = globalThis, startIndex, namespace } = options;
     const name = getWorkflowRunStreamId(this.runId, namespace);
-    return getExternalRevivers(global, ops).ReadableStream({
+    return getExternalRevivers(global, ops, this.runId).ReadableStream({
       name,
       startIndex,
     }) as ReadableStream<R>;
@@ -197,7 +197,7 @@ export class Run<TResult> {
         const run = await this.world.runs.get(this.runId);
 
         if (run.status === 'completed') {
-          return hydrateWorkflowReturnValue(run.output, [], globalThis);
+          return hydrateWorkflowReturnValue(run.output, [], this.runId);
         }
 
         if (run.status === 'cancelled') {
@@ -671,7 +671,7 @@ export const stepEntrypoint =
             }
             // Hydrate the step input arguments
             const ops: Promise<void>[] = [];
-            const args = hydrateStepArguments(step.input, ops);
+            const args = hydrateStepArguments(step.input, ops, workflowRunId);
 
             span?.setAttributes({
               ...Attribute.StepArgumentsCount(args.length),
@@ -698,7 +698,7 @@ export const stepEntrypoint =
               () => stepFn(...args)
             );
 
-            result = dehydrateStepReturnValue(result, ops);
+            result = dehydrateStepReturnValue(result, ops, workflowRunId);
 
             waitUntil(Promise.all(ops));
 

packages/core/src/runtime/resume-hook.ts

@@ -23,7 +23,7 @@ export async function getHookByToken(token: string): Promise<Hook> {
   const world = getWorld();
   const hook = await world.hooks.getByToken(token);
   if (typeof hook.metadata !== 'undefined') {
-    hook.metadata = hydrateStepArguments(hook.metadata as any, [], globalThis);
+    hook.metadata = hydrateStepArguments(hook.metadata as any, [], hook.runId);
   }
   return hook;
 }
@@ -79,7 +79,7 @@ export async function resumeHook<T = any>(
         const dehydratedPayload = dehydrateStepReturnValue(
           payload,
           ops,
-          globalThis
+          hook.runId
         );
         waitUntil(Promise.all(ops));
 

packages/core/src/runtime/start.ts

@@ -1,5 +1,6 @@
 import { waitUntil } from '@vercel/functions';
 import { WorkflowRuntimeError } from '@workflow/errors';
+import { withResolvers } from '@workflow/utils';
 import { Run } from '../runtime.js';
 import type { Serializable, WorkflowInvokePayload } from '../schemas.js';
 import { dehydrateWorkflowArguments } from '../serialization.js';
@@ -87,7 +88,14 @@ export async function start<TArgs extends unknown[], TResult>(
       const world = getWorld();
       const deploymentId = opts.deploymentId ?? (await world.getDeploymentId());
       const ops: Promise<void>[] = [];
-      const workflowArguments = dehydrateWorkflowArguments(args, ops);
+      const { promise: runIdPromise, resolve: resolveRunId } =
+        withResolvers<string>();
+
+      const workflowArguments = dehydrateWorkflowArguments(
+        args,
+        ops,
+        runIdPromise
+      );
       // Serialize current trace context to propagate across queue boundary
       const traceCarrier = await serializeTraceCarrier();
 
@@ -97,6 +105,9 @@ export async function start<TArgs extends unknown[], TResult>(
         input: workflowArguments,
         executionContext: { traceCarrier },
       });
+
+      resolveRunId(runResponse.runId);
+
       waitUntil(Promise.all(ops));
 
       span?.setAttributes({

packages/core/src/serialization.test.ts

@@ -15,6 +15,8 @@ import {
 import { STREAM_NAME_SYMBOL } from './symbols.js';
 import { createContext } from './vm/index.js';
 
+const mockRunId = 'wrun_mockidnumber0001';
+
 describe('getStreamType', () => {
   it('should return `undefined` for a regular stream', () => {
     const stream = new ReadableStream();
@@ -41,7 +43,7 @@ describe('workflow arguments', () => {
 
   it('should work with Date', () => {
     const date = new Date('2025-07-17T04:30:34.824Z');
-    const serialized = dehydrateWorkflowArguments(date, []);
+    const serialized = dehydrateWorkflowArguments(date, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -61,7 +63,7 @@ describe('workflow arguments', () => {
 
   it('should work with invalid Date', () => {
     const date = new Date('asdf');
-    const serialized = dehydrateWorkflowArguments(date, []);
+    const serialized = dehydrateWorkflowArguments(date, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -81,7 +83,7 @@ describe('workflow arguments', () => {
 
   it('should work with BigInt', () => {
     const bigInt = BigInt('9007199254740992');
-    const serialized = dehydrateWorkflowArguments(bigInt, []);
+    const serialized = dehydrateWorkflowArguments(bigInt, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -99,7 +101,7 @@ describe('workflow arguments', () => {
 
   it('should work with BigInt negative', () => {
     const bigInt = BigInt('-12345678901234567890');
-    const serialized = dehydrateWorkflowArguments(bigInt, []);
+    const serialized = dehydrateWorkflowArguments(bigInt, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -120,7 +122,7 @@ describe('workflow arguments', () => {
       [2, 'foo'],
       [6, 'bar'],
     ]);
-    const serialized = dehydrateWorkflowArguments(map, []);
+    const serialized = dehydrateWorkflowArguments(map, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -154,7 +156,7 @@ describe('workflow arguments', () => {
 
   it('should work with Set', () => {
     const set = new Set([1, '2', true]);
-    const serialized = dehydrateWorkflowArguments(set, []);
+    const serialized = dehydrateWorkflowArguments(set, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -180,7 +182,7 @@ describe('workflow arguments', () => {
 
   it('should work with WritableStream', () => {
     const stream = new WritableStream();
-    const serialized = dehydrateWorkflowArguments(stream, []);
+    const serialized = dehydrateWorkflowArguments(stream, [], mockRunId);
     const uuid = serialized[2];
     expect(serialized).toMatchInlineSnapshot(`
       [
@@ -205,7 +207,7 @@ describe('workflow arguments', () => {
 
   it('should work with ReadableStream', () => {
     const stream = new ReadableStream();
-    const serialized = dehydrateWorkflowArguments(stream, []);
+    const serialized = dehydrateWorkflowArguments(stream, [], mockRunId);
     const uuid = serialized[2];
     expect(serialized).toMatchInlineSnapshot(`
       [
@@ -233,7 +235,7 @@ describe('workflow arguments', () => {
     headers.set('foo', 'bar');
     headers.append('set-cookie', 'a');
     headers.append('set-cookie', 'b');
-    const serialized = dehydrateWorkflowArguments(headers, []);
+    const serialized = dehydrateWorkflowArguments(headers, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -281,7 +283,7 @@ describe('workflow arguments', () => {
         ['set-cookie', 'b'],
       ]),
     });
-    const serialized = dehydrateWorkflowArguments(response, []);
+    const serialized = dehydrateWorkflowArguments(response, [], mockRunId);
     const bodyUuid = serialized[serialized.length - 3];
     expect(serialized).toMatchInlineSnapshot(`
       [
@@ -372,7 +374,7 @@ describe('workflow arguments', () => {
   it('should work with URLSearchParams', () => {
     const params = new URLSearchParams('a=1&b=2&a=3');
 
-    const serialized = dehydrateWorkflowArguments(params, []);
+    const serialized = dehydrateWorkflowArguments(params, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -399,7 +401,7 @@ describe('workflow arguments', () => {
   it('should work with empty URLSearchParams', () => {
     const params = new URLSearchParams();
 
-    const serialized = dehydrateWorkflowArguments(params, []);
+    const serialized = dehydrateWorkflowArguments(params, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -420,7 +422,7 @@ describe('workflow arguments', () => {
   it('should work with empty ArrayBuffer', () => {
     const buffer = new ArrayBuffer(0);
 
-    const serialized = dehydrateWorkflowArguments(buffer, []);
+    const serialized = dehydrateWorkflowArguments(buffer, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -440,7 +442,7 @@ describe('workflow arguments', () => {
   it('should work with empty Uint8Array', () => {
     const array = new Uint8Array(0);
 
-    const serialized = dehydrateWorkflowArguments(array, []);
+    const serialized = dehydrateWorkflowArguments(array, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -461,7 +463,7 @@ describe('workflow arguments', () => {
   it('should work with empty Int32Array', () => {
     const array = new Int32Array(0);
 
-    const serialized = dehydrateWorkflowArguments(array, []);
+    const serialized = dehydrateWorkflowArguments(array, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -482,7 +484,7 @@ describe('workflow arguments', () => {
   it('should work with empty Float64Array', () => {
     const array = new Float64Array(0);
 
-    const serialized = dehydrateWorkflowArguments(array, []);
+    const serialized = dehydrateWorkflowArguments(array, [], mockRunId);
     expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -517,7 +519,7 @@ describe('workflow arguments', () => {
         duplex: 'half',
       } as RequestInit);
 
-      const serialized = dehydrateWorkflowArguments(request, []);
+      const serialized = dehydrateWorkflowArguments(request, [], mockRunId);
       expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -625,7 +627,7 @@ describe('workflow arguments', () => {
       const responseWritable = new WritableStream();
       request[Symbol.for('WEBHOOK_RESPONSE_WRITABLE')] = responseWritable;
 
-      const serialized = dehydrateWorkflowArguments(request, []);
+      const serialized = dehydrateWorkflowArguments(request, [], mockRunId);
       expect(serialized).toMatchInlineSnapshot(`
       [
         [
@@ -727,7 +729,7 @@ describe('workflow arguments', () => {
     class Foo {}
     let err: WorkflowRuntimeError | undefined;
     try {
-      dehydrateWorkflowArguments(new Foo(), []);
+      dehydrateWorkflowArguments(new Foo(), [], mockRunId);
     } catch (err_) {
       err = err_ as WorkflowRuntimeError;
     }
@@ -775,7 +777,7 @@ describe('step return value', () => {
     class Foo {}
     let err: WorkflowRuntimeError | undefined;
     try {
-      dehydrateStepReturnValue(new Foo(), []);
+      dehydrateStepReturnValue(new Foo(), [], mockRunId);
     } catch (err_) {
       err = err_ as WorkflowRuntimeError;
     }