feat(core): resolve run.returnValue via a World long poll instead of a 1s poll

.changeset/run-status-long-poll.md

@@ -0,0 +1,15 @@
+---
+'@workflow/core': minor
+'@workflow/world': minor
+'@workflow/world-local': minor
+'@workflow/world-postgres': minor
+'@workflow/world-vercel': minor
+---
+
+Resolve `await run.returnValue` as soon as a run finishes, via a new optional World long poll.
+
+`Storage['runs']` gains `waitForTerminalStatus(runId, { timeoutMs, signal, resolveData })`: one read the World holds open until the run reaches a terminal status, returning the same entity `runs.get()` returns (a budget that expires returns the latest snapshot, not an error). `Run#pollReturnValue` uses it when the World implements it, so a run that finishes mid-wait is reported immediately instead of at the next poll tick — previously up to a full `WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS` (1s) later.
+
+Implemented by `world-vercel` against workflow-server's new long-pollable `GET /v2/runs/:runId/status` route, by `world-postgres` with `LISTEN`/`NOTIFY` on run-terminal writes, and by `world-local` with an in-process signal over the run files. Every implementation re-reads the run before answering and backstops the wait with a periodic re-read, so a lost notification costs latency rather than correctness.
+
+The method is optional and the fast path is strictly additive: a World that omits it (`world-sim`, third-party adapters) keeps interval-polling `runs.get()` exactly as before, and `world-vercel` falls back to the plain read when the workflow-server it is talking to has no such route. `WORKFLOW_RETURN_VALUE_LONG_POLL=0` restores interval polling everywhere; `WORKFLOW_RETURN_VALUE_WAIT_MS` tunes the per-call budget (default 25s).

docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx

@@ -151,6 +151,40 @@ const run = await world.runs.get(runId); // [!code highlight]
 
 **Returns:** `WorkflowRun` (or `WorkflowRunWithoutData` when `resolveData: 'none'`)
 
