feat(streams): add lease fast path prototype

.changeset/bright-dolphins-laugh.md

@@ -0,0 +1,7 @@
+---
+'@workflow/core': patch
+'@workflow/world': patch
+'@workflow/world-vercel': patch
+---
+
+Add an opt-in replay-safe stream lease append prototype.

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

@@ -228,6 +228,11 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL
 
 These variables are primarily for tests, debugging, or unusual deployments.
 
+### `WORKFLOW_STREAM_LEASE_FAST_PATH`
+
+- Default: unset (off)
+- Experimental deploy-only stream-write prototype. Set to `1` only when targeting a workflow-server preview implementing the matching lease protocol. It gives each writable sink a stable writer identity and sequence so lease appends can safely retry 5xx responses; streams that cannot retain a sole-writer lease de-opt to the existing allocator. This does not enable connection persistence or write pipelining.
+
 ### `WORKFLOW_STREAM_FLUSH_INTERVAL_MS`
 
 - Default: `0` (dispatch the first chunk of an idle stream immediately)

docs/content/worlds/v5/vercel.mdx

@@ -196,6 +196,10 @@ Per-request timeout, in milliseconds, for Vercel World HTTP calls to workflow-se
 
 Maximum stream chunks written in one Vercel World request. Larger batches are split across multiple requests. Default: `1000`. Minimum: `1`.
 
+### `WORKFLOW_STREAM_LEASE_FAST_PATH`
+
+Experimental deploy-only stream-write prototype. Set to `1` only with a matching workflow-server preview. It enables replay-safe lease appends and 5xx retry while retaining the legacy allocator as a per-stream fallback. Default: off. It does not enable persistent connections or pipelining.
+
 ### `WORKFLOW_EVENTS_TRANSPORT`
 
 Experimental. Set `WORKFLOW_EVENTS_TRANSPORT=ws` to ship workflow run events to the Vercel World over a WebSocket instead of one HTTP request each. Default: `http`.

packages/core/src/serialization.ts

@@ -149,6 +149,10 @@ export type SerializationFormatType =
  * (e.g., when starting a workflow or handling step return values).
  */
 const defaultUlid = monotonicFactory();
+// Lease epochs fence writers across sink instances. Date.now() alone can repeat
+// for sinks constructed in the same millisecond, so retain a process-local
+// monotonic floor while keeping the token an ordinary number on the wire.
+let latestStreamLeaseEpoch = Date.now();
 
 /**
  * Detect if a readable stream is a byte stream.
@@ -1279,6 +1283,23 @@ export class WorkflowServerWritableStream extends WritableStream<Uint8Array> {
      * early-ack sink.
      */
     let sinkError: unknown;
+    // The lease protocol is deliberately opt-in: worlds without this optional
+    // transport (and every default deployment) retain the existing writes.
+    // Identity and ordering remain sink-local; this is not a connection.
+    const leaseFastPathEnabled =
+      process.env.WORKFLOW_STREAM_LEASE_FAST_PATH === '1';
+    const writerId = defaultUlid();
+    // The server fences this client-supplied token. A new sink always starts a
+    // new writer session, so this process-local monotonic token needs no
+    // lease-acquisition round trip.
+    const epoch = ++latestStreamLeaseEpoch;
+    let nextLeaseSeq = 0;
+    let leaseFastPathActive = true;
+    // A seq-gap names the first writer sequence the server lacks. Because it
+    // can be from an earlier flush, retain the per-sink sequence until this
+    // experimental path de-opts; a gap can then be repaired in FIFO order.
+    // This is intentionally scoped to the flag-gated measurement prototype.
+    const leaseHistory: Uint8Array[] = [];
     // Group-commit window. The env var, when set, overrides the World
     // option; otherwise `world.streamFlushIntervalMs` governs (default 0) —
     // including the very first chunk. When it must come from the world,
