[pull] main from vercel:main

.changeset/lazy-terminal-run-data.md

@@ -0,0 +1,5 @@
+---
+"@workflow/world": patch
+---
+
+Accept lazy completed and failed runs whose payload is represented by a remote reference.

.changeset/python-cancellable-steps.md

@@ -0,0 +1,4 @@
+---
+---
+
+Expand Python conformance coverage for cancellable steps.

.changeset/stream-read-key-prefetch.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Prefetch run encryption keys when reading workflow streams.

packages/core/e2e/e2e.test.ts

@@ -149,10 +149,10 @@ const e2e = (fn: string) => {
  * mean testing something else, so a non-JS app skips them instead of carrying
  * them as gaps.
  *
- * A handful of markers are weaker than that: health check, the webhook route, and
- * app-provided API routes are protocol-level and *ought* to travel, but no other
- * SDK serves them yet, so there is nothing to conform to. Those sites say so, and
- * should move back to plain `test` as soon as a second implementation lands.
+ * A handful of markers are weaker than that: the webhook route and app-provided
+ * API routes are protocol-level and *ought* to travel, but no other SDK serves
+ * them yet, so there is nothing to conform to. Those sites say so, and should
+ * move back to plain `test` as soon as a second implementation lands.
  *
  * Every test not marked here is in scope for cross-language conformance, and is
  * gated only by `e2e-conformance.json`. No-op for the JS workbench apps.
@@ -2929,12 +2929,7 @@ describe.concurrent('e2e', () => {
   // For production use on Vercel with Deployment Protection enabled, use the
   // queue-based `healthCheck(world, options)` function instead, which
   // bypasses protection by sending messages through the Queue infrastructure.
-  // JS-only for now, though no longer for want of a second implementation:
-  // vercel-py answers both probes as of vercel-py#292. What it omits is
-  // `workflowCoreVersion`, asserted below, on the grounds that it names a
-  // JavaScript package's version. Moving all three health-check tests out of
-  // js-only together means settling what a non-JS SDK reports there.
-  testJsOnly.skipIf(!isLocalDeployment())(
+  test.skipIf(!isLocalDeployment())(
     'health check endpoint (HTTP) - workflow endpoint responds to __health query parameter',
     { timeout: 30_000 },
     async () => {
@@ -2955,21 +2950,32 @@ describe.concurrent('e2e', () => {
       );
       expect(flowRes.status).toBe(200);
       expect(flowRes.headers.get('Content-Type')).toBe('application/json');
-      const flowBody = await flowRes.json();
+      const { workflowCoreVersion, ...flowBody } = await flowRes.json();
       expect(flowBody).toEqual({
         healthy: true,
         endpoint: '/.well-known/workflow/v1/flow',
         // specVersion comes from the World's declared specVersion (e.g. 3
         // for world-vercel) or falls back to SPEC_VERSION_CURRENT (2).
         specVersion: expect.any(Number),
-        workflowCoreVersion: expect.any(String),
       });
-      expect(flowBody.specVersion).toBeGreaterThanOrEqual(SPEC_VERSION_CURRENT);
+      // A JavaScript app is built from the same `@workflow/core` as this driver,
+      // so advertising an older spec version than the library it ships with is a
+      // regression. A second implementation's spec version is its own: it reports
+      // what it *writes*, so the only portable claim is that it is a real version.
+      if (isJsApp()) {
+        expect(flowBody.specVersion).toBeGreaterThanOrEqual(
+          SPEC_VERSION_CURRENT
+        );
+        // See comments in the next test about workflowCoreVersion
+        expect(typeof workflowCoreVersion).toBe('string');
+      } else {
+        expect(flowBody.specVersion).toBeGreaterThanOrEqual(1);
+      }
       // V2: no separate step endpoint — combined into the flow handler.
     }
   );
 
-  testJsOnly(
+  test(
     'health check (queue-based) - workflow endpoint responds to health check messages',
     { timeout: 60_000 },
     async () => {
@@ -2983,14 +2989,20 @@ describe.concurrent('e2e', () => {
         timeout: 30000,
       });
       expect(workflowResult.healthy).toBe(true);
-      // The deployed app advertises its `@workflow/core` version so
+      // A JavaScript app advertises its `@workflow/core` version so
       // callers can derive capability metadata (see `getRunCapabilities`
       // in `capabilities.ts`).
-      expect(typeof workflowResult.workflowCoreVersion).toBe('string');
+      // An SDK in another language has no such package, and the field is
+      // not advertised; cross-language capability detection should not
+      // be built on top of emulated `@workflow/core` version, thus needs
+      // further design and evolution.
+      if (isJsApp()) {
+        expect(typeof workflowResult.workflowCoreVersion).toBe('string');
+      }
     }
   );
 
-  testJsOnly(
+  test(
     'health check (CLI) - workflow health command reports healthy endpoints',
     { timeout: 60_000 },
     async () => {
@@ -3822,7 +3834,7 @@ describe.concurrent('e2e', () => {
   // AbortController / AbortSignal
   // ==========================================================================
 
-  describeJsOnly('AbortController', () => {
+  describe('AbortController', () => {
     test(
       'abortTimeoutWorkflow: timeout cancels long-running step',
       { timeout: 60_000 },
@@ -4734,7 +4746,13 @@ describe.concurrent('e2e', () => {
         // catch, with a message naming the violated rule and its limit.
         expect(outcomes.reserved).toMatch(/^FatalError: /);
         expect(outcomes.reserved).toContain('reserved prefix');
-        expect(outcomes.reserved).toContain('allowReservedAttributes');
+        // The message names the opt-out parameter, and each SDK names it in its
+        // own casing — `allowReservedAttributes` here, `allow_reserved_attributes`
+        // in Python. Assert that it points at the escape hatch, not how one
+        // language spells it.
+        expect(outcomes.reserved).toMatch(
+          /allow[_]?[rR]eserved[_]?[aA]ttributes/
+        );
         expect(outcomes.emptyKey).toContain('must not be empty');
         expect(outcomes.keyTooLong).toContain(
           'key length 257 exceeds limit 256'
@@ -4747,7 +4765,10 @@ describe.concurrent('e2e', () => {
           'byte length 400 exceeds limit 256'
         );
         expect(outcomes.overCap).toContain('exceed limit 64');
-        expect(outcomes.nonObject).toContain('requires a plain object');
+        // Same idea: "plain object" in JS is "mapping" in Python. What the test
+        // is for is that a non-object argument is rejected by name at the call
+        // site, which either wording satisfies.
+        expect(outcomes.nonObject).toMatch(/requires a (plain object|mapping)/);
 
         // No invalid write reached the run, and the run stayed healthy
         // enough to complete a valid write afterwards.

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

@@ -461,6 +461,164 @@ describe('createReconnectingFramedStream', () => {
     expect(chunks).toEqual([payloadFrame(1), payloadFrame(2), payloadFrame(3)]);
   });
 
+  it('starts key resolution concurrently with the first stream GET', async () => {
+    let resolveStream: (stream: ReadableStream<Uint8Array>) => void;
+    const streamPromise = new Promise<ReadableStream<Uint8Array>>((resolve) => {
+      resolveStream = resolve;
+    });
+    let resolveKey: () => void;
+    const keyPromise = new Promise<void>((resolve) => {
+      resolveKey = resolve;
+    });
+    const get = vi.fn().mockReturnValue(streamPromise);
+    const prefetchKey = vi.fn().mockReturnValue(keyPromise);
+    setWorld({
+      specVersion: SPEC_VERSION_CURRENT,
+      streams: { get },
+    } as unknown as World);
+
+    const read = readAll(
+      createReconnectingFramedStream(RUN_ID, 's', 0, prefetchKey)
+    );
+    await vi.waitFor(() => {
+      expect(get).toHaveBeenCalledOnce();
+      expect(prefetchKey).toHaveBeenCalledOnce();
+    });
+
+    resolveKey?.();
+    resolveStream?.(
+      scriptedStream([
+        { kind: 'value', value: payloadFrame(7) },
+        { kind: 'close' },
+      ])
+    );
+    await expect(read).resolves.toEqual([payloadFrame(7)]);
+  });
+
+  it('finishes key resolution before the first raw frame', async () => {
+    let releaseFrame: () => void;
+    const frameReady = new Promise<void>((resolve) => {
+      releaseFrame = resolve;
+    });
+    const prefetchKey = vi.fn().mockResolvedValue(undefined);
+    const { world } = makeWorldWithScriptedStreams({
+      0: () =>
+        new ReadableStream({
+          async pull(controller) {
+            await frameReady;
+            controller.enqueue(payloadFrame(7));
+            controller.close();
+          },
+        }),
+    });
+    setWorld(world);
+
+    const read = readAll(
+      createReconnectingFramedStream(RUN_ID, 's', 0, prefetchKey)
+    );
+    await vi.waitFor(() => expect(prefetchKey).toHaveBeenCalledOnce());
+    // The resolver has already settled by the time the raw frame is released.
+    await Promise.resolve();
+    releaseFrame?.();
+    await expect(read).resolves.toEqual([payloadFrame(7)]);
+    expect(prefetchKey).toHaveBeenCalledOnce();
+  });
+
+  it('prefetches one key promise across reconnects', async () => {
+    const prefetchKey = vi.fn().mockResolvedValue(undefined);
+    const { world, calls } = makeWorldWithScriptedStreams({
+      0: () =>
+        scriptedStream([
+          { kind: 'value', value: payloadFrame(1) },
+          { kind: 'error', err: new Error('connection reset') },
+        ]),
+      1: () =>
+        scriptedStream([
+          { kind: 'value', value: payloadFrame(2) },
+          { kind: 'close' },
+        ]),
+    });
+    setWorld(world);
+
+    await expect(
+      readAll(createReconnectingFramedStream(RUN_ID, 's', 0, prefetchKey))
+    ).resolves.toEqual([payloadFrame(1), payloadFrame(2)]);
+    expect(calls).toEqual([0, 1]);
+    expect(prefetchKey).toHaveBeenCalledOnce();
+  });
+
+  it('keeps a stream GET failure primary when its speculative key lookup also fails', async () => {
+    const streamError = new Error('stream connection failed');
+    const keyError = new Error('key lookup failed');
+    const unhandled = vi.fn();
+    process.once('unhandledRejection', unhandled);
+    setWorld({
+      specVersion: SPEC_VERSION_CURRENT,
+      streams: { get: vi.fn().mockRejectedValue(streamError) },
+    } as unknown as World);
+
+    await expect(
+      readAll(
+        createReconnectingFramedStream(RUN_ID, 's', -1, () =>
+          Promise.reject(keyError)
+        )
+      )
+    ).rejects.toThrow('stream connection failed');
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(unhandled).not.toHaveBeenCalled();
+  });
+
+  it('observes a rejected speculative key lookup after cancellation', async () => {
+    const keyError = new Error('key lookup failed');
+    const unhandled = vi.fn();
+    process.once('unhandledRejection', unhandled);
+    const stream = createReconnectingFramedStream(RUN_ID, 's', 0, () =>
+      Promise.reject(keyError)
+    );
+    const reader = stream.getReader();
+    const pending = reader.read();
+    await reader.cancel();
+    await expect(pending).resolves.toMatchObject({ done: true });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(unhandled).not.toHaveBeenCalled();
+  });
+
+  it('cancels an acquired underlying reader while a prefetched key is pending', async () => {
+    let cancelCount = 0;
+    let resolveKey: () => void;
+    const prefetchKey = vi.fn().mockReturnValue(
+      new Promise<void>((resolve) => {
+        resolveKey = resolve;
+      })
+    );
+    const source = new ReadableStream<Uint8Array>({
+      pull() {
+        // Keep the first raw read pending until the consumer cancels.
+      },
+      cancel() {
+        cancelCount++;
+      },
+    });
+    const get = vi.fn().mockResolvedValue(source);
+    setWorld({
+      specVersion: SPEC_VERSION_CURRENT,
+      streams: { get },
+    } as unknown as World);
+
+    const reader = createReconnectingFramedStream(
+      RUN_ID,
+      's',
+      0,
+      prefetchKey
+    ).getReader();
+    const pendingRead = reader.read();
+    await vi.waitFor(() => expect(get).toHaveBeenCalledOnce());
+    await reader.cancel();
+    await expect(pendingRead).resolves.toMatchObject({ done: true });
+    expect(cancelCount).toBe(1);
+    resolveKey?.();
+  });
+
   it('threads runId through to streams.get', async () => {
     const getSpy = vi.fn(
       async (_runId: string, _name: string, _startIndex?: number) =>

packages/core/src/runtime/run.ts

@@ -13,7 +13,7 @@ import {
   type PayloadKey,
 } from '../serialization/encryption.js';
 import {
-  getExternalRevivers,
+  getRunReadableStream,
   hydrateRunError,
   hydrateWorkflowReturnValue,
 } from '../serialization.js';
@@ -224,9 +224,9 @@ export class Run<TResult> {
   }
 
   /**
-   * Defer fetching the run and its encryption key until serialized stream data
-   * is actually read. An empty or metadata-only stream must not start an
-   * unobserved run lookup.
+   * Defers fetching the run and its encryption key until a readable is first
+   * consumed. The first pull prefetches it so an encrypted first frame can join
+   * the lookup; this also applies to an empty consumed stream.
    * @internal
    */
   #getEncryptionKeyLazily(): () => Promise<PayloadKey | undefined> {
@@ -370,15 +370,18 @@ export class Run<TResult> {
     'use step';
     const { ops = [], global = globalThis, startIndex, namespace } = options;
     const name = getWorkflowRunStreamId(this.runId, namespace);
-    // The resolver starts only when the deserialize stream sees its first
-    // chunk, so creating or probing an empty stream cannot reject in the
-    // background.
+    // The resolver starts only on the readable's first pull, so construction
+    // is inert. A consumed empty stream still performs the speculative lookup
+    // to keep the first encrypted-frame path concurrent.
     const encryptionKey = this.#getEncryptionKeyLazily();
-    const stream = getExternalRevivers(global, ops, this.runId, encryptionKey)
-      .ReadableStream!({
+    const stream = getRunReadableStream<R>(
+      global,
+      ops,
+      this.runId,
       name,
       startIndex,
-    }) as ReadableStream<R>;
+      encryptionKey
+    );
 
     const worldPromise = this.#lazyWorldPromise;
     const runId = this.runId;