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

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

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

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) builds upon the chat-core pipeline with additional transformation:

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

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