[pull] main from vercel:main

.changeset/calm-streams-stop.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Stop framed stream reconnects after the consumer cancels, including while completion checks or reconnect acquisition are still pending.

.changeset/retain-vms-across-waits.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Retain a live workflow VM across inline progress when sleeps are open or created at the same suspension boundary.

docs/content/docs/v5/configuration/runtime-tuning.mdx

@@ -185,10 +185,10 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL
 ### `WORKFLOW_RETAINED_VM`
 
 - Default: enabled
-- Keeps the suspended workflow VM alive across inline steps within one invocation, so each iteration of the inline loop appends only the newly written events instead of replaying the whole event log in a fresh VM.
-- A step-driven suspension can keep the VM retained even when hooks are open or created at the same boundary. Hook-only suspensions park the invocation. Suspensions involving waits or attributes, runs with an open wait, and any replay divergence fall back to a full replay.
+- Node.js VM engine only. Keeps the suspended workflow VM alive across inline steps within one invocation, so each iteration of the inline loop appends only the newly written events instead of replaying the whole event log in a fresh VM. QuickJS manages its own retained inline loop independently of this setting.
+- A step- or attribute-driven suspension can keep the VM retained even when hooks or waits are open or created at the same boundary. Hook- or wait-only suspensions park the invocation because nothing in the current delivery can advance them. Any replay divergence falls back to a full replay.
 - Step inputs made of plain data (objects, arrays, primitives) and standard built-ins (`Map`, `Set`, `Date`, `RegExp`, typed arrays, `ArrayBuffer`, `URL`, `Headers`) keep the VM retained. Patching or polyfilling built-in prototypes doesn't change that because serialization never calls them. A boundary falls back to a full replay only when serializing its arguments runs code the workflow controls, such as a getter, a proxy, or a custom class serializer, or computes an `Error`'s stack trace.
-- Set `0` or `false` to replay from scratch in a fresh VM on every iteration.
+- Set `0` or `false` to replay the Node.js workflow from scratch in a fresh VM on every iteration.
 
 ### `WORKFLOW_INLINE_OWNERSHIP`
 

docs/content/docs/v5/whats-new.mdx

@@ -33,7 +33,7 @@ The largest change in v5 has no API surface: the runtime does far less work per
 
 **The runtime avoids waiting on the persistence layer where it can determine that is safe for your workload.** The runtime skips many API calls when they aren't needed, such as requesting the event log on a run's first invocation. Step creation is folded into step execution rather than being its own round trip. The inline loop consumes the event-log delta from the previous step's write instead of re-listing events. Each optimization is gated on specific runtime conditions and can be turned off individually. See [Runtime tuning](/docs/configuration/runtime-tuning).
 
