# How to Handle SSE Streaming with Abort Signals in OmniRoute: A Complete Guide

> Master SSE streaming with AbortSignals in OmniRoute. Learn to implement client-initiated cancellation, idle timeouts, and HTTP error codes for robust Server-Sent Event handling.

- Repository: [Diego Rodrigues de Sa e Souza/OmniRoute](https://github.com/diegosouzapw/OmniRoute)
- Tags: how-to-guide
- Published: 2026-08-14

---

**Use `createSSEStream` with an `AbortSignal` to enable client-initiated cancellation, automatic idle timeouts, and proper HTTP error codes (499/504) for aborted Server-Sent Event streams in OmniRoute.**

OmniRoute provides a robust pipeline for handling Server-Sent Events (SSE) streaming with built-in support for **abort signals**, **idle timeouts**, and **provider-side failure handling**. This article explains how the `createSSEStream` transform in [`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts) manages request lifecycle and cancellation, with practical code examples you can use today.

## Core Components of SSE Abort Handling

### The `createSSEStream` Transform

The heart of OmniRoute's SSE handling is the **`createSSEStream`** transform, located in [[`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts)](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/utils/stream.ts#L6222-LL6350) (lines 6222–6350). This factory function returns a transform stream that:

- Wraps upstream fetch responses
- Monitors for idle timeouts
- Respects external `AbortSignal` instances
- Handles provider error events
- Manages cleanup of pending requests

## Client-Initiated Abort Patterns

### Passing an AbortSignal to Chat Handlers

Handlers in [[`open-sse/handlers/chat.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chat.ts)](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/handlers/chat.ts) forward the caller's `AbortSignal` directly to the underlying `fetch`. Here's how to use it:

```typescript
import { fetchChatCompletions } from "@/open-sse/handlers/chat.ts";

// 1️⃣ Create a request-level AbortSignal with timeout
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), 5_000);

// 2️⃣ Pass the signal into the handler
await fetchChatCompletions({
  model: "gpt-4o-mini",
  messages: [{ role: "user", content: "Explain abort handling." }],
  signal: controller.signal,            // <- forwarded to fetch
  onComplete: (payload) => console.log("✅ done", payload),
  onFailure:  (err) => console.warn("❌ failure", err),
});

clearTimeout(timeout);

```

### HTTP Status Codes for Aborted SSE Streams

OmniRoute distinguishes between two abort scenarios based on timing:

| Scenario | HTTP Status | Test Reference |
|----------|-------------|----------------|
| Abort **before** any data received | **499** | [`tests/unit/t3-chat-web.test.ts:312`](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/tests/unit/t3-chat-web.test.ts) |
| Abort **after** partial response | **504** | [`tests/unit/tryBackedChat.test.ts:137`](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/tests/unit/tryBackedChat.test.ts) |

The 499 status indicates a clean client cancellation, while 504 signals that a gateway timeout occurred with partial data already streamed.

## Server-Side Timeout Protection

### Upstream Body Timeout with `withBodyTimeout`

The helper function **`withBodyTimeout`** (lines [88–101](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/utils/stream.ts#L88-L101)) races the fetch body promise against a configurable timeout:

```typescript
import { withBodyTimeout } from "@/open-sse/utils/stream.ts";

const bodyPromise = fetch(url, { signal }).then(r => r.text());

// Timeout body read after 8 seconds
const safeBody = await withBodyTimeout(bodyPromise, 8_000);
throws new BodyTimeoutError();           // if timeout fires

```

### Idle Stream Watchdog

When `createSSEStream` starts, it creates an **interval timer** (lines [148–155](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/utils/stream.ts#L148-LL155)) that monitors `STREAM_IDLE_TIMEOUT_MS` (default 30s). If no chunk arrives in time:

1. The stream aborts internally
2. A warning is logged
3. The optional `onFailure` callback receives a synthetic `stream_idle_timeout` payload (lines [151–182](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/utils/stream.ts#L151-LL182))

```typescript
{
  onFailure: (failure) => {
    if (failure.code === "stream_idle_timeout") {
      console.error("No data from provider for 30s");
    }
  }
}

```

## Handling Provider-Side Failures

### Rate Limits and Usage Limits

`createSSEStream` listens for `response.failed` events from providers. These are forwarded to **`onFailure`**, where you can decide whether to swallow the error or let the stream error out:

```typescript
await readTransformed(
  [ `data: ${JSON.stringify({ 
      type: "response.failed", 
      response: { error: { code: "rate_limit_exceeded" } }
    })}\n\n` ],
  {
    mode: "translate",
    sourceFormat: "openai",
    targetFormat: "openai_responses",
    provider: "codex",
    model: "gpt-5.5",
    onFailure: (failure) => {
      if (failure.code === "rate_limit_exceeded") {
        console.warn("Provider throttled:", failure.message);
        return true;  // swallow error, don't emit generic error
      }
      return false;   // let stream error out
    },
  }
);

```

Tests demonstrating this behavior:
- **"translate mode aborts on Responses failure with rate limit error"** — lines 1270–1278
- **"passthrough aborts on Responses usage-limit failures and reports 429"** — lines 2109–2145

## Cleanup and Request Bookkeeping

### Pending Request Tracking

The transform tracks **`pendingRequestClearedFromStream`** and clears the pending-request entry in the usage database when:

- The stream ends normally
- The stream aborts (client or idle timeout)
- An upstream error occurs

This prevents dangling `usage` rows in the database (lines 149–155 in `createSSEStream`).

## Format-Specific SSE Behavior

### Conditional `[DONE]` Terminator

Not all clients expect the `data: [DONE]` terminator. `createSSEStream` computes **`shouldEmitDoneTerminator`** (lines [698–699](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/utils/stream.ts#L698-L699)) based on target format:

| Target Format | Emits `[DONE]`? |
|-------------|-----------------|
| OpenAI Chat | Yes |
| OpenAI Responses | No |
| Claude | No |
| Antigravity | Yes |

### Metadata Comments

When `OMNIROUTE_SSE_COMMENTS` is enabled, **`emitFinalSseMetadata`** (lines [1036–1054](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/utils/stream.ts#L1036-LL1054)) emits a final metadata comment:

```

: x-omniroute-request-id=req_abc123; x-omniroute-tokens-used=847

```

This is omitted for strict JSON-parsing clients.

## Testing Abort Signal Handling

### Unit Test: External Abort Before Any Chunk

From [[`tests/unit/tryBackedChat.test.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/tests/unit/tryBackedChat.test.ts)](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/tests/unit/tryBackedChat.test.ts):

