{"version":3,"file":"stream.cjs","names":["isNetworkError"],"sources":["../../src/utils/stream.ts"],"sourcesContent":["import { isNetworkError } from \"./error.js\";\n\n// in this case don't quite match.\ntype IterableReadableStreamInterface = ReadableStream & AsyncIterable;\n\n/**\n * Options for streaming with automatic retry logic.\n */\nexport interface StreamWithRetryOptions {\n /**\n * Maximum number of reconnection attempts. Default is 5.\n */\n maxRetries?: number;\n\n /**\n * AbortSignal to cancel the stream.\n */\n signal?: AbortSignal;\n\n /**\n * Callback invoked when a reconnection attempt is made.\n */\n onReconnect?: (options: {\n attempt: number;\n lastEventId?: string;\n cause: unknown;\n }) => void;\n}\n\n/**\n * Parameters for making a stream request\n */\nexport interface StreamRequestParams {\n /**\n * Last event ID to resume from, if available\n */\n lastEventId?: string;\n\n /**\n * Optional reconnection path from the Location header\n */\n reconnectPath?: string;\n}\n\n/**\n * Error thrown when maximum reconnection attempts are exceeded.\n */\nexport class MaxReconnectAttemptsError extends Error {\n constructor(maxAttempts: number, cause: unknown) {\n super(`Exceeded maximum SSE reconnection attempts (${maxAttempts})`);\n this.name = \"MaxReconnectAttemptsError\";\n this.cause = cause;\n }\n}\n\n/**\n * Stream with automatic retry logic for SSE connections.\n * Implements reconnection behavior similar to the Python SDK.\n *\n * @param makeRequest Function to make requests. When `params` is undefined/empty, it's the initial request.\n * When `params.reconnectPath` is provided, it's a reconnection request.\n * @param options Configuration options\n * @returns AsyncGenerator yielding stream events\n */\nexport async function* streamWithRetry(\n makeRequest: (params?: StreamRequestParams) => Promise<{\n response: Response;\n stream: ReadableStream;\n }>,\n options: StreamWithRetryOptions = {}\n): AsyncGenerator {\n const maxRetries = options.maxRetries ?? 5;\n let attempt = 0;\n let lastEventId: string | undefined;\n let reconnectPath: string | undefined;\n\n while (true) {\n let shouldRetry = false;\n let lastError: unknown;\n let reader: ReadableStreamDefaultReader | undefined;\n\n try {\n // Check if aborted before making request\n if (options.signal?.aborted) return;\n\n // Make request - initial if no reconnect path, reconnect otherwise\n const { response, stream } = await makeRequest(\n reconnectPath ? { lastEventId, reconnectPath } : undefined\n );\n\n // Check for Location header (server-provided reconnection path)\n const locationHeader = response.headers.get(\"location\");\n if (locationHeader) {\n reconnectPath = locationHeader;\n }\n\n // Verify content type\n const contentType = response.headers.get(\"content-type\")?.split(\";\")[0];\n if (contentType && !contentType.includes(\"text/event-stream\")) {\n throw new Error(\n `Expected response header Content-Type to contain 'text/event-stream', got '${contentType}'`\n );\n }\n\n reader = stream.getReader();\n\n try {\n while (true) {\n // Check abort signal before each read\n if (options.signal?.aborted) {\n await reader.cancel();\n return;\n }\n\n const { done, value } = await reader.read();\n\n if (done) {\n // Stream completed successfully\n break;\n }\n\n // Track last event ID for reconnection\n if (value.id) {\n lastEventId = value.id;\n }\n\n yield value;\n }\n\n // Stream completed successfully, exit retry loop\n break;\n } catch (error) {\n // Error during streaming - attempt reconnect if we have a location header\n if (reconnectPath && !options.signal?.aborted) {\n shouldRetry = true;\n } else {\n throw error;\n }\n } finally {\n if (reader) {\n try {\n reader.releaseLock();\n } catch {\n // Ignore errors when releasing lock\n }\n }\n }\n } catch (error) {\n lastError = error;\n\n // Only retry if we have reconnection capability and it's a network error\n if (isNetworkError(error) && reconnectPath && !options.signal?.aborted) {\n shouldRetry = true;\n } else {\n throw error;\n }\n }\n\n if (shouldRetry) {\n attempt += 1;\n if (attempt > maxRetries) {\n throw new MaxReconnectAttemptsError(maxRetries, lastError);\n }\n\n // Notify about reconnection attempt\n options.onReconnect?.({ attempt, lastEventId, cause: lastError });\n\n // Exponential backoff with jitter: min(1000 * 2^attempt, 5000) + random jitter\n const baseDelay = Math.min(1000 * 2 ** (attempt - 1), 5000);\n const jitter = Math.random() * 1000;\n const delay = baseDelay + jitter;\n\n await new Promise((resolve) => {\n setTimeout(resolve, delay);\n });\n\n continue;\n }\n\n // Successfully completed\n break;\n }\n}\n\n/*\n * Support async iterator syntax for ReadableStreams in all environments.\n * Source: https://github.com/MattiasBuelens/web-streams-polyfill/pull/122#issuecomment-1627354490\n */\nexport class IterableReadableStream\n extends ReadableStream\n implements IterableReadableStreamInterface\n{\n public reader: ReadableStreamDefaultReader;\n\n ensureReader() {\n if (!this.reader) {\n this.reader = this.getReader();\n }\n }\n\n async next(): Promise> {\n this.ensureReader();\n try {\n const result = await this.reader.read();\n if (result.done) {\n this.reader.releaseLock(); // release lock when stream becomes closed\n return {\n done: true,\n value: undefined,\n };\n } else {\n return {\n done: false,\n value: result.value,\n };\n }\n } catch (e) {\n this.reader.releaseLock(); // release lock when stream becomes errored\n throw e;\n }\n }\n\n async return(): Promise> {\n this.ensureReader();\n // If wrapped in a Node stream, cancel is already called.\n if (this.locked) {\n const cancelPromise = this.reader.cancel(); // cancel first, but don't await yet\n this.reader.releaseLock(); // release lock first\n await cancelPromise; // now await it\n }\n return { done: true, value: undefined };\n }\n\n // eslint-disable-next-line @typescript-eslint/no-explicit-any\n async throw(e: any): Promise> {\n this.ensureReader();\n if (this.locked) {\n const cancelPromise = this.reader.cancel(); // cancel first, but don't await yet\n this.reader.releaseLock(); // release lock first\n await cancelPromise; // now await it\n }\n throw e;\n }\n\n // eslint-disable-next-line @typescript-eslint/ban-ts-comment\n // @ts-ignore Not present in Node 18 types, required in latest Node 22\n async [Symbol.asyncDispose]() {\n await this.return();\n }\n\n [Symbol.asyncIterator]() {\n return this;\n }\n\n static fromReadableStream(stream: ReadableStream) {\n // From https://developer.mozilla.org/en-US/docs/Web/API/Streams_API/Using_readable_streams#reading_the_stream\n const reader = stream.getReader();\n return new IterableReadableStream({\n start(controller) {\n return pump();\n function pump(): Promise {\n return reader.read().then(({ done, value }) => {\n // When no more data needs to be consumed, close the stream\n if (done) {\n controller.close();\n return;\n }\n // Enqueue the next data chunk into our target stream\n controller.enqueue(value);\n return pump();\n });\n }\n },\n cancel() {\n reader.releaseLock();\n },\n });\n }\n\n static fromAsyncGenerator(generator: AsyncGenerator) {\n return new IterableReadableStream({\n async pull(controller) {\n const { value, done } = await generator.next();\n // When no more data needs to be consumed, close the stream\n if (done) {\n controller.close();\n }\n // Fix: `else if (value)` will hang the streaming when nullish value (e.g. empty string) is pulled\n controller.enqueue(value);\n },\n async cancel(reason) {\n await generator.return(reason);\n },\n });\n }\n}\n"],"mappings":";;;;;AA+CA,IAAa,4BAAb,cAA+C,MAAM;CACnD,YAAY,aAAqB,OAAgB;AAC/C,QAAM,+CAA+C,YAAY,GAAG;AACpE,OAAK,OAAO;AACZ,OAAK,QAAQ;;;;;;;;;;;;AAajB,gBAAuB,gBACrB,aAIA,UAAkC,EAAE,EACjB;CACnB,MAAM,aAAa,QAAQ,cAAc;CACzC,IAAI,UAAU;CACd,IAAI;CACJ,IAAI;AAEJ,QAAO,MAAM;EACX,IAAI,cAAc;EAClB,IAAI;EACJ,IAAI;AAEJ,MAAI;AAEF,OAAI,QAAQ,QAAQ,QAAS;GAG7B,MAAM,EAAE,UAAU,WAAW,MAAM,YACjC,gBAAgB;IAAE;IAAa;IAAe,GAAG,KAAA,EAClD;GAGD,MAAM,iBAAiB,SAAS,QAAQ,IAAI,WAAW;AACvD,OAAI,eACF,iBAAgB;GAIlB,MAAM,cAAc,SAAS,QAAQ,IAAI,eAAe,EAAE,MAAM,IAAI,CAAC;AACrE,OAAI,eAAe,CAAC,YAAY,SAAS,oBAAoB,CAC3D,OAAM,IAAI,MACR,8EAA8E,YAAY,GAC3F;AAGH,YAAS,OAAO,WAAW;AAE3B,OAAI;AACF,WAAO,MAAM;AAEX,SAAI,QAAQ,QAAQ,SAAS;AAC3B,YAAM,OAAO,QAAQ;AACrB;;KAGF,MAAM,EAAE,MAAM,UAAU,MAAM,OAAO,MAAM;AAE3C,SAAI,KAEF;AAIF,SAAI,MAAM,GACR,eAAc,MAAM;AAGtB,WAAM;;AAIR;YACO,OAAO;AAEd,QAAI,iBAAiB,CAAC,QAAQ,QAAQ,QACpC,eAAc;QAEd,OAAM;aAEA;AACR,QAAI,OACF,KAAI;AACF,YAAO,aAAa;YACd;;WAKL,OAAO;AACd,eAAY;AAGZ,OAAIA,cAAAA,eAAe,MAAM,IAAI,iBAAiB,CAAC,QAAQ,QAAQ,QAC7D,eAAc;OAEd,OAAM;;AAIV,MAAI,aAAa;AACf,cAAW;AACX,OAAI,UAAU,WACZ,OAAM,IAAI,0BAA0B,YAAY,UAAU;AAI5D,WAAQ,cAAc;IAAE;IAAS;IAAa,OAAO;IAAW,CAAC;GAKjE,MAAM,QAFY,KAAK,IAAI,MAAO,MAAM,UAAU,IAAI,IAAK,GAC5C,KAAK,QAAQ,GAAG;AAG/B,SAAM,IAAI,SAAS,YAAY;AAC7B,eAAW,SAAS,MAAM;KAC1B;AAEF;;AAIF;;;AAQJ,IAAa,yBAAb,MAAa,+BACH,eAEV;CACE;CAEA,eAAe;AACb,MAAI,CAAC,KAAK,OACR,MAAK,SAAS,KAAK,WAAW;;CAIlC,MAAM,OAAmC;AACvC,OAAK,cAAc;AACnB,MAAI;GACF,MAAM,SAAS,MAAM,KAAK,OAAO,MAAM;AACvC,OAAI,OAAO,MAAM;AACf,SAAK,OAAO,aAAa;AACzB,WAAO;KACL,MAAM;KACN,OAAO,KAAA;KACR;SAED,QAAO;IACL,MAAM;IACN,OAAO,OAAO;IACf;WAEI,GAAG;AACV,QAAK,OAAO,aAAa;AACzB,SAAM;;;CAIV,MAAM,SAAqC;AACzC,OAAK,cAAc;AAEnB,MAAI,KAAK,QAAQ;GACf,MAAM,gBAAgB,KAAK,OAAO,QAAQ;AAC1C,QAAK,OAAO,aAAa;AACzB,SAAM;;AAER,SAAO;GAAE,MAAM;GAAM,OAAO,KAAA;GAAW;;CAIzC,MAAM,MAAM,GAAoC;AAC9C,OAAK,cAAc;AACnB,MAAI,KAAK,QAAQ;GACf,MAAM,gBAAgB,KAAK,OAAO,QAAQ;AAC1C,QAAK,OAAO,aAAa;AACzB,SAAM;;AAER,QAAM;;CAKR,OAAO,OAAO,gBAAgB;AAC5B,QAAM,KAAK,QAAQ;;CAGrB,CAAC,OAAO,iBAAiB;AACvB,SAAO;;CAGT,OAAO,mBAAsB,QAA2B;EAEtD,MAAM,SAAS,OAAO,WAAW;AACjC,SAAO,IAAI,uBAA0B;GACnC,MAAM,YAAY;AAChB,WAAO,MAAM;IACb,SAAS,OAA+B;AACtC,YAAO,OAAO,MAAM,CAAC,MAAM,EAAE,MAAM,YAAY;AAE7C,UAAI,MAAM;AACR,kBAAW,OAAO;AAClB;;AAGF,iBAAW,QAAQ,MAAM;AACzB,aAAO,MAAM;OACb;;;GAGN,SAAS;AACP,WAAO,aAAa;;GAEvB,CAAC;;CAGJ,OAAO,mBAAsB,WAA8B;AACzD,SAAO,IAAI,uBAA0B;GACnC,MAAM,KAAK,YAAY;IACrB,MAAM,EAAE,OAAO,SAAS,MAAM,UAAU,MAAM;AAE9C,QAAI,KACF,YAAW,OAAO;AAGpB,eAAW,QAAQ,MAAM;;GAE3B,MAAM,OAAO,QAAQ;AACnB,UAAM,UAAU,OAAO,OAAO;;GAEjC,CAAC"}