# How OmniRoute Handles SSE Streaming Responses: A Deep Dive Into the TransformStream Pipeline

> Discover how OmniRoute handles SSE streaming responses. Learn about its TransformStream pipeline for PII sanitization, progress tracking, heartbeats, and alias injection.

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

---

**OmniRoute processes SSE (Server‑Sent Events) streaming responses through a composable `TransformStream` pipeline that sanitizes PII, tracks progress, injects heartbeats, and optionally echoes model aliases before returning the final stream to clients.**

The `diegosouzapw/OmniRoute` repository implements a production‑grade SSE streaming architecture for LLM proxying. Rather than passing raw provider responses directly to clients, the system assembles modular transformation stages that handle privacy, observability, and protocol compatibility. This article breaks down exactly how the pipeline works, with direct references to the source implementation.

## The Core Streaming Pipeline Architecture

OmniRoute's SSE handling centers on `assembleStreamingPipeline()` in **`open‑sse/handlers/chatCore/streamingPipeline.ts`**. This function orchestrates multiple `TransformStream` stages into a single processing chain.

The pipeline always begins with the raw provider response and ends with a `ReadableStream` suitable for `Response` construction. Between these endpoints, conditional stages may activate based on feature flags, request headers, and configuration options.

### Pipeline Stage Order

1. **Base transformation** – Provider‑specific JSON‑to‑SSE conversion via `pipeWithDisconnect`
2. **PII sanitization** – Optional redaction stage (feature‑flagged or custom)
3. **Progress tracking** – Periodic progress events when `OmniRoute-Progress: on` is present
4. **Heartbeat keep‑alive** – Configurable ping messages to prevent connection timeouts
5. **Model‑echo rewrite** – Final stage that overwrites `model` fields in SSE data lines

Each stage receives the output stream from the previous stage, creating a clean composable architecture that simplifies testing and modification.

## Stage 1: PII Sanitization Transform

OmniRoute conditionally applies PII redaction through `createPiiSseTransform`. The pipeline checks `PII_RESPONSE_SANITIZATION` to determine which implementation to use.

- If the feature flag is **enabled**: The built‑in sanitizer from `@/lib/streamingPiiTransform` wraps the stream
- If the feature flag is **disabled but a custom transform is provided**: The caller‑supplied `createPiiTransform` function is invoked
- If **neither**: This stage is skipped entirely

This design allows operators to enforce organization‑wide PII policies while still permitting custom sanitization logic for specific routes.

```typescript
// Simplified usage from streamingPipeline.ts
const piiTransform = PII_RESPONSE_SANITIZATION
  ? createPiiSseTransform()
  : (createPiiTransform ? createPiiTransform() : null);

if (piiTransform) {
  stream = stream.pipeThrough(piiTransform);
}

```

## Stage 2: Progress Tracking Transform

When clients need visibility into streaming progress, they send `OmniRoute-Progress: on` in request headers. The pipeline responds by:

- Activating `createProgressTransform` to inject progress events
- Setting the response header `omni‑progress: enabled` so clients can confirm the feature activated

Progress events appear as SSE comments or data lines depending on the client format, allowing UIs to render progress bars without parsing the actual LLM output.

## Stage 3: SSE Heartbeat Keep‑Alive

Long‑running LLM requests risk timing out at load balancers or client HTTP libraries. OmniRoute solves this with `createSseHeartbeatTransform` in **`open‑sse/utils/sseHeartbeat.ts`**.

The heartbeat system has two key behaviors:

| Configuration | Description |
|-------------|-------------|
| `intervalMs` | Defaults to `SSE_HEARTBEAT_INTERVAL_MS` (15 seconds) |
| `shape` | Determined by `shapeForClientFormat()` based on expected client format |

### Heartbeat Shape Variants

The `shapeForClientFormat()` function selects from four heartbeat formats in **`open‑sse/utils/sseHeartbeat.ts`**:

- **`comment`** – Simple SSE comment line (`:heartbeat`)
- **`anthropic‑ping`** – Anthropic‑compatible ping event
- **`openai‑chunk`** – Empty OpenAI‑style delta chunk
- **`openai‑responses‑in‑progress`** – Specialized format for the Responses API

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

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