```typescript
it("returns 504 when external AbortSignal is already aborted", async () => {
  const ac = new AbortController();
  ac.abort(); // abort immediately

  await assert.rejects(
    chatCore({
      model: "gpt-4o-mini",
      messages: [{ role: "user", content: "ping" }],
      signal: ac.signal,                 // passed straight to executor
    }),
    /Aborted|504/                         // handler translates to 504
  );
});

```

### Abort Mid-Stream Test

From [[`tests/unit/t3-chat-web.test.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/tests/unit/t3-chat-web.test.ts)](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/tests/unit/t3-chat-web.test.ts):

```typescript
it("execute: AbortSignal abort → returns 499", async () => {
  const controller = new AbortController();
  
  const promise = executeChat({
    signal: controller.signal,
    // ...
  });
  
  controller.abort();                     // abort after execution starts
  
  const result = await promise;
  expect(result.status).toBe(499);        // client closed request
});

```

## Key Files Reference

| File | Purpose |
|------|---------|
| [[`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts)](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/utils/stream.ts) | Core SSE transform, idle watchdog, abort handling, `[DONE]` logic |
| [[`open-sse/utils/streamErrorFormat.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamErrorFormat.ts)](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/utils/streamErrorFormat.ts) | Normalizes provider error payloads |
| [[`open-sse/handlers/chat.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chat.ts)](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/open-sse/handlers/chat.ts) | Bridges API routes to executor, forwards signals |
| [[`tests/unit/stream-utils.test.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/tests/unit/stream-utils.test.ts)](https://github.com/diegosouzapw/OmniRoute/blob/release/v3.8.50/tests/unit/stream-utils.test.ts) | Idle timeout, rate-limit, usage-limit, terminator tests |

## Summary

- **Use `AbortController`** to enable client-initiated cancellation in OmniRoute SSE streams
- **Expect HTTP 499** for aborts before data arrives, **504** for aborts after partial response
- **Configure `withBodyTimeout`** to protect against slow upstream body reads
- **Rely on the idle watchdog** (default 30s) to detect stalled provider connections
- **Implement `onFailure`** to handle rate limits, usage limits, and idle timeouts gracefully
- **Pass signals through handlers** in [`open-sse/handlers/chat.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chat.ts) for automatic propagation to fetch

## Frequently Asked Questions

### How do I cancel an in-progress SSE request in OmniRoute?

Create an `AbortController`, pass its `signal` property to any OmniRoute handler that accepts it (such as `fetchChatCompletions`), then call `controller.abort()` when you need to cancel. The handler forwards this signal to the underlying `fetch`, and the SSE pipeline handles cleanup automatically.

### What HTTP status code does OmniRoute return when an SSE stream is aborted?

It depends on timing. If the abort occurs **before any data is received**, OmniRoute returns **HTTP 499** (client closed request). If the abort happens **after partial data has streamed**, it returns **HTTP 504** (gateway timeout). This distinction helps callers understand whether to retry or treat the request as definitively cancelled.

### How does OmniRoute handle provider-side rate limits during SSE streaming?

The `createSSEStream` transform recognizes `response.failed` events from providers and forwards them to your `onFailure` callback. You can inspect the error code (e.g., `rate_limit_exceeded`) and return `true` to swallow the error gracefully, or `false` to let the stream error out. The tests in [`stream-utils.test.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/stream-utils.test.ts) demonstrate both patterns.