diff --git a/libs/sdk-js/src/client.ts b/libs/sdk-js/src/client.ts index 258567606..102b1ba95 100644 --- a/libs/sdk-js/src/client.ts +++ b/libs/sdk-js/src/client.ts @@ -1095,13 +1095,12 @@ export class RunsClient< /** * Stream output from a run in real-time, until the run is done. - * Output is not buffered, so any output produced before this call will - * not be received here. * - * @param threadId The ID of the thread. + * @param threadId The ID of the thread. Can be set to `null` | `undefined` for stateless runs. * @param runId The ID of the run. * @param options Additional options for controlling the stream behavior: * - signal: An AbortSignal that can be used to cancel the stream request + * - lastEventId: The ID of the last event received. Can be used to reconnect to a stream without losing events. * - cancelOnDisconnect: When true, automatically cancels the run if the client disconnects from the stream * - streamMode: Controls what types of events to receive from the stream (can be a single mode or array of modes) * Must be a subset of the stream modes passed when creating the run. Background runs default to having the union of all @@ -1109,16 +1108,17 @@ export class RunsClient< * @returns An async generator yielding stream parts. */ async *joinStream( - threadId: string, + threadId: string | undefined | null, runId: string, options?: | { signal?: AbortSignal; cancelOnDisconnect?: boolean; + lastEventId?: string; streamMode?: StreamMode | StreamMode[]; } | AbortSignal, - ): AsyncGenerator<{ event: StreamEvent; data: any }> { + ): AsyncGenerator<{ id?: string; event: StreamEvent; data: any }> { const opts = typeof options === "object" && options != null && @@ -1127,15 +1127,23 @@ export class RunsClient< : options; const response = await this.asyncCaller.fetch( - ...this.prepareFetchOptions(`/threads/${threadId}/runs/${runId}/stream`, { - method: "GET", - timeoutMs: null, - signal: opts?.signal, - params: { - cancel_on_disconnect: opts?.cancelOnDisconnect ? "1" : "0", - stream_mode: opts?.streamMode, + ...this.prepareFetchOptions( + threadId != null + ? `/threads/${threadId}/runs/${runId}/stream` + : `/runs/${runId}/stream`, + { + method: "GET", + timeoutMs: null, + signal: opts?.signal, + headers: opts?.lastEventId + ? { "Last-Event-ID": opts.lastEventId } + : undefined, + params: { + cancel_on_disconnect: opts?.cancelOnDisconnect ? "1" : "0", + stream_mode: opts?.streamMode, + }, }, - }), + ), ); const stream: ReadableStream<{ event: string; data: any }> = ( diff --git a/libs/sdk-js/src/types.stream.ts b/libs/sdk-js/src/types.stream.ts index 2234cf43a..5d69c2b9f 100644 --- a/libs/sdk-js/src/types.stream.ts +++ b/libs/sdk-js/src/types.stream.ts @@ -24,15 +24,21 @@ type MessageTupleMetadata = { [key: string]: unknown; }; -type AsSubgraph = { - event: TEvent["event"] | `${TEvent["event"]}|${string}`; - data: TEvent["data"]; -}; +type AsSubgraph = + { + id?: TEvent["id"]; + event: TEvent["event"] | `${TEvent["event"]}|${string}`; + data: TEvent["data"]; + }; /** * Stream event with values after completion of each step. */ -export type ValuesStreamEvent = { event: "values"; data: StateType }; +export type ValuesStreamEvent = { + id?: string; + event: "values"; + data: StateType; +}; /** @internal */ export type SubgraphValuesStreamEvent = AsSubgraph< @@ -57,6 +63,7 @@ export type SubgraphMessagesTupleStreamEvent = * Metadata stream event with information about the run and thread */ export type MetadataStreamEvent = { + id?: string; event: "metadata"; data: { run_id: string; thread_id: string }; }; @@ -65,6 +72,7 @@ export type MetadataStreamEvent = { * Stream event with error information. */ export type ErrorStreamEvent = { + id?: string; event: "error"; data: { error: string; message: string }; }; @@ -78,6 +86,7 @@ export type SubgraphErrorStreamEvent = AsSubgraph; * produced the update as well as the update. */ export type UpdatesStreamEvent = { + id?: string; event: "updates"; data: { [node: string]: UpdateType }; }; @@ -96,14 +105,17 @@ export type CustomStreamEvent = { event: "custom"; data: T }; export type SubgraphCustomStreamEvent = AsSubgraph>; type MessagesMetadataStreamEvent = { + id?: string; event: "messages/metadata"; data: { [messageId: string]: { metadata: unknown } }; }; type MessagesCompleteStreamEvent = { + id?: string; event: "messages/complete"; data: Message[]; }; type MessagesPartialStreamEvent = { + id?: string; event: "messages/partial"; data: Message[]; }; @@ -126,7 +138,7 @@ export type SubgraphMessagesStreamEvent = /** * Stream event with detailed debug information. */ -export type DebugStreamEvent = { event: "debug"; data: unknown }; +export type DebugStreamEvent = { id?: string; event: "debug"; data: unknown }; /** @internal */ export type SubgraphDebugStreamEvent = AsSubgraph; @@ -135,6 +147,7 @@ export type SubgraphDebugStreamEvent = AsSubgraph; * Stream event with events occurring during execution. */ export type EventsStreamEvent = { + id?: string; event: "events"; data: { event: @@ -157,6 +170,7 @@ export type SubgraphEventsStreamEvent = AsSubgraph; * the `RunsStreamPayload` to receive this event. */ export type FeedbackStreamEvent = { + id?: string; event: "feedback"; data: { [feedbackKey: string]: string }; }; diff --git a/libs/sdk-js/src/utils/sse.ts b/libs/sdk-js/src/utils/sse.ts index c3c525113..d0482f7e5 100644 --- a/libs/sdk-js/src/utils/sse.ts +++ b/libs/sdk-js/src/utils/sse.ts @@ -93,6 +93,7 @@ export class BytesLineDecoder extends TransformStream { } interface StreamPart { + id: string | undefined; event: string; data: unknown; } @@ -113,6 +114,7 @@ export class SSEDecoder extends TransformStream { if (!event && !data.length && !lastEventId && retry == null) return; const sse = { + id: lastEventId || undefined, event, data: data.length ? decodeArraysToJson(decoder, data) : null, }; @@ -151,6 +153,7 @@ export class SSEDecoder extends TransformStream { flush(controller) { if (event) { controller.enqueue({ + id: lastEventId || undefined, event, data: data.length ? decodeArraysToJson(decoder, data) : null, });