[core] Ack'd write-channel stream writes with framed-v2 writer markers

.changeset/framed-v2-stream-markers.md

@@ -0,0 +1,6 @@
+---
+"@workflow/core": minor
+"@workflow/world": minor
+---
+
+Stream chunks now carry per-writer framed-v2 markers so readers deduplicate retransmitted chunks, and stream writes can grant the backend retransmit-safe delivery via `writeMulti` options.

.changeset/ws-stream-write-channel.md

@@ -0,0 +1,5 @@
+---
+"@workflow/world-vercel": minor
+---
+
+Stream writes on Vercel now flow over a long-lived acknowledged WebSocket write channel instead of one request per batch, with per-chunk durability confirmation and exactly-once delivery across reconnects.

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

@@ -134,6 +134,66 @@ These variables are primarily for tests, debugging, or unusual deployments.
 - Group-commit window for the *leading* chunk of an idle stream. `0` sends it at once; a positive value holds it up to that many milliseconds to collect a group — an opt-in trade of first-chunk latency for larger batches, useful for slow-but-steady producers. Chunks arriving while a request is already in flight always coalesce into the next group regardless of this setting.
 - Also available as `streamFlushIntervalMs` on Worlds that expose it (the env var, when set, takes precedence over the World option).
 
+### `WORKFLOW_STREAM_WRITE_TRANSPORT`
+
+- Default: `auto`
+- Transport for acknowledged stream writes on the Vercel World: `auto` uses the WebSocket write channel and falls back to batched PUT after a fatal channel failure; `put` forces batched PUT (kill switch); `websocket` forces the channel with no fallback.
+
+### `WORKFLOW_STREAM_WRITE_RECYCLE_MS`
+
+- Default: `110000`
+- How long one stream write channel connection is used before it is proactively rotated.
+- Minimum: `250`.
+
+### `WORKFLOW_STREAM_WRITE_MAX_IN_FLIGHT_FRAMES`
+
+- Default: `64`
+- Frame-count cap on a stream write channel's unacknowledged window (producer backpressure).
+
+### `WORKFLOW_STREAM_WRITE_MAX_IN_FLIGHT_BYTES`
+
+- Default: `4194304` (4MB)
+- Byte cap on a stream write channel's unacknowledged window (producer backpressure).
+
+### `WORKFLOW_STREAM_WRITE_MAX_CONSECUTIVE_RECONNECTS`
+
+- Default: `5`
+- Consecutive failed connections (without an acknowledgement in between) before a stream write channel gives up.
+
+### `WORKFLOW_STREAM_WRITE_MAX_TOTAL_RECONNECTS`
+
+- Default: `64`
+- Lifetime unclean-reconnect budget for one stream write channel writer.
+
+### `WORKFLOW_STREAM_WRITE_CONNECT_TIMEOUT_MS`
+
+- Default: `10000`
+- Deadline for the stream write channel's WebSocket upgrade handshake.
+
+### `WORKFLOW_STREAM_WRITE_ACK_DEADLINE_MS`
+
+- Default: `15000`
+- How long a sent chunk may sit unacknowledged before the connection is presumed half-open and replaced.
+- Minimum: `1000`.
+
+### `WORKFLOW_STREAM_WRITE_IDLE_CLOSE_MS`
+
+- Default: `10000`
+- How long a fully-drained stream write channel is kept open before it is released; the next write reconnects.
+- Minimum: `250`.
+
+### `WORKFLOW_STREAM_WRITE_FIN_ACK_TIMEOUT_MS`
+
+- Default: `10000`
+- How long stream completion waits for the server's finalization confirmation (FIN_ACK) before proceeding.
+- Minimum: `100`.
+
+### `WORKFLOW_STREAM_WRITE_FALLBACK_COOLDOWN_MS`
+
+- Default: `60000`
+- How long acknowledged stream writes stay on the batched PUT fallback after a fatal write-channel failure.
+- Minimum: `1000`.
+
 ### `WORKFLOW_STREAM_MAX_INFLIGHT_CHUNKS`
 
 - Default: `1000`

packages/core/src/byte-stream-framing.test.ts

@@ -233,6 +233,139 @@ describe('getByteUnframingStream', () => {
   });
 });
 
