fix: version stream tail cursors across worlds

.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.

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

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

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
 

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
 

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
     }
   }
 }

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,
+        });
       });
     });
 

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,
         };

packages/world-postgres/package.json

@@ -70,6 +70,9 @@
     "typescript": "catalog:",
     "vitest": "catalog:"
   },
+  "peerDependencies": {
+    "workflow": "workspace:^"
+  },
   "keywords": [],
   "author": "",
   "packageManager": "pnpm@10.15.1"