You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Commit 66d49c0
Browse filesBrowse the repository at this point in the historyBrowse files
**BREAKING CHANGE**: Restructure stream methods on World interface to use `world.streams.*` namespace with `runId` as the first parameter. `writeToStream(name, runId, chunk)` → `streams.write(runId, name, chunk)`, `writeToStreamMulti` → `streams.writeMulti`, `closeStream` → `streams.close`, `readFromStream` → `streams.get(runId, name, startIndex?)`, `listStreamsByRunId` → `streams.list(runId)`.
Copy file name to clipboardExpand all lines: docs/content/docs/api-reference/workflow-api/world/streams.mdx
+35-33Lines changed: 35 additions & 33 deletions
Display the source diff
Display the rich diff
Original file line number
Diff line number
Diff line change
@@ -2,26 +2,26 @@
2
2
title: Streams
3
3
description: Read, write, and manage real-time data streams for workflow runs.
4
4
type: reference
5
-
summary: "Methods: writeToStream(), writeToStreamMulti(), readFromStream(), closeStream(), listStreamsByRunId(), getStreamChunks(), getStreamInfo(). Stream methods live directly on the world object."
5
+
summary: "Methods: streams.write(), streams.writeMulti(), streams.get(), streams.close(), streams.list(), streams.getChunks(), streams.getInfo(). Stream methods live on world.streams."
6
6
prerequisites:
7
7
- /docs/api-reference/workflow-api/get-world
8
8
related:
9
9
- /docs/foundations/streaming
10
10
- /docs/api-reference/workflow/get-writable
11
11
keywords:
12
-
- writeToStream
13
-
- writeToStreamMulti
14
-
- readFromStream
15
-
- closeStream
16
-
- listStreamsByRunId
17
-
- getStreamChunks
18
-
- getStreamInfo
12
+
- streams.write
13
+
- streams.writeMulti
14
+
- streams.get
15
+
- streams.close
16
+
- streams.list
17
+
- streams.getChunks
18
+
- streams.getInfo
19
19
- Streamer interface
20
20
- real-time streaming
21
21
- stream lifecycle
22
22
---
23
23
24
-
Stream methods live directly on the `world` object returned by `getWorld()`. Use them to write chunks, read streams, and manage stream lifecycle outside of the standard `getWritable()` pattern.
24
+
Stream methods live on `world.streams` (the `streams` sub-object of the `world` object returned by `getWorld()`). Use them to write chunks, read streams, and manage stream lifecycle outside of the standard `getWritable()` pattern.
25
25
26
26
<Callouttype="info">
27
27
For most streaming use cases, use [`getWritable()`](/docs/api-reference/workflow/get-writable) inside steps. Direct stream methods are for advanced scenarios like building custom stream consumers or managing streams from outside a workflow.
@@ -33,81 +33,82 @@ Stream methods live directly on the `world` object returned by `getWorld()`. Use
33
33
import { getWorld } from"workflow/runtime";
34
34
35
35
const world =getWorld(); // [!code highlight]
36
-
// Stream methods are called directly on world — e.g. world.writeToStream()
36
+
// Stream methods are called on world.streams — e.g. world.streams.write()
Write multiple chunks in a single operation. Optional optimization — not all World implementations support it. Falls back to sequential `writeToStream()` calls if unavailable.
59
+
Write multiple chunks in a single operation. Optional optimization — not all World implementations support it. Falls back to sequential `write()` calls if unavailable.
|`startIndex`|`number`| Optional. Positive values skip chunks from the start (0-based). Negative values read from the tail (e.g. `-3` starts 3 chunks from the end). Clamped to 0. |
Fetch stream chunks with cursor-based pagination. Unlike `readFromStream()` (which returns a live `ReadableStream`), this returns a snapshot of currently available chunks.
124
+
Fetch stream chunks with cursor-based pagination. Unlike `get()` (which returns a live `ReadableStream`), this returns a snapshot of currently available chunks.
124
125
125
126
```typescript lineNumbers
126
-
const result =awaitworld.getStreamChunks("default", runId, { // [!code highlight]
127
+
const result =awaitworld.streams.getChunks(runId, "default", { // [!code highlight]
/** Lightweight metadata: tail index and completion flag. */
206
-
getStreamInfo(
207
-
name:string,
208
-
runId:string
209
-
):Promise<{ tailIndex:number; done:boolean }>;
169
+
streamFlushIntervalMs?:number;
170
+
171
+
streams: {
172
+
write(
173
+
runId:string,
174
+
name:string,
175
+
chunk:string|Uint8Array
176
+
):Promise<void>;
177
+
178
+
writeMulti?(
179
+
runId:string,
180
+
name:string,
181
+
chunks: (string|Uint8Array)[]
182
+
):Promise<void>;
183
+
184
+
close(runId:string, name:string):Promise<void>;
185
+
186
+
get(
187
+
runId:string,
188
+
name:string,
189
+
startIndex?:number
190
+
):Promise<ReadableStream<Uint8Array>>;
191
+
192
+
list(runId:string):Promise<string[]>;
193
+
194
+
/** Paginated snapshot of stream chunks. */
195
+
getChunks(
196
+
runId:string,
197
+
name:string,
198
+
options?: { limit?:number; cursor?:string }
199
+
):Promise<{
200
+
data: { index:number; data:Uint8Array }[];
201
+
cursor:string|null;
202
+
hasMore:boolean;
203
+
done:boolean;
204
+
}>;
205
+
206
+
/** Lightweight metadata: tail index and completion flag. */
207
+
getInfo(
208
+
runId:string,
209
+
name:string
210
+
):Promise<{ tailIndex:number; done:boolean }>;
211
+
};
210
212
}
211
213
```
212
214
213
215
Streams are identified by a combination of `runId` and `name`. Each workflow run can have multiple named streams.
214
-
`writeToStreamMulti()` is an optional optimization for batching multiple writes.
216
+
`writeMulti()` is an optional optimization for batching multiple writes.
215
217
216
-
`getStreamChunks` returns a paginated snapshot of currently available chunks (unlike `readFromStream` which returns a live `ReadableStream` that waits for new chunks). `getStreamInfo` 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.
218
+
`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.
0 commit comments