-**The workflow VM is kept alive across inline steps.** Within one invocation, a step-driven suspension keeps the live VM and hydrated state, including when hooks are open or created at the same boundary, so the next iteration appends only the newly written events instead of rebuilding the sandbox and replaying the whole log. Step inputs made of plain data or standard built-ins keep this fast path; see [`WORKFLOW_RETAINED_VM`](/docs/configuration/runtime-tuning#workflow_retained_vm).
+**The workflow VM is kept alive across inline steps.** Within one invocation, a step- or attribute-driven suspension keeps the live VM and hydrated state, including when hooks or waits are open or created at the same boundary, so the next iteration appends only the newly written events instead of rebuilding the sandbox and replaying the whole log. Step inputs made of plain data or standard built-ins keep this fast path; see [`WORKFLOW_RETAINED_VM`](/docs/configuration/runtime-tuning#workflow_retained_vm).
 
 **Resuming a hook takes one round trip instead of two.** `resumeHook()` writes the `hook_received` event and dispatches the queue message concurrently, with a `(runId, resumeId)` dedup constraint keeping the two writers converging on exactly one event. See [Resilient hook resumption](/docs/changelog/resilient-resume).
 

packages/core/src/private.ts

@@ -138,12 +138,10 @@ export interface WorkflowOrchestratorContext {
   globalThis: typeof globalThis;
   /**
    * Increments when a suspension is accepted and on every retained-session
-   * resume. Step, hook, and attribute suspension signals capture it when
+   * resume. Step, hook, wait, and attribute suspension signals capture it when
    * scheduled and no-op if it moved, which drops same-boundary sibling signals
    * and timers queued at boundary N that would fire after the session resumed
-   * into boundary N+1. Sleep signals are intentionally unguarded: their
-   * presence makes the boundary unretainable, so a late signal correctly
-   * demotes the session (workflow.ts `onWorkflowError`).
+   * into boundary N+1.
    */
   suspensionGeneration: number;
   eventsConsumer: EventsConsumer;

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

@@ -329,6 +329,118 @@ describe('createReconnectingFramedStream', () => {
     expect(cancelSpy).toHaveBeenCalled();
   });
 
+  it('does not reconnect when canceled during completion verification', async () => {
+    const infoStarted = Promise.withResolvers<void>();
+    const infoGate = Promise.withResolvers<void>();
+    const { world, calls } = makeWorldWithScriptedStreams(
+      {
+        0: () =>
+          scriptedStream([
+            { kind: 'value', value: payloadFrame(1) },
+            { kind: 'close' },
+          ]),
+      },
+      async () => {
+        infoStarted.resolve();
+        await infoGate.promise;
+        return { tailIndex: 0, done: false };
+      }
+    );
+    setWorld(world);
+
+    const reader = createReconnectingFramedStream(RUN_ID, 's', 0).getReader();
+    expect((await reader.read()).value).toEqual(payloadFrame(1));
+    const pendingRead = reader.read();
+    await infoStarted.promise;
+
+    await reader.cancel('client abort');
+    infoGate.resolve();
+    await pendingRead;
+
+    expect(calls).toEqual([0]);
+  });
+
+  it('cancels a reconnect source acquired after the consumer cancels', async () => {
+    const reconnectStarted = Promise.withResolvers<void>();
+    const reconnectGate = Promise.withResolvers<void>();
+    const cancelSpy = vi.fn();
+    let connections = 0;
+    const world = {
+      specVersion: SPEC_VERSION_CURRENT,
+      streams: {
+        get: vi.fn(async () => {
+          connections++;
+          if (connections === 1) {
+            return scriptedStream([
+              { kind: 'value', value: payloadFrame(1) },
+              { kind: 'error', err: new Error('connection dropped') },
+            ]);
+          }
+          reconnectStarted.resolve();
+          await reconnectGate.promise;
+          return new ReadableStream<Uint8Array>({
+            async pull() {
+              await new Promise(() => {});
+            },
+            cancel(reason) {
+              cancelSpy(reason);
+            },
+          });
+        }),
+      },
+    } as unknown as World;
+    setWorld(world);
+
+    const reader = createReconnectingFramedStream(RUN_ID, 's', 0).getReader();
+    expect((await reader.read()).value).toEqual(payloadFrame(1));
+    const pendingRead = reader.read();
+    await reconnectStarted.promise;
+
+    await reader.cancel('client abort');
+    reconnectGate.resolve();
+    await pendingRead;
+
+    await vi.waitFor(() => {
+      expect(cancelSpy).toHaveBeenCalledWith('client abort');
+    });
+    expect(world.streams.get).toHaveBeenCalledTimes(2);
+  });
+
+  it('stops retrying when a pending reconnect rejects after cancellation', async () => {
+    const reconnectStarted = Promise.withResolvers<void>();
+    const reconnectGate = Promise.withResolvers<void>();
+    let connections = 0;
+    const world = {
+      specVersion: SPEC_VERSION_CURRENT,
+      streams: {
+        get: vi.fn(async () => {
+          connections++;
+          if (connections === 1) {
+            return scriptedStream([
+              { kind: 'value', value: payloadFrame(1) },
+              { kind: 'error', err: new Error('connection dropped') },
+            ]);
+          }
+          reconnectStarted.resolve();
+          await reconnectGate.promise;
+          throw new Error('reconnect failed');
+        }),
+      },
+    } as unknown as World;
+    setWorld(world);
+
+    const reader = createReconnectingFramedStream(RUN_ID, 's', 0).getReader();
+    expect((await reader.read()).value).toEqual(payloadFrame(1));
+    const pendingRead = reader.read();
+    await reconnectStarted.promise;
+
+    await reader.cancel('client abort');
+    reconnectGate.resolve();
+    await pendingRead;
+
+    expect(world.streams.get).toHaveBeenCalledTimes(2);
+  });
+
   it('emits every complete frame packed into a single read', async () => {
     // One transport read carrying three back-to-back frames must surface as
     // three separate downstream chunks — exercises the inner drain loop.