How to Monitor the OmniRoute Streaming Process: A Complete Guide to SSE Observability

Monitor OmniRoute streaming by polling the /api/monitoring/health endpoint for real-time system metrics and using the onComplete callback in createSSEStream to capture per-request summaries.

OmniRoute handles LLM responses through a custom Server-Sent Events (SSE) pipeline that requires specialized observability techniques. This guide explains how to monitor the OmniRoute streaming process using built-in health endpoints and stream-level hooks, based on the actual implementation in the diegosouzapw/OmniRoute repository.

Understanding the OmniRoute SSE Pipeline

The core streaming logic lives in open-sse/utils/stream.ts, where the createSSEStream function constructs a pipeline that normalizes incoming SSE chunks, extracts tool-call information, and tracks usage tokens. Each stream instance monitors its own health through internal timers and emits lifecycle events through the onComplete callback.

The createSSEStream Entry Point

When you initiate a streaming request, createSSEStream configures the SSE pipeline with provider-specific formatting rules. The function signature accepts an onComplete callback that receives a summary object containing the final status, token usage, and any errors encountered during the stream.

import { createSSEStream, STREAM_MODE, FORMATS } from "@omniroute/open-sse/utils/stream";

const stream = createSSEStream({
  mode: STREAM_MODE.PASSTHROUGH,
  sourceFormat: FORMATS.OPENAI,
  clientResponseFormat: FORMATS.OPENAI,
  provider: "openai",
  connectionId: "oc-123",
  body: { model: "gpt-4o-mini", stream: true },
  onComplete: ({ status, usage }) => {
    console.log("Stream finished – status:", status, "usage:", usage);
  },
});

Key Monitoring Hooks Inside the Stream

Inside open-sse/utils/stream.ts, three critical monitoring mechanisms guard stream integrity:

• idleTimer (around line 600): Tracks stream idle time and triggers streamTimedOut if data stops flowing
• shouldEmitDoneTerminator (lines 644-648): Determines whether to emit a final [DONE] marker based on client format (enabled for OpenAI, disabled for Claude/Responses API)
• applyTextualToolCallStreamingGuard (lines 554-583): Consolidates partial JSON fragments into proper tool_calls while monitoring parsing health

Real-Time Health Monitoring via the API

For system-wide observability, OmniRoute exposes a comprehensive health endpoint that aggregates circuit-breaker states, rate-limit status, and active session statistics.

The /api/monitoring/health Endpoint

The GET /api/monitoring/health endpoint implemented in src/app/api/monitoring/health/route.ts returns a JSON payload containing the router's current operational status. When called, the endpoint dynamically imports monitoring modules to avoid startup overhead, then aggregates metrics from the circuit-breaker, rate-limit manager, quota monitor, and session tracker.

import fetch from "node-fetch";

async function getHealth() {
  const resp = await fetch("http://localhost/api/monitoring/health");
  if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
  const health = await resp.json();

  console.log("Router status:", health.status);
  console.log("Active connections:", health.activeConnections);
  console.log("Provider breakers:", health.providerBreakers);
  console.log("Connection cooldowns:", health.connectionHealth);
  console.log("Quota monitor:", health.quotaMonitor);
}
getHealth().catch(console.error);

Aggregating Metrics with buildHealthPayload

The buildHealthPayload function in src/lib/monitoring/observability.ts assembles the health response by collecting:

• System info: Node version, uptime, and memory usage
• Provider breaker states: Circuit-breaker status for each LLM provider
• Connection cooldown summaries: Via summarizeConnectionCooldown (lines 179-200), showing rate-limited connections
• Quota monitor aggregates: Current states (active, alerting, exhausted, error) per provider
• Active session snapshots: Via buildSessionsSummary (lines 97-118), counting sticky-bound and top-active sessions

Critical Metrics to Track

Effective monitoring requires tracking specific metrics across the streaming lifecycle and infrastructure layers.

Stream Timeouts and Idle Detection

The stream idle timeout is monitored internally by createSSEStream through the idleTimer mechanism. If the timer fires due to lack of data, the stream closes and sets streamTimedOut to true in the completion callback. Monitor this flag to detect stalled provider connections or network interruptions.