@@ -1354,13 +1375,102 @@ export class WorkflowServerWritableStream extends WritableStream<Uint8Array> {
       await ensureRunReady();
       const world = await worldPromise;
       const dispatchAt = Date.now();
-      if (typeof world.streams.writeMulti === 'function' && group.length > 1) {
-        await world.streams.writeMulti(runId, name, group);
-      } else {
-        // Fall back to sequential writes
-        for (const chunk of group) {
-          await world.streams.write(runId, name, chunk);
+      const writeLegacy = async (): Promise<void> => {
+        if (
+          typeof world.streams.writeMulti === 'function' &&
+          group.length > 1
+        ) {
+          await world.streams.writeMulti(runId, name, group);
+        } else {
+          for (const chunk of group) {
+            await world.streams.write(runId, name, chunk);
+          }
         }
+      };
+
+      if (
+        leaseFastPathEnabled &&
+        leaseFastPathActive &&
+        typeof world.streams.writeLease === 'function'
+      ) {
+        const seqStart = nextLeaseSeq;
+        leaseHistory.push(...group);
+        const result = await world.streams.writeLease(runId, name, group, {
+          writerId,
+          epoch,
+          seqStart,
+        });
+        if (result.status === 'need-reserve') {
+          // De-opt is sticky: a multi-writer/expired lease stream remains on
+          // the known-safe allocator for this sink's remaining lifetime. A
+          // paged transport can have already acknowledged a leading prefix.
+          leaseFastPathActive = false;
+          const unacknowledged = group.slice(result.acknowledged ?? 0);
+          if (unacknowledged.length === group.length) {
+            await writeLegacy();
+          } else if (
+            typeof world.streams.writeMulti === 'function' &&
+            unacknowledged.length > 1
+          ) {
+            await world.streams.writeMulti(runId, name, unacknowledged);
+          } else {
+            for (const chunk of unacknowledged) {
+              await world.streams.write(runId, name, chunk);
+            }
+          }
+        } else if (result.status === 'seq-gap') {
+          // `expected` is the server's next sequence, so it precedes this
+          // request's seqStart. Replay every retained missing chunk in FIFO
+          // order, including this group, before accepting later groups.
+          if (result.expected === undefined) {
+            throw new Error('Stream lease seq-gap response omitted expected');
+          }
+          // `writeLease` can page one core group. In that case a later page
+          // reports its own expected sequence, which may lie inside the group
+          // even though no later core group has begun.
+          if (
+            result.expected < 0 ||
+            result.expected >= seqStart + group.length
+          ) {
+            throw new Error(
+              `Stream lease seq-gap expected ${result.expected} outside retained history`
+            );
+          }
+          const recovery = leaseHistory.slice(result.expected);
+          const recovered = await world.streams.writeLease(
+            runId,
+            name,
+            recovery,
+            {
+              writerId,
+              epoch,
+              seqStart: result.expected,
+            }
+          );
+          if (recovered.status === 'need-reserve') {
+            // A paged recovery may have acknowledged a leading prefix before
+            // it lost the lease. Resume legacy allocation from the remaining
+            // server-reported gap, never re-writing that acknowledged range.
+            leaseFastPathActive = false;
+            const unacknowledged = recovery.slice(recovered.acknowledged ?? 0);
+            if (
+              typeof world.streams.writeMulti === 'function' &&
+              unacknowledged.length > 1
+            ) {
+              await world.streams.writeMulti(runId, name, unacknowledged);
+            } else {
+              for (const chunk of unacknowledged) {
+                await world.streams.write(runId, name, chunk);
+              }
+            }
+          } else if (recovered.status === 'seq-gap') {
+            throw new Error('Stream lease seq-gap recovery did not advance');
+          }
+        }
+        // `ok` and `replay` both durably account for this entire group.
+        nextLeaseSeq += group.length;
+      } else {
+        await writeLegacy();
       }
       if (groupT0 !== undefined) {
         recordStreamWriteFlush(

packages/core/src/writable-stream.test.ts

@@ -33,6 +33,7 @@ describe('WorkflowServerWritableStream', () => {
   let mockStreams: {
     write: ReturnType<typeof vi.fn>;
     writeMulti: ReturnType<typeof vi.fn>;
+    writeLease: ReturnType<typeof vi.fn>;
     close: ReturnType<typeof vi.fn>;
   };
   let mockWorld: {
@@ -45,6 +46,11 @@ describe('WorkflowServerWritableStream', () => {
     mockStreams = {
       write: vi.fn().mockResolvedValue(undefined),
       writeMulti: vi.fn().mockResolvedValue(undefined),
+      writeLease: vi.fn().mockResolvedValue({
+        base: 0,
+        committed: 0,
+        status: 'ok',
+      }),
       close: vi.fn().mockResolvedValue(undefined),
     };
 
@@ -54,6 +60,7 @@ describe('WorkflowServerWritableStream', () => {
   });
 
   afterEach(() => {
+    delete process.env.WORKFLOW_STREAM_LEASE_FAST_PATH;
     setWorld(undefined);
     vi.clearAllMocks();
   });
@@ -78,6 +85,113 @@ describe('WorkflowServerWritableStream', () => {
     });
   });
 
+  describe('stream lease fast path', () => {
+    it('is off by default and leaves legacy writes unchanged', async () => {
+      const stream = new WorkflowServerWritableStream('run-123', 'test-stream');
+      const writer = stream.getWriter();
+      await writer.write(new Uint8Array([1]));
+      await writer.close();
+
+      expect(mockStreams.writeLease).not.toHaveBeenCalled();
+      expect(mockStreams.write).toHaveBeenCalledTimes(1);
+    });
+
+    it('keeps writer identity and increments sequence across flushes', async () => {
+      process.env.WORKFLOW_STREAM_LEASE_FAST_PATH = '1';
+      const stream = new WorkflowServerWritableStream('run-123', 'test-stream');
+      const writer = stream.getWriter();
+      await writer.write(new Uint8Array([1]));
+      await writer.write(new Uint8Array([2]));
+      await writer.close();
+
+      expect(mockStreams.writeLease).toHaveBeenCalledTimes(2);
+      const first = mockStreams.writeLease.mock.calls[0][3];
+      const second = mockStreams.writeLease.mock.calls[1][3];
+      expect(first.writerId).toBe(second.writerId);
+      expect(first.epoch).toBe(second.epoch);
+      expect([first.seqStart, second.seqStart]).toEqual([0, 1]);
+    });
+
+    it('treats replay as success and permanently de-opts after need-reserve', async () => {
+      process.env.WORKFLOW_STREAM_LEASE_FAST_PATH = '1';
+      mockStreams.writeLease
+        .mockResolvedValueOnce({ base: 0, committed: 0, status: 'replay' })
+        .mockResolvedValueOnce({
+          base: 1,
+          committed: 1,
+          status: 'need-reserve',
+        });
+      const stream = new WorkflowServerWritableStream('run-123', 'test-stream');
+      const writer = stream.getWriter();
+      await writer.write(new Uint8Array([1]));
+      await writer.write(new Uint8Array([2]));
+      await writer.write(new Uint8Array([3]));
+      await writer.close();
+
+      expect(mockStreams.writeLease).toHaveBeenCalledTimes(2);
+      const legacyChunks = [
+        ...mockStreams.write.mock.calls.map((call) => call[2]),
+        ...mockStreams.writeMulti.mock.calls.flatMap((call) => call[2]),
+      ];
+      expect(legacyChunks.map((chunk) => chunk[0])).toEqual([2, 3]);
+    });
+
+    it('repairs a seq-gap from retained history in FIFO order', async () => {
+      process.env.WORKFLOW_STREAM_LEASE_FAST_PATH = '1';
+      mockStreams.writeLease
+        .mockResolvedValueOnce({ base: 0, committed: 0, status: 'ok' })
+        .mockResolvedValueOnce({
+          base: 0,
+          committed: 0,
+          status: 'seq-gap',
+          expected: 0,
+        })
+        .mockResolvedValueOnce({ base: 0, committed: 0, status: 'ok' });
+      const stream = new WorkflowServerWritableStream('run-123', 'test-stream');
+      const writer = stream.getWriter();
+      await writer.write(new Uint8Array([1]));
+      await writer.write(new Uint8Array([2]));
+      await writer.close();
+
+      expect(mockStreams.writeLease).toHaveBeenCalledTimes(3);
+      expect(
+        mockStreams.writeLease.mock.calls.map((call) => call[3].seqStart)
+      ).toEqual([0, 1, 0]);
+      expect(mockStreams.writeLease.mock.calls[2][2]).toEqual([
+        new Uint8Array([1]),
+        new Uint8Array([2]),
+      ]);
+    });
+
+    it('falls back from the entire missing suffix after seq-gap recovery loses its lease', async () => {
+      process.env.WORKFLOW_STREAM_LEASE_FAST_PATH = '1';
+      mockStreams.writeLease
+        .mockResolvedValueOnce({ base: 0, committed: 0, status: 'ok' })
+        .mockResolvedValueOnce({
+          base: 0,
+          committed: 0,
+          status: 'seq-gap',
+          expected: 0,
+        })
+        .mockResolvedValueOnce({
+          base: 0,
+          committed: 0,
+          status: 'need-reserve',
+        });
+      const stream = new WorkflowServerWritableStream('run-123', 'test-stream');
+      const writer = stream.getWriter();
+      await writer.write(new Uint8Array([1]));
+      await writer.write(new Uint8Array([2]));
+      await writer.close();
+
+      expect(mockStreams.writeMulti).toHaveBeenCalledTimes(1);
+      expect(mockStreams.writeMulti.mock.calls[0][2]).toEqual([
+        new Uint8Array([1]),
+        new Uint8Array([2]),
+      ]);
+    });
+  });
+
   describe('group-commit write behavior', () => {
     it('write() resolves on buffer entry; the leading chunk dispatches immediately (no window tax)', async () => {
       const stream = new WorkflowServerWritableStream('run-123', 'test-stream');

packages/world-vercel/src/http-client.test.ts

@@ -19,6 +19,7 @@ import {
   isRecyclableTransportError,
   STREAM_AGENT_OPTIONS,
   STREAM_CLOSE_RETRY_OPTIONS,
+  STREAM_LEASE_RETRY_OPTIONS,
   STREAM_RETRY_OPTIONS,
 } from './http-client.js';
 
@@ -78,6 +79,13 @@ describe('getStreamDispatcher', () => {
   // unsafe close shapes awaiting in-flight backups) surface as retriable
   // 503s with the stream left durably closing. Without 5xx here, that 503
   // rejects writer.close() and the stream stays fenced until run expiry.
+  it('retries idempotent lease appends on 5xx', () => {
+    expect(STREAM_LEASE_RETRY_OPTIONS.methods).toEqual(['PUT']);
+    for (const code of [429, 500, 502, 503, 504]) {
+      expect(STREAM_LEASE_RETRY_OPTIONS.statusCodes).toContain(code);
+    }
+  });
+
   it('retries stream close on 5xx (idempotent, and the close barrier depends on it)', () => {
     expect(STREAM_CLOSE_RETRY_OPTIONS.methods).toEqual(['PUT']);
     for (const code of [429, 500, 502, 503, 504]) {

packages/world-vercel/src/http-client.ts

@@ -3,6 +3,7 @@ import type { APIConfig } from './utils.js';
 
 let _dispatcher: RetryAgent | undefined;
 let _streamDispatcher: RetryAgent | undefined;
+let _streamLeaseDispatcher: RetryAgent | undefined;
 let _streamCloseDispatcher: RetryAgent | undefined;
 
 /**
@@ -218,6 +219,16 @@ export const STREAM_RETRY_OPTIONS: RetryHandler.RetryOptions = {
   statusCodes: [429],
 };
 
+/**
+ * Lease appends carry a writer sequence and are replay-safe, so unlike legacy
+ * stream appends they can retry a 5xx response without duplicating chunks.
+ */
+export const STREAM_LEASE_RETRY_OPTIONS: RetryHandler.RetryOptions = {
+  retryAfter: true,
+  methods: ['PUT'],
+  statusCodes: [429, 500, 502, 503, 504],
+};
+
 /**
  * Retry options for stream CLOSE (the `X-Stream-Done` PUT). Unlike chunk
  * appends, close is idempotent on the server: a duplicate close of a
@@ -556,6 +567,10 @@ export function getStreamDispatcher(config?: APIConfig): unknown {
  * shared close agent whose retry policy includes 5xx — close is idempotent
  * (see STREAM_CLOSE_RETRY_OPTIONS), unlike chunk appends.
  */
+export function getStreamLeaseDispatcher(config?: APIConfig): unknown {
+  return config?.dispatcher ?? getDefaultStreamLeaseDispatcher();
+}
+
 export function getStreamCloseDispatcher(config?: APIConfig): unknown {
   return config?.dispatcher ?? getDefaultStreamCloseDispatcher();
 }
@@ -676,6 +691,11 @@ function getDefaultStreamDispatcher(): RetryAgent {
 }
 
 /** Shared agent for the idempotent stream close (5xx retriable). */
+function getDefaultStreamLeaseDispatcher(): RetryAgent {
+  _streamLeaseDispatcher ??= createStreamDispatcher(STREAM_LEASE_RETRY_OPTIONS);
+  return _streamLeaseDispatcher;
+}
+
 function getDefaultStreamCloseDispatcher(): RetryAgent {
   _streamCloseDispatcher ??= createStreamDispatcher(STREAM_CLOSE_RETRY_OPTIONS);
   return _streamCloseDispatcher;