How OmniRoute Handles SSE Streaming Responses: A Technical Deep Dive

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, 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:

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). 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) 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.

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, 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.

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

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.
• 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.
• Model-echo (feature #1311) rewrites model names in the stream using createModelEchoTransform from open-sse/services/responseModelEcho.ts.
• Responses API endpoints receive additional format transformation in 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 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.

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