diff --git a/.changeset/stream-tail-world-cursors.md b/.changeset/stream-tail-world-cursors.md new file mode 100644 index 0000000000..7e197ee18a --- /dev/null +++ b/.changeset/stream-tail-world-cursors.md @@ -0,0 +1,9 @@ +--- +'@workflow/world': patch +'@workflow/world-local': patch +'@workflow/world-postgres': patch +'@workflow/world-vercel': patch +'workflow': patch +--- + +Return a checkpoint cursor after every non-empty open-stream chunk page, use the versioned Vercel API contract, and release matching Workflow and World package versions. diff --git a/docs/content/docs/v4/api-reference/workflow-runtime/world/streams.mdx b/docs/content/docs/v4/api-reference/workflow-runtime/world/streams.mdx index b458b6c940..74c9e2307b 100644 --- a/docs/content/docs/v4/api-reference/workflow-runtime/world/streams.mdx +++ b/docs/content/docs/v4/api-reference/workflow-runtime/world/streams.mdx @@ -137,17 +137,19 @@ const result = await world.streams.getChunks(runId, "default", { // [!code highl | `runId` | `string` | The workflow run ID | | `name` | `string` | The stream name | | `options.limit` | `number` | Max chunks per page (default: 100, max: 1000) | -| `options.cursor` | `string` | Cursor from a previous response | +| `options.cursor` | `string` | Resume cursor from a previous non-empty response | **Returns:** `StreamChunksResponse` | Field | Type | Description | |-------|------|-------------| | `data` | `StreamChunk[]` | Chunks in index order. Each has `index` (0-based) and `data` (`Uint8Array`). | -| `cursor` | `string \| null` | Cursor for the next page | +| `cursor` | `string \| null` | Position immediately after the returned chunks. `null` for an empty page or a completed tail. | | `hasMore` | `boolean` | Whether more pages of already-written chunks exist | | `done` | `boolean` | Whether the stream is fully closed. When `false`, new chunks may appear in future requests even after `hasMore` is `false`. | +`cursor` and `hasMore` answer different questions. A non-empty page at the current tail of an open stream returns `hasMore: false` and a cursor so a later poll can resume after the chunks already returned. An empty page returns `cursor: null`; retain the cursor from the previous non-empty response when polling an open stream. + ### getInfo() Retrieve lightweight metadata about a stream without fetching chunks. @@ -199,14 +201,16 @@ import { getWorld } from "workflow/runtime"; const world = await getWorld(); let cursor: string | undefined; +let hasMore: boolean; do { const result = await world.streams.getChunks(runId, "default", { cursor }); // [!code highlight] for (const chunk of result.data) { console.log(`Chunk ${chunk.index}:`, chunk.data); } - cursor = result.cursor ?? undefined; -} while (cursor); + cursor = result.cursor ?? cursor; + hasMore = result.hasMore; +} while (hasMore); ``` ## Related diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/world/streams.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/world/streams.mdx index b458b6c940..74c9e2307b 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/world/streams.mdx +++ b/docs/content/docs/v5/api-reference/workflow-runtime/world/streams.mdx @@ -137,17 +137,19 @@ const result = await world.streams.getChunks(runId, "default", { // [!code highl | `runId` | `string` | The workflow run ID | | `name` | `string` | The stream name | | `options.limit` | `number` | Max chunks per page (default: 100, max: 1000) | -| `options.cursor` | `string` | Cursor from a previous response | +| `options.cursor` | `string` | Resume cursor from a previous non-empty response | **Returns:** `StreamChunksResponse` | Field | Type | Description | |-------|------|-------------| | `data` | `StreamChunk[]` | Chunks in index order. Each has `index` (0-based) and `data` (`Uint8Array`). | -| `cursor` | `string \| null` | Cursor for the next page | +| `cursor` | `string \| null` | Position immediately after the returned chunks. `null` for an empty page or a completed tail. | | `hasMore` | `boolean` | Whether more pages of already-written chunks exist | | `done` | `boolean` | Whether the stream is fully closed. When `false`, new chunks may appear in future requests even after `hasMore` is `false`. | +`cursor` and `hasMore` answer different questions. A non-empty page at the current tail of an open stream returns `hasMore: false` and a cursor so a later poll can resume after the chunks already returned. An empty page returns `cursor: null`; retain the cursor from the previous non-empty response when polling an open stream. + ### getInfo() Retrieve lightweight metadata about a stream without fetching chunks. @@ -199,14 +201,16 @@ import { getWorld } from "workflow/runtime"; const world = await getWorld(); let cursor: string | undefined; +let hasMore: boolean; do { const result = await world.streams.getChunks(runId, "default", { cursor }); // [!code highlight] for (const chunk of result.data) { console.log(`Chunk ${chunk.index}:`, chunk.data); } - cursor = result.cursor ?? undefined; -} while (cursor); + cursor = result.cursor ?? cursor; + hasMore = result.hasMore; +} while (hasMore); ``` ## Related diff --git a/docs/content/worlds/v4/building-a-world.mdx b/docs/content/worlds/v4/building-a-world.mdx index bd0b1a047d..ec76d4fbb3 100644 --- a/docs/content/worlds/v4/building-a-world.mdx +++ b/docs/content/worlds/v4/building-a-world.mdx @@ -215,7 +215,7 @@ interface Streamer { Streams are identified by a combination of `runId` and `name`. Each workflow run can have multiple named streams. `writeMulti()` is an optional optimization for batching multiple writes. -`getChunks` returns a paginated snapshot of currently available chunks (unlike `get` which returns a live `ReadableStream` that waits for new chunks). `getInfo` returns the tail index (last chunk index, 0-based, or `-1` when empty) and whether the stream is complete — useful for resolving negative `startIndex` values into absolute positions. +`getChunks` returns a paginated snapshot of currently available chunks (unlike `get` which returns a live `ReadableStream` that waits for new chunks). `hasMore` is true only when another page of already-written chunks exists. Every non-empty page from an open stream returns a cursor positioned immediately after its last chunk, including the current tail where `hasMore` is false. Empty pages and completed final pages return `cursor: null`. A request with a cursor must not return chunks before that position. `getInfo` returns the tail index (last chunk index, 0-based, or `-1` when empty) and whether the stream is complete — useful for resolving negative `startIndex` values into absolute positions. ## Reference Implementations diff --git a/docs/content/worlds/v5/building-a-world.mdx b/docs/content/worlds/v5/building-a-world.mdx index fc1024283c..db8b5226a7 100644 --- a/docs/content/worlds/v5/building-a-world.mdx +++ b/docs/content/worlds/v5/building-a-world.mdx @@ -248,7 +248,7 @@ interface Streamer { Streams are identified by a combination of `runId` and `name`. Each workflow run can have multiple named streams. `writeMulti()` is an optional optimization for batching multiple writes. -`getChunks` returns a paginated snapshot of currently available chunks (unlike `get` which returns a live `ReadableStream` that waits for new chunks). `getInfo` returns the tail index (last chunk index, 0-based, or `-1` when empty) and whether the stream is complete — useful for resolving negative `startIndex` values into absolute positions. +`getChunks` returns a paginated snapshot of currently available chunks (unlike `get` which returns a live `ReadableStream` that waits for new chunks). `hasMore` is true only when another page of already-written chunks exists. Every non-empty page from an open stream returns a cursor positioned immediately after its last chunk, including the current tail where `hasMore` is false. Empty pages and completed final pages return `cursor: null`. A request with a cursor must not return chunks before that position. `getInfo` returns the tail index (last chunk index, 0-based, or `-1` when empty) and whether the stream is complete — useful for resolving negative `startIndex` values into absolute positions. ## Reference Implementations diff --git a/packages/workflow/package.json b/packages/workflow/package.json index ed72924759..27bda0d9c6 100644 --- a/packages/workflow/package.json +++ b/packages/workflow/package.json @@ -88,11 +88,19 @@ "typescript": "catalog:" }, "peerDependencies": { - "@opentelemetry/api": "1" + "@opentelemetry/api": "1", + "@workflow/world-local": "workspace:^", + "@workflow/world-vercel": "workspace:^" }, "peerDependenciesMeta": { "@opentelemetry/api": { "optional": true + }, + "@workflow/world-local": { + "optional": true + }, + "@workflow/world-vercel": { + "optional": true } } } diff --git a/packages/world-local/src/streamer.test.ts b/packages/world-local/src/streamer.test.ts index 41eebf5eb6..44d3b5ecbb 100644 --- a/packages/world-local/src/streamer.test.ts +++ b/packages/world-local/src/streamer.test.ts @@ -1,3 +1,4 @@ +import assert from 'node:assert/strict'; import { promises as fs } from 'node:fs'; import os from 'node:os'; import path from 'node:path'; @@ -955,35 +956,34 @@ describe('streamer', () => { await new Promise((resolve) => setTimeout(resolve, 2)); await streamer.streams.write(TEST_RUN_ID, streamName, 'b'); await new Promise((resolve) => setTimeout(resolve, 2)); - await streamer.streams.write(TEST_RUN_ID, streamName, 'c'); await streamer.streams.close(TEST_RUN_ID, streamName); - // Page 1: limit=2 + // Page 1: limit=1 const page1 = await streamer.streams.getChunks( TEST_RUN_ID, streamName, { - limit: 2, + limit: 1, } ); - expect(page1.data).toHaveLength(2); + expect(page1.data).toHaveLength(1); expect(page1.data[0].index).toBe(0); - expect(page1.data[1].index).toBe(1); expect(page1.hasMore).toBe(true); - expect(page1.cursor).not.toBeNull(); + assert(page1.cursor); - // Page 2: remaining chunks + // Page 2: a full final page const page2 = await streamer.streams.getChunks( TEST_RUN_ID, streamName, { - limit: 2, - cursor: page1.cursor!, + limit: 1, + cursor: page1.cursor, } ); expect(page2.data).toHaveLength(1); - expect(page2.data[0].index).toBe(2); + expect(page2.data[0].index).toBe(1); expect(page2.hasMore).toBe(false); + expect(page2.cursor).toBeNull(); expect(page2.done).toBe(true); }); @@ -1035,6 +1035,18 @@ describe('streamer', () => { ); expect(result.data).toHaveLength(1); expect(result.done).toBe(false); + expect(result.hasMore).toBe(false); + assert(result.cursor); + + await streamer.streams.write(TEST_RUN_ID, streamName, 'more'); + const resumed = await streamer.streams.getChunks( + TEST_RUN_ID, + streamName, + { cursor: result.cursor } + ); + expect( + resumed.data.map((chunk) => Buffer.from(chunk.data).toString()) + ).toEqual(['more']); }); it('should return empty data for nonexistent stream', async () => { @@ -1045,6 +1057,7 @@ describe('streamer', () => { 'nonexistent' ); expect(result.data).toEqual([]); + expect(result.cursor).toBeNull(); expect(result.hasMore).toBe(false); }); @@ -1065,6 +1078,40 @@ describe('streamer', () => { ); expect(result.data).toHaveLength(1); expect(result.data[0].index).toBe(0); + + const invalidShape = Buffer.from( + JSON.stringify({ i: 'not-a-number' }) + ).toString('base64'); + const shapeResult = await streamer.streams.getChunks( + TEST_RUN_ID, + streamName, + { cursor: invalidShape } + ); + expect(shapeResult.data[0].index).toBe(0); + }); + + it('reports a closed stream after a cursor past its final chunk', async () => { + const { streamer } = await setupStreamer(); + const streamName = 'cursor-past-end'; + + await streamer.streams.write(TEST_RUN_ID, streamName, 'data'); + await streamer.streams.close(TEST_RUN_ID, streamName); + + const cursor = Buffer.from(JSON.stringify({ i: 50 })).toString( + 'base64' + ); + const result = await streamer.streams.getChunks( + TEST_RUN_ID, + streamName, + { cursor } + ); + + expect(result).toMatchObject({ + data: [], + cursor: null, + hasMore: false, + done: true, + }); }); }); diff --git a/packages/world-local/src/streamer.ts b/packages/world-local/src/streamer.ts index ccc99e1b3a..f66754231b 100644 --- a/packages/world-local/src/streamer.ts +++ b/packages/world-local/src/streamer.ts @@ -1,11 +1,12 @@ import { EventEmitter } from 'node:events'; import fs from 'node:fs/promises'; import path from 'node:path'; -import type { - GetChunksOptions, - StreamChunksResponse, - Streamer, - StreamInfoResponse, +import { + type GetChunksOptions, + type StreamChunksResponse, + StreamCursorPositionSchema, + type Streamer, + type StreamInfoResponse, } from '@workflow/world'; import { monotonicFactory } from 'ulid'; import { z } from 'zod'; @@ -349,42 +350,33 @@ export function createStreamer(basedir: string, tag?: string): Streamer { let startIndex = 0; if (options?.cursor) { try { - const decoded = JSON.parse( - Buffer.from(options.cursor, 'base64').toString('utf-8') - ); - startIndex = decoded.i; + startIndex = StreamCursorPositionSchema.parse( + JSON.parse(Buffer.from(options.cursor, 'base64').toString()) + ).i; } catch { startIndex = 0; } } - // Walk from startIndex, reading only the files we need. - // Files before the cursor are skipped entirely. let streamDone = false; + let hasMore = false; const resultChunks: { index: number; data: Uint8Array }[] = []; - let dataIndex = 0; // running count of data (non-EOF) files seen - for (const file of chunkFiles) { + for ( + let fileIndex = startIndex; + fileIndex < chunkFiles.length; + fileIndex++ + ) { + const file = chunkFiles[fileIndex]; const ext = fileExtMap.get(file) ?? '.bin'; const filePath = path.join(chunksDir, `${file}${ext}`); - // Before the cursor: only need to check EOF (1 byte), skip content - if (dataIndex < startIndex) { - if (isEofByte(await readFirstByte(filePath))) { - streamDone = true; - break; - } - dataIndex++; - continue; - } - // Collected enough data chunks — peek at the next file for EOF/hasMore if (resultChunks.length >= limit) { if (isEofByte(await readFirstByte(filePath))) { streamDone = true; } else { - // More data files exist beyond this page - dataIndex++; + hasMore = true; } break; } @@ -396,23 +388,29 @@ export function createStreamer(basedir: string, tag?: string): Streamer { break; } resultChunks.push({ - index: dataIndex, + index: startIndex + resultChunks.length, data: Uint8Array.from(chunk.chunk), }); - dataIndex++; } - // hasMore = we know there are data files beyond this page - const hasMore = - !streamDone && dataIndex > startIndex + resultChunks.length; - const nextIndex = startIndex + resultChunks.length; - const nextCursor = hasMore - ? Buffer.from(JSON.stringify({ i: nextIndex })).toString('base64') - : null; + if (!streamDone && startIndex >= chunkFiles.length) { + const file = chunkFiles.at(-1); + if (file) { + const ext = fileExtMap.get(file) ?? '.bin'; + streamDone = isEofByte( + await readFirstByte(path.join(chunksDir, `${file}${ext}`)) + ); + } + } return { data: resultChunks, - cursor: nextCursor, + cursor: + resultChunks.length > 0 && (hasMore || !streamDone) + ? Buffer.from( + JSON.stringify({ i: startIndex + resultChunks.length }) + ).toString('base64') + : null, hasMore, done: streamDone, }; diff --git a/packages/world-postgres/package.json b/packages/world-postgres/package.json index 3da50d6555..78ddddef16 100644 --- a/packages/world-postgres/package.json +++ b/packages/world-postgres/package.json @@ -70,6 +70,9 @@ "typescript": "catalog:", "vitest": "catalog:" }, + "peerDependencies": { + "workflow": "workspace:^" + }, "keywords": [], "author": "", "packageManager": "pnpm@10.15.1" diff --git a/packages/world-postgres/src/streamer.ts b/packages/world-postgres/src/streamer.ts index a79c3ebd75..0ea057ebe8 100644 --- a/packages/world-postgres/src/streamer.ts +++ b/packages/world-postgres/src/streamer.ts @@ -1,9 +1,10 @@ import { EventEmitter } from 'node:events'; -import type { - GetChunksOptions, - StreamChunksResponse, - Streamer, - StreamInfoResponse, +import { + type GetChunksOptions, + type StreamChunksResponse, + StreamCursorPositionSchema, + type Streamer, + type StreamInfoResponse, } from '@workflow/world'; import { and, asc, eq, gt, sql } from 'drizzle-orm'; import { Client, type Pool } from 'pg'; @@ -17,6 +18,10 @@ const StreamPublishMessage = z.object({ chunkId: z.templateLiteral(['chnk_', z.string()]), }); +const StreamCursorSchema = StreamCursorPositionSchema.extend({ + c: z.templateLiteral(['chnk_', z.string()]), +}); + interface StreamChunkEvent { id: `chnk_${string}`; data: Uint8Array; @@ -238,14 +243,12 @@ export function createStreamer(pool: Pool, drizzle: Drizzle): PostgresStreamer { ): Promise { const limit = options?.limit ?? 100; - // Decode cursor to get the last seen chunkId - let cursorChunkId: string | null = null; + let cursor: z.infer | undefined; if (options?.cursor) { try { - const decoded = JSON.parse( - Buffer.from(options.cursor, 'base64').toString('utf-8') + cursor = StreamCursorSchema.parse( + JSON.parse(Buffer.from(options.cursor, 'base64').toString()) ); - cursorChunkId = decoded.c; } catch { // Invalid cursor, start from beginning } @@ -264,9 +267,7 @@ export function createStreamer(pool: Pool, drizzle: Drizzle): PostgresStreamer { and( eq(streams.streamId, name), eq(streams.eof, false), - ...(cursorChunkId - ? [gt(streams.chunkId, cursorChunkId as `chnk_${string}`)] - : []) + ...(cursor ? [gt(streams.chunkId, cursor.c)] : []) ) ) .orderBy(asc(streams.chunkId)) @@ -275,51 +276,30 @@ export function createStreamer(pool: Pool, drizzle: Drizzle): PostgresStreamer { const hasMore = rows.length > limit; const pageRows = rows.slice(0, limit); - // Check if stream is complete via a separate EOF query - let streamDone = false; const [eofRow] = await drizzle .select({ eof: streams.eof }) .from(streams) .where(and(eq(streams.streamId, name), eq(streams.eof, true))) .limit(1); - if (eofRow) { - streamDone = true; - } - - // Build the cursor index: we need a running index across pages. - // Decode the current start index from the cursor. - let baseIndex = 0; - if (options?.cursor) { - try { - const decoded = JSON.parse( - Buffer.from(options.cursor, 'base64').toString('utf-8') - ); - if (typeof decoded.i === 'number') { - baseIndex = decoded.i; - } - } catch { - // Invalid cursor - } - } + const streamDone = eofRow !== undefined; + const baseIndex = cursor?.i ?? 0; const chunks = pageRows.map((row, i) => ({ index: baseIndex + i, data: new Uint8Array(row.data), })); - const nextCursor = - hasMore && pageRows.length > 0 - ? Buffer.from( - JSON.stringify({ - c: pageRows[pageRows.length - 1].chunkId, - i: baseIndex + pageRows.length, - }) - ).toString('base64') - : null; - return { data: chunks, - cursor: nextCursor, + cursor: + pageRows.length > 0 && (hasMore || !streamDone) + ? Buffer.from( + JSON.stringify({ + c: pageRows[pageRows.length - 1].chunkId, + i: baseIndex + pageRows.length, + }) + ).toString('base64') + : null, hasMore, done: streamDone, }; diff --git a/packages/world-postgres/test/storage.test.ts b/packages/world-postgres/test/storage.test.ts index 4882e17c00..91e513043d 100644 --- a/packages/world-postgres/test/storage.test.ts +++ b/packages/world-postgres/test/storage.test.ts @@ -1,3 +1,4 @@ +import assert from 'node:assert/strict'; import { execSync } from 'node:child_process'; import { PostgreSqlContainer } from '@testcontainers/postgresql'; import type { @@ -30,6 +31,7 @@ import { createRunsStorage, createStepsStorage, } from '../src/storage.js'; +import { createStreamer } from '../src/streamer.js'; // Helper types for events storage type EventsStorage = ReturnType; @@ -142,7 +144,7 @@ describe('Storage (Postgres integration)', () => { async function truncateTables() { await pool.query( - 'TRUNCATE TABLE workflow.workflow_events, workflow.workflow_event_slots, workflow.workflow_steps, workflow.workflow_hooks, workflow.workflow_runs RESTART IDENTITY CASCADE' + 'TRUNCATE TABLE workflow.workflow_events, workflow.workflow_event_slots, workflow.workflow_steps, workflow.workflow_hooks, workflow.workflow_runs, workflow.workflow_stream_chunks RESTART IDENTITY CASCADE' ); } @@ -180,6 +182,37 @@ describe('Storage (Postgres integration)', () => { await container.stop(); }); + describe('streams', () => { + it('resumes an open stream from its current tail', async () => { + const streamer = createStreamer(pool, drizzle); + const runId = `wrun_${ulid()}`; + const streamId = `strm_${ulid()}`; + + try { + await streamer.streams.write(runId, streamId, 'first'); + const first = await streamer.streams.getChunks(runId, streamId); + + expect(first.hasMore).toBe(false); + assert(first.cursor); + + await streamer.streams.write(runId, streamId, 'second'); + await streamer.streams.close(runId, streamId); + const resumed = await streamer.streams.getChunks(runId, streamId, { + cursor: first.cursor, + limit: 1, + }); + + expect( + resumed.data.map((chunk) => Buffer.from(chunk.data).toString()) + ).toEqual(['second']); + expect(resumed.cursor).toBeNull(); + expect(resumed.done).toBe(true); + } finally { + await streamer.close(); + } + }); + }); + describe('runs', () => { describe('create', () => { it('should create a new workflow run', async () => { diff --git a/packages/world-sim/src/streams.test.ts b/packages/world-sim/src/streams.test.ts new file mode 100644 index 0000000000..7f57914e1d --- /dev/null +++ b/packages/world-sim/src/streams.test.ts @@ -0,0 +1,31 @@ +import assert from 'node:assert/strict'; +import { describe, expect, it } from 'vitest'; +import { createSimStreamer } from './streams.js'; + +describe('SimStreamer.getChunks', () => { + it('resumes an open stream after its current tail', async () => { + const streamer = createSimStreamer(); + await streamer.streams.write('run', 'stream', 'first'); + + const first = await streamer.streams.getChunks('run', 'stream'); + expect(first).toMatchObject({ hasMore: false, done: false }); + expect(first.cursor).toBe('1'); + assert(first.cursor); + + await streamer.streams.write('run', 'stream', 'second'); + const second = await streamer.streams.getChunks('run', 'stream', { + cursor: first.cursor, + }); + + expect(second.data.map(({ index }) => index)).toEqual([1]); + }); + + it('omits the cursor at a completed tail', async () => { + const streamer = createSimStreamer(); + await streamer.streams.write('run', 'stream', 'only'); + await streamer.streams.close('run', 'stream'); + + const page = await streamer.streams.getChunks('run', 'stream'); + expect(page).toMatchObject({ cursor: null, hasMore: false, done: true }); + }); +}); diff --git a/packages/world-sim/src/streams.ts b/packages/world-sim/src/streams.ts index 78e717b6f1..a962d5d315 100644 --- a/packages/world-sim/src/streams.ts +++ b/packages/world-sim/src/streams.ts @@ -134,7 +134,10 @@ export function createSimStreamer(): SimStreamer { const next = from + slice.length; return { data: slice.map((data, i) => ({ index: from + i, data })), - cursor: next < state.chunks.length ? String(next) : null, + cursor: + slice.length > 0 && (next < state.chunks.length || !state.closed) + ? String(next) + : null, hasMore: next < state.chunks.length, done: state.closed, }; diff --git a/packages/world-vercel/src/streamer.test.ts b/packages/world-vercel/src/streamer.test.ts index f7ddac8d77..3f428faeb1 100644 --- a/packages/world-vercel/src/streamer.test.ts +++ b/packages/world-vercel/src/streamer.test.ts @@ -179,6 +179,12 @@ vi.mock('./utils.js', () => ({ baseUrl: 'https://test.example.com', headers: new Headers(), }), + makeRequest: vi.fn().mockResolvedValue({ + data: [], + cursor: null, + hasMore: false, + done: true, + }), })); describe('streams.get', () => { @@ -208,6 +214,23 @@ describe('streams.get', () => { expect(url.pathname).toBe('/v3/runs/run-123/stream/my-stream'); }); + it('reads chunk pages from the v3 endpoint', async () => { + const { makeRequest } = await import('./utils.js'); + const streamer = await getStreamer(); + + await streamer.streams.getChunks('run-123', 'my-stream', { + limit: 500, + cursor: 'next', + }); + + expect(makeRequest).toHaveBeenCalledWith( + expect.objectContaining({ + endpoint: + '/v3/runs/run-123/streams/my-stream/chunks?limit=500&cursor=next', + }) + ); + }); + it('throws a typed terminal error with the retention details on 410', async () => { const expiredAt = '2026-08-10T14:40:00.000Z'; vi.spyOn(globalThis, 'fetch').mockImplementation(async () => diff --git a/packages/world-vercel/src/streamer.ts b/packages/world-vercel/src/streamer.ts index 59538ea6d2..efc5e5ae78 100644 --- a/packages/world-vercel/src/streamer.ts +++ b/packages/world-vercel/src/streamer.ts @@ -75,8 +75,8 @@ function getStreamUrl(name: string, runId: string, httpConfig: HttpConfig) { // (`createReconnectingFramedStream`) resume from the next chunk rather than // treating the timeout as end-of-stream. Reading from v2 would silently // truncate long-lived streams at the server's 2-minute limit. Only the live -// read is affected by the timeout — writes, completion, and snapshot reads -// (chunks/info/list) stay on v2. +// read is affected by the timeout. Chunk snapshots also use v3 for resumable +// open-tail cursors; writes, completion, stream info, and list stay on v2. function getStreamReadUrl(name: string, runId: string, httpConfig: HttpConfig) { return new URL( `${httpConfig.baseUrl}/v3/runs/${encodeURIComponent(runId)}/stream/${encodeURIComponent(name)}` @@ -372,7 +372,7 @@ export function createStreamer(config?: APIConfig): Streamer { params.set('cursor', options.cursor); } const qs = params.toString(); - const endpoint = `/v2/runs/${encodeURIComponent(runId)}/streams/${encodeURIComponent(name)}/chunks${qs ? `?${qs}` : ''}`; + const endpoint = `/v3/runs/${encodeURIComponent(runId)}/streams/${encodeURIComponent(name)}/chunks${qs ? `?${qs}` : ''}`; return makeRequest({ endpoint, config, diff --git a/packages/world-vercel/src/utils.test.ts b/packages/world-vercel/src/utils.test.ts index 002c1e1a0f..8a63d071ea 100644 --- a/packages/world-vercel/src/utils.test.ts +++ b/packages/world-vercel/src/utils.test.ts @@ -291,7 +291,7 @@ describe('getHttpConfig (proxied path)', () => { describe('makeRequest stream expiry errors', () => { it.each([ - ['/v2/runs/wrun_test/streams/stream-test/chunks'], + ['/v3/runs/wrun_test/streams/stream-test/chunks'], ['/v2/runs/wrun_test/streams/stream-test/info'], ])('preserves stream-expired details from %s', async (endpoint) => { vi.spyOn(globalThis, 'fetch').mockImplementation(async () => diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts index a8325cc44c..1d953ab55e 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -111,6 +111,7 @@ export type { } from './shared.js'; export { PaginatedResponseSchema, + StreamCursorPositionSchema, StructuredErrorSchema, } from './shared.js'; export { diff --git a/packages/world/src/shared.ts b/packages/world/src/shared.ts index 9c3d6d0e7f..6c0684c51c 100644 --- a/packages/world/src/shared.ts +++ b/packages/world/src/shared.ts @@ -85,10 +85,14 @@ export interface StreamChunk { export interface GetChunksOptions { /** Maximum number of chunks to return per page (default: 100, max: 1000) */ limit?: number; - /** Opaque cursor from a previous response to fetch the next page */ + /** Opaque cursor from a previous response at which to resume */ cursor?: string; } +export const StreamCursorPositionSchema = z.object({ + i: z.number().int().nonnegative(), +}); + /** * Metadata about a stream, returned by {@link Streamer.getStreamInfo}. */ @@ -113,7 +117,7 @@ export interface StreamInfoResponse { export interface StreamChunksResponse { /** Array of stream chunks in index order */ data: StreamChunk[]; - /** Cursor for the next page, or `null` when no more pages are available */ + /** Position after the returned chunks, or `null` when none were returned or the stream ended */ cursor: string | null; /** Whether additional pages of already-written chunks exist */ hasMore: boolean; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 392d99262c..1d038a07e1 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1321,6 +1321,12 @@ importers: '@workflow/utils': specifier: workspace:* version: link:../utils + '@workflow/world-local': + specifier: workspace:^ + version: link:../world-local + '@workflow/world-vercel': + specifier: workspace:^ + version: link:../world-vercel ms: specifier: 2.1.3 version: 2.1.3 @@ -1447,6 +1453,9 @@ importers: ulid: specifier: 'catalog:' version: 3.0.1 + workflow: + specifier: workspace:^ + version: link:../workflow zod: specifier: 'catalog:' version: 4.3.6 diff --git a/skills/workflow/SKILL.md b/skills/workflow/SKILL.md index d5a0e5207c..20bb62afa8 100644 --- a/skills/workflow/SKILL.md +++ b/skills/workflow/SKILL.md @@ -3,7 +3,7 @@ name: workflow description: Creates durable, resumable workflows using Vercel's Workflow SDK. Use when building workflows that need to survive restarts, pause for external events, retry on failure, or coordinate multi-step operations over time. Triggers on mentions of "workflow", "durable functions", "resumable", "workflow sdk", "queue", "event", "push", "subscribe", or step-based orchestration. metadata: author: Vercel Inc. - version: '1.10' + version: '1.11' --- ## *CRITICAL*: Always Use Correct `workflow` Documentation @@ -691,7 +691,13 @@ await world.streams.writeMulti?.(runId, name, chunks); const readable = await world.streams.get(runId, name, startIndex); await world.streams.close(runId, name); const streamNames = await world.streams.list(runId); -const chunks = await world.streams.getChunks(runId, name, { limit, cursor }); +let cursor: string | undefined; +let hasMore: boolean; +do { + const page = await world.streams.getChunks(runId, name, { limit, cursor }); + cursor = page.cursor ?? cursor; + hasMore = page.hasMore; +} while (hasMore); const info = await world.streams.getInfo(runId, name); // Queue (methods live directly on world — internal SDK infrastructure)