+// ----------------------------------------------------------------------------
+// framed-v2: per-writer markers for the single-request streaming writer.
+// ----------------------------------------------------------------------------
+
+describe('framed-v2 byte framing', () => {
+  const WRITER_ID_SIZE = 8;
+  const FRAME_MARKER_SIZE = WRITER_ID_SIZE + 8;
+
+  function writerId(seed: number): Uint8Array {
+    const id = new Uint8Array(WRITER_ID_SIZE);
+    id[0] = seed;
+    return id;
+  }
+
+  it('round-trips payloads and strips the marker', async () => {
+    const id = writerId(0xab);
+    const chunks = [
+      new Uint8Array([1, 2, 3]),
+      new Uint8Array([4, 5]),
+      new Uint8Array([6]),
+    ];
+    const result = await readAll(
+      readableFromChunks(chunks)
+        .pipeThrough(getByteFramingStream(id))
+        .pipeThrough(getByteUnframingStream('framed-v2'))
+    );
+    expect(result).toEqual(chunks);
+  });
+
+  it('frames carry a parseable marker with a monotonic per-writer seq', async () => {
+    const id = writerId(0x07);
+    const frames = await readAll(
+      readableFromChunks([
+        new Uint8Array([10]),
+        new Uint8Array([20]),
+        new Uint8Array([30]),
+      ]).pipeThrough(getByteFramingStream(id))
+    );
+    frames.forEach((frame, i) => {
+      const view = new DataView(
+        frame.buffer,
+        frame.byteOffset,
+        frame.byteLength
+      );
+      const frameLength = view.getUint32(0, false);
+      expect(frameLength).toBe(FRAME_MARKER_SIZE + 1);
+      // writerId occupies the bytes right after the 4-byte length prefix.
+      expect(frame[FRAME_HEADER_SIZE]).toBe(0x07);
+      // seq is the 8-byte big-endian uint64 after the writerId.
+      const seq = view.getBigUint64(FRAME_HEADER_SIZE + WRITER_ID_SIZE, false);
+      expect(seq).toBe(BigInt(i));
+    });
+  });
+
+  it('dedupes a replayed frame (same writerId + seq) on the read side', async () => {
+    // One writer frames chunks 0..3 in order (seq 0..3). Recovery re-sends
+    // already-persisted frames 1 and 2 verbatim before the new frame 3, so the
+    // stored tail the reader sees is: 0,1,2, 1,2, 3.
+    const id = writerId(0x42);
+    const [f0, f1, f2, f3] = await readAll(
+      readableFromChunks([
+        new Uint8Array([0]),
+        new Uint8Array([1]),
+        new Uint8Array([2]),
+        new Uint8Array([3]),
+      ]).pipeThrough(getByteFramingStream(id))
+    );
+
+    const delivered = await readAll(
+      readableFromChunks([f0, f1, f2, f1, f2, f3]).pipeThrough(
+        getByteUnframingStream('framed-v2')
+      )
+    );
+
+    // Replays (seq 1, 2) are dropped; 0,1,2,3 each delivered exactly once.
+    expect(delivered).toEqual([
+      new Uint8Array([0]),
+      new Uint8Array([1]),
+      new Uint8Array([2]),
+      new Uint8Array([3]),
+    ]);
+  });
+
+  it('keeps interleaved frames from two writers apart (per-writer dedupe)', async () => {
+    // Concurrent writers on one stream are documented behavior — and a
+    // crashed invocation's writer (A) can interleave with its retry's
+    // writer (B). Seq spaces are per-writer: B's seq 0/1 must not collide
+    // with A's, and a replay of A's frame after B started is still deduped.
+    const idA = writerId(0xa1);
+    const idB = writerId(0xb2);
+    const [a0, a1] = await readAll(
+      readableFromChunks([
+        new Uint8Array([10]),
+        new Uint8Array([11]),
+      ]).pipeThrough(getByteFramingStream(idA))
+    );
+    const [b0, b1] = await readAll(
+      readableFromChunks([
+        new Uint8Array([20]),
+        new Uint8Array([21]),
+      ]).pipeThrough(getByteFramingStream(idB))
+    );
+
+    const delivered = await readAll(
+      readableFromChunks([a0, b0, a1, a1, b1]).pipeThrough(
+        getByteUnframingStream('framed-v2')
+      )
+    );
+
+    // Every unique frame from both writers delivered exactly once, in
+    // arrival order; only A's replayed frame is dropped.
+    expect(delivered).toEqual([
+      new Uint8Array([10]),
+      new Uint8Array([20]),
+      new Uint8Array([11]),
+      new Uint8Array([21]),
+    ]);
+  });
+
+  it('errors on a framed-v2 frame too small to hold a marker', async () => {
+    // A framed-v1-shaped frame (no marker) read as framed-v2 must be rejected
+    // rather than silently mis-stripping payload bytes as a marker.
+    const tooSmall = concat(header(2), new Uint8Array([0xaa, 0xbb]));
+    await expect(
+      readAll(
+        readableFromChunks([tooSmall]).pipeThrough(
+          getByteUnframingStream('framed-v2')
+        )
+      )
+    ).rejects.toThrow(/smaller than the .*writer marker/);
+  });
+});
+
 // ----------------------------------------------------------------------------
 // End-to-end: dehydrate + hydrate carries the framing decision through the
 // stream ref, and round-trips byte data correctly in both modes.
@@ -449,6 +582,68 @@ describe('byte-stream framing end-to-end through dehydrate/hydrate', () => {
     expect(userBytes).toEqual(expected);
   });
 
