How the OmniRoute Responses API Transformer Converts OpenAI‑style SSE Streams into the Codex Responses API Format

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

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

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)

function normalizeResponsesUsage(
  incoming: Record<string, any>,
  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):

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:

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

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:

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:

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.

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