# How OmniRoute Handles SSE Streaming Responses: A Deep Dive into the Transform Pipeline

> Discover how OmniRoute's TransformStream pipeline handles SSE streaming, including PII sanitization, progress tracking, and keep-alives for efficient client delivery.

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

---

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

OmniRoute's SSE implementation, as built in the `diegosouzapw/OmniRoute` repository, treats streaming responses as a chain of modular transformations rather than a single pass-through. This architecture allows feature flags, client preferences, and provider formats to be handled elegantly without coupling concerns.

## The Five-Stage SSE Streaming Pipeline

The core pipeline is assembled in **[`open-sse/handlers/chatCore/streamingPipeline.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chatCore/streamingPipeline.ts)**. Each stage is a `TransformStream` that can be conditionally injected based on configuration and request headers.

### Stage 1: Base Provider Stream Connection

All SSE flows begin with `pipeWithDisconnect`, a utility that bridges the raw upstream HTTP response to a controllable stream. The handler receives the provider's `ReadableStream` and wraps it with an `AbortController` so client disconnects propagate cleanly.

This foundation ensures that backpressure and cancellation work correctly regardless of which transformations follow.

### Stage 2: PII Sanitization (Optional)

When the `PII_RESPONSE_SANITIZATION` environment feature flag is enabled, the pipeline injects `createPiiSseTransform` from **`@/lib/streamingPiiTransform`**. For custom deployments, callers may supply their own PII transform function via the `createPiiTransform` parameter.

The sanitization runs line-by-line on SSE `data:` payloads, parsing JSON, scrubbing detected fields, and re-serializing—introducing minimal latency due to streaming processing.

### Stage 3: Progress Tracking (Opt-In)

Clients request progress events by sending the header `OmniRoute-Progress: on`. When detected, **[`open-sse/utils/progressTracker.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/progressTracker.ts)** provides `createProgressTransform`, which emits synthetic SSE messages containing metadata about stream advancement.

The pipeline also mutates the response headers to include `omni-progress: enabled`, confirming the feature activation to the caller.

### Stage 4: Heartbeat Keep-Alive

Long-running LLM streams risk proxy timeouts. OmniRoute solves this through **[`open-sse/utils/sseHeartbeat.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/sseHeartbeat.ts)** and its `createSseHeartbeatTransform` function.

Key configuration via `shapeForClientFormat`:

| Shape | Use Case | Output Format |
|-------|----------|---------------|
| `comment` | Generic SSE clients | `:keep-alive\n\n` |
| `anthropic-ping` | Anthropic SDK compatibility | `{type: "ping"}` |
| `openai-chunk` | OpenAI streaming format | `{object: "chat.completion.chunk", ..., choices: []}` |
| `openai-responses-in-progress` | Responses API | `{status: "in_progress"}` |

Default heartbeat interval is **15 seconds** (`SSE_HEARTBEAT_INTERVAL_MS`).

### Stage 5: Model Echo (Feature #1311)

When `echoModel` is provided in the request, `createModelEchoTransform` from **[`open-sse/services/responseModelEcho.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/services/responseModelEcho.ts)** becomes the final stage. This transform rewrites the `model` field in every SSE `data:` line to match the alias the client originally requested, masking internal routing decisions.

Example transformation:

```json
// Original upstream
{"id":"cmpl-abc","model":"internal-gpt-4-turbo","choices":[...]}

// After model-echo with alias "gpt-5.5-alias"
{"id":"cmpl-abc","model":"gpt-5.5-alias","choices":[...]}

```

## Implementation Example: Assembling the Pipeline

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

const pipeline = assembleStreamingPipeline({
  providerResponse,           // raw fetch response body (ReadableStream)
  transformStream,            // JSON-to-SSE transformer for the provider
  streamController,           // AbortController for client disconnects
  createPiiTransform: null,   // feature flag controls PII
  clientRawRequestHeaders,    // inspected for OmniRoute-Progress
  clientResponseFormat: 'openai',
  echoModel: 'gpt-5.5-alias', // optional model alias override
  responseHeaders: {},        // mutated with progress flag if enabled
});

// Returns ReadableStream ready for Response construction
return new Response(pipeline, {
  headers: {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    ...responseHeaders,
  },
});

```

## Responses API: Extended SSE Handling

The newer OpenAI-compatible Responses API endpoint (**[`open-sse/handlers/responsesHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/responsesHandler.ts)**) builds upon the chat-core pipeline with additional transformation:

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

const transformed = result.response.body
  .pipeThrough(createResponsesApiTransformStream(null, undefined, { customToolNames }))
  .pipeThrough(createSseHeartbeatTransform({
    shape: HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS,
  }));

```

This converts the Chat Completions format to Responses API format before applying the same heartbeat infrastructure.

## Performance Instrumentation

Every pipeline stage is wrapped with `performance.mark` and `performance.measure` calls. Developers can observe `omni-pipeline` measurements in browser DevTools or server telemetry to identify latency bottlenecks between stages.

## Summary

- **Modular architecture**: Five composable `TransformStream` stages in **[`streamingPipeline.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/streamingPipeline.ts)** handle all SSE concerns.
- **Security-first**: PII sanitization is feature-flagged and runs inline without buffering the entire response.
- **Client-aware**: Heartbeat shapes adapt to Anthropic, OpenAI, or generic SSE consumers automatically.
- **Observable**: Built-in `performance.mark` instrumentation enables pipeline latency analysis.
- **Extensible**: Model echo and custom PII transforms allow deployment-specific customization without core changes.

## Frequently Asked Questions

### How does OmniRoute prevent proxy timeouts on long LLM streams?

OmniRoute injects configurable heartbeat keep-alives via `createSseHeartbeatTransform` in **[`open-sse/utils/sseHeartbeat.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/sseHeartbeat.ts)**. By default, a heartbeat fires every 15 seconds (`SSE_HEARTBEAT_INTERVAL_MS`), with the format automatically selected based on the client's expected response type.

### Can clients opt out of PII sanitization in OmniRoute SSE streams?

PII sanitization is controlled server-side by the `PII_RESPONSE_SANITIZATION` feature flag; clients cannot directly disable it. However, deployers can supply a custom PII transform or `null` to bypass sanitization entirely when assembling the pipeline.

### What happens when a client disconnects mid-stream in OmniRoute?

The pipeline's `pipeWithDisconnect` utility propagates the `AbortController` signal through all `TransformStream` stages. Upstream fetch requests cancel immediately, preventing wasted compute on responses that will never be consumed.

### How does OmniRoute's Responses API differ from standard chat streaming for SSE?

The Responses API handler (**[`responsesHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/responsesHandler.ts)**) first pipes the chat-core stream through `createResponsesApiTransformStream` to reshape the JSON structure, then applies the standard heartbeat transform with the `OPENAI_RESPONSES_IN_PROGRESS` shape. This maintains format compatibility while reusing the core pipeline infrastructure.