# How OmniRoute Handles SSE Streaming Responses: A Technical Deep Dive

> Discover how OmniRoute expertly manages SSE streaming responses using a TransformStream pipeline for PII sanitization, progress tracking, and more.

- Repository: [Diego Rodrigues de Sa e Souza/OmniRoute](https://github.com/diegosouzapw/OmniRoute)
- Tags: deep-dive
- Published: 2026-08-29

---

**OmniRoute processes Server-Sent Events (SSE) through a composable `TransformStream` pipeline that applies PII sanitization, progress tracking, heartbeat keep-alives, and model-echo transformations before delivering data to the client.**

The **OmniRoute** open-source project (available at `diegosouzapw/OmniRoute`) implements a sophisticated streaming architecture for LLM responses. Rather than proxying raw provider streams directly, the system constructs a multi-stage transformation pipeline that enriches the SSE stream with enterprise features like privacy controls and observability.

## The Streaming Pipeline Architecture

OmniRoute's SSE handling centers on **[`open-sse/handlers/chatCore/streamingPipeline.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chatCore/streamingPipeline.ts)**, which orchestrates a chain of `TransformStream` instances. Each stage mutates or augments the stream according to feature flags and client headers, producing a final `ReadableStream` returned as a `Response` with `Content-Type: text/event-stream`.

### Pipeline Assembly

The `assembleStreamingPipeline` function serves as the entry point. It accepts the raw provider response, transformation logic, and configuration options, then wires together the necessary stages:

```typescript
import { assembleStreamingPipeline } from '@/open-sse/handlers/chatCore/streamingPipeline';

const pipeline = assembleStreamingPipeline({
  providerResponse,             // Raw fetch response body (ReadableStream)
  transformStream,              // Provider-specific JSON-to-SSE transformer
  streamController,            // AbortController for client disconnects
  createPiiTransform: null,    // Feature flag determines PII handling
  clientRawRequestHeaders,     // Incoming request headers
  clientResponseFormat: 'openai',
  echoModel: 'gpt-5.5-alias',   // Optional model alias replacement
  responseHeaders: {},          // Mutated to include progress flags
});

```

### Stage 1: Base Provider Transformation

The pipeline begins with `pipeWithDisconnect`, which bridges the raw HTTP response from the upstream LLM into the transformation chain. This stage handles backpressure and client-initiated aborts, ensuring that disconnecting the consumer properly signals cancellation to the provider.

### Stage 2: PII Sanitization

When the `PII_RESPONSE_SANITIZATION` feature flag is enabled, the stream passes through `createPiiSseTransform` imported from `@/lib/streamingPiiTransform`. This stage scrubs personally identifiable information from SSE data lines before they reach the client. Alternatively, callers can supply a custom PII transform function to override the default behavior.

### Stage 3: Progress Tracking

For clients requiring visibility into long-running generations, OmniRoute supports opt-in progress tracking. When the request includes the header `OmniRoute-Progress: on`, the pipeline injects `createProgressTransform` (implemented in **[`open-sse/utils/progressTracker.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/progressTracker.ts)**). This stage emits periodic progress events and sets the response header `omni-progress: enabled` to confirm activation.

### Stage 4: Heartbeat Keep-Alive

To prevent connection timeouts during slow generation, `createSseHeartbeatTransform` (from **[`open-sse/utils/sseHeartbeat.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/sseHeartbeat.ts)**) injects periodic keep-alive signals. The heartbeat interval defaults to **15 seconds** (`SSE_HEARTBEAT_INTERVAL_MS`), but the shape of the heartbeat varies by client format:

- **`comment`**: Standard SSE comment lines (`:ping`)
- **`anthropic-ping`**: Anthropic-compatible ping events
- **`openai-chunk`**: Empty OpenAI-style data chunks
- **`openai-responses-in-progress`**: Responses API in-progress indicators

The `shapeForClientFormat` utility selects the appropriate format based on the client's expected response type.

```typescript
import { createSseHeartbeatTransform, HEARTBEAT_SHAPES } from '@/open-sse/utils/sseHeartbeat';

const heartbeat = createSseHeartbeatTransform({
  intervalMs: 15_000,
  shape: HEARTBEAT_SHAPES.OPENAI_CHUNK,
});
rawStream.pipeThrough(heartbeat);

```

### Stage 5: Model-Echo Transformation

Implemented in **[`open-sse/services/responseModelEcho.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/services/responseModelEcho.ts)**, the optional `createModelEchoTransform` addresses feature request #1311. When the `echoModel` parameter is supplied, this final stage rewrites the `model` field in every SSE `data:` line to show the client-provided alias rather than the underlying provider model name. This allows API consumers to see consistent model identifiers (e.g., `gpt-5.5-alias`) even when the actual inference uses different upstream models.

```typescript
import { createModelEchoTransform } from '@/open-sse/services/responseModelEcho';

const modelEcho = createModelEchoTransform('gpt-5.5-alias');
sseStream.pipeThrough(modelEcho);

```

## Responses API SSE Handling

For OpenAI-compatible Responses API endpoints, OmniRoute applies additional transformations in **[`open-sse/handlers/responsesHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/responsesHandler.ts)**. The `handleResponsesCore` function passes the chat-core response through `createResponsesApiTransformStream` to convert Chat Completions format into Responses API format, then pipes the result through the same heartbeat transform using the `OPENAI_RESPONSES_IN_PROGRESS` shape.

```typescript
import { handleResponsesCore } from '@/open-sse/handlers/responsesHandler';
import { createResponsesApiTransformStream } from '@/open-sse/transformer/responsesTransformer';

const result = await handleResponsesCore({ /* ... */ });
if (result.success && result.response?.body) {
  const transformed = result.response.body
    .pipeThrough(createResponsesApiTransformStream(null, undefined, { customToolNames }))
    .pipeThrough(createSseHeartbeatTransform({
      shape: HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS,
    }));
  return new Response(transformed, { headers: { 'Content-Type': 'text/event-stream' } });
}

```

## Performance Instrumentation

Every pipeline stage includes `performance.mark` and `performance.measure` calls, creating a traceable `omni-pipeline` measurement. Developers can profile the latency introduced by specific transformations (such as PII sanitization or model-echo rewriting) by examining these performance entries in the runtime.

## Summary

- **OmniRoute** constructs SSE streams using composable `TransformStream` stages defined in [`open-sse/handlers/chatCore/streamingPipeline.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chatCore/streamingPipeline.ts).
- **PII sanitization** runs conditionally based on the `PII_RESPONSE_SANITIZATION` feature flag.
- **Progress tracking** activates via the `OmniRoute-Progress: on` header and emits incremental updates.
- **Heartbeat keep-alives** prevent timeouts every 15 seconds by default, with shapes adapted to client formats via [`open-sse/utils/sseHeartbeat.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/sseHeartbeat.ts).
- **Model-echo** (feature #1311) rewrites model names in the stream using `createModelEchoTransform` from [`open-sse/services/responseModelEcho.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/services/responseModelEcho.ts).
- **Responses API** endpoints receive additional format transformation in [`open-sse/handlers/responsesHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/responsesHandler.ts) before SSE delivery.

## Frequently Asked Questions

### How does OmniRoute prevent SSE connection timeouts during slow LLM responses?

OmniRoute injects heartbeat keep-alive signals every 15 seconds (configurable via `SSE_HEARTBEAT_INTERVAL_MS`) using `createSseHeartbeatTransform`. The heartbeat format adapts to the client's expected response type—for example, OpenAI clients receive empty data chunks while Anthropic clients receive ping events—ensuring the connection remains active without breaking client parsers.

### Can OmniRoute sanitize sensitive information from streaming responses?

Yes. When the `PII_RESPONSE_SANITIZATION` feature flag is enabled, the pipeline wraps the stream in `createPiiSseTransform` from `@/lib/streamingPiiTransform`. This scans each SSE data line and redacts personally identifiable information before transmission. Developers can also inject custom PII transforms by passing a `createPiiTransform` function to the pipeline assembly.

### What is the "model-echo" feature in OmniRoute's SSE pipeline?

Model-echo (feature #1311) allows the API to display a user-friendly model alias instead of the actual upstream provider model name. When the `echoModel` parameter is provided to `assembleStreamingPipeline`, the `createModelEchoTransform` stage rewrites the `model` field in every SSE `data:` line. This ensures clients see consistent identifiers like `gpt-5.5-alias` even when the backend routes to different underlying models.

### How does the Responses API endpoint differ from standard chat completions in OmniRoute?

The Responses API handler in [`open-sse/handlers/responsesHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/responsesHandler.ts) first converts the standard chat completion format into the OpenAI Responses API structure using `createResponsesApiTransformStream`. It then applies the standard SSE pipeline including heartbeat transforms, but uses the `OPENAI_RESPONSES_IN_PROGRESS` heartbeat shape to maintain compatibility with OpenAI's streaming expectations.