+  it('round-trips a framed-v2 byte stream: frames carry per-writer markers that the unframer strips', async () => {
+    setWorld(makeMockWorld());
+    const { readFrameMarker } = await import('./serialization/frame-marker.js');
+
+    const original = [new Uint8Array([1, 2, 3]), new Uint8Array([4, 5])];
+    const expected = original.map((u) => new Uint8Array(u));
+    let i = 0;
+    const stream = new ReadableStream<Uint8Array>({
+      type: 'bytes',
+      pull(c) {
+        if (i < original.length) {
+          c.enqueue(original[i++]);
+        } else {
+          c.close();
+        }
+      },
+    });
+
+    const ops: Promise<void>[] = [];
+    const dehydrated = await dehydrateStepReturnValue(
+      stream,
+      'wrun_test',
+      undefined,
+      ops,
+      globalThis,
+      false,
+      true,
+      false,
+      undefined,
+      // framedStreamMarkers: the target run can decode framed-v2
+      true
+    );
+    await Promise.all(ops);
+
+    // The ref records framed-v2, so the consumer picks the marker-stripping
+    // unframer.
+    const text = new TextDecoder().decode(dehydrated as Uint8Array);
+    expect(text).toContain('framed-v2');
+
+    const name = extractStreamName(dehydrated as Uint8Array);
+    const world = await (await import('./runtime/world.js')).getWorld();
+
+    // Every wire frame carries the same writerId with increasing seq.
+    const wire = await readBytes(await world.streams.get('wrun_test', name));
+    const markers = wire.map((f) =>
+      readFrameMarker(f.subarray(FRAME_HEADER_SIZE))
+    );
+    expect(markers[0].writerId).toEqual(markers[1].writerId);
+    expect(markers.map((m) => m.seq)).toEqual([0n, 1n]);
+
+    // The framed-v2 unframer strips the markers — and drops a frame the
+    // write transport resent (duplicate writerId+seq) instead of delivering
+    // it twice.
+    const replayed = [...wire, wire[wire.length - 1]];
+    const got = await readBytes(
+      readableFromChunks(replayed).pipeThrough(
+        getByteUnframingStream('framed-v2')
+      )
+    );
+    expect(got).toEqual(expected);
+  });
+
   it('round-trips a raw byte stream: producer writes raw, consumer reads raw', async () => {
     setWorld(makeMockWorld());
 

packages/core/src/capabilities.test.ts

@@ -1,5 +1,8 @@
 import { describe, expect, it } from 'vitest';
-import { getRunCapabilities } from './capabilities.js';
+import {
+  framedStreamMarkersEnabled,
+  getRunCapabilities,
+} from './capabilities.js';
 import { SerializationFormat } from './serialization.js';
 
 describe('getRunCapabilities', () => {
@@ -103,4 +106,48 @@ describe('getRunCapabilities', () => {
       expect(getRunCapabilities(version).framedByteStreams).toBe(true);
     });
   });
+
+  describe('framedStreamMarkers (framed-v2 + streaming writes)', () => {
+    it('is false when version is undefined', () => {
+      expect(getRunCapabilities(undefined).framedStreamMarkers).toBe(false);
+    });
+
+    it.each([
+      // framed-v2 is newer than framed-v1 (beta.15), so anything at or below
+      // the framed-v1 cutoff must read as unsupported.
+      '4.2.0',
+      '5.0.0-beta.15',
+      '5.0.0-beta.25',
+    ])('is false for pre-framed-v2 version %s', (version) => {
+      expect(getRunCapabilities(version).framedStreamMarkers).toBe(false);
+    });
+
+    it('is true at the exact cutoff version (5.0.0-beta.26)', () => {
+      expect(getRunCapabilities('5.0.0-beta.26').framedStreamMarkers).toBe(
+        true
+      );
+    });
+
+    it.each([
+      '5.0.0-beta.27',
+      '5.0.0',
+      '5.1.0',
+      '6.0.0',
+    ])('is true for post-framed-v2 version %s', (version) => {
+      expect(getRunCapabilities(version).framedStreamMarkers).toBe(true);
+    });
+
+    it('does not enable framed-v2 while framed-v1 byte streams remain enabled below the cutoff', () => {
+      // A run between the two cutoffs can unframe framed-v1 but NOT framed-v2.
+      const caps = getRunCapabilities('5.0.0-beta.20');
+      expect(caps.framedByteStreams).toBe(true);
+      expect(caps.framedStreamMarkers).toBe(false);
+    });
+
+    it('framedStreamMarkersEnabled follows the capability', () => {
+      expect(framedStreamMarkersEnabled('5.0.0-beta.20')).toBe(false);
+      expect(framedStreamMarkersEnabled('5.0.0-beta.26')).toBe(true);
+      expect(framedStreamMarkersEnabled(undefined)).toBe(false);
+    });
+  });
 });