Merge branch 'sse-and-backoff' into 'development'

SSE Session and examples

See merge request GeneralProtocols/xo/utils!8

eslint.config.mjs

@@ -2,7 +2,7 @@ import baseConfig from '@xo-cash/eslint-config';
 
 export default [
     {
-        ignores: [ 'scripts/**', 'docs/**' ],
+        ignores: [ 'scripts/**' ],
     },
     ...baseConfig,
 ];

package-lock.json

@@ -1,12 +1,12 @@
 {
     "name": "@xo-cash/utils",
-    "version": "0.0.7",
+    "version": "0.0.8",
     "lockfileVersion": 3,
     "requires": true,
     "packages": {
         "": {
             "name": "@xo-cash/utils",
-            "version": "0.0.7",
+            "version": "0.0.8",
             "license": "MIT",
             "dependencies": {
                 "@bitauth/libauth": "^3.1.0-next.8",

package.json

@@ -1,6 +1,6 @@
 {
     "name": "@xo-cash/utils",
-    "version": "0.0.7",
+    "version": "0.0.8",
     "description": "XO Cash utilities",
     "type": "module",
     "types": "./dist/index.d.mts",

sandbox/sandbox-llm.ts

@@ -0,0 +1,93 @@
+import { SSESession } from '../source/sse-session/index.ts';
+
+// Because this is in a library, and we don't want to add the node types to this as it is intended to be used in a browser
+// we will just declare the process object here locally so we don't get type errors from this script
+declare const process: {
+    env: Record<string, string>;
+    stdout: {
+        write: (data: string) => void;
+    };
+};
+
+// Recommended URL: https://openrouter.ai/api/v1/chat/completions
+// Recommended Model: nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free
+// Example Command: API_KEY="your-api-key" URL="https://openrouter.ai/api/v1/chat/completions" MODEL="ibm-granite/granite-4.1-8b" PROMPT="Hello, Tell me a joke about robots?" npx tsx ./sandbox/sandbox-llm.ts
+
+// Read the Environment Variables for url, model, prompt and api key
+const url = process.env.URL ?? 'https://openrouter.ai/api/v1/chat/completions';
+const model = process.env.MODEL ?? 'nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free';
+const prompt = process.env.PROMPT ?? 'Hello, Tell me a joke about robots?';
+
+const apiKey = process.env.API_KEY ?? '';
+
+// Throw an error if the api key is not set
+if (!apiKey) {
+    throw new Error('API key is required');
+}
+
+// Create a function to get the auth header
+const getAuthHeader = (): string => {
+    return `Bearer ${apiKey}`;
+};
+
+// Create our sse session
+const sseSession = new SSESession(url, {
+    // LLMs use requests
+    method: 'POST',
+
+    // Create the body of the request
+    body: JSON.stringify({
+        model: model,
+        messages: [
+            {
+                role: 'user',
+                content: prompt,
+            },
+        ],
+        stream: true,
+    }),
+
+    // Create a custom request handler to set the headers
+    onRequest: async (requestInit): Promise<RequestInit> => {
+        requestInit.headers ??= {} as HeadersInit;
+
+        // Handle typescript annoyances
+        const headers = requestInit.headers as Record<string, string>;
+
+        // Set our headers - We could also do this using the `headers` property in the SSESession constructor
+        // Doing it here to demonstrate dynamic headers, for example a signed timestamp could be used to authenticate the request.
+        headers.Authorization = getAuthHeader();
+        headers['Content-Type'] = 'application/json';
+
+        return requestInit;
+    },
+});
+
+// Connect to the SSESession
+await sseSession.connect();
+
+// Loop over the message chunks using `for await`
+for await (const message of sseSession.messages) {
+    // Handle `[DONE]` (this may be specific to OpenRouter)
+    if (message.data === '[DONE]') {
+        continue;
+    }
+
+    // First, we will parse the event to JSON
+    const responseJson = JSON.parse(message.data);
+
+    // Then we will grab the relevant part of the response (we want to first grab the choices array)
+    const choices = responseJson.choices;
+
+    // Then we will grab the first choice
+    const firstChoice = choices[0];
+
+    // Then we will grab the next chunk of text
+    const messageContent = firstChoice.delta?.content;
+
+    // Then we will append the text to the console
+    process.stdout.write(messageContent || '');
+}
+
+// Just a terminal/node thing. If we dont put a new line, the console will overwrite the text with the `cwd` or next command input
+process.stdout.write('\n');

sandbox/sandbox-price-oracle.ts

@@ -0,0 +1,17 @@
+import { SSESession, type SSEvent } from '../source/sse-session/index.ts';
+
+// Command: npx tsx ./sandbox/sandbox-price-oracle.ts
+
+// Use the GP Price Oracle API
+const url = 'https://oracles.generalprotocols.com/sse/v1/messages';
+
+// Create our sse session
+const sseSession = await SSESession.create(url);
+
+// Create a message event handler
+sseSession.on('message', (message: SSEvent) => {
+    console.log(message);
+});
+
+// Connect to the SSESession
+await sseSession.connect();

source/index.ts

@@ -2,6 +2,7 @@ export * from './errors.ts';
 export * from './event-emitter.ts';
 export * from './exponential-backoff/index.ts';
 export * from './extended-json.ts';
+export * from './misc.ts';
 export * from './script.ts';
 export * from './sse-session/index.ts';
 export * from './template/errors.ts';

source/misc.ts

@@ -37,3 +37,15 @@ export const isWithinBounds = (value: number, min: number, max: number): boolean
 
     return true;
 };
+
+/**
+ * Converts a non-error to an error
+ * @param error - The error to convert
+ * @returns The error
+ */
+export const normalizeError = (error: unknown): Error => {
+    // Convert the error to an error if it is not already an error
+    const errorInstance = error instanceof Error ? error : new Error(`${error}`);
+
+    return errorInstance;
+};

source/sse-session/constants.ts

@@ -32,3 +32,26 @@ export const SSE_TRAILING_NEWLINE_REGEX = /\n$/;
  * buffered partial lines between streamed chunks before the next parse call.
  */
 export const NEW_LINE = '\n';
+
+/**
+ * Default exponential backoff options for the SSE session.
+ */
+export const SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS = {
+    baseDelay: 250,
+    maxDelay: 10000,
+    maxAttempts: 0,
+    growthRate: 1.3,
+    jitter: 0.3,
+} as const;
+
+/**
+ * Default attempt reconnect flag for the SSE session.
+ * @default true
+ */
+export const SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT = true;
+
+/**
+ * Default persistent flag for the SSE session.
+ * @default false
+ */
+export const SSE_SESSION_PERSISTENT_DEFAULT = false;

source/sse-session/errors.ts

@@ -0,0 +1,31 @@
+/* eslint-disable max-classes-per-file */
+
+/**
+ * Error thrown when a response body is null
+ */
+export class ResponseBodyNullError extends Error {
+    constructor() {
+        super('Response body is null');
+        this.name = 'ResponseBodyNullError';
+    }
+}
+
+/**
+ * Error thrown when an HTTP error occurs
+ */
+export class HTTPError extends Error {
+    constructor(status: number, message: string) {
+        super(`HTTP error! Status: ${status} - ${message}`);
+        this.name = 'HTTPError';
+    }
+}
+
+/**
+ * Error thrown when a plugin's unsubscribe function is not a function
+ */
+export class SSESessionUnsubscribePluginNotAFunctionError extends Error {
+    constructor(pluginName: string) {
+        super(`${pluginName}'s unsubscribe function is not a function`);
+        this.name = 'SSESessionUnsubscribePluginNotAFunctionError';
+    }
+}

source/sse-session/index.ts

@@ -1,2 +1,4 @@
 export * from './async-push-iterator.ts';
+export * from './types.ts';
+export * from './sse-session.ts';
 export * from './sse-event-parser.ts';

source/sse-session/sse-session.ts

@@ -0,0 +1,635 @@
+import type { SSESessionOptions, SSESessionEventMap, SSEvent, SSESessionEventCallbacks, SSESessionOperationContext } from './types.ts';
+import { SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS, SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT, SSE_SESSION_PERSISTENT_DEFAULT } from './constants.ts';
+
+import { EventEmitter, type OffCallback } from '../event-emitter.ts';
+import { ExponentialBackoff } from '../exponential-backoff/index.ts';
+import { normalizeError } from '../misc.ts';
+
+import { SSEEventParser } from './sse-event-parser.ts';
+import { AsyncPushIterator } from './async-push-iterator.ts';
+import { HTTPError, ResponseBodyNullError, SSESessionUnsubscribePluginNotAFunctionError } from './errors.ts';
+
+/**
+ * A fetch-based Server-Sent Events (SSE) client with reconnect and optional
+ * browser tab visibility handling.
+ *
+ * Each session maintains one HTTP streaming connection at a time. Incoming
+ * bytes are parsed into {@link SSEvent} objects and delivered through two
+ * surfaces:
+ *
+ * - **Events** — `"connected"`, `"message"`, `"disconnected"`, `"error"`,
+ *   and `"closed"` on the session itself (extends {@link EventEmitter}).
+ * - **Messages** — {@link messages}, an async iterable for `for await...of`
+ *   consumers.
+ *
+ * Typical usage:
+ *
+ * ```ts
+ * const session = await SSESession.create("/events");
+ *
+ * session.on("message", (event) => console.log(event.data));
+ *
+ * for await (const event of session.messages) {
+ *   handle(event);
+ * }
+ * ```
+ *
+ * ## Lifecycle
+ *
+ * - {@link connect} opens (or reopens) the transport. It resolves once the
+ *   HTTP stream is established; reading continues in the background.
+ * - {@link disconnect} stops the in-flight fetch without ending the session.
+ *   Used internally for tab visibility. The {@link messages} iterator stays
+ *   open so an existing consumer resumes when the tab becomes visible again.
+ * - {@link close} aborts the transport, closes {@link messages}, and emits
+ *   `"closed"`.
+ *
+ * Automatic reconnect is controlled by {@link SSESessionOptions.persistent}
+ * (server closed the stream) and
+ * {@link SSESessionOptions.attemptReconnect} (transport error).
+ *
+ * ## Connection supersession
+ *
+ * Each {@link connect} will create a new controller if one does not exist. Otherwise,
+ * it will return immediately.
+ */
+export class SSESession extends EventEmitter<SSESessionEventMap> {
+    /**
+     * Creates a session and waits until the first connection is established.
+     *
+     * @param url - The SSE endpoint URL.
+     * @param options - Configuration merged with instance defaults.
+     * @returns A connected session.
+     * @throws When the initial connection cannot be established.
+     */
+    static async create(url: string, options: Partial<SSESessionOptions & SSESessionEventCallbacks> = {}): Promise<SSESession> {
+        const session = new SSESession(url, options);
+        await session.connect();
+
+        return session;
+    }
+
+    /**
+     * Enables SSE resume semantics by sending `Last-Event-ID` on reconnect.
+     *
+     * Listens for incoming `"message"` events and remembers the most recent
+     * {@link SSEvent.id}. On every subsequent connect or reconnect, the session's
+     * {@link onRequest} hook is wrapped so that header is attached when an id is
+     * known, allowing the server to replay only events the client has not yet
+     * received.
+     *
+     * The existing {@link onRequest} callback is preserved and runs after the
+     * header is applied, so auth or other header mutations continue to work.
+     *
+     * Since this relies on listening to the `"message"` event, this should ideally
+     * be called before "{@link connect} is called so no messages (and no ids) are missed.
+     *
+     * ```ts
+     * const session = new SSESession(url);
+     * SSESession.addLastEventIdReconnect(session);
+     * await session.connect();
+     * // Reconnects send Last-Event-ID once an event with an id is received.
+     * ```
+     *
+     * @param session - The session to instrument.
+     * @returns The same session, for chaining.
+     */
+    static addLastEventIdReconnect(session: SSESession): { session: SSESession; removeListener?: OffCallback } {
+        // If the plugin is already installed, we return the session and the remove listener function.
+        if (session.plugins.has(SSESession.addLastEventIdReconnect)) {
+            // Get the remove listener function from the session.
+            const removeListener = session.plugins.get(SSESession.addLastEventIdReconnect) as OffCallback;
+
+            // If the plugin was added with a non-function listener, we throw an error.
+            if (typeof removeListener !== 'function') {
+                throw new SSESessionUnsubscribePluginNotAFunctionError('addLastEventIdReconnect');
+            }
+
+            // If the plugin is already installed, we return the session and the remove listener function.
+            return { session, removeListener };
+        }
+
+        // Store the last event id so we can send it on reconnect.
+        let lastEventId: string | undefined;
+
+        // Define the message listener that will store the last event id and begin listening for 'message' events.
+        const messageListener = (event: SSEvent): void => {
+            // If the event has no id, we should not update the lastEventId and instead keep the old one.
+            lastEventId = event.id ?? lastEventId;
+        };
+
+        session.on('message', messageListener);
+
+        // Store the original onRequest callback so we can extend its behavior to write the reconnect header.
+        const originalOnRequest = session.options.onRequest;
+
+        // Whether we actively send the Last-Event-ID header on reconnect.
+        // Because the `onRequest` callback may be extended further by other code, we can't just roll-back to the `originalOnRequest`
+        // Instead, we just use this flag to determine if we should write the Last-Event-ID header.
+        let enabled = true;
+
+        // Extend the onRequest callback to write the Last-Event-ID header if it is enabled.
+        session.options.onRequest = async (request: RequestInit): Promise<RequestInit> => {
+            if (lastEventId && enabled) {
+                request.headers = { ...request.headers, 'Last-Event-ID': lastEventId };
+            }
+
+            return originalOnRequest(request);
+        };
+
+        // Define the remove listener function that will disable the Last-Event-ID header on reconnect.
+        const removeListener = (): void => {
+            enabled = false;
+            session.off('message', messageListener);
+
+            // If the remove listener function is the same as the existing remove listener function, we remove the plugin from the session.
+            // This protects against the `removeListener` function being called onto after a new instance of this plugin is added
+            const existingRemoveListener = session.plugins.get(SSESession.addLastEventIdReconnect) as OffCallback;
+            if (existingRemoveListener === removeListener) {
+                session.plugins.delete(SSESession.addLastEventIdReconnect);
+            }
+        };
+
+        // Store the plugin in the session.
+        session.plugins.set(SSESession.addLastEventIdReconnect, removeListener);
+
+        return { session, removeListener };
+    }
+
+    /**
+     * Pauses and resumes a session based on browser tab visibility.
+     *
+     * Uses the Page Visibility API (`document.visibilitychange`):
+     *
+     * - **hidden** — {@link disconnect} stops the active fetch. {@link messages}
+     *   stays open; `"disconnected"` fires but `"closed"` does not.
+     * - **visible** — {@link connect} re-establishes the stream if needed.
+     *
+     * This controller will not re-connect if the session was disconnected by something else.
+     *
+     * No-op in non-browser environments where `document` is undefined.
+     *
+     * @param session - The session to manage.
+     */
+    static addBrowserVisibilityHandler(session: SSESession): { session: SSESession; removeListener?: OffCallback } {
+        if (typeof document === 'undefined') return { session };
+
+        // If the plugin is already installed, we return the session and the remove listener function.
+        if (session.plugins.has(SSESession.addBrowserVisibilityHandler)) {
+            // Get the remove listener function from the session.
+            const removeListener = session.plugins.get(SSESession.addBrowserVisibilityHandler) as OffCallback;
+
+            // If the plugin was added with a non-function listener, we throw an error.
+            if (typeof removeListener !== 'function') {
+                throw new SSESessionUnsubscribePluginNotAFunctionError('addBrowserVisibilityHandler');
+            }
+
+            // If the plugin is already installed, we return the session and the remove listener function.
+            return { session, removeListener };
+        }
+
+        // Create a listener that will reconnect the session
+        // This listener is added when the hidden->disconnect listener disconnects the session, but will not be added
+        // if the session was disconnected by something else, like a manual disconnect, server-disconnect, or another plugin.
+        const reconnectOnVisible = async (): Promise<void> => {
+            if (document.visibilityState === 'visible') {
+                // Remove this listener
+                document.removeEventListener('visibilitychange', reconnectOnVisible);
+
+                // Reconnect the session
+                await session.connect();
+            }
+        };
+
+        // Create a listener that will disable the reconnect listener if the session is disconnected by something else.
+        const disableOnExternalDisconnect = ({ source }: SSESessionOperationContext): void => {
+            // If the source is not the browser visibility handler, we remove the listener.
+            if (source !== SSESession.addBrowserVisibilityHandler) {
+                document.removeEventListener('visibilitychange', reconnectOnVisible);
+            }
+        };
+
+        // Create a listener that will live as-long as this session is not closed
+        // This listener will create a new 'on visible' listener that will reconnect the session when it disconnects the session.
+        // This prevents it from trying to reconnect the session if the session was disconnected by something else
+        const handleVisibilityChange = async (): Promise<void> => {
+            // If the session is not connected, we do nothing.
+            if (!session.active) {
+                return;
+            }
+
+            // If the session is connected and the visibility is hidden, we disconnect it.
+            if (document.visibilityState === 'hidden') {
+                // Add the listener for connected event to reconnect the session.
+                document.addEventListener('visibilitychange', reconnectOnVisible);
+
+                // Disconnect the session.
+                session.disconnect({ source: SSESession.addBrowserVisibilityHandler });
+            }
+        };
+
+        // Add the listener for visibility change.
+        document.addEventListener('visibilitychange', handleVisibilityChange);
+
+        // Add a listener for `beforeDisconnect` event to disable the reconnect listener if the session is disconnected by something else.
+        session.on('beforeDisconnect', disableOnExternalDisconnect);
+
+        // Remove the listener for visibility change.
+        const removeListener = (): void => {
+            document.removeEventListener('visibilitychange', handleVisibilityChange);
+            document.removeEventListener('visibilitychange', reconnectOnVisible);
+
+            session.off('beforeDisconnect', disableOnExternalDisconnect);
+
+            // If the remove listener function is the same as the existing remove listener function, we remove the plugin from the session.
+            // This protects against the `removeListener` function being called onto after a new instance of this plugin is added
+            const existingRemoveListener = session.plugins.get(SSESession.addBrowserVisibilityHandler) as OffCallback;
+            if (existingRemoveListener === removeListener) {
+                session.plugins.delete(SSESession.addBrowserVisibilityHandler);
+            }
+        };
+
+        // Store the plugin in the session.
+        session.plugins.set(SSESession.addBrowserVisibilityHandler, removeListener);
+
+        // Return the session and the remove listener function.
+        return { session, removeListener };
+    }
+
+    /** SSE endpoint URL for this session. */
+    readonly #url: string;
+
+    /**
+     * Per-instance configuration.
+     *
+     * Defaults live on the instance field (not a shared static) so each session
+     * gets its own {@link SSEEventParser} and {@link ExponentialBackoff}.
+     */
+    public options: SSESessionOptions = {
+        // Fetch options
+        fetch: (...args) => fetch(...args),
+        method: 'GET',
+        headers: {
+            Accept: 'text/event-stream',
+            'Cache-Control': 'no-cache',
+        },
+        body: new FormData(),
+
+        // callback to mutate the fetch options before the fetch is made.
+        onRequest: (request) => Promise.resolve(request),
+
+        // Retry the initial fetch.
+        // Use the spread operator so it's impossible for us to override the default options. (default is exponential backoff)
+        retry: new ExponentialBackoff({
+            ...SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS,
+        }),
+
+        // Reconnection options
+        attemptReconnect: SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT,
+        persistent: SSE_SESSION_PERSISTENT_DEFAULT,
+
+        // Event parser options
+        eventParser: new SSEEventParser(),
+    };
+
+    /**
+     * Registered plugins for the session.
+     * This can be used to prevent duplicate plugins from being added to the session.
+     * The value for a plugin is arbitrary, for example a function that removes the plugin.
+     */
+    public readonly plugins: Map<unknown, unknown> = new Map();
+
+    /** AbortController for the currently active fetch, if any. */
+    #connectionController: AbortController | null = null;
+
+    /** The server's requested retry interval in milliseconds (per SSE spec) if any */
+    #retryInterval: number | null = null;
+
+    /**
+     * Asynchronous stream of parsed SSE events for the active connection.
+     *
+     * Stays open across {@link disconnect} and automatic reconnects so an existing
+     * `for await` consumer keeps receiving events after visibility resumes.
+     *
+     * Closes when:
+     * - the server ends the stream and {@link SSESessionOptions.persistent}
+     *   is false,
+     * - {@link close} is called, or
+     * - a transport error occurs with
+     *   {@link SSESessionOptions.attemptReconnect} disabled.
+     *
+     * A later {@link connect} replaces this with a new iterator when the
+     * previous one was closed. Consumers should read from `session.messages`
+     * rather than caching a reference across terminal disconnects.
+     */
+    public messages: AsyncPushIterator<SSEvent> = new AsyncPushIterator<SSEvent>();
+
+    public constructor(url: string, options: Partial<SSESessionOptions & SSESessionEventCallbacks> = {}) {
+        super();
+
+        const { onConnected, onDisconnected, onError, ...restOptions } = options;
+
+        this.#url = url;
+        this.options = {
+            ...this.options,
+            ...restOptions,
+            // Shallow merge would drop default headers when options.headers is set.
+            headers: { ...this.options.headers, ...options.headers },
+        };
+
+        // Add the onConnected, onDisconnected, and onError as event listeners to the session.
+        // We accept these as options to the constructor so that we can both instantiate the session and add the event listeners at the same time,
+        // avoiding any potential race-conditions or async/await issues.
+        if (onConnected) {
+            this.on('connected', onConnected);
+        }
+
+        if (onDisconnected) {
+            this.on('disconnected', onDisconnected);
+        }
+
+        if (onError) {
+            this.on('error', onError);
+        }
+    }
+
+    /**
+     * Returns true if the session is active by checking if the controller is not null.
+     */
+    public get active(): boolean {
+        return this.#connectionController !== null;
+    }
+
+    /**
+     * Connects or reconnects to the SSE endpoint.
+     *
+     * Resolves once the HTTP stream is established and `"connected"` has been
+     * emitted. Body reading continues asynchronously in the background via the
+     * internal `#readStream` method.
+     *
+     * @throws When the fetch retry policy exhausts attempts or the connection
+     *   is superseded before the reader is handed off (in the latter case the
+     *   promise resolves without throwing).
+     */
+    public async connect(): Promise<void> {
+        // If there is already a controller present, we are already connected.
+        if (this.#connectionController) return;
+
+        // Prepare for a fresh transport. Parser state from an abandoned connection
+        // must not bleed into the next one; reopen messages if a prior terminal
+        // close ended the consumer's iteration loop.
+        this.#resetEventParser();
+        this.#ensureMessageStreamOpen();
+
+        const connectionController = new AbortController();
+        this.#connectionController = connectionController;
+
+        const { method, headers, body } = this.options;
+
+        // Create the fetch options.
+        const fetchOptions: RequestInit = {
+            method,
+            headers: headers || {},
+            signal: connectionController.signal,
+            cache: 'no-store',
+        };
+
+        // If the method is POST, we will set the body of the fetch options.
+        // NOTE: RequestInit doesn't allow body to be `undefined`, so if its false-y we will set it to null.
+        if (method === 'POST') {
+            fetchOptions.body = body || null;
+        }
+
+        // Create a reader for the response body.
+        let reader: ReadableStreamDefaultReader<Uint8Array>;
+
+        try {
+            // Retry the fetch using the provided retry policy. (default is exponential backoff)
+            reader = await this.options.retry.run(() => this.#createReader(fetchOptions), {
+                signal: connectionController.signal,
+            });
+        } catch (error) {
+            // A newer `connect()` superseded this attempt — leave state to the winner.
+            if (this.#connectionController !== connectionController) return;
+
+            // Reset the controller to null to allow for reconnection if needed.
+            this.#connectionController = null;
+
+            // Normalize the error so it can be passed to the event listeners.
+            const normalizedError = normalizeError(error);
+
+            // Emit the disconnected and error events.
+            this.emit('disconnected', { reason: 'error', error: normalizedError });
+            this.emit('error', normalizedError);
+
+            // Close the message stream so `for await...of` consumers resolve.
+            this.#closeMessageStream();
+
+            // Throw the error so the caller can handle it.
+            throw error;
+        }
+
+        // Connection succeeded but was already replaced (for example disconnect during fetch).
+        if (this.#connectionController !== connectionController) {
+            await reader.cancel();
+
+            return;
+        }
+
+        this.emit('connected', undefined);
+
+        // Fire-and-forget: connect() resolves while the stream is consumed.
+        // Data is emitted as a `message` event and from the `messages` iterator.
+        this.#readStream(reader, connectionController).catch((error) => {
+            this.emit('error', error);
+        });
+    }
+
+    /**
+     * Disconnects only the currently active transport.
+     *
+     * `beforeDisconnect` is emitted on every invocation, even when no
+     * transport is currently active. This allows observers to react to
+     * an explicit disconnect operation.
+     *
+     * @param context - Context describing the operation initiator.
+     * @emits `beforeDisconnect` with the supplied operation context.
+     * @emits `disconnected` with reason `"disconnect"` when an active
+     * transport is actually disconnected.
+     */
+    public disconnect(context: SSESessionOperationContext = {}): void {
+        this.emit('beforeDisconnect', context);
+
+        if (this.#connectionController) {
+            // Grab the current controller to ensure we are aborting the correct one.
+            const connectionController = this.#connectionController;
+            this.#connectionController = null;
+
+            // Invalidate any in-flight read loop and fetch for this transport.
+            connectionController.abort();
+
+            // Emit the disconnected event.
+            this.emit('disconnected', {
+                reason: 'disconnect',
+                source: context.source,
+            });
+        }
+
+        // Reset the event parser buffer so data is not carried through resulting in malformed events.
+        this.#resetEventParser();
+    }
+
+    /**
+     * Terminates the session and disables attached visibility handling until
+     * the same instance is manually {@link connect connected} again.
+     *
+     * Closes {@link messages} and emits `"closed"`.
+     */
+    public close(): void {
+        // Disconnect the session.
+        this.disconnect();
+
+        // Close the message stream so `for await...of` consumers resolve.
+        this.#closeMessageStream();
+
+        // Emit the closed event.
+        this.emit('closed', undefined);
+    }
+
+    /**
+     * Performs the HTTP request and returns a reader for the response body.
+     *
+     * {@link SSESessionOptions.onRequest} may mutate headers (for example auth
+     * tokens or `Last-Event-ID`) before the fetch runs.
+     */
+    async #createReader(fetchOptions: RequestInit): Promise<ReadableStreamDefaultReader<Uint8Array>> {
+        const requestOptions = await this.options.onRequest(fetchOptions);
+        const response = await this.options.fetch(this.#url, requestOptions);
+
+        // Handle a bad response from the server
+        if (!response.ok) {
+            // Get the response code and text.
+            const responseCode = response.status;
+            const responseText = await response.text();
+
+            // Create a new error and emit it.
+            const error = new HTTPError(responseCode, responseText);
+            this.emit('error', error);
+
+            // Throw the error so the caller can handle it.
+            throw error;
+        }
+
+        // Handle a response with no body (SSE still uses the response body)
+        if (!response.body) {
+            const error = new ResponseBodyNullError();
+            this.emit('error', error);
+            throw error;
+        }
+
+        // Return the reader for the response body.
+        return response.body.getReader();
+    }
+
+    /**
+     * Reads bytes from an established stream until it ends, errors, or is
+     * superseded by a newer connection.
+     */
+    async #readStream(reader: ReadableStreamDefaultReader<Uint8Array>, connectionController: AbortController): Promise<void> {
+        try {
+            while (this.#connectionController === connectionController) {
+                const { done, value } = await reader.read();
+
+                // `disconnect()` or a newer connect() may have landed while we were awaiting.
+                if (this.#connectionController !== connectionController) return;
+
+                // If the stream is done, we've been disconnected.
+                // We need to clean up and either complete or reconnect the session.
+                if (done) {
+                    // Reset the controller to null to allow for reconnection if needed.
+                    this.#connectionController = null;
+
+                    // Emit the disconnected event.
+                    this.emit('disconnected', { reason: 'remote' });
+
+                    // If the session is persistent, we will attempt to reconnect.
+                    // NOTE: in very rare cases, a malfunctioning server can make this code turn into an infinite loop.
+                    if (this.options.persistent) {
+                        // Server closed gracefully — reopen unless the consumer opted out.
+                        await this.connect();
+                    } else {
+                        // Close the message stream so `for await...of` consumers resolve.
+                        this.#closeMessageStream();
+                    }
+
+                    return;
+                }
+
+                // Some environments yield `{ done: false, value: undefined }`.
+                if (!value) continue;
+
+                // Emit the parsed events to the event listeners and the messages iterator.
+                for (const event of this.options.eventParser.parseEvents(value)) {
+                    // If the event has a retry interval, we will store it.
+                    if (event.retry) {
+                        this.#retryInterval = event.retry;
+                    }
+
+                    this.emit('message', event);
+                    this.messages.push(event);
+                }
+            }
+        } catch (error) {
+            // If the controller is different, we already started a new connection and it would be confusing to handle this error.
+            // The controller can also be null here if the session was disconnected.
+            if (connectionController !== this.#connectionController) return;
+
+            // Invalidate the current controller to allow for reconnection if needed
+            this.#connectionController = null;
+
+            const normalizedError = normalizeError(error);
+            this.emit('disconnected', { reason: 'error', error: normalizedError });
+
+            // Expected path for `disconnect()` — do not treat as an error or reconnect.
+            if (connectionController.signal.aborted) return;
+
+            this.emit('error', normalizedError);
+
+            // If the session is configured to attempt reconnect, we will attempt to reconnect.
+            if (this.options.attemptReconnect) {
+                // If the server has requested a retry interval, we will wait for it before attempting to reconnect.
+                if (this.#retryInterval) {
+                    await new Promise((resolve) => setTimeout(resolve, this.#retryInterval!));
+                }
+
+                // Attempt to reconnect.
+                await this.connect();
+            } else {
+                // Close the message stream so `for await...of` consumers resolve.
+                this.#closeMessageStream();
+            }
+        }
+    }
+
+    /** Clears partial SSE frames left over from an abandoned transport. */
+    #resetEventParser(): void {
+        this.options.eventParser.reset();
+    }
+
+    /**
+     * Creates a new {@link messages} iterator when the previous one was closed
+     * by a terminal disconnect or server stream end.
+     */
+    #ensureMessageStreamOpen(): void {
+        if (!this.messages.closed) return;
+
+        this.messages = new AsyncPushIterator<SSEvent>();
+    }
+
+    /** Ends the message iteration loop for the current connection span. */
+    #closeMessageStream(): void {
+        if (this.messages.closed) return;
+
+        this.messages.close();
+    }
+}

source/sse-session/types.ts

@@ -1,3 +1,173 @@
+import type { ExponentialBackoffRunOptions } from '../exponential-backoff/exponential-backoff';
+
+export type SSERequestInit = {
+
+    /**
+     * The HTTP method to use.
+     */
+    method: 'GET' | 'POST';
+
+    /**
+     * Request headers sent on every connect and reconnect.
+     */
+    headers?: Record<string, string>;
+
+    /**
+     * Request body for POST-based SSE endpoints.
+     */
+    body?: string | FormData;
+};
+
+/**
+ * The fetch function to use.
+ *
+ * NOTE: This is compatible with Browser/Node's native "fetch" function.
+ *       We use this in place of "typeof fetch" so that we can accept non-standard URLs ("url" is a "string" here).
+ *       For example, a LibP2P adapter might not use a standardized URL format (and might only include "path").
+ *       This would cause a type error as native fetch expects type "URL".
+ */
+export type SSERequestFunction = {
+    fetch: (url: string, options: RequestInit) => Promise<Response>;
+};
+
+/**
+ * Lifecycle hooks invoked by {@link SSESession} during connect, read, and teardown.
+ */
+export type SSESessionCallbacks = {
+
+    /**
+     * Called before each fetch so callers can attach auth headers, cookies, or
+     * a `Last-Event-ID` for resume semantics.
+     */
+    onRequest: (request: RequestInit) => Promise<RequestInit>;
+};
+
+export type SSESessionEventCallbacks = {
+
+    /**
+     * Called after the HTTP stream is established and before body reading begins.
+     */
+    onConnected: () => void;
+
+    /**
+     * Called when the active transport ends — including {@link SSESession.disconnect},
+     * server stream completion, and errors. Not paired with {@link SSESessionEventCallbacks.onConnected}
+     * when the initial connect never succeeds.
+     */
+    onDisconnected: () => void;
+
+    /**
+     * Called on fetch or read failures. Not invoked for intentional
+     * {@link SSESession.disconnect} disconnects.
+     */
+    onError: (error: Error) => void;
+};
+
+export type SSESessionRetryInterface = {
+
+    /**
+     * Retry policy used while establishing the HTTP connection in
+     * {@link SSESession.connect}. Defaults to {@link ExponentialBackoff} with
+     * unlimited attempts.
+     */
+    retry: {
+        run<T>(fn: () => Promise<T>, options?: ExponentialBackoffRunOptions): Promise<T>;
+    };
+};
+
+export interface SSEParser {
+
+    /**
+     * Incremental SSE frame parser for the response body.
+     *
+     * {@link SSEEventParser.reset} is called by the session when abandoning a
+     * transport so partial frames do not carry over to the next connection.
+     */
+    eventParser: {
+        parseEvents(buffer: Uint8Array): SSEvent[];
+        reset(): void;
+    };
+}
+
+export type SSELifecycleOptions = {
+
+    /**
+     * When true, {@link SSESession} calls {@link SSESession.connect} again after
+     * a transport **error** (not an intentional disconnect).
+     * @default true
+     */
+    attemptReconnect: boolean;
+
+    /**
+     * When true, {@link SSESession} calls {@link SSESession.connect} again after
+     * the **server** closes the stream normally (`done`).
+     * @default false
+     */
+    persistent: boolean;
+};
+
+/**
+ * Context from the source that triggered the event.
+ */
+export type SSESessionOperationContext = {
+    source?: unknown;
+};
+
+/**
+ * Discriminated context for the reason the session was disconnected internally.
+ */
+export type SSESessionDisconnectInternalContext = {
+    reason: 'disconnect';
+} & SSESessionOperationContext;
+
+/**
+ * Discriminated context for the reason the session was disconnected remotely.
+ */
+export type SSESessionDisconnectRemoteContext = {
+    reason: 'remote';
+};
+
+/**
+ * Discriminated context for the reason the session was disconnected due to an error.
+ */
+export type SSESessionDisconnectErrorContext = {
+    reason: 'error';
+    error: Error;
+};
+
+/**
+ * Discriminated context for the reason the session was disconnected.
+ */
+export type SSESessionDisconnectContext = SSESessionDisconnectInternalContext | SSESessionDisconnectRemoteContext | SSESessionDisconnectErrorContext;
+
+/**
+ * Events emitted by {@link SSESession}.
+ *
+ * - `"connected"` — HTTP stream established.
+ * - `"message"` — A complete SSE event was parsed.
+ * - `"disconnected"` — The active transport ended (including {@link SSESession.disconnect}).
+ * - `"error"` — An unexpected fetch or read failure.
+ * - `"closed"` — {@link SSESession.close} was called.
+ */
+export type SSESessionEventMap = {
+    connected: void;
+    beforeDisconnect: SSESessionOperationContext;
+    disconnected: SSESessionDisconnectContext;
+    error: Error;
+    message: SSEvent;
+    closed: void;
+};
+
+/**
+ * Configuration for {@link SSESession}.
+ */
+export type SSESessionOptions = SSESessionCallbacks
+    & SSERequestInit
+    & SSERequestFunction
+    & SSESessionRetryInterface
+    & SSELifecycleOptions
+    & SSEParser;
+
 /**
  * Represents a Server-Sent Event.
  */

test/event-emitter.test.ts

@@ -101,6 +101,9 @@ const testEventEmitterEmitContinuesAfterListenerThrows = (): void => {
     const emitter = new EventEmitter<TestEvents>();
     const secondListener = vi.fn();
 
+    // Spy on the console.error method
+    const consoleErrorSpy = vi.spyOn(console, 'error').mockImplementation(() => {});
+
     emitter.on('message', (): void => {
         throw new Error('listener failure');
     });
@@ -110,6 +113,12 @@ const testEventEmitterEmitContinuesAfterListenerThrows = (): void => {
 
     expect(secondListener).toHaveBeenCalledOnce();
     expect(secondListener).toHaveBeenCalledWith('hello');
+
+    expect(consoleErrorSpy).toHaveBeenCalledOnce();
+    expect(consoleErrorSpy).toHaveBeenCalledWith(new Error('listener failure'));
+
+    // Remove the mock on console.error
+    consoleErrorSpy.mockRestore();
 };
 
 /**

test/misc.test.ts

@@ -0,0 +1,59 @@
+import { expect, test } from 'vitest';
+import { normalizeError } from '../source/misc.ts';
+
+/** Test cases for the normalizeError function. */
+const normalizeErrorTestCases = [
+    {
+        input: new Error('test error'),
+        expected: new Error('test error'),
+    },
+    {
+        input: 'test error',
+        expected: new Error('test error'),
+    },
+    {
+        input: 1,
+        expected: new Error('1'),
+    },
+    {
+        input: undefined,
+        expected: new Error('undefined'),
+    },
+    {
+        input: null,
+        expected: new Error('null'),
+    },
+    {
+        input: false,
+        expected: new Error('false'),
+    },
+    {
+        input: true,
+        expected: new Error('true'),
+    },
+    {
+        input: {},
+        expected: new Error('[object Object]'),
+    },
+    {
+        input: [],
+        expected: new Error(),
+    },
+];
+
+/**
+ * Tests that normalizeError converts non-Error throws to Error instances.
+ */
+const testNormalizeError = async (): Promise<void> => {
+    for (const testCase of normalizeErrorTestCases) {
+        const result = normalizeError(testCase.input);
+        expect(result).toBeInstanceOf(Error);
+        expect(result.message).toBe(testCase.expected.message);
+    }
+};
+
+const runTests = async (): Promise<void> => {
+    test('normalizeError: converts non-Error throws to Error instances', testNormalizeError);
+};
+
+await runTests();

test/sse-session/helpers/sse-stream.ts

@@ -0,0 +1,65 @@
+export type SseTestStreamOptions = {
+
+    /** Milliseconds to wait before enqueueing each chunk after the first. */
+    chunkDelayMs?: number;
+
+    /**
+     * When true, closes the body as soon as all initial chunks have been sent.
+     * Use this to simulate a server that sends events and then ends the stream.
+     */
+    closeWhenDone?: boolean;
+};
+
+/**
+ * Test double for an SSE HTTP response body.
+ *
+ * Enqueues fixture chunks in order and stays open until {@link close} is called,
+ * matching real servers that keep the connection alive after each event's trailing
+ * `\n\n` frame boundary.
+ */
+export class SseTestStream {
+    readonly stream: ReadableStream<Uint8Array>;
+
+    private controller: ReadableStreamDefaultController<Uint8Array> | null = null;
+
+    private closed = false;
+
+    /**
+     * @param chunks - Fixture `raw` strings, whole or split, to simulate chunk boundaries.
+     * @param options - Delivery timing and optional auto-close after the initial chunks.
+     */
+    constructor(chunks: string[], options: SseTestStreamOptions = {}) {
+        const { chunkDelayMs = 0, closeWhenDone = false } = options;
+        const encoder = new TextEncoder();
+
+        this.stream = new ReadableStream({
+            start: async (controller): Promise<void> => {
+                this.controller = controller;
+
+                for (let i = 0; i < chunks.length; i++) {
+                    if (chunkDelayMs > 0 && i > 0) {
+                        await new Promise((resolve) => setTimeout(resolve, chunkDelayMs));
+                    }
+
+                    if (this.closed) return;
+
+                    controller.enqueue(encoder.encode(chunks[i]!));
+                }
+
+                if (closeWhenDone) {
+                    this.close();
+                }
+            },
+        });
+    }
+
+    /**
+     * Ends the HTTP body the way a server closing the SSE connection would.
+     */
+    close(): void {
+        if (this.closed) return;
+
+        this.closed = true;
+        this.controller?.close();
+    }
+}