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

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.

// 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

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:

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

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

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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}