test: reduce e2e timing flakes

.changeset/e2e-stream-flush.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Harden inline step completion against stream flush and transient persistence races.

packages/core/e2e/bench.bench.ts

@@ -1,10 +1,12 @@
+import fs from 'node:fs';
+import path from 'node:path';
 import { createVercelWorld } from '@workflow/world-vercel';
-import fs from 'fs';
-import path from 'path';
 import { bench, describe } from 'vitest';
 import { getTrustedSourcesHeaders } from '../../../scripts/trusted-sources-headers.mjs';
 import type { Run } from '../src/runtime';
 import { setWorld, start } from '../src/runtime';
+import { getWorldLazy } from '../src/runtime/get-world-lazy';
+import { STREAM_CLIENT_NAME_SYMBOL } from '../src/symbols';
 import { getWorkbenchAppPath, isLocalDeployment } from './utils';
 
 const deploymentUrl = process.env.DEPLOYMENT_URL;
@@ -99,6 +101,10 @@ async function getWorkflowMetadata(
 
 const benchWf = (fn: string) =>
   getWorkflowMetadata('workflows/97_bench.ts', fn);
+const STREAM_READ_TIMEOUT_MS = process.env.WORKFLOW_VERCEL_ENV
+  ? 90_000
+  : 30_000;
+const STREAM_READ_RETRY_INTERVAL_MS = 1_000;
 
 // Store workflow execution times for each benchmark
 const workflowTimings: Record<
@@ -328,6 +334,73 @@ async function consumeStreamWithMetrics(
   return { firstByteTimeMs, slurpTimeMs, totalBytes, chunks };
 }
 
+/**
+ * Rehydrate a returned stream until enough persisted bytes are visible.
+ * Vercel stream chunks can lag the completed run output by a short window;
+ * opening the returned stream in that window can produce an empty reader.
+ */
+async function consumeReturnedStreamWithMetrics(
+  value: unknown,
+  run: Run<unknown>,
+  startedAt: string | undefined,
+  options: { minBytes?: number; waitForDone?: boolean } = {}
+): ReturnType<typeof consumeStreamWithMetrics> {
+  const { minBytes = 1, waitForDone = minBytes > 1 } = options;
+  const deadline = Date.now() + STREAM_READ_TIMEOUT_MS;
+  let result: Awaited<ReturnType<typeof consumeStreamWithMetrics>> | undefined;
+
+  while (Date.now() < deadline) {
+    result = await consumeStreamWithMetrics(value, startedAt);
+    if (result.totalBytes >= minBytes) {
+      return result;
+    }
+    await waitForReturnedStreamRetry(value, run, deadline, waitForDone);
+    value = await run.returnValue;
+  }
+
+  throw new Error(
+    `Timed out waiting for returned stream bytes: expected at least ${minBytes}, got ${result?.totalBytes ?? 0}`
+  );
+}
+
+async function waitForReturnedStreamRetry(
+  value: unknown,
+  run: Run<unknown>,
+  deadline: number,
+  waitForDone: boolean
+): Promise<void> {
+  const name =
+    value instanceof ReadableStream
+      ? (value as Record<symbol, unknown>)[STREAM_CLIENT_NAME_SYMBOL]
+      : undefined;
+
+  if (typeof name !== 'string') {
+    await sleepUntilNextRetry(deadline);
+    return;
+  }
+
+  const world = await getWorldLazy();
+  while (Date.now() < deadline) {
+    const info = await world.streams.getInfo(run.runId, name).catch(() => null);
+    if (info && (waitForDone ? info.done : info.tailIndex >= 0)) {
+      return;
+    }
+    await sleepUntilNextRetry(deadline);
+  }
+}
+
+function sleepUntilNextRetry(deadline: number): Promise<void> {
+  return new Promise((resolve) =>
+    setTimeout(
+      resolve,
+      Math.max(
+        0,
+        Math.min(STREAM_READ_RETRY_INTERVAL_MS, deadline - Date.now())
+      )
+    )
+  );
+}
+
 describe('Workflow Performance Benchmarks', () => {
   bench(
     'workflow with no steps',
@@ -389,7 +462,7 @@ describe('Workflow Performance Benchmarks', () => {
       const value = await run.returnValue;
       const timings = await getRunTimings(run);
       const { firstByteTimeMs, slurpTimeMs, totalBytes } =
-        await consumeStreamWithMetrics(value, timings.startedAt);
+        await consumeReturnedStreamWithMetrics(value, run, timings.startedAt);
       // Correctness: stream should produce ~5KB (50 chunks * ~100 bytes)
       if (totalBytes === 0) {
         throw new Error(
@@ -505,8 +578,7 @@ describe('Workflow Performance Benchmarks', () => {
       skip: false,
       time: 60000,
       expectedTotalBytes: 1024 * 1024,
-      // Pipeline returns the actual data stream
-      summaryStream: false,
+      summaryStream: true,
     },
     {
       name: 'stream pipeline with 10 transform steps (1MB)',
@@ -515,7 +587,7 @@ describe('Workflow Performance Benchmarks', () => {
       skip: !fullSuite,
       time: 120000,
       expectedTotalBytes: 1024 * 1024,
-      summaryStream: false,
+      summaryStream: true,
     },
     {
       name: '10 parallel streams (1MB each)',
@@ -573,7 +645,15 @@ describe('Workflow Performance Benchmarks', () => {
         const value = await run.returnValue;
         const timings = await getRunTimings(run);
         const { firstByteTimeMs, slurpTimeMs, totalBytes, chunks } =
-          await consumeStreamWithMetrics(value, timings.startedAt);
+          await consumeReturnedStreamWithMetrics(
+            value,
+            run,
+            timings.startedAt,
+            {
+              minBytes: summaryStream ? 1 : expectedTotalBytes,
+              waitForDone: summaryStream === true,
+            }
+          );
 
         if (summaryStream) {
           // Parallel/fan-out workflows return a JSON summary stream;
@@ -595,7 +675,7 @@ describe('Workflow Performance Benchmarks', () => {
             );
           }
         } else {
-          // Pipeline workflows return the actual data stream
+          // Non-summary workflows return the actual data stream.
           if (totalBytes !== expectedTotalBytes) {
             throw new Error(
               `Stream correctness failure: expected ${expectedTotalBytes} bytes but got ${totalBytes}`