// Pipe any SSE stream through to inject keep‑alives
const streamWithKeepAlive = rawSseStream.pipeThrough(heartbeat);

```

## Stage 4: Model‑Echo Transform (Feature #1311)

The final optional stage addresses a common proxy scenario: clients request a model alias (e.g., `gpt‑5.5‑alias`) but the upstream provider returns a concrete model name in the SSE stream. OmniRoute's **model‑echo** feature rewrites every `data:` line so the client sees the alias it originally requested.

Implementation resides in **`open‑sse/services/responseModelEcho.ts`**:

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

// Create transform that rewrites "model" fields in each SSE event
const modelEcho = createModelEchoTransform('gpt-5.5-alias');

// Apply as final pipeline stage
finalStream = sseStream.pipeThrough(modelEcho);

```

The transform parses each SSE line as JSON, overwrites the `model` property, and re‑serializes—preserving all other fields exactly.

## Responses API: Extended SSE Handling

OmniRoute's **Responses API** endpoint (**`open‑sse/handlers/responsesHandler.ts`**) builds on the core pipeline with additional transformations. Rather than returning standard Chat Completions SSE, it:

1. Receives the chat‑core response stream
2. Pipes through `createResponsesApiTransformStream` to convert format
3. Applies the heartbeat transform with `HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS`

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

const result = await handleResponsesCore({ /* config */ });

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',
      'Cache-Control': 'no-cache',
    },
  });
}

```

This layered approach allows the Responses API to reuse core pipeline components while applying format‑specific adaptations.

## Performance Instrumentation

Every pipeline stage includes `performance.mark()` and `performance.measure()` calls. Developers can profile the complete transformation latency using the `omni‑pipeline` measurement name.

The instrumentation captures:

- Time from raw stream start to first byte sent to client
- Per‑stage overhead for PII, progress, heartbeat, and model‑echo transforms

## Complete Pipeline Assembly Example

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

const pipeline = assembleStreamingPipeline({
  providerResponse: upstreamFetchResponse.body!,
  transformStream: openAiToSseTransform,
  streamController: abortController,
  createPiiTransform: null, // defer to feature flag
  clientRawRequestHeaders: request.headers,
  clientResponseFormat: 'openai',
  echoModel: 'enterprise-gpt-alias', // enable model‑echo
  responseHeaders: {}, // mutated to add 'omni-progress: enabled' if applicable
});

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

```

## Summary

- **OmniRoute's SSE streaming** uses composable `TransformStream` stages in **`open‑sse/handlers/chatCore/streamingPipeline.ts`**
- **PII sanitization** activates via `PII_RESPONSE_SANITIZATION` feature flag or custom transform injection
- **Progress tracking** enables when clients send `OmniRoute-Progress: on` header
- **Heartbeat keep‑alive** defaults to 15‑second intervals with four shape variants for different client formats
- **Model‑echo** rewrites SSE `model` fields when `echoModel` is provided, implemented in **`open‑sse/services/responseModelEcho.ts`**
- **Responses API** extends the pipeline with `createResponsesApiTransformStream` and specialized heartbeat formatting
- **Performance profiling** is built in via `performance.mark/measure` on the `omni‑pipeline` measurement

## Frequently Asked Questions

### How does OmniRoute prevent SSE connection timeouts?

OmniRoute injects configurable heartbeat messages using `createSseHeartbeatTransform` in **`open‑sse/utils/sseHeartbeat.ts`**. The default interval is 15 seconds (`SSE_HEARTBEAT_INTERVAL_MS`), and the heartbeat format adapts to match the client's expected protocol—whether OpenAI, Anthropic, or standard SSE comments.

### Can I disable PII sanitization for specific requests?

Yes. The pipeline accepts an optional `createPiiTransform` parameter. When set to `null`, the system checks `PII_RESPONSE_SANITIZATION`. You can also pass a custom transform function to apply request‑specific sanitization logic, bypassing the global feature flag entirely.

### What is model‑echo and why would I use it?

Model‑echo (feature #1311) rewrites the `model` field in every SSE `data:` line so clients see the alias they requested rather than the underlying provider's concrete model name. This preserves abstraction layers when using model aliases or routing rules, implemented via `createModelEchoTransform` in **`open‑sse/services/responseModelEcho.ts`**.