Spaces:
Runtime error
Runtime error
| title: "open-sse Architecture" | |
| version: 3.8.16 | |
| lastUpdated: 2026-06-08 | |
| # open-sse Architecture | |
| > **TL;DR**: `open-sse/` is the core streaming engine that powers every LLM request in OmniRoute. It contains ~406 files implementing the request pipeline, executors, services, MCP server, and translation layer. This guide explains how the pieces fit together. | |
| **Source:** `open-sse/` (workspace package, ~143K LOC across 406 files) | |
| --- | |
| ## Why a Separate Workspace Package? | |
| `open-sse/` is a **standalone workspace** in the OmniRoute monorepo for several reasons: | |
| 1. **Reusability** β `open-sse` is published as `@omniroute/open-sse` on npm, so other projects can use it independently | |
| 2. **Clean boundaries** β the streaming engine is decoupled from the OmniRoute-specific UI/DB layer | |
| 3. **Performance** β the engine has no Next.js dependencies, enabling faster cold starts in CLI/serverless contexts | |
| 4. **Versioning** β `open-sse` can release on its own cadence | |
| ```json | |
| // package.json | |
| "workspaces": ["open-sse"] | |
| ``` | |
| --- | |
| ## Top-Level Structure | |
| ``` | |
| open-sse/ | |
| βββ index.ts # Public entry point | |
| βββ types.d.ts # Public type exports | |
| βββ package.json # @omniroute/open-sse | |
| βββ config/ # Provider configs, constants, registries | |
| βββ executors/ # Per-provider HTTP executors (59 files) | |
| βββ handlers/ # Request handlers (chatCore, responses, etc.) | |
| βββ lib/ # Internal utilities | |
| βββ mcp-server/ # Model Context Protocol server | |
| βββ services/ # ~114 service modules | |
| βββ transformer/ # Responses API format transformer | |
| βββ translator/ # Format translation (OpenAI β Claude β Gemini) | |
| βββ utils/ # Shared utilities (logging, error, stream, etc.) | |
| ``` | |
| ### Module Counts | |
| | Directory | Files | Purpose | | |
| | `executors/` | 62 | Per-provider HTTP executors (unified via DefaultExecutor factory) | | |
| | `handlers/` | ~15 | Request entry points (chatCore, responses, embeddings) | | |
| | `services/` | ~114 | Routing, caching, rate limiting, refresh, etc. | | |
| | `translator/` | ~10 | Format conversion (OpenAI β Claude β Gemini) | | |
| | `mcp-server/` | 30 | MCP tools and transports | | |
| | `utils/` | ~30 | Cross-cutting utilities (logging, error, stream) | | |
| | `config/` | ~10 | Provider configs, constants, registries | | |
| --- | |
| ## The Request Pipeline | |
| Every LLM request flows through a **5-stage pipeline**: | |
| ``` | |
| ββββββββββββββββ | |
| HTTP request β 1. ROUTE β combo resolution, model selection | |
| (Next.js route) ββββββββ¬ββββββββ | |
| β | |
| βΌ | |
| ββββββββββββββββ | |
| β 2. TRANSLATEβ format conversion (OpenAI β Claude β Gemini) | |
| ββββββββ¬ββββββββ | |
| β | |
| βΌ | |
| ββββββββββββββββ | |
| β 3. EXECUTE β provider executor, HTTP, retry, breaker | |
| ββββββββ¬ββββββββ | |
| β | |
| βΌ | |
| ββββββββββββββββ | |
| β 4. STREAM β SSE transformation, backpressure | |
| ββββββββ¬ββββββββ | |
| β | |
| βΌ | |
| ββββββββββββββββ | |
| β 5. RECORD β usage tracking, call log, error classification | |
| ββββββββ¬ββββββββ | |
| β | |
| βΌ | |
| HTTP response (SSE or JSON) | |
| ``` | |
| ### Stage 1: Route (services/combo.ts) | |
| **Entry point**: `handleComboChat()` in `services/combo.ts` | |
| Resolves the request to a concrete `(provider, model, account, credentials)` tuple: | |
| - Look up the combo by ID (or build a virtual combo for `auto/*` models) | |
| - Apply routing strategy (priority, weighted, round-robin, etc.) | |
| - Filter out unhealthy providers (circuit breaker) | |
| - Pick the next viable target | |
| For `auto/*` models, this stage also: | |
| - Runs the **9-factor scoring** algorithm (`services/autoCombo/`) | |
| - Selects a `provider+model` pair based on health, cost, latency, etc. | |
| ### Stage 2: Translate (translator/) | |
| If the source format (e.g., OpenAI) differs from the target format (e.g., Claude), the request is **translated**: | |
| - System prompt β system message | |
| - Tool definitions β provider-specific tool format | |
| - Reasoning/thinking parameters β provider-specific equivalents | |
| - Message role normalization (`developer` β `system` for non-OpenAI) | |
| The `translator/index.ts` exposes: | |
| ```ts | |
| translateRequest(body, sourceFormat, targetFormat): TranslatedRequest | |
| needsTranslation(source, target): boolean | |
| ``` | |
| ### Stage 3: Execute (executors/) | |
| **Entry point**: `getExecutor(providerId).execute(request, options)` | |
| All providers use `DefaultExecutor` (`executors/default.ts`) via the `getExecutor()` factory fallback. The executor: | |
| - Builds the upstream URL (`buildUrl()`) | |
| - Adds provider-specific headers (`buildHeaders()`) | |
| - Transforms the request body (`transformRequest()`) | |
| - Sends the HTTP request with retry + exponential backoff | |
| - Handles auth refresh if needed (OAuth providers) | |
| All executors extend `BaseExecutor` (`executors/base.ts`, 1170 LOC) which provides: | |
| - Common retry logic | |
| - Proxy integration | |
| - Circuit breaker integration | |
| - Usage recording hooks | |
| ### Stage 4: Stream (utils/stream.ts) | |
| For streaming responses, the executor returns a **ReadableStream**. The handler: | |
| - Pipes through an SSE transform (`createSSETransformStreamWithLogger`) | |
| - Applies heartbeat pings to detect dead connections | |
| - Handles client disconnect gracefully (`pipeWithDisconnect`) | |
| - Transforms SSE β JSON for non-streaming clients | |
| For non-streaming responses, the executor returns a parsed JSON object that is passed through unchanged. | |
| ### Stage 5: Record (services/usage.ts) | |
| After the response (success or failure), usage is recorded: | |
| - `prompt_tokens`, `completion_tokens`, `cached_tokens` from the response | |
| - `cost_usd` computed from pricing data | |
| - `latency_ms`, `status`, `error_class` if failed | |
| - Persisted to `usage_history` table | |
| Call log artifacts (if enabled) are written to `${DATA_DIR}/call_logs/`. | |
| --- | |
| ## Key Files Deep-Dive | |
| ### chatCore.ts (5977 lines) | |
| The **main request handler**. Despite its size, it has a clear structure: | |
| ```ts | |
| // Pseudo-structure of chatCore.ts | |
| export async function handleChat(request: NextRequest) { | |
| // 1. Auth + CORS | |
| await authenticateRequest(request); | |
| applyCorsHeaders(response); | |
| // 2. Body validation | |
| const body = await parseRequestBody(request); | |
| // 3. Format detection + translation | |
| const sourceFormat = detectFormat(request); | |
| const targetFormat = getTargetFormat(providerId); | |
| if (needsTranslation(sourceFormat, targetFormat)) { | |
| body = translateRequest(body, sourceFormat, targetFormat); | |
| } | |
| // 4. Combo routing | |
| const targets = await resolveComboTargets(comboId, body); | |
| for (const target of targets) { | |
| try { | |
| const result = await executeOnTarget(target, body); | |
| await recordUsage(result); | |
| return result; | |
| } catch (err) { | |
| // Continue to next target | |
| } | |
| } | |
| // 5. Emergency fallback | |
| return await emergencyFallback(body); | |
| } | |
| ``` | |
| Despite being one giant function, it's organized into **commented sections** that map to the 5-stage pipeline. | |
| ### combo.ts (4456 LOC) | |
| The **routing engine** that resolves a combo to ordered targets. | |
| ```ts | |
| // services/combo.ts | |
| export async function handleComboChat(body, comboId): Promise<ChatResult> { | |
| const targets = await resolveComboTargets(comboId, body); | |
| for (const target of targets) { | |
| try { | |
| return await handleSingleModel(target, body); | |
| } catch (err) { | |
| log.warn("target failed, trying next", { target, err }); | |
| } | |
| } | |
| throw new ComboExhaustedError("All targets failed"); | |
| } | |
| ``` | |
| Supports **17 routing strategies** (see `src/shared/constants/routingStrategies.ts`): | |
| | Strategy | Behavior | | |
| |----------|----------| | |
| | `priority` | First-target ordered list | | |
| | `weighted` | Probabilistic by per-target weight | | |
| | `round-robin` | Cycle through targets in order | | |
| | `context-relay` | Hand off context across targets | | |
| | `fill-first` | Fill quota before moving to next | | |
| | `p2c` | Power of two choices | | |
| | `random` | Uniform random | | |
| | `least-used` | Pick the one with fewest recent uses | | |
| | `cost-optimized` | Cheapest healthy target first | | |
| | `reset-aware` | Aware of provider reset windows | | |
| | `reset-window` | Reset window-based routing | | |
| | `headroom` | Most remaining quota headroom first | | |
| | `strict-random` | Truly uniform (no quality weighting) | | |
| | `auto` | Use 9-factor scoring (`autoCombo/`) | | |
| | `lkgp` | Last known good provider first | | |
| | `context-optimized` | Best for long-context requests | | |
| | `fusion` | Fan out to a panel in parallel, then synthesize via a judge (`fusion.ts`) | | |
| ### base.ts (1170 LOC) | |
| The **abstract executor** that all 59 executors extend. It contains: | |
| - `buildUrl()` β default URL construction (subclasses override for custom) | |
| - `buildHeaders()` β default headers (auth, content-type) | |
| - `transformRequest()` β pass-through by default | |
| - `execute()` β the main HTTP loop with retry/backoff/breaker | |
| ```ts | |
| // open-sse/executors/default.ts | |
| export class DefaultExecutor extends BaseExecutor { | |
| // Handles all OpenAI/Anthropic-compatible providers | |
| // Providers register configurations (URL, auth, headers) but share executor logic | |
| } | |
| ``` | |
| Provider-specific behavior (auth headers, base URL, version headers) is configured via the provider registry, not separate executor classes. | |
| ``` | |
| --- | |
| ## Services (117 modules) | |
| Services are **focused, single-purpose modules** that handlers compose. The big categories: | |
| ### Routing & Combo | |
| - `combo.ts` β entry point for combo-routed requests | |
| - `services/autoCombo/` β 9-factor scoring, 8 auto routing strategies | |
| - `wildcardRouter.ts` β matches wildcard routes (`gpt-*`) | |
| - `modelFamilyFallback.ts` β T5 intra-family fallback | |
| ### Rate Limiting & Quota | |
| - `rateLimitManager.ts` β token bucket per key+provider | |
| - `usage.ts` β usage recording | |
| - `quotaCache.ts` β in-memory quota snapshots | |
| ### Account & Token | |
| - `tokenRefresh.ts` β OAuth refresh on 401 | |
| - `accountFallback.ts` β switch to alternate account | |
| - `sessionManager.ts` β multi-turn session state | |
| ### Intelligence | |
| - `intentClassifier.ts` β classify request intent | |
| - `taskAwareRouter.ts` β route by task type | |
| - `thinkingBudget.ts` β allocate thinking tokens | |
| - `contextManager.ts` β inject routing context | |
| ### Resilience | |
| - `resilience.ts` β retry, backoff, breaker orchestration | |
| - `emergencyFallback.ts` β last-resort fallback | |
| - `modelDeprecation.ts` β auto-route to successor models | |
| ### State | |
| - `signatureCache.ts` β dedup by request signature | |
| - `volumeDetector.ts` β load shedding | |
| - `contextHandoff.ts` β session serialization | |
| ### Compression | |
| - `compression/` (subdirectory) β full compression pipeline | |
| - 39 files covering engines, rule packs, adapters | |
| ### Skills | |
| - (covered in [SKILLS.md](./SKILLS.md)) | |
| ### Memory | |
| - (covered in [MEMORY.md](./MEMORY.md)) | |
| --- | |
| ## Executors (75+ files) | |
| One file per provider. They all extend `BaseExecutor` and override what differs. | |
| ### Common Patterns | |
| Providers are resolved via `getExecutor(providerId)`, which returns the configured executor. OpenAI/Anthropic-compatible providers use `DefaultExecutor` (`executors/default.ts`). Provider-specific behavior (base URL, auth headers, API version) is configured in `open-sse/config/providers/`, while request body transformations are handled in `open-sse/translator/`. | |
| **Custom URL** is set via provider configuration: | |
| ```ts | |
| // Provider config in open-sse/config/providers/ | |
| export default { | |
| id: "together", | |
| baseURL: "https://api.together.xyz/v1/chat/completions", | |
| } | |
| ``` | |
| **Custom auth** is handled through the provider registry's auth configuration (API key, OAuth, header profiles). | |
| **Custom request body** transformations (e.g., Anthropic separating `system` from `messages`) are registered per-provider in `open-sse/translator/`. | |
| ``` | |
| ### The Executor Factory | |
| `executors/index.ts` exports `getExecutor(providerId)`: | |
| ```ts | |
| import { getExecutor } from "@omniroute/open-sse/executors"; | |
| const executor = getExecutor("anthropic"); | |
| const result = await executor.execute({ | |
| model: "claude-sonnet-4-5", | |
| messages: [...], | |
| }); | |
| ``` | |
| The factory is generated from `config/providerRegistry.ts` which lists all 212+ providers and their executor class. | |
| --- | |
| ## Translators | |
| Translate between **3 formats**: OpenAI, Anthropic, Gemini, plus the new Responses API. | |
| ### When Translation Happens | |
| ```ts | |
| import { needsTranslation, translateRequest } from "@omniroute/open-sse/translator"; | |
| if (needsTranslation(sourceFormat, targetFormat)) { | |
| body = translateRequest(body, sourceFormat, targetFormat); | |
| } | |
| ``` | |
| Common translations: | |
| - `OpenAI β Anthropic`: separate `system` field, `x-api-key` header | |
| - `OpenAI β Gemini`: `contents` instead of `messages`, `systemInstruction` | |
| - `OpenAI β Responses API`: `input` array, `previous_response_id` state | |
| ### Edge Cases Handled | |
| - `developer` role β `system` for non-OpenAI | |
| - `system` role β merged into first user message for GLM/ERNIE | |
| - `json_schema` β Gemini's `responseMimeType` + `responseSchema` | |
| - `tools` β provider-specific tool format | |
| - Thinking parameters (o1, Claude) β provider-specific equivalents | |
| --- | |
| ## MCP Server | |
| `open-sse/mcp-server/` implements the **Model Context Protocol** server: | |
| - **30+ tools** (provider management, combos, memory, cache, compression, 1proxy, skills) | |
| - **3 transports**: stdio, SSE, Streamable HTTP | |
| - **13 scopes** for fine-grained authorization | |
| ### Tool Registration | |
| Tools are registered as standalone files in `open-sse/mcp-server/tools/`, each exporting a name, schema, handler, and scope: | |
| ```ts | |
| // open-sse/mcp-server/tools/getHealth.ts | |
| import { z } from "zod"; | |
| export default { | |
| name: "omniroute_get_health", | |
| description: "Get system health snapshot", | |
| scope: "read:health", | |
| inputSchema: z.object({}), | |
| handler: async (_args, ctx) => { | |
| return await getSystemHealth(); | |
| }, | |
| }; | |
| ``` | |
| ### Transports | |
| ```ts | |
| // stdio (CLI usage) | |
| startMcpStdio(server); | |
| // SSE (HTTP-based streaming) | |
| startMcpSse(server, port); | |
| // Streamable HTTP (modern MCP) | |
| startMcpStreamable(server, port); | |
| ``` | |
| ### Authorization | |
| Every tool call goes through scope checks (`open-sse/mcp-server/auth/`): | |
| ```ts | |
| if (!hasScope(apiKey, "providers:read")) { | |
| throw new Error("Insufficient scope"); | |
| } | |
| ``` | |
| --- | |
| ## Transformers | |
| `open-sse/transformer/` converts between **Chat Completions** and **Responses API** formats. | |
| ### Why a Separate Transformer? | |
| The Responses API is OpenAI's new format with **stateful conversations** (`previous_response_id`). When a client sends a Responses request, OmniRoute: | |
| 1. Converts Responses β Chat Completions internally | |
| 2. Sends to provider (any provider that supports Chat Completions) | |
| 3. Converts the response back to Responses format | |
| 4. Streams the converted response to the client | |
| The transformer (`transformer/responsesTransformer.ts`) provides: | |
| ```ts | |
| createResponsesApiTransformStream(): TransformStream | |
| ``` | |
| This handles: | |
| - `response.output_item.added` events | |
| - `response.output_text.delta` events | |
| - `response.completed` event | |
| - Tool call mapping (`function_call` β `tool_calls`) | |
| --- | |
| ## Configuration | |
| `open-sse/config/` holds the configuration layer: | |
| | File | Purpose | | |
| |------|---------| | |
| | `providerRegistry.ts` | 212+ provider definitions | | |
| | `providerModels.ts` | Model aliases, format mapping | | |
| | `constants.ts` | Timeouts, limits, status codes | | |
| | `defaultThinkingSignature.ts` | Default Claude thinking signature | | |
| | `modelStrip.ts` (in services) | Per-provider field stripping | | |
| ### Provider Registry Schema | |
| ```ts | |
| interface ProviderConfig { | |
| id: string; | |
| name: string; | |
| baseUrl: string; | |
| authType: "bearer" | "api-key" | "oauth" | "cookie"; | |
| executorClass: string; | |
| defaultModel: string; | |
| capabilities: ProviderCapabilities; | |
| models: ModelDefinition[]; | |
| } | |
| ``` | |
| Zod validation at module load ensures all provider configs are valid. | |
| --- | |
| ## Performance Constraints | |
| The routing engine has strict performance budgets: | |
| | Operation | Target | Measurement | | |
| |-----------|--------|-------------| | |
| | Combo resolution | <10ms | For 50 targets | | |
| | Rate limit check | <1ms | In-memory token bucket | | |
| | Model family fallback | <5ms | Cached family definitions | | |
| | Request routing dispatch | <2ms | Hot path | | |
| | **No blocking I/O in routing hot path** | β | All async | | |
| --- | |
| ## Anti-Patterns | |
| β **Synchronous DB calls in `combo.ts`** β pre-compute and cache | |
| β **Retry logic in handlers** β use `retry()` from resilience service | |
| β **Direct provider config access** β use `providerRegistry` getters | |
| β **Hardcoded fallback chains** β define in `modelFamilyFallback.ts` | |
| β **State mutations across concurrent requests** β use request-scoped context only | |
| --- | |
| ## Adding a New Component | |
| ### Adding a New Service | |
| 1. Create `open-sse/services/[serviceName].ts` with focused responsibility | |
| 2. Export main handler function and any constants | |
| 3. Add unit tests in `tests/unit/services/[serviceName].test.mjs` | |
| 4. Integrate into request pipeline in `handlers/chatCore.ts` (if routing-related) | |
| 5. Update routing logic in `combo.ts` if service affects target selection | |
| 6. Document in this file | |
| ### Adding a New Executor | |
| 1. Create `open-sse/executors/[provider].ts` extending `BaseExecutor` | |
| 2. Register in `config/providerRegistry.ts` | |
| 3. Add to `executors/index.ts` factory | |
| 4. Add unit tests for the executor | |
| 5. Document in `docs/architecture/ARCHITECTURE.md` | |
| ### Adding a New MCP Tool | |
| 1. Create or update `open-sse/mcp-server/tools/[category]Tools.ts` | |
| 2. Define Zod schema for inputs | |
| 3. Register tool in `mcp-server/index.ts` | |
| 4. Add to scope matrix in `mcp-server/auth/` | |
| 5. Add unit tests | |
| --- | |
| ## See Also | |
| - [ARCHITECTURE.md](../architecture/ARCHITECTURE.md) β high-level architecture | |
| - [CODEBASE_DOCUMENTATION.md](../architecture/CODEBASE_DOCUMENTATION.md) β engineering reference | |
| - [REPOSITORY_MAP.md](../architecture/REPOSITORY_MAP.md) β directory-by-directory | |
| - [AUTO-COMBO.md](../routing/AUTO-COMBO.md) β 9-factor scoring | |
| - [MCP-SERVER.md](./MCP-SERVER.md) β MCP server | |
| - [A2A-SERVER.md](./A2A-SERVER.md) β A2A server | |
| - Source: `open-sse/` (400+ files, ~143K LOC) | |