+### runs.waitForTerminalStatus()
+
+Optional. Long poll for a run to reach a terminal status (`completed`,
+`failed`, or `cancelled`) instead of re-reading it on an interval. This is what
+`await run.returnValue` uses, so a run's result reaches the awaiting side as
+soon as it finishes rather than at the next poll tick.
+
+```typescript lineNumbers
+const run = await world.runs.waitForTerminalStatus?.(runId, { // [!code highlight]
+  timeoutMs: 25_000, // [!code highlight]
+}); // [!code highlight]
+```
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `runId` | `string` | The workflow run ID |
+| `params.timeoutMs` | `number` | Upper bound on the wait — the call returns earlier, the moment the run is terminal |
+| `params.signal` | `AbortSignal` | Abandons the wait |
+| `params.resolveData` | `'all' \| 'none'` | Whether to include input/output data. Default: `'all'` |
+
+**Returns:** the same `WorkflowRun` as `runs.get()` — terminal if the run
+finished within the budget, otherwise the latest snapshot. An expired budget is
+a normal return, not an error, and a missing run throws
+`WorkflowRunNotFoundError` exactly as `runs.get()` does.
+
+<Callout>
+  Not every backend can hold a read open, so this method is optional and may
+  also return a non-terminal snapshot before `timeoutMs` is up. Callers pace
+  their own retries — `await run.returnValue` keeps consecutive non-terminal
+  observations at least one `WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS` apart — and
+  worlds that omit the method are polled on that interval instead. Set
+  `WORKFLOW_RETURN_VALUE_LONG_POLL=0` to force interval polling everywhere.
+</Callout>
+
 ### runs.list()
 
 ```typescript lineNumbers

docs/content/worlds/v5/building-a-world.mdx

@@ -88,6 +88,9 @@ interface Storage {
   runs: {
     get(id: string, params?: GetWorkflowRunParams): Promise<WorkflowRun>;
     list(params?: ListWorkflowRunsParams): Promise<PaginatedResponse<WorkflowRun>>;
+
+    // Optional: long poll for a terminal status (see below)
+    waitForTerminalStatus?(id: string, params?: WaitForTerminalRunStatusParams): Promise<WorkflowRun>;
   };
 
   steps: {
@@ -129,6 +132,21 @@ Keep the owning Run available for at least as long as its token remains unavaila
 
 **Automatic Hook Cleanup:** When a run ends, remove its live Hooks. Make each token available unless its `tokenRetentionUntil` is still in the future. A `hook_disposed` event always makes the token available immediately.
 
+### Optional: Waiting for a Terminal Run Status
+
+`await run.returnValue` has to find out when a run finished. Without help it re-reads the run every second, so a run that finishes just after a read is reported up to a second late. Implement `runs.waitForTerminalStatus(id, { timeoutMs, signal, resolveData })` and the runtime asks once and is answered the moment the run ends.
+
+The contract is deliberately forgiving, because "wait" means something different in every store:
+
+- Resolve as soon as the run's status is terminal, returning the same entity `get` returns.
+- Resolve no later than roughly `timeoutMs` with the latest snapshot, whatever its status. **A timeout is a normal return, never an error** — a run that is still running is a legitimate answer, and the runtime simply asks again.
+- `timeoutMs` is an upper bound, not a lower one: returning a non-terminal snapshot early is allowed, and the runtime paces its own retries.
+- Fail exactly like `get` — a missing run throws `WorkflowRunNotFoundError`.
+
+How you wait is up to your store. The reference worlds use, respectively, a server-side long poll (`world-vercel` holds `GET /v2/runs/:runId/status` open), `LISTEN`/`NOTIFY` (`world-postgres`), and an in-process emitter over the run files (`world-local`). Whatever the mechanism, treat the notification as a *signal only* and re-read the run before answering, and back the wait with a periodic re-read so a lost notification costs latency rather than hanging until the budget expires.
+
+Omitting the method is a supported choice — a store with no change notification, or a deterministic simulator like `world-sim` where a real wait would stall a virtual clock, simply leaves it off and the runtime keeps interval-polling `get`. Nothing else degrades; there is no capability to declare.
+
 ### Event ID Allocation
 
 <Callout type="warn">

packages/core/src/runtime/run-return-value-long-poll.test.ts

@@ -0,0 +1,293 @@
+import { WorkflowRunCancelledError } from '@workflow/errors';
+import { SPEC_VERSION_CURRENT, type World } from '@workflow/world';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+// Mock version module to avoid missing generated file
+vi.mock('../version.js', () => ({ version: '0.0.0-test' }));
+
+import { dehydrateWorkflowReturnValue } from '../serialization.js';
+import {
+  getReturnValueWaitTimeoutMs,
+  isReturnValueLongPollEnabled,
+  Run,
+} from './run.js';
+import { setWorld } from './world.js';
+
+/**
+ * `await run.returnValue` and the World's optional long poll
+ * (`runs.waitForTerminalStatus`).
+ *
+ * The behavior being pinned down here is the *pacing*, not the values: which
+ * read the runtime issues, and how long it waits between two non-terminal
+ * observations. All of it runs on fake timers so the assertions are about the
+ * scheduling itself rather than wall-clock luck.
+ */
+
+const RUN_ID = 'wrun_01JB0000000000000000000000';
+
+const baseRun = {
+  runId: RUN_ID,
+  workflowName: 'test-workflow',
+  specVersion: 2,
+  input: [],
+  createdAt: new Date(0),
+  updatedAt: new Date(0),
+  startedAt: new Date(0),
+  deploymentId: 'test-deployment',
+};
+
+const runningRun = { ...baseRun, status: 'running' as const };
+const cancelledRun = {
+  ...baseRun,
+  status: 'cancelled' as const,
+  completedAt: new Date(0),
+};
+
+function createWorld(runs: Partial<World['runs']>): World {
+  return {
+    specVersion: SPEC_VERSION_CURRENT,
+    runs: {
+      get: vi.fn().mockResolvedValue(runningRun),
+      ...runs,
+    },
+    events: {
+      list: vi
+        .fn()
+        .mockResolvedValue({ data: [], hasMore: false, cursor: null }),
+      create: vi.fn(),
+    },
+    queue: vi.fn().mockResolvedValue(undefined),
+  } as unknown as World;
+}
+
+/** Resolve after `ms` on the *fake* clock. */
+const sleepFake = (ms: number) =>
+  new Promise<void>((resolve) => setTimeout(resolve, ms));
+
+describe('Run.returnValue long poll', () => {
+  const envNames = [
+    'WORKFLOW_RETURN_VALUE_LONG_POLL',
+    'WORKFLOW_RETURN_VALUE_WAIT_MS',
+    'WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS',
+  ] as const;
+  const originalEnv = new Map(
+    envNames.map((name) => [name, process.env[name]])
+  );
+
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    for (const [name, value] of originalEnv) {
+      if (value === undefined) delete process.env[name];
+      else process.env[name] = value;
+    }
+    setWorld(undefined as unknown as World);
+  });
+
+  it('waits on the World long poll instead of polling runs.get', async () => {
+    const waitForTerminalStatus = vi.fn().mockResolvedValue(cancelledRun);
+    const world = createWorld({ waitForTerminalStatus });
+    setWorld(world);
+
+    await expect(new Run(RUN_ID).returnValue).rejects.toBeInstanceOf(
+      WorkflowRunCancelledError
+    );
+
+    expect(waitForTerminalStatus).toHaveBeenCalledTimes(1);
+    expect(waitForTerminalStatus).toHaveBeenCalledWith(RUN_ID, {
+      timeoutMs: getReturnValueWaitTimeoutMs(),
+    });
+    // The status read went through the long poll — no interval poll happened.
+    expect(world.runs.get).not.toHaveBeenCalled();
+  });
+
+  it('forwards the configured wait budget', async () => {
+    process.env.WORKFLOW_RETURN_VALUE_WAIT_MS = '3000';
+    const waitForTerminalStatus = vi.fn().mockResolvedValue(cancelledRun);
+    setWorld(createWorld({ waitForTerminalStatus }));
+
+    await expect(new Run(RUN_ID).returnValue).rejects.toBeInstanceOf(
+      WorkflowRunCancelledError
+    );
+
+    expect(waitForTerminalStatus).toHaveBeenCalledWith(RUN_ID, {
+      timeoutMs: 3_000,
+    });
+  });
+
+  it('resolves the hydrated return value from a completed long poll', async () => {
+    vi.useRealTimers();
+    const output = await dehydrateWorkflowReturnValue(
+      { ok: true },
+      RUN_ID,
+      undefined
+    );
+    const completedRun = {
+      ...baseRun,
+      status: 'completed' as const,
+      completedAt: new Date(0),
+      output,
+    };
+    const waitForTerminalStatus = vi.fn().mockResolvedValue(completedRun);
+    setWorld(
+      createWorld({
+        waitForTerminalStatus,
+        get: vi.fn().mockResolvedValue(completedRun),
+      })
+    );
+
+    await expect(new Run(RUN_ID).returnValue).resolves.toEqual({ ok: true });
+  });
+
+  it('paces a World whose wait returns a non-terminal run early', async () => {
+    // A World that cannot actually hold the wait open (e.g. world-vercel
+    // talking to a server without the long-poll route) answers immediately
+    // with a non-terminal run. The loop must fall back to interval polling
+    // rather than spinning on it.
+    process.env.WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS = '1000';
+    const waitForTerminalStatus = vi
+      .fn()
+      .mockResolvedValueOnce(runningRun)
+      .mockResolvedValue(cancelledRun);
+    setWorld(createWorld({ waitForTerminalStatus }));
+
+    const pending = new Run(RUN_ID).returnValue;
+    const assertion = expect(pending).rejects.toBeInstanceOf(
+      WorkflowRunCancelledError
+    );
+
+    await vi.advanceTimersByTimeAsync(0);
+    expect(waitForTerminalStatus).toHaveBeenCalledTimes(1);
+
+    // Still inside the interval — no second attempt yet.
+    await vi.advanceTimersByTimeAsync(999);
+    expect(waitForTerminalStatus).toHaveBeenCalledTimes(1);
+
+    await vi.advanceTimersByTimeAsync(1);
+    expect(waitForTerminalStatus).toHaveBeenCalledTimes(2);
+
+    await assertion;
+  });
+
+  it('does not add an interval sleep when the wait already outlasted it', async () => {
+    // The whole point of the long poll: a wait that blocked for longer than
+    // the poll interval re-issues immediately instead of sleeping again.
+    process.env.WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS = '1000';
+    const waitForTerminalStatus = vi
+      .fn()
+      .mockImplementationOnce(async () => {
+        await sleepFake(1_500);
+        return runningRun;
+      })
+      .mockResolvedValue(cancelledRun);
+    setWorld(createWorld({ waitForTerminalStatus }));
+
+    const pending = new Run(RUN_ID).returnValue;
+    const assertion = expect(pending).rejects.toBeInstanceOf(
+      WorkflowRunCancelledError
+    );
+
+    await vi.advanceTimersByTimeAsync(1_500);
+    expect(waitForTerminalStatus).toHaveBeenCalledTimes(2);
+
+    await assertion;
+  });
+
+  it('interval-polls runs.get when the World has no long poll', async () => {
+    process.env.WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS = '1000';
+    const get = vi
+      .fn()
+      .mockResolvedValueOnce(runningRun)
+      .mockResolvedValue(cancelledRun);
+    const world = createWorld({ get });
+    setWorld(world);
+
+    const pending = new Run(RUN_ID).returnValue;
+    const assertion = expect(pending).rejects.toBeInstanceOf(
+      WorkflowRunCancelledError
+    );
+
+    await vi.advanceTimersByTimeAsync(0);
+    expect(get).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(1_000);
+    expect(get).toHaveBeenCalledTimes(2);
+
+    await assertion;
+  });
+
+  it('kill switch restores the fixed-interval poll', async () => {
+    process.env.WORKFLOW_RETURN_VALUE_LONG_POLL = '0';
+    process.env.WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS = '1000';
+    const waitForTerminalStatus = vi.fn().mockResolvedValue(cancelledRun);
+    const get = vi
+      .fn()
+      .mockResolvedValueOnce(runningRun)
+      .mockResolvedValue(cancelledRun);
+    setWorld(createWorld({ get, waitForTerminalStatus }));
+
+    const pending = new Run(RUN_ID).returnValue;
+    const assertion = expect(pending).rejects.toBeInstanceOf(
+      WorkflowRunCancelledError
+    );
+
+    await vi.advanceTimersByTimeAsync(0);
+    await vi.advanceTimersByTimeAsync(1_000);
+
+    await assertion;
+    expect(waitForTerminalStatus).not.toHaveBeenCalled();
+    expect(get).toHaveBeenCalledTimes(2);
+  });
+});
+
+describe('isReturnValueLongPollEnabled', () => {
+  const envName = 'WORKFLOW_RETURN_VALUE_LONG_POLL';
+  const originalValue = process.env[envName];
+
+  afterEach(() => {
+    if (originalValue === undefined) delete process.env[envName];
+    else process.env[envName] = originalValue;
+  });
+
+  it('defaults to on', () => {
+    delete process.env[envName];
+    expect(isReturnValueLongPollEnabled()).toBe(true);
+  });
+
+  it('treats an empty value as unset', () => {
+    process.env[envName] = '';
+    expect(isReturnValueLongPollEnabled()).toBe(true);
+  });
+
+  it.each(['0', 'false', 'FALSE'])('is disabled by %s', (value) => {
+    process.env[envName] = value;
+    expect(isReturnValueLongPollEnabled()).toBe(false);
+  });
+
+  it.each(['1', 'true'])('stays enabled for %s', (value) => {
+    process.env[envName] = value;
+    expect(isReturnValueLongPollEnabled()).toBe(true);
+  });
+});
+
+describe('getReturnValueWaitTimeoutMs', () => {
+  const envName = 'WORKFLOW_RETURN_VALUE_WAIT_MS';
+  const originalValue = process.env[envName];
+
+  afterEach(() => {
+    if (originalValue === undefined) delete process.env[envName];
+    else process.env[envName] = originalValue;
+  });
+
+  it('defaults to a budget under the adapter request timeout', () => {
+    delete process.env[envName];
+    expect(getReturnValueWaitTimeoutMs()).toBe(25_000);
+  });
+
+  it('accepts a runtime override', () => {
+    process.env[envName] = '5000';
+    expect(getReturnValueWaitTimeoutMs()).toBe(5_000);
+  });
+});