Provider Connection Cooldowns

The summarizeConnectionCooldown function tracks how many connections for each provider remain rate-limited. Check the coolingDown count and soonestRetryAfterMs values to understand when specific providers become available again.

function logCoolingProviders(health) {
  const cooldowns = health.connectionHealth;
  for (const [provider, info] of Object.entries(cooldowns)) {
    console.log(
      `${provider}: ${info.coolingDown}/${info.total} connections cooling down – next retry in ${info.soonestRetryAfterMs} ms`
    );
  }
}

Quota and Rate Limit Status

The health payload includes quotaMonitorSummary and quotaMonitorMonitors fields that indicate per-provider quota exhaustion. Combined with rateLimitStatus from the rate-limit manager, these metrics reveal whether failures stem from provider quotas or OmniRoute's internal throttling.

Active Session Statistics

The buildSessionsSummary function generates counts of total active sessions, sticky-bound sessions (locked to specific providers), and the most active individual sessions. Use these metrics to identify load distribution imbalances and sticky-session congestion.

Implementation Examples

Fetching Health Metrics Programmatically

Poll the health endpoint every 30 seconds to detect degrade conditions before they cause failures:

setInterval(async () => {
  const health = await fetch("http://localhost/api/monitoring/health")
    .then(r => r.json());
    
  if (health.status !== "healthy") {
    alert(`OmniRoute degraded: ${health.status}`);
  }
}, 30000);

Handling Stream Completion Events

Always implement the onComplete callback to capture per-request outcomes, including token usage and timeout status:

onComplete: ({ status, usage, streamTimedOut, error }) => {
  if (streamTimedOut) {
    console.error("Stream idle timeout detected");
  }
  // Log to observability platform
  metrics.record("omniroute.stream.complete", { status, usage });
}

Interpreting Connection Cooldown Data

Connection cooldown metrics help diagnose provider-specific rate limiting. A high coolingDown count relative to total indicates the provider is currently throttling requests, while soonestRetryAfterMs tells you exactly when to retry.

Summary

• Poll /api/monitoring/health to retrieve real-time system status, provider health, and quota states aggregated by buildHealthPayload in src/lib/monitoring/observability.ts
• Use the onComplete callback in createSSEStream to capture per-request summaries including token usage and timeout flags
• Monitor streamTimedOut to detect stalled streams and shouldEmitDoneTerminator to verify protocol compliance
• Check connectionHealth cooldowns to understand provider availability and rate-limit recovery times
• Track quotaMonitor states to distinguish between provider quota exhaustion and internal routing issues

Frequently Asked Questions

How do I detect if an OmniRoute stream has timed out?

The stream timeout status is exposed through the streamTimedOut boolean in the onComplete callback object. When the internal idleTimer (implemented around line 600 in open-sse/utils/stream.ts) fires due to inactivity, the stream closes and sets this flag to true, allowing you to distinguish between completed streams and network stalls.

What does the coolingDown metric indicate in the health payload?

The coolingDown metric returned by summarizeConnectionCooldown in src/lib/monitoring/observability.ts indicates how many connections for a specific provider are currently rate-limited and unavailable for new requests. Alongside soonestRetryAfterMs, this metric shows exactly when the provider will accept new traffic again, helping you implement client-side backoff strategies.

Can I monitor OmniRoute streaming without using the health endpoint?

Yes, you can monitor individual streams through the onComplete callback provided to createSSEStream, which reports usage statistics, duration, and completion status. However, for system-wide observability including circuit-breaker states and provider health, the /api/monitoring/health endpoint remains necessary as it aggregates data from dynamically loaded modules like the quota monitor and rate-limit manager.

How does OmniRoute handle tool-call extraction monitoring?

Tool-call extraction is monitored by applyTextualToolCallStreamingGuard (lines 554-583 in open-sse/utils/stream.ts), which consolidates partial JSON fragments into proper tool_calls objects. If parsing fails or malformed tool-call data arrives, the guard prevents stream corruption and reports the issue through the standard error handling mechanism, ensuring that monitoring systems receive clean, actionable error states rather than raw parse failures.

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