Add prepare script

eslint.config.mjs

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

package-lock.json

@@ -11,6 +11,7 @@
             "dependencies": {
                 "@bitauth/libauth": "^3.1.0-next.8",
                 "@xo-cash/types": "0.0.3",
+                "eventemitter3": "^5.0.4",
                 "zod": "^4.3.6"
             },
             "devDependencies": {
@@ -5690,6 +5691,12 @@
                 "node": ">= 0.6"
             }
         },
+        "node_modules/eventemitter3": {
+            "version": "5.0.4",
+            "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-5.0.4.tgz",
+            "integrity": "sha512-mlsTRyGaPBjPedk6Bvw+aqbsXDtoAyAzm5MO7JgU+yVRyMQ5O8bD4Kcci7BS85f93veegeCPkL8R4GLClnjLFw==",
+            "license": "MIT"
+        },
         "node_modules/execa": {
             "version": "7.2.0",
             "resolved": "https://registry.npmjs.org/execa/-/execa-7.2.0.tgz",

package.json

@@ -23,7 +23,8 @@
         "spellcheck": "cspell 'source/**' 'test/**' 'playground/**'",
         "style": "eslint",
         "syntax": "tsc --noEmit",
-        "test": "vitest --dir test/ --test-timeout=15000 --passWithNoTests --run --coverage"
+        "test": "vitest --dir test/ --test-timeout=15000 --passWithNoTests --run --coverage",
+        "prepare": "npm run build"
     },
     "files": [
         "dist"
@@ -45,6 +46,7 @@
     "dependencies": {
         "@bitauth/libauth": "^3.1.0-next.8",
         "@xo-cash/types": "0.0.3",
+        "eventemitter3": "^5.0.4",
         "zod": "^4.3.6"
     },
     "overrides": {

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
+// Exmaple 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 Environemt 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 relavent 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/errors.ts

@@ -0,0 +1,42 @@
+/**
+ * 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 the maximum number of retries is hit in an exponential backoff
+ */
+export class ExponentialBackoffMaxRetriesHitError extends Error {
+    constructor(errors: Array<Error>) {
+        super('Exponential backoff: Max retries hit', { cause: errors });
+        this.name = 'ExponentialBackoffMaxRetriesHitError';
+    }
+}
+
+/**
+ * Error thrown when the exponential backoff retries are stopped
+ */
+export class ExponentialBackoffStoppedRetriesError extends Error {
+    constructor(reason: unknown) {
+        // Convert the reason to an error if it is not an error
+        const reasonError = reason instanceof Error ? reason : new Error(`${reason}`);
+
+        super(`Exponential backoff was aborted: "${reasonError.message}"`, { cause: reasonError });
+        this.name = 'ExponentialBackoffStoppedRetriesError';
+    }
+}

source/exponential-backoff.ts

@@ -0,0 +1,275 @@
+import { ExponentialBackoffStoppedRetriesError, ExponentialBackoffMaxRetriesHitError } from './errors.ts';
+
+/**
+ * Exponential backoff is a technique used to retry a function after a delay.
+ *
+ * The delay increases exponentially with each attempt, up to a maximum delay.
+ *
+ * The jitter is a random amount of time subtracted from the delay to prevent thundering herd problems.
+ *
+ * The growth rate is the factor by which the delay increases with each attempt.
+ */
+export class ExponentialBackoff {
+    readonly #options: ExponentialBackoffOptions;
+
+    /**
+     * Creates a new exponential-backoff instance.
+     *
+     * Unspecified options use the defaults listed below.
+     *
+     * @param options - Exponential-backoff configuration overrides.
+     * @param options.maxDelay - Maximum delay between retries. Default: `10_000` ms.
+     * @param options.maxAttempts - Maximum number of attempts; `0` retries indefinitely. Default: `10`.
+     * @param options.baseDelay - Delay used as the basis for the first retry. Default: `1_000` ms.
+     * @param options.growthRate - Multiplier applied to the delay after each attempt. Default: `2`.
+     * @param options.jitter - Maximum proportional reduction subtracted from each delay (0–1). Default: `0.1`.
+     */
+    constructor(options: Partial<ExponentialBackoffOptions> = {}) {
+        this.#options = {
+            maxDelay: 10_000,
+            maxAttempts: 10,
+            baseDelay: 1_000,
+            growthRate: 2,
+            jitter: 0.1,
+            ...options,
+        };
+
+        ExponentialBackoff.validateOptions(this.#options);
+    }
+
+    /**
+     * Create a new ExponentialBackoff instance
+     *
+     * @param config - The configuration for the exponential backoff
+     * @returns The ExponentialBackoff instance
+     */
+    static from(config?: Partial<ExponentialBackoffOptions>): ExponentialBackoff {
+        const backoff = new ExponentialBackoff(config);
+
+        return backoff;
+    }
+
+    /**
+     * Run the function with exponential backoff
+     *
+     * @param taskFn - The function to run
+     * @param onError - The callback to call when an error occurs
+     * @param options - The configuration for the exponential backoff
+     *
+     * @throws An {@link ExponentialBackoffMaxRetriesHitError} with all the errors that were thrown by the task function
+     * @throws An {@link ExponentialBackoffStoppedRetriesError} if the abort signal is activated
+     *
+     * @returns The result of the function
+     */
+    static run<T>(
+        taskFn: (callbackParameters: ExponentialBackoffCallbackParameters) => Promise<T>,
+        onError = (_error: Error): void => {},
+        options?: Partial<ExponentialBackoffOptions>,
+    ): Promise<T> {
+        const backoff = ExponentialBackoff.from(options);
+
+        return backoff.run(taskFn, onError);
+    }
+
+    /**
+     * Calculate the delay before we should attempt to retry
+     *
+     * @param options - The configuration for the exponential backoff
+     * @param attempt - The current attempt number
+     * @returns The time in milliseconds before another attempt should be made
+     */
+    public static calculateDelay(options: ExponentialBackoffOptions, attempt: number): number {
+        // Get the power of the growth rate
+        const power = options.growthRate ** attempt;
+
+        // Get the delay before jitter or limit
+        const rawDelay = options.baseDelay * power;
+
+        // Cap the delay to the maximum. Do this before the jitter so jitter does not become larger than delay
+        const cappedDelay = Math.min(rawDelay, options.maxDelay);
+
+        // Get a random number for the amount to "jitter" the delay by
+        const jitterAmount = Math.random();
+
+        // Calculate the jitter
+        const jitter = jitterAmount * options.jitter * cappedDelay;
+
+        // Subtract the jitter from the delay
+        return cappedDelay - jitter;
+    }
+
+    /**
+     * Validate the options for the exponential backoff
+     *
+     * @param options - The options to validate
+     *
+     * @throws An error if the options are invalid
+     */
+    public static validateOptions(options: ExponentialBackoffOptions): void {
+        // Validate the max delay is a finite number not less than 0
+        if (!Number.isFinite(options.maxDelay)) {
+            throw new Error('maxDelay must be a finite number');
+        }
+
+        if (options.maxDelay < 0) {
+            throw new Error('maxDelay must be not less than 0');
+        }
+
+        // Validate the max attempts is a finite number not less than 0
+        if (!Number.isFinite(options.maxAttempts)) {
+            throw new Error('maxAttempts must be a finite number');
+        }
+
+        if (options.maxAttempts < 0) {
+            throw new Error('maxAttempts must be not less than 0');
+        }
+
+        // Validate the base delay is a finite number not less than 0
+        if (!Number.isFinite(options.baseDelay)) {
+            throw new Error('baseDelay must be a finite number');
+        }
+
+        if (options.baseDelay < 0) {
+            throw new Error('baseDelay must be not less than 0');
+        }
+
+        // Validate the growth rate is a finite number not less than 0
+        if (!Number.isFinite(options.growthRate)) {
+            throw new Error('growthRate must be a finite number');
+        }
+
+        if (options.growthRate < 0) {
+            throw new Error('growthRate must be not less than 0');
+        }
+
+        // Validate the jitter is a finite number not less than 0 or greater than 1
+        if (!Number.isFinite(options.jitter)) {
+            throw new Error('jitter must be a finite number');
+        }
+
+        if (options.jitter < 0 || options.jitter > 1) {
+            throw new Error('jitter must be not less than 0 or greater than 1');
+        }
+    }
+
+    /**
+     * Run the function with exponential backoff
+     *
+     * If the function fails but we have not hit the max attempts, the error will be passed to the onError callback
+     * and the function will be retried with an exponential delay
+     *
+     * If the function fails and we have hit the max attempts, an ExponentialBackoffMaxRetriesHitError will be thrown with all the errors that were thrown by the task function
+     *
+     * @param taskFn - The function to run
+     * @param onError - The callback to call when an error occurs
+     *
+     * @throws An {@link ExponentialBackoffMaxRetriesHitError} with all the errors that were thrown by the task function
+     * @throws An {@link ExponentialBackoffStoppedRetriesError} if the abort signal is activated
+     *
+     * @returns The result of the function
+     */
+    async run<T>(
+        taskFn: (callbackParameters: ExponentialBackoffCallbackParameters) => Promise<T>,
+        onError = (_error: Error): void => {},
+    ): Promise<T> {
+        // Initialize an abort signal to allow the task function to be aborted
+        const abortController = new AbortController();
+        const stopRetries = abortController.abort.bind(abortController);
+
+        // Initialize an empty array to store the errors
+        const errors: Error[] = [];
+
+        // Initialize the attempt counter
+        let attempt = 0;
+
+        // If the max attempts is 0, we should continue indefinitely.
+        const unlimitedAttempts = this.#options.maxAttempts === 0;
+
+        // Loop until we succeed, hit the max attempts, or the abort signal is activated
+        while (true) {
+            try {
+                // Await the promise before returning so its execution context remains in the try-catch
+                // If we didn't await, this `run` function would successfully return and any errors would not be caught here.
+                return await taskFn({ stopRetries });
+            } catch (error) {
+                // Store the error in case we fail every attempt
+                const errorInstance = error instanceof Error ? error : new Error(`${error}`);
+                onError(errorInstance);
+
+                // If we have unlimited attempts, don't append this to the errors array to prevent a memory leak.
+                if (!unlimitedAttempts) {
+                    errors.push(errorInstance);
+                }
+            }
+
+            // Calculate the count for next attempt. Do this now so we can exit before waiting and before running the next attempt.
+            const nextAttemptCount = attempt + 1;
+            const nextAttemptExceedsMaxAttempts = nextAttemptCount >= this.#options.maxAttempts;
+
+            // If the next attempt exceeds the max attempts, break out of the loop
+            if (!unlimitedAttempts && nextAttemptExceedsMaxAttempts) {
+                break;
+            }
+
+            // Check if the abort signal has been aborted
+            if (abortController.signal.aborted) {
+                // Throw an error if the abort signal has been aborted
+                throw new ExponentialBackoffStoppedRetriesError(abortController.signal.reason);
+            }
+
+            // Wait before going to the next attempt
+            const delay = ExponentialBackoff.calculateDelay(this.#options, attempt);
+            await new Promise((resolve) => setTimeout(resolve, delay));
+
+            attempt++;
+        }
+
+        // We completed the loop without ever succeeding. Throw an ExponentialBackoffMaxRetriesHitError with all the errors we got
+        throw new ExponentialBackoffMaxRetriesHitError(errors);
+    }
+}
+
+export type ExponentialBackoffOptions = {
+
+    /**
+     * The maximum delay between attempts in milliseconds
+     */
+    maxDelay: number;
+
+    /**
+     * The maximum number of attempts. Passing 0 will result in infinite attempts.
+     */
+    maxAttempts: number;
+
+    /**
+     * The base delay between attempts in milliseconds
+     */
+    baseDelay: number;
+
+    /**
+     * The growth rate of the delay
+     */
+    growthRate: number;
+
+    /**
+     * The jitter of the delay as a percentage of growthRate. The jitter is subtracted from the delay.
+     */
+    jitter: number;
+};
+
+/**
+ * The function to call to stop the retries.
+ * This mimics the AbortSignal.abort function by taking in a reason for stopping
+ *
+ * @param reason - The reason for stopping the retries.
+ */
+export type ExponentialBackoffStopRetriesFunction = (reason: unknown) => void;
+
+/**
+ * The parameters for the task function
+ *
+ * @param stopRetries - The function to call to stop the retries
+ */
+export type ExponentialBackoffCallbackParameters = {
+    stopRetries: ExponentialBackoffStopRetriesFunction;
+};

source/index.ts

@@ -1,4 +1,6 @@
+export * from './exponential-backoff.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

@@ -0,0 +1,15 @@
+/**
+ * Tries to execute an async function and handles any errors that occur.
+ * @param fn - The function to execute.
+ * @param onError - The callback to call if the function fails.
+ * @returns The result of the function.
+ */
+export const tryAsync = async (fn: () => unknown, onError?: (error: Error) => void): Promise<void> => {
+    try {
+        await fn();
+    } catch (error) {
+        const errorInstance = error instanceof Error ? error : new Error(`${error}`);
+
+        onError?.(errorInstance);
+    }
+};

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,478 @@
+import type { SSESessionOptions, SSESessionEventMap, SSEvent } from './types.ts';
+
+import { EventEmitter } from 'eventemitter3';
+
+import { HTTPError, ResponseBodyNullError } from '../errors.ts';
+import { tryAsync } from '../misc.ts';
+import { ExponentialBackoff } from '../exponential-backoff.ts';
+
+import { SSEEventParser } from './sse-event-parser.ts';
+import { AsyncPushIterator } from './async-push-iterator.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 abort} 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 disconnect} aborts the transport, closes {@link messages}, emits
+ *   `"closed"`, and disables attached visibility handlers until the next
+ *   manual {@link connect}.
+ *
+ * Automatic reconnect is controlled by {@link SSESessionOptions.persistent}
+ * (server closed the stream) and
+ * {@link SSESessionOptions.attemptReconnect} (transport error).
+ *
+ * ## Connection supersession
+ *
+ * Each {@link connect} or {@link abort} bumps an internal `connectionId`.
+ * Background read loops capture their id at start and exit quietly when a
+ * newer connection supersedes them, avoiding duplicate events or errors from
+ * stale transports.
+ */
+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> = {}): Promise<SSESession> {
+        const client = new SSESession(url, options);
+        await client.connect();
+
+        return client;
+    }
+
+    /**
+     * 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.
+     *
+     * Attach as early in the session lifetime as possible. When added after
+     * {@link create}, the initial connection omits the header (no id yet);
+     * all later reconnects include it. To instrument before the first connect,
+     * call this on the session returned from {@link withBrowserVisibility}
+     * before awaiting a separate {@link connect} when the tab starts hidden.
+     *
+     * ```ts
+     * const session = await SSESession.create(url);
+     * await SSESession.addLastEventIdReconnect(session);
+     * // Reconnects send Last-Event-ID once an event with an id is received.
+     * ```
+     *
+     * @param client - The session to instrument.
+     * @returns The same session, for chaining.
+     */
+    static async addLastEventIdReconnect(client: SSESession): Promise<SSESession> {
+        let lastEventId: string | undefined;
+
+        client.on('message', (event) => {
+            lastEventId = event.id;
+        });
+
+        const originalOnRequest = client.options.onRequest;
+
+        client.options.onRequest = async (request: RequestInit): Promise<RequestInit> => {
+            if (lastEventId) {
+                request.headers = { ...request.headers, 'Last-Event-ID': lastEventId };
+            }
+
+            return originalOnRequest(request);
+        };
+
+        return client;
+    }
+
+    /**
+     * Pauses and resumes a session based on browser tab visibility.
+     *
+     * Uses the Page Visibility API (`document.visibilitychange`):
+     *
+     * - **hidden** — {@link abort} stops the active fetch. {@link messages}
+     *   stays open; `"disconnected"` fires but `"closed"` does not.
+     * - **visible** — {@link connect} re-establishes the stream if needed.
+     *
+     * The listener is removed when {@link disconnect} emits `"closed"`, and
+     * re-attached automatically on the next `"connected"` event.
+     *
+     * No-op in non-browser environments where `document` is undefined.
+     *
+     * @param client - The session to manage.
+     */
+    static addBrowserVisibilityHandler(client: SSESession): SSESession {
+        if (typeof document === 'undefined') return client;
+
+        const handleVisibilityChange = (): void => {
+            if (document.visibilityState === 'hidden') {
+                void client.abort();
+
+                return;
+            }
+
+            client.connect().catch(() => {
+                // connect() reports failures via onError and the "error" event.
+            });
+        };
+
+        document.addEventListener('visibilitychange', handleVisibilityChange);
+
+        // Stop managing visibility after an explicit disconnect; re-register
+        // when the same instance is manually connected again.
+        client.once('closed', () => {
+            document.removeEventListener('visibilitychange', handleVisibilityChange);
+
+            client.once('connected', () => {
+                SSESession.addBrowserVisibilityHandler(client);
+            });
+        });
+
+        return client;
+    }
+
+    /** SSE endpoint URL for this session. */
+    private 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: (...args) => fetch(...args),
+        method: 'GET',
+        headers: {
+            Accept: 'text/event-stream',
+            'Cache-Control': 'no-cache',
+        },
+        body: new FormData(),
+
+        onRequest: (request) => Promise.resolve(request),
+        onConnected: () => {},
+        onDisconnected: () => {},
+        onError: (error) => console.error('SSEClient error:', error),
+
+        // Retry the initial fetch until it succeeds (maxAttempts: 0 = unlimited).
+        retry: new ExponentialBackoff({
+            baseDelay: 1000,
+            maxDelay: 10000,
+            maxAttempts: 0,
+            growthRate: 1.3,
+            jitter: 0.3,
+        }),
+
+        attemptReconnect: true,
+        persistent: false,
+
+        eventParser: new SSEEventParser(),
+    };
+
+    /** AbortController for the currently active fetch, if any. */
+    private controller: AbortController | null = null;
+
+    /**
+     * Asynchronous stream of parsed SSE events for the active connection.
+     *
+     * Stays open across {@link abort} 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 disconnect} 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> = {}) {
+        super();
+
+        this.url = url;
+        this.options = {
+            ...this.options,
+            ...options,
+            // Shallow merge would drop default headers when options.headers is set.
+            headers: { ...this.options.headers, ...options.headers },
+        };
+    }
+
+    /**
+     * 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
+     * {@link readStream}.
+     *
+     * @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.controller) 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 controller = new AbortController();
+        this.controller = controller;
+
+        const { method, headers, body } = this.options;
+
+        const fetchBody = method === 'POST' ? body : null;
+
+        const fetchOptions: RequestInit = {
+            method,
+            headers: headers || {},
+            body: fetchBody,
+            signal: controller.signal,
+            cache: 'no-store',
+        };
+
+        let reader: ReadableStreamDefaultReader<Uint8Array>;
+
+        try {
+            reader = await this.options.retry.run(() => this.createReader(fetchOptions));
+        } catch (error) {
+            // A newer abort/connect superseded this attempt — leave state to the winner.
+            if (this.controller !== controller) return;
+
+            this.controller = null;
+
+            await this.notifyDisconnected();
+            await this.notifyError(error);
+            this.closeMessageStream();
+
+            throw error;
+        }
+
+        // Connection succeeded but was already replaced (for example abort during fetch).
+        if (this.controller !== controller) {
+            await reader.cancel();
+
+            return;
+        }
+
+        await tryAsync(
+            () => this.options.onConnected(),
+            (error) => this.options.onError(error),
+        );
+        this.emit('connected', undefined);
+
+        // Fire-and-forget: connect() resolves while the stream is consumed.
+        this.readStream(reader, controller).catch((error) => {
+            this.options.onError(error);
+        });
+    }
+
+    /**
+     * Aborts only the currently active transport.
+     *
+     * The session remains reusable: {@link messages} stays open, visibility
+     * handling stays attached, and {@link connect} can reopen the stream.
+     * Partial parser state from the abandoned transport is discarded.
+     *
+     * Emits `"disconnected"` but not `"closed"`.
+     */
+    public async abort(): Promise<void> {
+        if (!this.controller) return;
+
+        // Grab the current controller to ensure we are aborting the correct one.
+        const controller = this.controller;
+        this.controller = null;
+
+        // Invalidate any in-flight read loop and fetch for this transport.
+        controller.abort();
+        this.resetEventParser();
+
+        await this.notifyDisconnected();
+    }
+
+    /**
+     * 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 async disconnect(): Promise<void> {
+        this.closeMessageStream();
+        this.emit('closed', undefined);
+
+        if (this.controller) {
+            await this.abort();
+        } else {
+            this.resetEventParser();
+        }
+    }
+
+    /**
+     * 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.
+     */
+    private async createReader(fetchOptions: RequestInit): Promise<ReadableStreamDefaultReader<Uint8Array>> {
+        const requestOptions = await this.options.onRequest(fetchOptions);
+        const response = await this.options.fetch(this.url, requestOptions);
+
+        if (!response.ok) {
+            const responseCode = response.status;
+            const responseText = await response.text();
+
+            const error = new HTTPError(responseCode, responseText);
+            void this.notifyError(error);
+            throw error;
+        }
+
+        if (!response.body) {
+            const error = new ResponseBodyNullError();
+            void this.notifyError(error);
+            throw error;
+        }
+
+        return response.body.getReader();
+    }
+
+    /**
+     * Reads bytes from an established stream until it ends, errors, or is
+     * superseded by a newer connection.
+     */
+    private async readStream(reader: ReadableStreamDefaultReader<Uint8Array>, controller: AbortController): Promise<void> {
+        try {
+            while (this.controller === controller) {
+                const { done, value } = await reader.read();
+
+                // abort() or a newer connect() may have landed while we were awaiting.
+                if (this.controller !== controller) return;
+
+                if (done) {
+                    this.controller = null;
+
+                    await this.notifyDisconnected();
+
+                    if (this.options.persistent) {
+                        // Server closed gracefully — reopen unless the consumer opted out.
+                        await this.connect();
+                    } else {
+                        this.closeMessageStream();
+                    }
+
+                    return;
+                }
+
+                // Some environments yield `{ done: false, value: undefined }`.
+                if (!value) continue;
+
+                for (const event of this.options.eventParser.parseEvents(value)) {
+                    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.
+            if (controller !== this.controller) return;
+
+            // Invalidate the current controller to allow for reconnection if needed
+            this.controller = null;
+
+            await this.notifyDisconnected();
+
+            // Expected path for abort() — do not treat as an error or reconnect.
+            if (controller.signal.aborted) return;
+
+            await this.notifyError(error);
+
+            if (this.options.attemptReconnect) {
+                await this.connect();
+            } else {
+                this.closeMessageStream();
+            }
+        }
+    }
+
+    /** Clears partial SSE frames left over from an abandoned transport. */
+    private 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.
+     */
+    private ensureMessageStreamOpen(): void {
+        if (!this.messages.closed) return;
+
+        this.messages = new AsyncPushIterator<SSEvent>();
+    }
+
+    /** Ends the message iteration loop for the current connection span. */
+    private closeMessageStream(): void {
+        if (this.messages.closed) return;
+
+        this.messages.close();
+    }
+
+    /** Invokes {@link SSESessionOptions.onDisconnected} and emits `"disconnected"`. */
+    private async notifyDisconnected(): Promise<void> {
+        await tryAsync(
+            () => this.options.onDisconnected(),
+            (error) => this.options.onError(error),
+        );
+        this.emit('disconnected', undefined);
+    }
+
+    /** Invokes {@link SSESessionOptions.onError} and emits `"error"`. */
+    private async notifyError(error: unknown): Promise<void> {
+        const errorInstance = error instanceof Error ? error : new Error(String(error));
+
+        await tryAsync(
+            () => this.options.onError(errorInstance),
+            (callbackError) => console.error('SSESession error:', callbackError),
+        );
+        this.emit('error', errorInstance);
+    }
+}

source/sse-session/types.ts

@@ -1,3 +1,131 @@
+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>;
+
+    /**
+     * Called after the HTTP stream is established and before body reading begins.
+     */
+    onConnected: () => void;
+
+    /**
+     * Called when the active transport ends — including {@link SSESession.abort},
+     * server stream completion, and errors. Not paired with {@link SSESessionCallbacks.onConnected}
+     * when the initial connect never succeeds.
+     */
+    onDisconnected: () => void;
+
+    /**
+     * Called on fetch or read failures. Not invoked for intentional
+     * {@link SSESession.abort} aborts.
+     */
+    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>, onError?: (error: Error) => void): 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 abort).
+     */
+    attemptReconnect: boolean;
+
+    /**
+     * When true, {@link SSESession} calls {@link SSESession.connect} again after
+     * the **server** closes the stream normally (`done`).
+     */
+    persistent: boolean;
+};
+
+/**
+ * Events emitted by {@link SSESession}.
+ *
+ * - `"connected"` — HTTP stream established.
+ * - `"message"` — A complete SSE event was parsed.
+ * - `"disconnected"` — The active transport ended (including {@link SSESession.abort}).
+ * - `"error"` — An unexpected fetch or read failure.
+ * - `"closed"` — {@link SSESession.disconnect} was called; visibility handling is detached.
+ */
+export type SSESessionEventMap = {
+    connected: void;
+    disconnected: void;
+    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/exponential-backoff.test.ts

@@ -0,0 +1,595 @@
+import { expect, test, vi } from 'vitest';
+import { ExponentialBackoff } from '../source/exponential-backoff.ts';
+import { ExponentialBackoffMaxRetriesHitError, ExponentialBackoffStoppedRetriesError } from '../source/errors.ts';
+
+/**
+ * A valid options object that satisfies {@link ExponentialBackoff.validateOptions}.
+ */
+const validExponentialBackoffOptions = {
+    maxDelay: 10_000,
+    maxAttempts: 10,
+    baseDelay: 1_000,
+    growthRate: 2,
+    jitter: 0.1,
+};
+
+/**
+ * Tests that the static {@link ExponentialBackoff.run} helper creates a throwaway instance
+ * with library defaults (including the default 1000ms base delay) when no options are passed.
+ */
+const testExponentialBackoffRunUsesDefaultOptions = async (): Promise<void> => {
+    // Fake timers let us advance time without waiting real seconds between retries.
+    vi.useFakeTimers();
+    // Pin Math.random to 0 so jitter does not reduce the default delay.
+    vi.spyOn(Math, 'random').mockReturnValue(0);
+
+    try {
+        // The wrapped function fails on its first invocation and succeeds on the second.
+        // That forces ExponentialBackoff.run down the retry path using default options.
+        const rejectThenResolveFn = vi.fn().mockRejectedValueOnce(new Error('retry me'))
+.mockResolvedValueOnce('static-result');
+
+        // Call the static helper with no onError and no options — defaults apply entirely.
+        const promise = ExponentialBackoff.run(rejectThenResolveFn);
+
+        // Yield one microtask so the first (immediate) attempt completes and schedules the retry timer.
+        await Promise.resolve();
+        expect(rejectThenResolveFn).toHaveBeenCalledTimes(1);
+
+        // Default baseDelay is 1000ms; advancing less would not trigger the retry yet.
+        await vi.advanceTimersByTimeAsync(1_000);
+
+        // The retry should have succeeded and returned the resolved value from the mock.
+        await expect(promise).resolves.toBe('static-result');
+        expect(rejectThenResolveFn).toHaveBeenCalledTimes(2);
+    } finally {
+        vi.useRealTimers();
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Tests that {@link ExponentialBackoff.run} accepts a partial options object and merges it
+ * with defaults, still retrying when only some fields are overridden.
+ */
+const testExponentialBackoffRunWithPartialOptions = async (): Promise<void> => {
+    // Same fail-then-succeed pattern; we only care that partial options still enable a retry.
+    const rejectThenResolveFn = vi.fn().mockRejectedValueOnce(new Error('retry me'))
+.mockResolvedValueOnce('done');
+
+    // baseDelay/jitter of 0 skip real waiting; maxAttempts: 3 gives headroom for one retry.
+    // onError is explicitly undefined to verify the default no-op handler is used.
+    const result = await ExponentialBackoff.run(rejectThenResolveFn, undefined, {
+        baseDelay: 0,
+        jitter: 0,
+        maxAttempts: 3,
+    });
+
+    expect(result).toBe('done');
+    expect(rejectThenResolveFn).toHaveBeenCalledTimes(2);
+};
+
+/**
+ * Tests that calling {@link ExponentialBackoff.run} on a constructed instance applies
+ * the instance's stored options when no per-run options are supplied.
+ */
+const testExponentialBackoffInstanceRunUsesDefaultOnError = async (): Promise<void> => {
+    const rejectThenResolveFn = vi.fn().mockRejectedValueOnce(new Error('retry me'))
+.mockResolvedValueOnce('instance-result');
+
+    // Options live on the instance; run(fn) should read them instead of static defaults.
+    const backoff = new ExponentialBackoff({ baseDelay: 0, jitter: 0, maxAttempts: 3 });
+
+    const result = await backoff.run(rejectThenResolveFn);
+
+    expect(result).toBe('instance-result');
+    expect(rejectThenResolveFn).toHaveBeenCalledTimes(2);
+};
+
+/**
+ * Tests the happy path: the wrapped function succeeds immediately and no retry machinery runs.
+ */
+const testExponentialBackoffSucceedsOnFirstAttempt = async (): Promise<void> => {
+    // Always resolves — never enters the catch/retry branch.
+    const resolveFn = vi.fn(async () => 'success');
+    const onError = vi.fn();
+
+    const result = await ExponentialBackoff.run(resolveFn, onError, {
+        baseDelay: 0,
+        jitter: 0,
+    });
+
+    expect(result).toBe('success');
+    expect(resolveFn).toHaveBeenCalledOnce();
+    expect(onError).not.toHaveBeenCalled();
+};
+
+/**
+ * Tests that retries continue across multiple failures until the function eventually resolves.
+ */
+const testExponentialBackoffRetriesUntilSuccess = async (): Promise<void> => {
+    // Three invocations: two rejections then a success on the third call.
+    const tripleRejectFn = vi
+        .fn()
+        .mockRejectedValueOnce(new Error('attempt 1'))
+        .mockRejectedValueOnce(new Error('attempt 2'))
+        .mockResolvedValueOnce('success');
+
+    // maxAttempts: 5 is high enough that we stop because fn succeeded, not because we hit the cap.
+    const result = await ExponentialBackoff.run(tripleRejectFn, () => {}, {
+        baseDelay: 0,
+        jitter: 0,
+        maxAttempts: 5,
+    });
+
+    expect(result).toBe('success');
+    expect(tripleRejectFn).toHaveBeenCalledTimes(3);
+};
+
+/**
+ * Tests that the onError callback is invoked once for every failed attempt, including the last one
+ * before an ExponentialBackoffMaxRetriesHitError is thrown to the caller.
+ */
+const testExponentialBackoffCallsOnErrorForEachFailure = async (): Promise<void> => {
+    const error = new Error('temporary failure');
+
+    // Always rejects with the same error — we will exhaust all attempts.
+    const rejectFn = vi.fn().mockRejectedValue(error);
+    const onError = vi.fn();
+
+    // maxAttempts: 3 means three tries total, all of which will fail.
+    await expect(ExponentialBackoff.run(rejectFn, onError, {
+            baseDelay: 0,
+            jitter: 0,
+            maxAttempts: 3,
+        })).rejects.toThrow(ExponentialBackoffMaxRetriesHitError);
+
+    expect(onError).toHaveBeenCalledTimes(3);
+    expect(onError).toHaveBeenCalledWith(error);
+};
+
+/**
+ * Tests that when all attempts are exhausted the caller receives an ExponentialBackoffMaxRetriesHitError
+ * with every task error preserved in order on the cause.
+ */
+const testExponentialBackoffThrowsMaxRetriesHitErrorWhenExhausted = async (): Promise<void> => {
+    const firstError = new Error('first');
+    const lastError = new Error('last');
+
+    // Two distinct errors so we can prove both are collected, not just the last one.
+    const doubleRejectFn = vi.fn().mockRejectedValueOnce(firstError)
+.mockRejectedValueOnce(lastError);
+
+    try {
+        await ExponentialBackoff.run(doubleRejectFn, () => {}, {
+            baseDelay: 0,
+            jitter: 0,
+            maxAttempts: 2,
+        });
+        expect.fail('Expected ExponentialBackoffMaxRetriesHitError to be thrown');
+    } catch (error) {
+        expect(error).toBeInstanceOf(ExponentialBackoffMaxRetriesHitError);
+        expect((error as ExponentialBackoffMaxRetriesHitError).cause).toEqual([ firstError, lastError ]);
+    }
+
+    expect(doubleRejectFn).toHaveBeenCalledTimes(2);
+};
+
+/**
+ * Tests that rejections which are not Error instances are coerced to Error before onError runs,
+ * so callers always observe a consistent error type in the callback.
+ */
+const testExponentialBackoffWrapsNonErrorThrows = async (): Promise<void> => {
+    // Reject with a plain string — not an Error subclass.
+    const rejectedFn = vi.fn().mockRejectedValue('not-an-error');
+    const onError = vi.fn();
+
+    // Single attempt — we fail fast and inspect what onError received.
+    try {
+        await ExponentialBackoff.run(rejectedFn, onError, {
+            baseDelay: 0,
+            jitter: 0,
+            maxAttempts: 1,
+        });
+        expect.fail('Expected ExponentialBackoffMaxRetriesHitError to be thrown');
+    } catch (error) {
+        expect(error).toBeInstanceOf(ExponentialBackoffMaxRetriesHitError);
+        const [ wrappedError ] = (error as ExponentialBackoffMaxRetriesHitError).cause as Error[];
+        expect(wrappedError).toBeInstanceOf(Error);
+        expect(wrappedError.message).toBe('not-an-error');
+    }
+
+    expect(onError).toHaveBeenCalledOnce();
+    expect(onError.mock.calls[0][0]).toBeInstanceOf(Error);
+    expect(onError.mock.calls[0][0].message).toBe('not-an-error');
+};
+
+/**
+ * Tests that when the task function succeeds and the abort signal is aborted, the result is returned
+ * and the onError callback is not called.
+ */
+const testExponentialBackoffRunSuccessAndAbortSignal = async (): Promise<void> => {
+    // Define the function which aborts the exponential backoff and succeeds
+    const abortAndSucceedFn = vi.fn(({ stopRetries }) => {
+        stopRetries(new Error('retry me'));
+
+        return Promise.resolve('success');
+    });
+    const onErrorFn = vi.fn();
+
+    // Run the exponential backoff with the function and the onError callback
+    const result = await ExponentialBackoff.run(abortAndSucceedFn, onErrorFn, {
+        baseDelay: 0,
+        jitter: 0,
+    });
+
+    // Expect the result to be the success message
+    expect(result).toBe('success');
+    expect(abortAndSucceedFn).toHaveBeenCalledOnce();
+
+    // Expect the onError callback to not have been called
+    expect(onErrorFn).not.toHaveBeenCalled();
+};
+
+/**
+ * Tests that when the abort signal is aborted with an error, an ExponentialBackoffStoppedRetriesError is thrown
+ * with the error as the message.
+ */
+const testExponentialBackoffRunWithAbortSignal = async (): Promise<void> => {
+    // Define the function which aborts the exponential backoff and throws an error
+    const abortAndThrowFn = vi.fn(({ stopRetries }) => {
+        stopRetries(new Error('exponential backoff aborted'));
+        throw new Error('error message');
+    });
+    const onErrorFn = vi.fn();
+
+    // Define the expected error
+    const expectedError = new ExponentialBackoffStoppedRetriesError(new Error('exponential backoff aborted'));
+
+    // Run the exponential backoff with the function and the onError callback and expect the error to be thrown
+    await expect(ExponentialBackoff.run(abortAndThrowFn, onErrorFn, {
+            baseDelay: 0,
+            jitter: 0,
+        })).rejects.toThrow(expectedError);
+
+    // Expect the onError callback to have been called once with the error
+    expect(onErrorFn).toHaveBeenCalledOnce();
+    expect(onErrorFn.mock.calls[0][0]).toBeInstanceOf(Error);
+    expect(onErrorFn.mock.calls[0][0].message).toBe('error message');
+
+    // Expect the function to have been called once and not to have resolved
+    expect(abortAndThrowFn).toHaveBeenCalledOnce();
+    expect(abortAndThrowFn).not.toHaveResolved();
+};
+
+/**
+ * Tests that when the abort signal is aborted with a string, an ExponentialBackoffStoppedRetriesError is thrown
+ * with the string as the message.
+ */
+const testExponentialBackoffRunAbortedStringCreatesError = async (): Promise<void> => {
+    // Define the function which aborts the exponential backoff and throws an error
+    const abortAndThrowStringFn = vi.fn(({ stopRetries }) => {
+        stopRetries('exponential backoff aborted');
+
+        // eslint-disable-next-line
+        throw 'error message';
+    });
+    const onErrorFn = vi.fn();
+
+    // Define the expected error, Note that we "stopRetries" with just a string, not an error. They are treated equivalently.
+    const expectedError = new ExponentialBackoffStoppedRetriesError(new Error('exponential backoff aborted'));
+
+    // Run the exponential backoff with the function and the onError callback and expect the error to be thrown
+    await expect(ExponentialBackoff.run(abortAndThrowStringFn, onErrorFn, {
+            baseDelay: 0,
+            jitter: 0,
+        })).rejects.toThrow(expectedError);
+
+    // Expect the onError callback to have been called once with the error
+    expect(onErrorFn).toHaveBeenCalledOnce();
+    expect(onErrorFn.mock.calls[0][0]).toBeInstanceOf(Error);
+    expect(onErrorFn.mock.calls[0][0].message).toBe('error message');
+
+    // Expect the function to have been called once and not to have resolved
+    expect(abortAndThrowStringFn).toHaveBeenCalledOnce();
+    expect(abortAndThrowStringFn).not.toHaveResolved();
+};
+
+/**
+ * Tests the {@link ExponentialBackoff.from} factory and subsequent instance {@link ExponentialBackoff.run}
+ * as an alternative to the static helper.
+ */
+const testExponentialBackoffFromAndInstanceRun = async (): Promise<void> => {
+    const successfullyResolve = vi.fn(async () => 42);
+
+    // from() is a convenience constructor; run() on the result should behave like the static path.
+    const backoff = ExponentialBackoff.from({
+        baseDelay: 0,
+        jitter: 0,
+    });
+
+    const result = await backoff.run(successfullyResolve);
+
+    expect(result).toBe(42);
+    expect(successfullyResolve).toHaveBeenCalledOnce();
+};
+
+/**
+ * Tests that maxAttempts: 0 disables the attempt cap so retries continue until the function succeeds.
+ */
+const testExponentialBackoffRetriesIndefinitelyWhenMaxAttemptsIsZero = async (): Promise<void> => {
+    // Four invocations: three failures then success — would exceed a cap of 3 if one existed.
+    const tripleRejectThenResolveFn = vi
+        .fn()
+        .mockRejectedValueOnce(new Error('attempt 1'))
+        .mockRejectedValueOnce(new Error('attempt 2'))
+        .mockRejectedValueOnce(new Error('attempt 3'))
+        .mockResolvedValueOnce('eventually');
+
+    const result = await ExponentialBackoff.run(tripleRejectThenResolveFn, () => {}, {
+        baseDelay: 0,
+        jitter: 0,
+        maxAttempts: 0,
+    });
+
+    expect(result).toBe('eventually');
+    expect(tripleRejectThenResolveFn).toHaveBeenCalledTimes(4);
+};
+
+/**
+ * Tests the delay formula: each retry waits baseDelay * growthRate^attemptIndex milliseconds
+ * (with jitter disabled so the math is exact).
+ */
+const testExponentialBackoffIncreasesDelayExponentially = async (): Promise<void> => {
+    vi.useFakeTimers();
+    vi.spyOn(Math, 'random').mockReturnValue(0.5);
+
+    try {
+        const doubleRejectThenResolveFn = vi
+            .fn()
+            .mockRejectedValueOnce(new Error('attempt 1'))
+            .mockRejectedValueOnce(new Error('attempt 2'))
+            .mockResolvedValueOnce('success');
+
+        const promise = ExponentialBackoff.run(doubleRejectThenResolveFn, () => {}, {
+            baseDelay: 100,
+            growthRate: 2,
+            jitter: 0,
+            maxDelay: 10_000,
+            maxAttempts: 5,
+        });
+
+        // Attempt 0 fires synchronously on the first microtask tick.
+        await Promise.resolve();
+        expect(doubleRejectThenResolveFn).toHaveBeenCalledTimes(1);
+
+        // After attempt 0 fails, delay = 100 * 2^0 = 100ms before attempt 1.
+        await vi.advanceTimersByTimeAsync(100);
+        expect(doubleRejectThenResolveFn).toHaveBeenCalledTimes(2);
+
+        // After attempt 1 fails, delay = 100 * 2^1 = 200ms before attempt 2.
+        await vi.advanceTimersByTimeAsync(200);
+        expect(doubleRejectThenResolveFn).toHaveBeenCalledTimes(3);
+
+        await expect(promise).resolves.toBe('success');
+    } finally {
+        vi.useRealTimers();
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Tests that computed delay never exceeds maxDelay even when exponential growth would go higher.
+ */
+const testExponentialBackoffCapsDelayAtMaxDelay = async (): Promise<void> => {
+    vi.useFakeTimers();
+    vi.spyOn(Math, 'random').mockReturnValue(0.5);
+
+    try {
+        const doubleRejectThenResolveFn = vi
+            .fn()
+            .mockRejectedValueOnce(new Error('attempt 1'))
+            .mockRejectedValueOnce(new Error('attempt 2'))
+            .mockResolvedValueOnce('success');
+
+        const promise = ExponentialBackoff.run(doubleRejectThenResolveFn, () => {}, {
+            baseDelay: 1_000,
+            growthRate: 4,
+            jitter: 0,
+            maxDelay: 2_000,
+            maxAttempts: 5,
+        });
+
+        await Promise.resolve();
+        expect(doubleRejectThenResolveFn).toHaveBeenCalledTimes(1);
+
+        // attempt 0: 1000 * 4^0 = 1000ms, below the 2000ms cap.
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(doubleRejectThenResolveFn).toHaveBeenCalledTimes(2);
+
+        // attempt 1: uncapped would be 4000ms but maxDelay clamps to 2000ms.
+        await vi.advanceTimersByTimeAsync(2_000);
+        expect(doubleRejectThenResolveFn).toHaveBeenCalledTimes(3);
+
+        await expect(promise).resolves.toBe('success');
+    } finally {
+        vi.useRealTimers();
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Tests that jitter subtracts up to jitter * cappedDelay from the capped delay based on Math.random.
+ */
+const testExponentialBackoffAppliesJitter = async (): Promise<void> => {
+    vi.useFakeTimers();
+    // random = 1 → full 10% reduction: 1000 - (1 * 0.1 * 1000) = 900ms.
+    vi.spyOn(Math, 'random').mockReturnValue(1);
+
+    try {
+        const rejectThenResolveFn = vi.fn().mockRejectedValueOnce(new Error('attempt 1'))
+.mockResolvedValueOnce('success');
+
+        const promise = ExponentialBackoff.run(rejectThenResolveFn, () => {}, {
+            baseDelay: 1_000,
+            growthRate: 1,
+            jitter: 0.1,
+            maxDelay: 10_000,
+            maxAttempts: 3,
+        });
+
+        await Promise.resolve();
+        expect(rejectThenResolveFn).toHaveBeenCalledTimes(1);
+
+        // Advancing 899ms is one ms short of the jittered delay; 900ms triggers the retry.
+        await vi.advanceTimersByTimeAsync(899);
+        expect(rejectThenResolveFn).toHaveBeenCalledTimes(1);
+
+        await vi.advanceTimersByTimeAsync(1);
+        expect(rejectThenResolveFn).toHaveBeenCalledTimes(2);
+
+        await expect(promise).resolves.toBe('success');
+    } finally {
+        vi.useRealTimers();
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Tests that {@link ExponentialBackoff.validateOptions} accepts valid options, including boundary values of 0 and 1.
+ */
+const testExponentialBackoffValidateOptionsAcceptsValidOptions = (): void => {
+    const validCases = [
+        validExponentialBackoffOptions,
+        {
+            ...validExponentialBackoffOptions,
+            maxDelay: 0,
+            maxAttempts: 0,
+            baseDelay: 0,
+            growthRate: 0,
+            jitter: 0,
+        },
+        {
+            ...validExponentialBackoffOptions,
+            jitter: 1,
+        },
+    ] as const;
+
+    for (const options of validCases) {
+        expect(() => ExponentialBackoff.validateOptions(options)).not.toThrow();
+    }
+};
+
+/**
+ * Tests that {@link ExponentialBackoff.validateOptions} rejects negative numeric options.
+ */
+const testExponentialBackoffValidateOptionsRejectsNegativeValues = (): void => {
+    // Define our test cases with each value being less than 0
+    const negativeCases = [
+        { field: 'maxDelay', value: -1 },
+        { field: 'maxAttempts', value: -1 },
+        { field: 'baseDelay', value: -1 },
+        { field: 'growthRate', value: -1 },
+    ] as const;
+
+    // Iterate through the test cases and expect an error to be thrown
+    for (const { field, value } of negativeCases) {
+        expect(() =>
+            ExponentialBackoff.validateOptions({
+                ...validExponentialBackoffOptions,
+                [field]: value,
+            })).toThrow(`${field} must be not less than 0`);
+    }
+};
+
+/**
+ * Tests that {@link ExponentialBackoff.validateOptions} rejects jitter below 0 or above 1.
+ */
+const testExponentialBackoffValidateOptionsRejectsInvalidJitter = (): void => {
+    // Define our test cases with each value being less than 0 or greater than 1
+    const invalidJitterCases: Array<{ value: number }> = [{ value: -0.1 }, { value: 1.1 }];
+
+    // Iterate through the test cases and expect an error to be thrown
+    for (const { value } of invalidJitterCases) {
+        expect(() =>
+            ExponentialBackoff.validateOptions({
+                ...validExponentialBackoffOptions,
+                jitter: value,
+            })).toThrow('jitter must be not less than 0 or greater than 1');
+    }
+};
+
+/**
+ * Tests that {@link ExponentialBackoff.validateOptions} rejects non-finite values such as Infinity.
+ */
+const testExponentialBackoffValidateOptionsRejectsNonFiniteValues = (): void => {
+    // Define our test cases with each value being Infinity
+    const nonFiniteCases = [
+        { field: 'maxDelay', value: Infinity },
+        { field: 'maxAttempts', value: Infinity },
+        { field: 'baseDelay', value: Infinity },
+        { field: 'growthRate', value: Infinity },
+        { field: 'jitter', value: Infinity },
+    ] as const;
+
+    // Iterate through the test cases and expect an error to be thrown
+    for (const { field, value } of nonFiniteCases) {
+        expect(() =>
+            ExponentialBackoff.validateOptions({
+                ...validExponentialBackoffOptions,
+                [field]: value,
+            })).toThrow(`${field} must be a finite number`);
+    }
+};
+
+/**
+ * Tests that {@link ExponentialBackoff.validateOptions} rejects NaN, which is also non-finite.
+ */
+const testExponentialBackoffValidateOptionsRejectsNaN = (): void => {
+    // Define our test cases with each value being NaN
+    const nanCases = [
+        { field: 'maxDelay', value: Number.NaN },
+        { field: 'maxAttempts', value: Number.NaN },
+        { field: 'baseDelay', value: Number.NaN },
+        { field: 'growthRate', value: Number.NaN },
+        { field: 'jitter', value: Number.NaN },
+    ] as const;
+
+    // Iterate through the test cases and expect an error to be thrown
+    for (const { field, value } of nanCases) {
+        expect(() =>
+            ExponentialBackoff.validateOptions({
+                ...validExponentialBackoffOptions,
+                [field]: value,
+            })).toThrow(`${field} must be a finite number`);
+    }
+};
+
+const runTests = async (): Promise<void> => {
+    test('ExponentialBackoff.run: delegates to a new instance using default options', testExponentialBackoffRunUsesDefaultOptions);
+    test('ExponentialBackoff.run: retries and succeeds with partial options', testExponentialBackoffRunWithPartialOptions);
+    test('ExponentialBackoff.run: uses the instance default onError when omitted', testExponentialBackoffInstanceRunUsesDefaultOnError);
+    test('ExponentialBackoff: returns the result on first success', testExponentialBackoffSucceedsOnFirstAttempt);
+    test('ExponentialBackoff: retries until the function succeeds', testExponentialBackoffRetriesUntilSuccess);
+    test('ExponentialBackoff: calls onError for each failed attempt', testExponentialBackoffCallsOnErrorForEachFailure);
+    test(
+        'ExponentialBackoff: throws ExponentialBackoffMaxRetriesHitError when max attempts are exhausted',
+        testExponentialBackoffThrowsMaxRetriesHitErrorWhenExhausted,
+    );
+    test('ExponentialBackoff: wraps non-Error throws before calling onError', testExponentialBackoffWrapsNonErrorThrows);
+    test('ExponentialBackoff: succeeds and aborts with abort signal', testExponentialBackoffRunSuccessAndAbortSignal);
+    test('ExponentialBackoff: aborts with abort signal', testExponentialBackoffRunWithAbortSignal);
+    test('ExponentialBackoff: aborts with aborted string creates error', testExponentialBackoffRunAbortedStringCreatesError);
+    test('ExponentialBackoff: works via from and instance run', testExponentialBackoffFromAndInstanceRun);
+    test('ExponentialBackoff: retries indefinitely when maxAttempts is 0', testExponentialBackoffRetriesIndefinitelyWhenMaxAttemptsIsZero);
+    test('ExponentialBackoff: increases delay exponentially between attempts', testExponentialBackoffIncreasesDelayExponentially);
+    test('ExponentialBackoff: caps delay at maxDelay', testExponentialBackoffCapsDelayAtMaxDelay);
+    test('ExponentialBackoff: subtracts jitter from the capped delay', testExponentialBackoffAppliesJitter);
+    test('ExponentialBackoff.validateOptions: accepts valid options', testExponentialBackoffValidateOptionsAcceptsValidOptions);
+    test('ExponentialBackoff.validateOptions: rejects negative values', testExponentialBackoffValidateOptionsRejectsNegativeValues);
+    test('ExponentialBackoff.validateOptions: rejects invalid jitter', testExponentialBackoffValidateOptionsRejectsInvalidJitter);
+    test('ExponentialBackoff.validateOptions: rejects Infinity', testExponentialBackoffValidateOptionsRejectsNonFiniteValues);
+    test('ExponentialBackoff.validateOptions: rejects NaN', testExponentialBackoffValidateOptionsRejectsNaN);
+};
+
+await runTests();

test/misc.test.ts

@@ -0,0 +1,70 @@
+import { expect, test, vi } from 'vitest';
+import { tryAsync } from '../source/misc.ts';
+
+/** Spy used to confirm the wrapped async function ran successfully. */
+const successFlagFn = vi.fn();
+
+/** Spy used to confirm the error callback was invoked on failure. */
+const errorFlagFn = vi.fn();
+
+/**
+ * Tests that tryAsync invokes the function and skips the error callback on success.
+ */
+const testTryAsyncCallsFunctionOnSuccess = async (): Promise<void> => {
+    // Reset spies so prior test runs do not affect call counts.
+    vi.clearAllMocks();
+
+    const successFn = async (): Promise<void> => {
+        successFlagFn();
+    };
+
+    await tryAsync(successFn);
+
+    // The wrapped function should run and no error handler should be called.
+    expect(successFlagFn).toHaveBeenCalledOnce();
+    expect(errorFlagFn).not.toHaveBeenCalled();
+};
+
+/**
+ * Tests that tryAsync invokes the error callback when the function throws.
+ */
+const testTryAsyncCallsErrorCallbackOnFailure = async (): Promise<void> => {
+    vi.clearAllMocks();
+
+    const errorFn = async (): Promise<void> => {
+        throw new Error('test');
+    };
+
+    await tryAsync(errorFn, errorFlagFn);
+
+    // The success path should not run; the error callback should receive the failure.
+    expect(successFlagFn).not.toHaveBeenCalled();
+    expect(errorFlagFn).toHaveBeenCalledOnce();
+};
+
+/**
+ * Tests that tryAsync wraps non-Error throws in Error instances before calling the error callback.
+ */
+const testTryAsyncConvertsNonErrorThrows = async (): Promise<void> => {
+    vi.clearAllMocks();
+
+    const errorFn = async (): Promise<void> => {
+        /* eslint-disable-next-line */
+        throw 'test';
+    };
+
+    await tryAsync(errorFn, errorFlagFn);
+
+    // Non-Error throws must be normalized to Error before onError is called.
+    expect(successFlagFn).not.toHaveBeenCalled();
+    expect(errorFlagFn).toHaveBeenCalledOnce();
+    expect(errorFlagFn).toHaveBeenCalledWith(new Error('test'));
+};
+
+const runTests = async (): Promise<void> => {
+    test('tryAsync: calls the function and skips the error callback on success', testTryAsyncCallsFunctionOnSuccess);
+    test('tryAsync: calls the error callback when the function fails', testTryAsyncCallsErrorCallbackOnFailure);
+    test('tryAsync: converts non-Error throws to Error instances', testTryAsyncConvertsNonErrorThrows);
+};
+
+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();
+    }
+}

test/sse-session/sse-session.test.ts

@@ -0,0 +1,633 @@
+import { expect, test, vi, type Mock } from 'vitest';
+
+import { SSESession } from '../../source/sse-session/sse-session.ts';
+import { ExponentialBackoff } from '../../source/exponential-backoff.ts';
+import type { SSESessionOptions, SSEvent } from '../../source/sse-session/types.ts';
+
+import { SseTestStream, type SseTestStreamOptions } from './helpers/sse-stream.ts';
+import { priceOracleEvents, storageEvents } from './fixtures/events.fixtures.ts';
+import { ExponentialBackoffMaxRetriesHitError, HTTPError, ResponseBodyNullError } from '../../source/errors.ts';
+
+/** URL passed to every session under test. */
+const EVENTS_URL = '/events';
+
+/** Headers required for a valid SSE response in these tests. */
+const SSE_HEADERS = { 'Content-Type': 'text/event-stream' };
+
+type FetchFn = SSESessionOptions['fetch'];
+
+/**
+ * Builds a minimal ExponentialBackoff so reconnect and retry paths finish quickly in tests.
+ * Real production delays would make vi.waitFor-based assertions time out.
+ *
+ * @param maxAttempts - Maximum retry attempts; defaults to 1.
+ */
+const testRetry = (maxAttempts = 1): ExponentialBackoff => {
+    return new ExponentialBackoff({
+        baseDelay: 1,
+        maxDelay: 1,
+        maxAttempts,
+        growthRate: 1,
+        jitter: 0,
+    });
+};
+
+/**
+ * Wraps fixture chunks in a Response backed by {@link SseTestStream}.
+ * SseTestStream simulates a real HTTP body: chunks arrive over time and the stream
+ * can optionally close itself when all chunks are sent.
+ *
+ * @param chunks - Raw SSE payload strings to stream.
+ * @param options - Optional stream timing and close behavior.
+ */
+const sseFetchResponse = (chunks: string[], options: SseTestStreamOptions = {}): Response => {
+    return new Response(new SseTestStream(chunks, options).stream, {
+        status: 200,
+        headers: SSE_HEADERS,
+    });
+};
+
+/**
+ * Creates a vitest mock fetch that delegates to the given responder.
+ * SSESession requires fetch injection so tests never hit the network.
+ *
+ * @param responder - Function that returns the Response for each fetch call.
+ */
+const createFetchMock = (responder: (url: string, init: RequestInit) => Response | Promise<Response>): Mock<FetchFn> => {
+    return vi.fn(async (url: string, init: RequestInit) => responder(url, init));
+};
+
+/**
+ * Session defaults that disable reconnect noise unless a test opts in.
+ * attemptReconnect: false — transport errors should not auto-retry by default.
+ * persistent: false — server closing the stream should close the message iterator.
+ */
+const defaultSessionOptions: Partial<SSESessionOptions> = {
+    attemptReconnect: false,
+    persistent: false,
+    retry: testRetry(),
+};
+
+/**
+ * Creates an SSESession wired to the injected fetch mock.
+ * SSESession.create immediately calls connect(), so fetch is invoked during creation.
+ *
+ * @param fetch - Mock fetch implementation.
+ * @param options - Per-test session overrides merged on top of defaults.
+ */
+const createSession = async (fetch: FetchFn, options: Partial<SSESessionOptions> = {}): Promise<SSESession> => {
+    return SSESession.create(EVENTS_URL, {
+        ...defaultSessionOptions,
+        onError: vi.fn(),
+        ...options,
+        fetch,
+    });
+};
+
+/**
+ * Runs a callback against a session and always disconnects afterward.
+ * Most tests use this so session lifecycle (connect on create, disconnect on exit)
+ * is consistent and resources are not leaked between cases.
+ *
+ * @param fetch - Mock fetch for the session.
+ * @param options - Session options.
+ * @param run - Test body receiving the connected session.
+ */
+const withSession = async <T>(fetch: FetchFn, options: Partial<SSESessionOptions>, run: (session: SSESession) => Promise<T>): Promise<T> => {
+    const session = await createSession(fetch, options);
+
+    try {
+        return await run(session);
+    } finally {
+        await session.disconnect();
+    }
+};
+
+/**
+ * Reads up to `count` messages from the session's async iterator.
+ * Breaking out of the for-await loop early leaves the underlying stream open,
+ * which is intentional for tests that inspect post-read session state.
+ *
+ * @param session - Connected SSE session.
+ * @param count - Number of events to collect before stopping.
+ */
+const readMessages = async (session: SSESession, count: number): Promise<SSEvent[]> => {
+    const events: SSEvent[] = [];
+
+    for await (const event of session.messages) {
+        events.push(event);
+
+        if (events.length >= count) {
+            break;
+        }
+    }
+
+    return events;
+};
+
+/**
+ * Builds a ReadableStream that emits one chunk then errors after a delay.
+ * Simulates a mid-stream network failure: the client receives partial data,
+ * then the connection drops before the server finishes sending.
+ *
+ * @param raw - SSE payload to enqueue before the error.
+ * @param delayMs - Milliseconds to wait before erroring (gives the parser time to process the chunk).
+ */
+const failingStreamAfter = (raw: string, delayMs = 50): ReadableStream<Uint8Array> => {
+    const encoder = new TextEncoder();
+
+    return new ReadableStream({
+        async start(controller): Promise<void> {
+            controller.enqueue(encoder.encode(raw));
+            await new Promise((resolve) => setTimeout(resolve, delayMs));
+            controller.error(new Error('network failure'));
+        },
+    });
+};
+
+/**
+ * Scenario: a client opens a standard GET SSE connection.
+ * Verifies the session passes the correct fetch options for a spec-compliant SSE request.
+ */
+const testSseSessionConnectCallsFetchWithExpectedOptions = async (): Promise<void> => {
+    // Return one valid storage fixture event so connect succeeds and the stream stays open briefly.
+    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
+
+    try {
+        // withSession → createSession → SSESession.create → connect → fetchMock is called once.
+        await withSession(fetchMock, {}, async () => {
+            expect(fetchMock).toHaveBeenCalledWith(
+                EVENTS_URL,
+                expect.objectContaining({
+                    method: 'GET',
+                    cache: 'no-store',
+                    signal: expect.any(AbortSignal),
+                    headers: expect.objectContaining({
+                        Accept: 'text/event-stream',
+                        'Cache-Control': 'no-cache',
+                    }),
+                }),
+            );
+
+            // While the stream is active the abort signal must not yet be triggered.
+            expect(fetchMock.mock.calls[0]?.[1]?.signal?.aborted).toBe(false);
+        });
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: a client connects and the application wants a lifecycle callback when the stream is ready.
+ * Verifies onConnected fires after the transport is established.
+ */
+const testSseSessionConnectInvokesOnConnected = async (): Promise<void> => {
+    const onConnected = vi.fn();
+    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
+
+    try {
+        await withSession(fetchMock, { onConnected }, async () => {
+            expect(onConnected).toHaveBeenCalledOnce();
+        });
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: a client is already connected and something calls connect() again (e.g. a duplicate init).
+ * Verifies the session does not open a second HTTP transport.
+ */
+const testSseSessionConnectDoesNotOpenSecondTransport = async (): Promise<void> => {
+    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
+
+    try {
+        await withSession(fetchMock, {}, async (session) => {
+            // Pull one event so the first transport is fully established and reading.
+            await readMessages(session, 1);
+
+            // Idempotent connect — should be a no-op at the fetch layer.
+            await session.connect();
+
+            expect(fetchMock).toHaveBeenCalledOnce();
+        });
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: the application attaches auth or other headers via onRequest before each fetch.
+ * Verifies mutations from onRequest reach the actual fetch call.
+ */
+const testSseSessionConnectPassesOnRequestMutations = async (): Promise<void> => {
+    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
+
+    try {
+        await withSession(
+            fetchMock,
+            {
+                // onRequest runs during connect and can rewrite headers/body before fetch sees them.
+                onRequest: async (request) => ({
+                    ...request,
+                    headers: { ...request.headers, Authorization: 'Bearer test-token' },
+                }),
+            },
+            async () => {
+                expect(fetchMock).toHaveBeenCalledWith(
+                    EVENTS_URL,
+                    expect.objectContaining({
+                        headers: expect.objectContaining({
+                            Authorization: 'Bearer test-token',
+                        }),
+                    }),
+                );
+            },
+        );
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: some SSE endpoints require POST with a form body instead of a plain GET.
+ * Verifies method and body are forwarded to fetch.
+ */
+const testSseSessionConnectSendsPostBody = async (): Promise<void> => {
+    const body = new FormData();
+    body.set('topic', 'prices');
+
+    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
+
+    try {
+        await withSession(
+            fetchMock,
+            {
+                method: 'POST',
+                body,
+            },
+            async () => {
+                expect(fetchMock).toHaveBeenCalledWith(
+                    EVENTS_URL,
+                    expect.objectContaining({
+                        method: 'POST',
+                        body,
+                    }),
+                );
+            },
+        );
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: a single long-lived connection delivers many events from different domains (LLM, oracle, storage).
+ * Verifies the parser and session deliver every fixture event in order through one stream.
+ */
+const testSseSessionDeliversMultipleFixtureEvents = async (): Promise<void> => {
+    const fixtures = [ ...priceOracleEvents, ...storageEvents ];
+    const expected = fixtures.flatMap(({ parsed }) => parsed ?? []);
+
+    // All raw payloads are concatenated into one SseTestStream response.
+    const fetchMock = createFetchMock(() => sseFetchResponse(fixtures.map(({ raw }) => raw)));
+
+    try {
+        await withSession(fetchMock, {}, async (session) => {
+            const received = await readMessages(session, expected.length);
+
+            expect(received).toEqual(expected);
+        });
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: application code listens via session.on('message') instead of the async iterator.
+ * Verifies the EventEmitter path receives the same parsed events as the iterator.
+ */
+const testSseSessionEmitsMessageEvents = async (): Promise<void> => {
+    const { raw, parsed } = storageEvents[0]!;
+    const emitted: SSEvent[] = [];
+
+    // ':\n\n' is an SSE comment/heartbeat; chunkDelayMs forces it to arrive as a separate chunk
+    // so incremental parsing is exercised before the real data frame.
+    const fetchMock = createFetchMock(() => sseFetchResponse([ ':\n\n', raw ], { chunkDelayMs: 50 }));
+
+    // Use createSession directly (not withSession) so we control disconnect timing in finally.
+    const session = await createSession(fetchMock, {});
+
+    try {
+        session.on('message', (event) => emitted.push(event));
+
+        await readMessages(session, 1);
+
+        expect(emitted).toEqual([ parsed![0] ]);
+    } finally {
+        await session.disconnect();
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: network chunks split an SSE frame at an arbitrary byte boundary.
+ * Verifies the internal parser buffers partial data and still emits a complete event.
+ */
+const testSseSessionParsesEventSplitAcrossChunks = async (): Promise<void> => {
+    const { raw, parsed } = storageEvents[0]!;
+    const mid = Math.floor(raw.length / 2);
+
+    // SseTestStream sends each array element as a separate chunk — the event is cut in half.
+    const fetchMock = createFetchMock(() => sseFetchResponse([ raw.slice(0, mid), raw.slice(mid) ]));
+
+    try {
+        await withSession(fetchMock, {}, async (session) => {
+            const [ event ] = await readMessages(session, 1);
+
+            expect(event).toEqual(parsed![0]);
+        });
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: server closes the stream after one event and the client is not in persistent mode.
+ * Verifies the message iterator closes and onDisconnected fires.
+ */
+const testSseSessionClosesOnServerCloseWhenNotPersistent = async (): Promise<void> => {
+    // closeWhenDone: true makes SseTestStream end the body after sending the chunk.
+    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true }));
+    const disconnected = vi.fn();
+
+    try {
+        // persistent defaults to false — terminal server close should shut down messages.
+        await withSession(fetchMock, { onDisconnected: disconnected }, async (session) => {
+            await readMessages(session, 1);
+
+            await vi.waitFor(() => expect(session.messages.closed).toBe(true));
+            expect(disconnected).toHaveBeenCalled();
+        });
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: server closes the stream, the message iterator closes, then the app manually reconnects.
+ * Verifies a second fetch opens and events flow again (non-persistent manual reconnect path).
+ */
+const testSseSessionReconnectsAfterTerminalClose = async (): Promise<void> => {
+    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true }));
+
+    // Override the default responder: first fetch closes after one event, second stays open.
+    fetchMock
+        .mockResolvedValueOnce(sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true }))
+        .mockResolvedValueOnce(sseFetchResponse([ storageEvents[0]!.raw ]));
+
+    try {
+        await withSession(fetchMock, {}, async (session) => {
+            await readMessages(session, 1);
+
+            await vi.waitFor(() => expect(session.messages.closed).toBe(true));
+
+            // App-initiated reconnect after the first stream ended.
+            await session.connect();
+
+            const [ event ] = await readMessages(session, 1);
+            expect(event).toEqual(storageEvents[0]!.parsed![0]);
+            expect(fetchMock).toHaveBeenCalledTimes(2);
+        });
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: persistent client — server closes the stream but the session should auto-reconnect.
+ * Verifies two fetch calls and events from both streams arrive on the same message iterator.
+ */
+const testSseSessionPersistentReconnectsOnServerClose = async (): Promise<void> => {
+    const fetchMock = createFetchMock(() => sseFetchResponse([ priceOracleEvents[0]!.raw ], { closeWhenDone: true }));
+
+    fetchMock
+        .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[0]!.raw ], { closeWhenDone: true }))
+        .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[1]!.raw ]));
+
+    try {
+        await withSession(
+            fetchMock,
+            {
+                persistent: true,
+                retry: testRetry(2),
+            },
+            async (session) => {
+                const [ first, second ] = await readMessages(session, 2);
+
+                expect(first).toEqual(priceOracleEvents[0]!.parsed![0]);
+                expect(second).toEqual(priceOracleEvents[1]!.parsed![0]);
+                await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
+            },
+        );
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: mid-stream network error with attemptReconnect enabled.
+ * Verifies the session fetches again and the iterator continues delivering events.
+ */
+const testSseSessionReconnectsOnTransportError = async (): Promise<void> => {
+    const fetchMock = createFetchMock(() => sseFetchResponse([ priceOracleEvents[0]!.raw ]));
+
+    fetchMock
+        // First connection: delivers one event then the stream errors (failingStreamAfter).
+        .mockResolvedValueOnce(new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }))
+        // Second connection: clean stream with the next fixture event.
+        .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[1]!.raw ]));
+
+    try {
+        await withSession(
+            fetchMock,
+            {
+                attemptReconnect: true,
+                retry: testRetry(2),
+            },
+            async (session) => {
+                const [ beforeError, recovered ] = await readMessages(session, 2);
+
+                expect(beforeError).toEqual(priceOracleEvents[0]!.parsed![0]);
+                expect(recovered).toEqual(priceOracleEvents[1]!.parsed![0]);
+                await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
+            },
+        );
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: mid-stream network error with attemptReconnect disabled (default).
+ * Verifies the session terminates: messages close, onError runs, and an error event is emitted.
+ */
+const testSseSessionClosesOnTransportErrorWhenNotReconnecting = async (): Promise<void> => {
+    const onError = vi.fn();
+    const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS });
+    const fetchMock = createFetchMock(() => response);
+
+    try {
+        await withSession(fetchMock, { onError }, async (session) => {
+            const errorEvent = new Promise<Error>((resolve) => session.once('error', resolve));
+
+            await vi.waitFor(() => expect(session.messages.closed).toBe(true));
+            expect(onError).toHaveBeenCalled();
+            await expect(errorEvent).resolves.toBeInstanceOf(Error);
+        });
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: the user aborts an active connection (e.g. navigation away) but may reconnect later.
+ * Verifies onDisconnected fires, fetch is aborted, but the message iterator stays open.
+ */
+const testSseSessionAbortEmitsDisconnectedAndKeepsMessagesOpen = async (): Promise<void> => {
+    const onDisconnected = vi.fn();
+    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
+
+    try {
+        await withSession(fetchMock, { onDisconnected }, async (session) => {
+            await readMessages(session, 1);
+            await session.abort();
+
+            expect(onDisconnected).toHaveBeenCalled();
+            expect(session.messages.closed).toBe(false);
+        });
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: the application fully tears down the session (logout, component unmount, etc.).
+ * Verifies disconnect closes the message iterator and emits the closed event.
+ */
+const testSseSessionDisconnectClosesMessagesAndEmitsClosed = async (): Promise<void> => {
+    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
+
+    const session = await createSession(fetchMock, {});
+    const closed = new Promise<void>((resolve) => session.once('closed', () => resolve()));
+
+    try {
+        await session.disconnect();
+
+        expect(session.messages.closed).toBe(true);
+        await expect(closed).resolves.toBeUndefined();
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: the server returns a non-2xx HTTP status (500).
+ * Verifies create rejects, onError is invoked, and the error is a real Error instance.
+ */
+const testSseSessionHttpErrorCallsOnErrorAndThrows = async (): Promise<void> => {
+    const onError = vi.fn();
+    const fetchMock = createFetchMock(() =>
+            new Response('nope', {
+                status: 500,
+                statusText: 'Internal Server Error',
+            }));
+
+    const expectedError = new ExponentialBackoffMaxRetriesHitError([ new HTTPError(500, 'nope') ]);
+
+    try {
+        await expect(createSession(fetchMock, { onError })).rejects.toThrow(expectedError);
+
+        expect(onError).toHaveBeenCalled();
+        expect(onError.mock.calls[0]?.[0]).toBeInstanceOf(Error);
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: fetch returns 200 but with a null body (misconfigured proxy or server bug).
+ * Verifies create rejects because SSE requires a readable stream body.
+ */
+const testSseSessionRejectsWhenResponseBodyIsNull = async (): Promise<void> => {
+    const onError = vi.fn();
+    const fetchMock = createFetchMock(() => new Response(null, { status: 200, headers: SSE_HEADERS }));
+
+    const expectedError = new ExponentialBackoffMaxRetriesHitError([ new ResponseBodyNullError() ]);
+
+    try {
+        await expect(createSession(fetchMock, { onError, retry: testRetry() })).rejects.toThrow(expectedError);
+
+        expect(onError).toHaveBeenCalled();
+    } finally {
+        vi.restoreAllMocks();
+    }
+};
+
+/**
+ * Scenario: SSE spec resume — after receiving an event with an id, reconnect should send Last-Event-ID.
+ * Verifies the second fetch includes the id from the first event (1234 in storageEvents[0]).
+ */
+const testSseSessionSendsLastEventIdOnReconnect = async (): Promise<void> => {
+    const { raw } = storageEvents[0]!;
+    let reconnectHeaders: Record<string, string> | undefined;
+    const fetchMock = createFetchMock(() => sseFetchResponse([ raw ]));
+
+    fetchMock
+        // First connection: heartbeat chunk, then the event (with id: 1234), then server closes.
+        .mockResolvedValueOnce(sseFetchResponse([ ':\n\n', raw ], { chunkDelayMs: 10, closeWhenDone: true }))
+        // Second connection: capture whatever headers the reconnect logic attached.
+        .mockImplementationOnce(async (_url, init) => {
+            reconnectHeaders = init.headers as Record<string, string>;
+
+            return sseFetchResponse([]);
+        });
+
+    const session = await createSession(fetchMock, { persistent: true, retry: testRetry(2) });
+
+    try {
+        // Registers an onRequest hook that copies the last seen event id into reconnect headers.
+        await SSESession.addLastEventIdReconnect(session);
+
+        await readMessages(session, 1);
+
+        await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
+        expect(reconnectHeaders?.['Last-Event-ID']).toBe('1234');
+    } finally {
+        await session.disconnect();
+        vi.restoreAllMocks();
+    }
+};
+
+const runTests = async (): Promise<void> => {
+    test('SSESession.connect: calls injected fetch with method, headers, and abort signal', testSseSessionConnectCallsFetchWithExpectedOptions);
+    test('SSESession.connect: invokes onConnected when the stream is established', testSseSessionConnectInvokesOnConnected);
+    test('SSESession.connect: does not open a second transport when connect is called again', testSseSessionConnectDoesNotOpenSecondTransport);
+    test('SSESession.connect: passes request mutations from onRequest to fetch', testSseSessionConnectPassesOnRequestMutations);
+    test('SSESession.connect: sends POST bodies for POST-based SSE endpoints', testSseSessionConnectSendsPostBody);
+    test('SSESession: delivers multiple fixture events through a single session', testSseSessionDeliversMultipleFixtureEvents);
+    test('SSESession: emits message events for incoming SSE frames', testSseSessionEmitsMessageEvents);
+    test('SSESession: parses a fixture event split across chunk boundaries', testSseSessionParsesEventSplitAcrossChunks);
+    test('SSESession: closes messages and emits disconnected when persistent is false', testSseSessionClosesOnServerCloseWhenNotPersistent);
+    test('SSESession: opens a new message stream after reconnecting following a terminal close', testSseSessionReconnectsAfterTerminalClose);
+    test('SSESession: reconnects when the server closes the stream and persistent is true', testSseSessionPersistentReconnectsOnServerClose);
+    test('SSESession: reconnects when attemptReconnect is true', testSseSessionReconnectsOnTransportError);
+    test('SSESession: closes messages and emits error when attemptReconnect is false', testSseSessionClosesOnTransportErrorWhenNotReconnecting);
+    test('SSESession.abort: aborts fetch, emits disconnected, and keeps messages open', testSseSessionAbortEmitsDisconnectedAndKeepsMessagesOpen);
+    test('SSESession.disconnect: closes messages and emits closed', testSseSessionDisconnectClosesMessagesAndEmitsClosed);
+    test('SSESession: calls onError, emits error, and throws from create on HTTP error', testSseSessionHttpErrorCallsOnErrorAndThrows);
+    test('SSESession: rejects when the response body is null', testSseSessionRejectsWhenResponseBodyIsNull);
+    test('SSESession: sends Last-Event-ID on reconnect after receiving an event with an id', testSseSessionSendsLastEventIdOnReconnect);
+};
+
+await runTests();