# How the OmniRoute Responses API Transformer Converts OpenAI‑style SSE Streams into the Codex Responses API Format > Discover how the OmniRoute Responses API transformer seamlessly converts OpenAI SSE streams into the Codex Responses API format. Learn about its multi-stage pipeline. - Repository: [Diego Rodrigues de Sa e Souza/OmniRoute](https://github.com/diegosouzapw/OmniRoute) - Tags: deep-dive - Published: 2026-08-19 --- **The OmniRoute Responses API transformer is a custom `TransformStream` that consumes OpenAI‑compatible Chat Completions SSE streams and emits the Codex Responses API format through a multi‑stage pipeline in `open‑sse/transformer/responsesTransformer.ts`.** The **OmniRoute** repository implements a translation layer that bridges different large‑language‑model API formats. At the heart of this system sits the **Responses API transformer**—a specialized stream processor that rewrites real‑time server‑sent events (SSE) from one schema to another without buffering entire responses. This article examines how that transformer converts OpenAI‑style Chat Completions streams into the Codex Responses API format, drawing directly from the source code implementation. ## Where the Transformer Lives The implementation resides in a single, focused module: - **File path:** `open‑sse/transformer/responsesTransformer.ts` - **Core class:** A `TransformStream` subclass with custom `start`, `transform`, `flush`, and `cancel` handlers - **Lines 62‑85:** Stream initialization and keep‑alive timer setup This location follows the project's convention of placing SSE‑related adapters under the `open‑sse/transformer/` directory, separating transformation concerns from transport and routing logic. ## Stage 1: Stream Setup and Keep‑Alive Management The transformer initializes with three critical components: 1. **Persistent `TextDecoder`** — Maintains state across chunks to handle multi‑byte UTF‑8 characters that might split across TCP packets 2. **Keep‑alive timer** — Periodically writes `: keepalive` SSE comments to prevent client timeouts during slow generation 3. **Internal state object** — Tracks whether the stream has started, accumulated usage statistics, and partial content buffers ```typescript // Conceptual structure based on lines 62-85 new TransformStream({ start(controller) { this.decoder = new TextDecoder(); this.keepAliveInterval = setInterval(() => { controller.enqueue(': keepalive\n\n'); }, 15000); this.state = { started: false, usage: null, buffer: '' }; }, transform(chunk, controller) { /* ... */ }, flush(controller) { /* ... */ }, cancel() { clearInterval(this.keepAliveInterval); } }) ``` The keep‑alive mechanism demonstrates a production‑ready consideration: many HTTP clients and proxies terminate idle connections after 30‑60 seconds, so the transformer proactively signals liveness without emitting spurious data events. ## Stage 2: Chunk Parsing and SSE Message Extraction The `transform` method (lines 85‑95) processes raw bytes into discrete SSE messages through a precise pipeline: ```typescript transform(chunk: Uint8Array, controller: TransformStreamDefaultController) { // Decode with state preservation for split multi-byte characters const text = this.decoder.decode(chunk, { stream: true }); // Optional debug logging this.logger?.debug('[ResponsesTransformer] raw chunk', { bytes: chunk.length }); // Split on SSE message boundaries const messages = text.split('\n\n'); for (const message of messages) { if (!message.startsWith('data: ')) continue; const payload = message.slice(6); // Remove "data: " prefix if (payload === '[DONE]') continue; // OpenAI stream terminator try { const json = JSON.parse(payload); this.processMessage(json, controller); } catch (e) { this.logger?.warn('Failed to parse SSE payload', { payload }); } } } ``` Key parsing decisions visible in this stage: - **Streaming decode:** The `stream: true` option ensures that partial UTF‑8 sequences at chunk boundaries accumulate rather than corrupt - **Message delimiting:** SSE mandates double‑newline (`\n\n`) separation; the transformer honors this strictly - **`[DONE]` filtering:** OpenAI streams terminate with this literal string rather than JSON; it is stripped rather than forwarded ## Stage 3: Usage Normalization Across Provider Formats Before emitting any events, the transformer reconciles disparate token‑counting schemes through **`normalizeResponsesUsage`** (lines 34‑60): | Provider field | Codex Responses field | |---------------|----------------------| | `prompt_tokens` | `input_tokens` | | `completion_tokens` | `output_tokens` | | `completion_tokens_details.reasoning_tokens` | `reasoning_tokens` | | `total_tokens` | `total_tokens` (computed if missing) | ```typescript function normalizeResponsesUsage( incoming: Record, accumulated: ResponsesUsage | null ): ResponsesUsage { const normalized: ResponsesUsage = { input_tokens: incoming.prompt_tokens ?? accumulated?.input_tokens ?? 0, output_tokens: incoming.completion_tokens ?? accumulated?.output_tokens ?? 0, reasoning_tokens: incoming.completion_tokens_details?.reasoning_tokens ?? accumulated?.reasoning_tokens ?? 0, }; // Accumulate across chunks rather than replace if (accumulated) { normalized.input_tokens += accumulated.input_tokens; normalized.output_tokens += accumulated.output_tokens; normalized.reasoning_tokens += accumulated.reasoning_tokens; } return normalized; } ``` This normalization is **incremental**: usage objects may arrive attached to any chunk in the stream, and the transformer maintains running totals rather than waiting for a final summary. This design supports providers that emit usage statistics at irregular intervals or split across multiple chunks. ## Stage 4: First‑Chunk Bootstrap Events The Codex Responses API requires explicit lifecycle events that OpenAI streams omit. The transformer synthesizes these on first data encounter (lines 332‑349): ```typescript private processFirstChunk( json: any, controller: TransformStreamDefaultController ): void { if (this.state.started) return; this.state.started = true; const responseId = `resp_${generateId()}`; const createdAt = Math.floor(Date.now() / 1000); // Emit response.created to establish the response record controller.enqueue(formatSse({ type: 'response.created', response: { id: responseId, created_at: createdAt, status: 'in_progress', model: json.model, // ... additional metadata } })); // Immediately transition to in_progress state controller.enqueue(formatSse({ type: 'response.in_progress', response: { id: responseId, created_at: createdAt, status: 'in_progress' } })); } ``` Two distinct events fire in rapid succession: 1. **`response.created`** — Initializes the response with a generated UUID, timestamp, and model identifier 2. **`response.in_progress`** — Signals that token generation has begun This dual‑event pattern matches the Codex protocol's state machine, where "created" and "in_progress" are distinct phases that clients may observe separately. ## Stage 5: Reasoning Content Handling A distinctive feature of the Responses API is explicit support for **reasoning tokens**—intermediate chain‑of‑thought content that models like o1 emit before final answers. The transformer detects and extracts this through delimiter‑based parsing: ``` `\\qquad\\tag` reasoning content `\\qquad\\rag` ``` When such delimited blocks appear in `content` or `delta` fields: ```typescript private extractReasoning(text: string): { reasoning: string; content: string } { const reasoningMatch = text.match(/\\qquad\\tag([\s\S]*?)\\qquad\\rag/); if (!reasoningMatch) { return { reasoning: '', content: text }; } return { reasoning: reasoningMatch[1].trim(), content: text.replace(reasoningMatch[0], '').trim() }; } ``` Extracted reasoning populates the `reasoning` field in `response.output_item.added` events, while stripped content continues to `response.content_part.added` events. This separation allows downstream clients to render reasoning traces distinctly from final answers, or to omit them entirely based on user preferences. ## Stage 6: Content Delta Transformation Standard text generation flows through a mapping from OpenAI's `delta` structure to Codex's `content_part` events: | OpenAI Chat Completions | Codex Responses API | |------------------------|---------------------| | `choices[0].delta.content` | `item.content[0].text` | | `choices[0].delta.role` | `item.role` | | `choices[0].index` | `output_index` | ```typescript private transformDelta( delta: any, responseId: string, outputIndex: number ): ResponsesEvent { const { reasoning, content } = this.extractReasoning(delta.content ?? ''); if (reasoning) { return { type: 'response.output_item.added', output_index: outputIndex, item: { type: 'reasoning', id: `rs_${generateId()}`, content: [{ type: 'reasoning_content', text: reasoning }] } }; } return { type: 'response.content_part.added', output_index: outputIndex, content_index: 0, part: { type: 'output_text', text: content, annotations: [] } }; } ``` Each delta typically generates exactly one Responses event, preserving the streaming granularity that allows clients to render tokens incrementally. ## Stage 7: Stream Termination and Finalization When the source stream ends, the `flush` handler emits mandatory completion events: ```typescript flush(controller: TransformStreamDefaultController) { // Flush any remaining decoder state const final = this.decoder.decode(); if (final) this.processBuffer(final, controller); // Emit completion events if stream progressed past initialization if (this.state.started) { controller.enqueue(formatSse({ type: 'response.completed', response: { id: this.state.responseId, status: 'completed', usage: this.state.usage } })); } // Clean up keep-alive timer clearInterval(this.keepAliveInterval); } ``` The **`response.completed`** event carries the final accumulated usage statistics, ensuring clients receive authoritative token counts even if intermediate chunks lacked usage data. ## Error Handling and Cancellation The transformer implements graceful degradation through its `cancel` handler: ```typescript cancel() { clearInterval(this.keepAliveInterval); // If we had started but not completed, emit response.incomplete if (this.state.started && !this.state.completed) { // Enqueue via alternative mechanism or log for observability this.logger?.warn('Stream cancelled mid-generation', { responseId: this.state.responseId, tokensEmitted: this.state.tokenCount }); } } ``` Cancellation may occur due to client disconnect, upstream errors, or timeout. The transformer cleans up resources without leaking timers or leaving the decoder in an invalid state. ## Summary - **The OmniRoute Responses API transformer** resides in `open‑sse/transformer/responsesTransformer.ts` and implements the `TransformStream` interface for streaming data conversion. - **Seven transformation stages** process OpenAI‑style SSE: stream setup, chunk parsing, usage normalization, first‑chunk bootstrap, reasoning extraction, content delta mapping, and stream finalization. - **`normalizeResponsesUsage`** (lines 34‑60) reconciles provider‑specific token fields into the canonical Codex shape with incremental accumulation. - **Lifecycle synthesis** generates `response.created` and `response.in_progress` events (lines 332‑349) that the source Chat Completions format lacks. - **Reasoning token extraction** uses delimiter parsing to separate chain‑of‑thought content from final answers, populating distinct Responses event types. - **Production‑ready features** include keep‑alive timers, streaming UTF‑8 decoding, and graceful cancellation handling. ## Frequently Asked Questions ### What is the OmniRoute Responses API transformer used for? The transformer enables interoperability between OpenAI‑compatible API endpoints and clients expecting the Codex Responses API format. It runs as a real‑time stream processor, converting Server‑Sent‑Events on‑the‑fly without requiring full response buffering, which makes it suitable for latency‑sensitive applications like chat interfaces. ### How does the transformer handle multi‑byte UTF‑8 characters split across network chunks? It maintains a persistent `TextDecoder` instance with the `stream: true` option across all `transform` calls. This decoder statefully buffers incomplete byte sequences at chunk boundaries, ensuring that characters like emoji or CJK glyphs decode correctly even when split by TCP packetization. ### Why does the transformer emit `: keepalive` SSE comments? Many HTTP clients, proxies, and load balancers terminate idle connections after 30‑60 seconds of inactivity. The keep‑alive timer (configured in lines 62‑85) writes periodic SSE comment lines that signal connection liveness without generating spurious data events that clients would need to filter. ### Can the transformer handle providers that emit usage statistics differently than OpenAI? Yes. The **`normalizeResponsesUsage`** function (lines 34‑60) maps multiple provider conventions—`prompt_tokens`, `input_tokens`, `completion_tokens_details.reasoning_tokens`, and others—into the canonical Codex format. It also accumulates usage incrementally, supporting providers that split statistics across chunks or emit them at non‑standard positions in the stream.