feat(sdk-js): expose ID for SSE events, update joinStream

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

libs/sdk-js/src/types.stream.ts

@@ -24,15 +24,21 @@ type MessageTupleMetadata = {
   [key: string]: unknown;
 };
 
-type AsSubgraph<TEvent extends { event: string; data: unknown }> = {
-  event: TEvent["event"] | `${TEvent["event"]}|${string}`;
-  data: TEvent["data"];
-};
+type AsSubgraph<TEvent extends { id?: string; event: string; data: unknown }> =
+  {
+    id?: TEvent["id"];
+    event: TEvent["event"] | `${TEvent["event"]}|${string}`;
+    data: TEvent["data"];
+  };
 
 /**
  * Stream event with values after completion of each step.
  */
-export type ValuesStreamEvent<StateType> = { event: "values"; data: StateType };
+export type ValuesStreamEvent<StateType> = {
+  id?: string;
+  event: "values";
+  data: StateType;
+};
 
 /** @internal */
 export type SubgraphValuesStreamEvent<StateType> = 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<ErrorStreamEvent>;
  * produced the update as well as the update.
  */
 export type UpdatesStreamEvent<UpdateType> = {
+  id?: string;
   event: "updates";
   data: { [node: string]: UpdateType };
 };
@@ -96,14 +105,17 @@ export type CustomStreamEvent<T> = { event: "custom"; data: T };
 export type SubgraphCustomStreamEvent<T> = AsSubgraph<CustomStreamEvent<T>>;
 
 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<DebugStreamEvent>;
@@ -135,6 +147,7 @@ export type SubgraphDebugStreamEvent = AsSubgraph<DebugStreamEvent>;
  * Stream event with events occurring during execution.
  */
 export type EventsStreamEvent = {
+  id?: string;
   event: "events";
   data: {
     event:
@@ -157,6 +170,7 @@ export type SubgraphEventsStreamEvent = AsSubgraph<EventsStreamEvent>;
  * the `RunsStreamPayload` to receive this event.
  */
 export type FeedbackStreamEvent = {
+  id?: string;
   event: "feedback";
   data: { [feedbackKey: string]: string };
 };

libs/sdk-js/src/utils/sse.ts

@@ -93,6 +93,7 @@ export class BytesLineDecoder extends TransformStream<Uint8Array, Uint8Array> {
 }
 
 interface StreamPart {
+  id: string | undefined;
   event: string;
   data: unknown;
 }
@@ -113,6 +114,7 @@ export class SSEDecoder extends TransformStream<Uint8Array, StreamPart> {
           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<Uint8Array, StreamPart> {
       flush(controller) {
         if (event) {
           controller.enqueue({
+            id: lastEventId || undefined,
             event,
             data: data.length ? decodeArraysToJson(decoder, data) : null,
           });