feat(routing): adaptive feedback loop v2 — operational/semantic quality, confidence, TTFT/ITL, end-to-end test (#10881)

Obrigado — feature substancial e bem estruturada: separa qualidade operacional (comportamento de wire: 4xx/5xx, 429, respostas malformadas, stream interrompido) de qualidade semântica (só setada por avaliadores externos, nunca inferida do sucesso HTTP), com confidence/sample-awareness para não deixar poucos sucessos de sorte dominarem o ranking. Instrumentação de streaming (TTFT/ITL) threaded até RoutingEvent, endpoint de explicabilidade, e teste E2E determinístico cobrindo degradação→recuperação→blip.

Validação (worktree própria a partir de origin/release/v3.8.50, merge limpo, 0 conflitos):
- typecheck:core limpo, complexity/cognitive-complexity dentro do baseline
- 59/59 testes passando (mlx-provider, routing-adaptive-e2e, routing-events(-concurrency), routing-otel, routing-quality, routing-scoring-quality, stream-timing, auto-combo-scoring-clamp)
This commit is contained in:
3g0r1ch
2026-08-20 23:28:30 +03:00
committed by GitHub
parent 25ba4f2a34
commit d87b97a786
32 changed files with 3078 additions and 33 deletions

View File

@@ -0,0 +1,350 @@
---
title: "Adaptive Routing: Routing Events, Quality Feedback & Explainability"
version: 3.8.50
lastUpdated: 2026-08-20
---
# Adaptive Routing: Routing Events, Quality Feedback & Explainability
This document describes the feedback-driven adaptive routing foundation added to
OmniRoute. It is deliberately small: it introduces a typed routing-outcome
channel, an online quality signal that feeds the existing auto-combo scorer, an
optional OpenTelemetry exporter, and an explainability endpoint. It does **not**
replace the existing resilience stack (circuit breaker, connection cooldown,
model lockout, health matrix, autopilot) — it complements it.
## 1. Architectural context
OmniRoute is a data plane with a **request hot path** and a **control/intelligence
plane**. The hot path must stay fast, memory-efficient, asynchronous, resilient and
predictable. Evaluation, quality scoring, experiments and historical analysis belong
to the control plane.
```
AI Agent / IDE
┌─────────────────────┐
│ OmniRoute │ data plane (fast, sync, in-memory)
│ routing / failover │
│ health / guardrail │
│ cache / streaming │
└──────────┬──────────┘
│ RoutingEvent (fire-and-forget, ~0.2µs)
┌─────────────────────┐
│ Feedback sinks │ control plane (async, best-effort)
│ quality tracker │
│ OTel exporter │
│ explain store │
└──────────┬──────────┘
▼ quality score
auto-combo scorer
```
### What was already there (audited, not duplicated)
| Concept | Existing implementation |
| ----------------------------------- | -------------------------------------------------------------------------------------------------- |
| Availability (can we send traffic?) | Circuit breaker (CLOSED/DEGRADED/OPEN/HALF_OPEN, DB-persisted), connection cooldown, model lockout |
| Health reporting | `providerHealthMatrix.ts`, `providerHealthAutopilot.ts` |
| Shadow traffic | `open-sse/services/combo/shadowRouting.ts` |
| Guardrails | `src/lib/guardrails/` (pre/post hooks) |
| Exact cache | `src/lib/semanticCache.ts` (signature-based) |
| Evaluators / eval-driven routing | `src/lib/evals/`, `open-sse/services/evalRouting.ts` |
| Combo decision explainability | `open-sse/services/combo/decisionTrace.ts` |
| Dashboard real-time events | `src/lib/events/eventBus.ts` (UI notification channel, `unknown` payloads, 100-entry history) |
The routing-event layer is **not** a re-implementation of `eventBus`: that bus is
the dashboard's real-time notification channel (typed _event names_, opaque
payloads, UI consumers). `RoutingEvent` is a typed _outcome_ struct
(latency/tokens/cost/outcome/finish-reason) consumed by the control plane's
feedback sinks (quality tracker, OTel exporter, explain store).
### What was missing (added here)
1. A **typed routing-outcome event + sink abstraction** (`RoutingEvent` /
`RoutingEventSink`). `decisionTrace` is combo-scoped and in-memory-only;
`comboMetrics` are cumulative counters; `call_logs` is raw async persistence.
None is a typed, sink-based outcome channel that a quality tracker, an OTel
exporter, or a Future-AGI-style evaluator can subscribe to.
2. An **online quality signal** (EWMA) for output quality — the scorer previously
proxied "quality" only through static task fitness and opt-in eval pass-rates.
3. An **optional, dependency-free OTel exporter** using GenAI semantic conventions.
4. An **explainability endpoint** returning the real routing decisions + quality state.
## 2. Routing Events (feedback foundation)
Files: `open-sse/services/routing/events.ts`, `.../index.ts`
A `RoutingEvent` carries only routing metadata:
```ts
interface RoutingEvent {
requestId: string;
provider: string;
model: string;
strategy: string; // "auto" | "priority" | "direct" | ...
latencyMs: number;
ttftMs: number | null;
inputTokens: number | null;
outputTokens: number | null;
cost: number | null;
retries: number;
fallbackUsed: boolean;
outcome: RoutingOutcome; // allowlisted union
status: number | null;
finishReason: string | null;
connectionId: string | null;
ts: number;
}
```
`RoutingEventSink` is a `Send+Sync`-style trait in TypeScript:
```ts
interface RoutingEventSink {
readonly name: string;
record(event: RoutingEvent): void; // must be O(1), no sync I/O
}
```
The hot path calls `emitRoutingEvent(event)` once per completed request
(the streaming-completion callback, the non-streaming success path, and the
malformed-200 failure path in `handleChatCore`). Dispatch is synchronous fan-out
to registered sinks, but each sink only enqueues/updates in-memory state. **No
synchronous database writes, no network I/O on the hot path.**
Default sinks:
- `MemoryRoutingEventStore` — bounded (500) ring buffer, newest-first, for the
explain endpoint.
- `QualityTracker` consumer — updates the EWMA quality estimate.
- `OtlpHttpsEventSink` — optional, enabled only when `OMNIROUTE_OTEL_ENDPOINT`
(or `OTEL_EXPORTER_OTLP_ENDPOINT`) is set.
### Measured overhead (honest comparison)
`npm run bench:routing-events` on this workstation (100k iterations; sub-µs ops
measured as aggregate µs/op because per-op percentiles are below
`performance.now()` timer resolution):
| Scenario | µs/op | ops/s |
| --------------------------------- | ------ | ------ |
| baseline (scoring only) | ~0.045 | ~22 M |
| baseline + RoutingEvent (2 sinks) | ~0.168 | ~5.9 M |
| baseline + event + OTel enqueue | ~0.163 | ~6.1 M |
| concurrent (8 interleaved bursts) | ~0.18 | — |
The event-dispatch delta over baseline scoring is ~0.12 µs/request; the OTel sink
only enqueues (O(1) buffer push), adding nothing measurable. These numbers are
machine-specific and relative — not a production guarantee. The v1 "~0.2 µs"
figure was an aggregate estimate; this methodology separates the scoring baseline
from the event-dispatch cost.
## 3. Quality Signal (feedback-driven provider state)
Files: `open-sse/services/routing/quality.ts`
v2 separates **operational** from **semantic** quality:
- **Operational** — derived from the routing hot path (HTTP 4xx/5xx, connection
failures, 429s, malformed responses, stream interruptions, `finish_reason=length`,
zero-output successes, latency/TTFT EWMA). A 200 is NOT treated as semantic
quality.
- **Semantic** — the actual value of the generated output. ONLY ever produced by
an evaluator via `setSemanticQuality()`. It is `null` until one provides it and
never leaks into the operational score.
Per-(provider, model) state (EWMA + bounded counters):
- `successEwma` — EWMA (α=0.2) of outcome success.
- `latencyEwma` / `ttftEwma` — EWMA of latency (α=0.1).
- `samples`, `anomalies`, `rateLimited`, `semantic`, `semanticConfidence`.
- `recencyMs` — how recently the model was last observed.
### Confidence / sample awareness
`confidence = clamp01(samples / 50)`, and the score returned to the scorer is
blended toward the neutral midpoint:
```
score = 0.5 + confidence * (operational - 0.5)
```
Consequences (verified by tests):
- A cold provider (0 samples) scores **0.5** — not unfairly penalized, but
unable to dominate a provider with thousands of solid observations.
- A provider with 7 lucky successes is pulled toward 0.5 (never dominates from
optimistic initialization).
- A provider with 50+ samples converges to its true operational score.
- Degradation and recovery are gradual (EWMA), and one isolated failure does
not destroy a healthy provider.
`ProviderQuality` exposes `{ operational, semantic, confidence, samples, anomalies,
rateLimited, successEwma, latencyEwmaMs, ttftEwmaMs, recencyMs }`.
This feeds the auto-combo scorer as the `quality` scoring factor:
- `ScoringFactors.quality` / `ScoringWeights.quality` in
`open-sse/services/autoCombo/scoring.ts`.
- `DEFAULT_WEIGHTS`: `health` 0.1905 → 0.1605, `quality` 0.03. Sum stays 1.0.
- `buildAutoCandidates` populates `candidate.quality` from the tracker; candidates
without data default to neutral **0.5** (a cold candidate is neither boosted nor
penalized).
The closed loop:
```
RoutingEvent → QualityTracker → getQualityScore → auto-combo quality factor
↑ │
└────── request outcome (handleChatCore) ←────────────┘
```
### Hard exclusion vs soft penalty
The quality signal is a **soft adaptive preference** only. Hard exclusion stays
with the existing resilience stack: circuit breaker OPEN, quota exhausted,
auth failure, model lockout — none of these are affected by the quality score.
A provider whose quality score dips temporarily is de-preferenced, never
hard-disabled.
## 3b. Canonical stream timing (TTFT / ITL)
Files: `open-sse/utils/streamTiming.ts`
`createStreamTiming()` is the single instrumentation seam for the streaming path,
wired into `createSSEStream` (open-sse/utils/stream.ts):
- `markByte()` — first upstream chunk received.
- `markForward()` — first chunk forwarded to the client (used for TTFT).
- `markInterrupted()` — stream timeout/abort/error before a clean finish.
- `ttft()` = first-forwarded-SSE-chunk latency. **This is NOT token-level TTFT**
a single SSE chunk may carry zero/one/many tokens. Documented precisely.
- `avgItlMs()` = mean inter-chunk gap (a chunk-latency proxy for ITL).
TTFT/ITL/interrupted flow into the `RoutingEvent` (`ttftMs`, `itlMs`) and are
exported as GenAI/OmniRoute span attributes by the OTel sink.
## 4. OpenTelemetry / GenAI observability
Files: `open-sse/services/routing/otel.ts`
- Dependency-free OTLP/HTTP JSON exporter (uses global `fetch`, no
`@opentelemetry/*` SDK).
- Spans follow GenAI semantic conventions (`gen_ai.provider.name`,
`gen_ai.request.model`, `gen_ai.usage.input_tokens/output_tokens`,
`gen_ai.completion.finish_reason`, `gen_ai.system`) plus OmniRoute routing
attributes (outcome, status, ttft, retries, fallback).
- `record()` only enqueues into a bounded buffer (O(1)); a background timer
flushes via `POST {endpoint}/v1/traces` asynchronously. Under overload the
oldest events are dropped (`dropped` counter) — never backpressure the data
plane.
- **Disabled unless configured.** `OMNIROUTE_OTEL_ENDPOINT` (or
`OTEL_EXPORTER_OTLP_ENDPOINT`) must be set; otherwise the sink is not
registered and zero OTel code runs.
## 5. Explainability
- `GET /v1/explain/routing` returns the recent `RoutingEvent`s (the real
decisions, newest first) and the per-provider/model quality snapshot.
- Auth mirrors `/v1/combos` (Bearer API key or dashboard session; anonymous on
single-user local deployments with `REQUIRE_API_KEY=false`).
- Combo-level per-invocation traces remain available via the existing
`decisionTrace.ts` (header `X-OmniRoute-Combo-Trace`).
- Safety: events carry only routing metadata, never prompts/bodies/credentials.
## 6. Evaluation-plane integration (Future AGI readiness)
OmniRoute treats Future AGI (or any evaluator) as a **potential
intelligence/evaluation backend, not a dependency**. The seams:
- A `RoutingEventSink` can forward events to an evaluator asynchronously.
- The `MemoryRoutingEventStore` + quality snapshot give an evaluator the raw
decision stream.
- A future `Evaluator` (deterministic, local judge, HTTP, WASM) would consume
events/traces and return a `QualityScore` that feeds the same
`getQualityScore`/quality-factor path.
- Existing eval-driven routing (`open-sse/services/evalRouting.ts`) already
re-orders combo targets by `eval_runs` pass-rates when enabled.
No evaluation runs synchronously on the request path, and the gateway operates
fully with the evaluator absent.
## 7. Final architectural review
1. **What remains on the synchronous hot path?** Routing/scoring, guardrail
pre-checks, cache lookup, and one `emitRoutingEvent` fan-out (~0.12 µs over
baseline scoring) to in-memory sinks.
2. **What moved to asynchronous processing?** OTel export (timer + fetch),
`call_logs`/usage persistence, semantic-cache writes, quality is in-memory
and O(1) (no async needed).
3. **How does a routing outcome become feedback?** `handleChatCore` emits a
`RoutingEvent``QualityTracker` updates EWMA state → `getQualityScore`
feeds the auto-combo `quality` factor.
4. **How does quality influence future routing?** A low quality score reduces
the weighted score of that provider/model in `scoreAutoTargets`, so degraded
models are gradually de-preferenced and recover as their EWMA improves.
5. **How can Future AGI integrate without becoming a dependency?** Via the
`RoutingEventSink` interface / a future `Evaluator` adapter — no hardcoded
dependency.
6. **What happens when the evaluator is unavailable?** Routing is unaffected;
quality falls back to neutral (1.0) for models with no observed signal.
7. **What happens when telemetry is unavailable?** The OTel sink simply isn't
registered; the rest of the routing layer runs unchanged.
8. **What happens under overload?** The OTel buffer drops oldest events; quality
and the ring buffer are bounded by construction; no backpressure.
9. **How does provider state recover after degradation?** EWMA re-converges as
successes accumulate; warmup keeps cold models neutral; the circuit breaker
independently recovers via HALF_OPEN probes.
10. **Which proposed features were intentionally NOT implemented, and why?**
- Shadow traffic / experiments — already implemented
(`combo/shadowRouting.ts`); not re-built.
- Guardrails — already implemented (`src/lib/guardrails/`); not duplicated.
- Semantic cache — already implemented (`src/lib/semanticCache.ts`); not
duplicated.
- A full experiment-management platform, dataset tooling, prompt-optimization
platform, vector DB, or mandatory external OTel infrastructure — out of
scope for a lean data plane.
- A Rust `RoutingEvent` struct — the data plane is TypeScript; the TS type
is the adapted equivalent.
## 8. Configuration reference
| Variable | Default | Effect |
| ----------------------------- | ----------- | ------------------------------------------------------------------------------- |
| `OMNIROUTE_OTEL_ENDPOINT` | unset | When set, enables the OTLP/HTTP traces exporter (e.g. `http://collector:4318`). |
| `OTEL_EXPORTER_OTLP_ENDPOINT` | unset | Fallback alias for the OTLP endpoint. |
| `OTEL_SERVICE_NAME` | `omniroute` | `service.name` resource attribute. |
## 9. Tests
- `tests/unit/routing-events.test.ts` — event normalization, status
classification, bounded ring buffer, sink fan-out + isolation.
- `tests/unit/routing-quality.test.ts` — EWMA warmup, failure/success recovery,
anomaly penalties, 429 transient handling, snapshot, reset.
- `tests/unit/routing-scoring-quality.test.ts` — weight integrity, neutral
default, quality factor ranking.
- `tests/unit/routing-otel.test.ts` — enable gating, GenAI span payload, async
flush, drop-under-overload.
- `tests/unit/routing-events-concurrency.test.ts` — thousands of events, ring
buffer boundedness, throwing-sink isolation, interleaved async bursts,
reset-during-inserts.
- `tests/unit/routing-adaptive-e2e.test.ts` — deterministic end-to-end loop via
the real `scoreAutoTargets` scorer: healthy → degrade → recover → blip, plus
cold-start and lucky-cold-provider scenarios.
- `tests/unit/stream-timing.test.ts` — TTFT (first-forwarded-chunk), ITL,
first-byte vs first-forward, interruption, malformed/empty chunk safety.
## 10. Pre-existing issues status (Phase 18)
| Issue | Status | Notes |
| ----------------------------------------------------- | ------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `omniglyph` export mismatch | **FIXED (environmental)** | `node_modules` was out of sync with `package-lock.json` (installed 1.3.1 vs locked 1.4.0). Running `npm install omniglyph@1.4.0` restored the locked version; type errors dropped to 0. Manifests unchanged. |
| Stale `getKnownContextOverflow` tests | **KNOWN — not fixed** | `combo-context-overflow-compression-probe.test.ts` imports a function that no longer exists in `open-sse/services/combo.ts` (only comments reference it). Fixing requires re-implementing or re-writing those tests — unrelated architectural churn. |
| `combo-runtime-unit-concurrency.test.ts` DB isolation | **KNOWN — not fixed** | Test-harness SQLite-isolation assertion fails when run directly; fails identically on the base branch. |
| i18n `llm.txt` drift | **KNOWN — not fixed** | `docs/i18n/*/llm.txt` differ from root; pre-existing, blocks the docs-sync pre-commit gate. |
Environmental vs code issues are kept distinct; no unrelated failures are hidden
behind changed test filters.

View File

@@ -12,6 +12,7 @@
"ROUTER_BACKENDS",
"admission-lanes",
"cluster-decisions",
"persistence-backend-boundary"
"persistence-backend-boundary",
"ADAPTIVE_ROUTING"
]
}

View File

@@ -52,6 +52,8 @@ safely retry only the failures after a partial result.
- **Pollinations** — Free GPT-5, Claude, Gemini (no key needed)
- **LongCat** — 10M tokens free (one-time grant, requires account + KYC)
- **Cloudflare AI** — 50+ models, 10K neurons/day
- **MLX Gemma 26B** — Local Apple Silicon model (~38.5 tok/s, ~15.9GB RAM)
- **MLX Qwen 3.8 27B** — Local Apple Silicon model (~9.1 tok/s, ~13.1GB RAM)
4. Click **Connect**
5. Done! You now have free AI access.
@@ -79,6 +81,94 @@ safely retry only the failures after a partial result.
5. Login with your account
6. Done! You now have access to your subscription models.
### Option D: Local MLX Models (Apple Silicon)
For Apple Silicon Macs with unified memory, OmniRoute supports connecting to local MLX models running via `mlx-lm.server` as regular OpenAI-compatible local providers.
#### Prerequisites
- **Apple Silicon Mac** (M1/M2/M3/M4) with 24GB+ unified memory recommended
- **uv** package manager: `curl -LsSf https://astral.sh/uv/install.sh | sh`
- **mlx-lm**: `uv pip install mlx-lm`
#### Quick Start
1. **Install dependencies**:
```bash
# Install uv if not already installed
curl -LsSf https://astral.sh/uv/install.sh | sh
# Install mlx-lm
uv pip install mlx-lm
```
2. **Start MLX servers manually** (in separate terminals):
```bash
# Terminal 1: Gemma 4 26B A4B IT-QAT (port 11435)
uv run mlx_lm.server --model mlx-community/gemma-4-26B-A4B-it-qat-q4_0-mlx-aligned --port 11435 --host 127.0.0.1
# Terminal 2: Qwen 3.8 27B MLX Mixed (port 11436)
uv run mlx_lm.server --model maglun/Qwen3.8-27B-MLX-Mixed-3.80bpw --port 11436 --host 127.0.0.1
```
3. **Connect in OmniRoute Dashboard**:
- Go to **Providers** → **Add Provider**
- Select **MLX Gemma 26B** or **MLX Qwen 3.8 27B**
- Click **Connect** (no API key needed)
4. **Use with OpenCode**:
```bash
# Configure OpenCode to use OmniRoute
opencode config set api.base_url http://localhost:20128/v1
opencode config set api.key <your-omniroute-api-key>
# Use MLX models
opencode run --model mlx-gemma/gemma-4-26b
opencode run --model mlx-qwen/qwen3.8-27b
```
#### Memory Management
**Important**: With 24GB unified memory, only **one large MLX model can run at a time**.
- Gemma 26B: ~15.9GB peak memory
- Qwen 3.8 27B: ~13.1GB peak memory
You must manage this manually:
- Run only one MLX server at a time, or
- Run both on separate machines, or
- Stop one before starting the other
OmniRoute does not automatically manage MLX server processes — it only routes requests to the OpenAI-compatible endpoints you configure.
#### Tool Calling Support
Both models support OpenAI-compatible tool calling. Test with:
```bash
curl -X POST http://localhost:20128/v1/chat/completions \
-H "Authorization: Bearer <key>" \
-H "Content-Type: application/json" \
-d '{
"model": "mlx-gemma/gemma-4-26b",
"messages": [{"role": "user", "content": "What is 2+2? Use the calculator tool."}],
"tools": [{"type": "function", "function": {"name": "calculator", "description": "Calculate", "parameters": {"type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"]}}}]
}'
```
#### Troubleshooting
| Issue | Solution |
| ------------------ | ----------------------------------------------------------------------------- |
| Server won't start | Check `uv run mlx_lm.server --help` and verify model IDs |
| Out of memory | Ensure only one model runs; close other apps; check Activity Monitor |
| Connection refused | Verify server is running on correct port (11435/11436) |
| Slow responses | First request loads model into memory (~30-60s); subsequent requests are fast |
| Tool calling fails | Ensure model supports tools; check OmniRoute logs for translation errors |
---
## Best Free Providers

View File

@@ -355,7 +355,7 @@ Use the dashboard at `/dashboard/providers` to enable, configure, and test each
| `zerolimitai` | `zerolimitai` | ZeroLimitAI | API key, aggregator | [link](https://www.zerolimitai.com) | Temporary free trial is advertised, but official pages conflict between 3 and 7 days; a 100-calls/day claim is not treated as permanent. |
| `zylo-api` | `zylo` | Zylo API | API key, aggregator | [link](https://zyloai.net) | Basic plan: 10 RPM, 7,200 requests/day and 200,000 tokens/day; limited to Basic text models. |
## Local Providers (12)
## Local Providers (14)
| ID | Alias | Name | Tags | Website | Notes |
|----|-------|------|------|---------|-------|
@@ -365,6 +365,8 @@ Use the dashboard at `/dashboard/providers` to enable, configure, and test each
| `llama-cpp` | `llamacpp` | llama.cpp | Local, self-hosted | [link](https://github.com/ggml-org/llama.cpp) | API key optional (use any value, e.g. sk-no-key-required). Configure the llama-server OpenAI-compatible base URL (default: http://127.0.0.1:8080/v1). Note: if Llamafile is also installed, both default to port 8080 — run only one at a time or override the port. |
| `llamafile` | `llamafile` | Llamafile | Local, self-hosted | [link](https://github.com/Mozilla-Ocho/llamafile) | API key optional. Configure the local Llamafile OpenAI-compatible base URL (default: http://127.0.0.1:8080/v1). |
| `lm-studio` | `lmstudio` | LM Studio | Local, self-hosted | [link](https://lmstudio.ai) | API key optional. Configure the local LM Studio OpenAI-compatible base URL (default: http://localhost:1234/v1). |
| `mlx-gemma` | `mlx-gemma` | MLX Gemma 26B | Local, self-hosted | [link](https://github.com/ml-explore/mlx) | No API key required. Runs mlx-lm server locally on port 11435. Requires `uv` and `mlx-lm` installed. Model: `mlx-community/gemma-4-26B-A4B-it-qat-q4_0-mlx-aligned` (~15.9GB peak memory). |
| `mlx-qwen` | `mlx-qwen` | MLX Qwen 3.8 27B | Local, self-hosted | [link](https://github.com/ml-explore/mlx) | No API key required. Runs mlx-lm server locally on port 11436. Requires `uv` and `mlx-lm` installed. Model: `maglun/Qwen3.8-27B-MLX-Mixed-3.80bpw` (~13.1GB peak memory). |
| `ollama-local` | `ollama` | Ollama | Local, self-hosted | [link](https://ollama.com) | No API key required. Ollama runs locally — configure its OpenAI-compatible base URL (default: http://localhost:11434/v1) and make sure Ollama is running before connecting. |
| `oobabooga` | `ooba` | oobabooga | Local, self-hosted | [link](https://github.com/oobabooga/text-generation-webui) | API key optional. Configure the local oobabooga OpenAI-compatible base URL (default: http://localhost:5000/v1). |
| `sdwebui` | `sdwebui` | SD WebUI | Local | [link](https://github.com/AUTOMATIC1111/stable-diffusion-webui) | No API key required. Configure the local WebUI base URL (default: http://localhost:7860). |

View File

@@ -3,6 +3,8 @@ import { unorouterProvider } from "./registry/unorouter/index.ts";
import { aimlapiProvider } from "./registry/aimlapi/index.ts";
import { byteplusProvider } from "./registry/byteplus/index.ts";
import { mlxGemmaProvider } from "./registry/mlx/index.ts";
import { mlxQwenProvider } from "./registry/mlx/index.ts";
import { ollama_cloudProvider } from "./registry/ollama-cloud/index.ts";
import { syntheticProvider } from "./registry/synthetic/index.ts";
import { ideogramProvider } from "./registry/ideogram/index.ts";
@@ -264,6 +266,8 @@ import { helixmindProvider } from "./registry/helixmind/index.ts";
export const REGISTRY: Record<string, RegistryEntry> = {
aimlapi: aimlapiProvider,
"mlx-gemma": mlxGemmaProvider,
"mlx-qwen": mlxQwenProvider,
"ollama-cloud": ollama_cloudProvider,
synthetic: syntheticProvider,
ideogram: ideogramProvider,

View File

@@ -0,0 +1,66 @@
import type { RegistryEntry } from "../../shared.ts";
import { buildOpenAiCompatibleRegistryEntry } from "../../shared.ts";
// MLX ports (deterministic, documented)
const MLX_GEMMA_PORT = 11435;
const MLX_QWEN_PORT = 11436;
// ─────────────────────────────────────────────────────────────────────────────
// Memory-aware context windows for MLX models on 24GB unified memory.
// Based on verified peak memory: Gemma 26B ~15.9GB, Qwen 27B ~13.1GB.
// KV cache estimate: 2 * 2 * layers * kv_heads * head_dim * num_ctx bytes.
// Conservative context windows to leave headroom for OS/other processes.
export const MLX_DEFAULT_CONTEXT_LIMIT = 32768;
const CONTEXT_GEMMA_26B = 8192; // 15.9GB weights + ~3.5GB KV @ 8k = ~19.4GB (safe for 24GB)
const CONTEXT_QWEN_27B = 8192; // 13.1GB weights + ~3.5GB KV @ 8k = ~16.6GB (safe for 24GB)
// ─────────────────────────────────────────────────────────────────────────────
// MLX Gemma 26B Provider
// Model: mlx-community/gemma-4-26B-A4B-it-qat-q4_0-mlx-aligned
// Verified speed: ~38.5 tok/s, peak memory: ~15.9 GB
export const mlxGemmaProvider: RegistryEntry = buildOpenAiCompatibleRegistryEntry({
id: "mlx-gemma",
alias: "mlx-gemma",
baseUrl: `http://localhost:${MLX_GEMMA_PORT}/v1`,
modelsUrl: `http://localhost:${MLX_GEMMA_PORT}/v1/models`,
passthroughModels: false,
defaultContextLength: MLX_DEFAULT_CONTEXT_LIMIT,
models: [
{
id: "mlx-community/gemma-4-26B-A4B-it-qat-q4_0-mlx-aligned",
name: "Gemma 4 26B A4B IT-QAT (MLX)",
toolCalling: true,
supportsVision: false,
supportsReasoning: false,
contextLength: CONTEXT_GEMMA_26B,
maxOutputTokens: 8192,
},
],
timeoutMs: 120000, // Longer timeout for model loading
});
// ─────────────────────────────────────────────────────────────────────────────
// MLX Qwen3.8 27B Provider
// Model: maglun/Qwen3.8-27B-MLX-Mixed-3.80bpw
// Verified speed: ~9.1 tok/s, peak memory: ~13.1 GB
export const mlxQwenProvider: RegistryEntry = buildOpenAiCompatibleRegistryEntry({
id: "mlx-qwen",
alias: "mlx-qwen",
baseUrl: `http://localhost:${MLX_QWEN_PORT}/v1`,
modelsUrl: `http://localhost:${MLX_QWEN_PORT}/v1/models`,
passthroughModels: false,
defaultContextLength: MLX_DEFAULT_CONTEXT_LIMIT,
models: [
{
id: "maglun/Qwen3.8-27B-MLX-Mixed-3.80bpw",
name: "Qwen 3.8 27B MLX Mixed 3.80bpw",
toolCalling: true,
supportsVision: false,
supportsReasoning: false,
contextLength: CONTEXT_QWEN_27B,
maxOutputTokens: 8192,
},
],
timeoutMs: 120000, // Longer timeout for model loading
});

View File

@@ -34,6 +34,38 @@ import { storeStreamingSemanticCacheResponse } from "./chatCore/streamingSemanti
import { assembleStreamingPipeline } from "./chatCore/streamingPipeline.ts";
import { sanitizeChatRequestBody } from "./chatCore/sanitization.ts";
import { applyResponsesInputPolicy } from "../services/responsesInputPolicy.ts";
import {
createRoutingEvent,
emitRoutingEvent,
outcomeFromStatus,
} from "../services/routing/index.ts";
/**
* Best-effort finish_reason extraction from a (possibly translated) response
* body for routing-event telemetry. Returns null when the shape is unknown.
*/
function routingFinishReason(body: unknown): string | null {
if (!body || typeof body !== "object") return null;
const record = body as Record<string, unknown>;
const choices = record.choices;
if (Array.isArray(choices)) {
const first = choices[0];
if (first && typeof first === "object") {
const fr = (first as Record<string, unknown>).finish_reason;
if (typeof fr === "string") return fr;
}
}
const output = record.output;
if (Array.isArray(output)) {
for (const item of output) {
if (item && typeof item === "object") {
const fr = (item as Record<string, unknown>).finish_reason;
if (typeof fr === "string") return fr;
}
}
}
return null;
}
import {
getHeaderValueCaseInsensitive,
isNoMemoryRequested,
@@ -5054,6 +5086,27 @@ export async function handleChatCore({
});
persistFailureUsage(HTTP_STATUS.BAD_GATEWAY, "malformed_translated_response");
trackPendingRequest(model, provider, pendingConnId, false);
// Routing event (feedback foundation) — record the malformed outcome so
// the quality tracker de-prioritizes this model over time.
void emitRoutingEvent(
createRoutingEvent({
requestId: traceId || pendingRequestId || "unknown",
provider: provider || "unknown",
model: model || "unknown",
strategy: isCombo ? (comboStrategy ?? "combo") : "direct",
latencyMs: Date.now() - startTime,
ttftMs: null,
inputTokens: null,
outputTokens: null,
cost: null,
retries: 0,
fallbackUsed: false, // combo-level fallback tracked by decisionTrace
outcome: "malformed",
status: HTTP_STATUS.BAD_GATEWAY,
finishReason: routingFinishReason(translatedResponse),
connectionId: credentials?.connectionId ?? null,
})
);
return createErrorResult(
HTTP_STATUS.BAD_GATEWAY,
malformedMessage,
@@ -5154,6 +5207,43 @@ export async function handleChatCore({
response: { status: 200, data: translatedResponse },
});
// Routing event (feedback foundation) — fire-and-forget, cheap.
void emitRoutingEvent(
createRoutingEvent({
requestId: traceId || pendingRequestId || "unknown",
provider: provider || "unknown",
model: model || "unknown",
strategy: isCombo ? (comboStrategy ?? "combo") : "direct",
latencyMs: Date.now() - startTime,
ttftMs: null,
inputTokens:
usage && typeof usage === "object"
? (() => {
const promptTokens = (usage as Record<string, unknown>).prompt_tokens;
return typeof promptTokens === "number" && Number.isFinite(promptTokens)
? promptTokens
: null;
})()
: null,
outputTokens:
usage && typeof usage === "object"
? (() => {
const completionTokens = (usage as Record<string, unknown>).completion_tokens;
return typeof completionTokens === "number" && Number.isFinite(completionTokens)
? completionTokens
: null;
})()
: null,
cost: Number.isFinite(estimatedCost) ? estimatedCost : null,
retries: 0,
fallbackUsed: false, // combo-level fallback tracked by decisionTrace
outcome: "success",
status: 200,
finishReason: routingFinishReason(translatedResponse),
connectionId: credentials?.connectionId ?? null,
})
);
return {
success: true,
response: buildNonStreamingJsonResponse(translatedResponse, responseHeaders),
@@ -5274,6 +5364,8 @@ export async function handleChatCore({
error: streamError,
errorCode: streamErrorCode,
ttft,
itlMs: streamItlMs,
interrupted: streamInterrupted,
}) => {
const normalizedStreamStatus = streamStatus || 200;
if (streamCompletionRecorded) return;
@@ -5377,6 +5469,53 @@ export async function handleChatCore({
endpoint: endpointPath,
});
// Routing event (feedback foundation) — fire-and-forget, cheap, never blocks
// the stream. Feeds the quality tracker + optional OTel exporter.
void emitRoutingEvent(
createRoutingEvent({
requestId: traceId || pendingRequestId || "unknown",
provider: provider || "unknown",
model: model || "unknown",
strategy: isCombo ? (comboStrategy ?? "combo") : "direct",
latencyMs: Date.now() - startTime,
ttftMs: typeof ttft === "number" && Number.isFinite(ttft) && ttft >= 0 ? ttft : null,
itlMs:
typeof streamItlMs === "number" && Number.isFinite(streamItlMs) && streamItlMs >= 0
? streamItlMs
: null,
inputTokens:
streamUsage && typeof streamUsage === "object"
? (() => {
const promptTokens = (streamUsage as Record<string, unknown>).prompt_tokens;
return typeof promptTokens === "number" && Number.isFinite(promptTokens)
? promptTokens
: null;
})()
: null,
outputTokens:
streamUsage && typeof streamUsage === "object"
? (() => {
const completionTokens = (streamUsage as Record<string, unknown>).completion_tokens;
return typeof completionTokens === "number" && Number.isFinite(completionTokens)
? completionTokens
: null;
})()
: null,
cost: null,
retries: 0,
fallbackUsed: false, // combo-level fallback tracked by decisionTrace
outcome:
normalizedStreamStatus === 200
? "success"
: streamErrorCode === "stream_interrupted" || streamErrorCode === "aborted"
? "stream_interrupted"
: outcomeFromStatus(normalizedStreamStatus),
status: normalizedStreamStatus,
finishReason: routingFinishReason(streamResponseBody),
connectionId: streamConnectionId ?? credentials?.connectionId ?? null,
})
);
persistAttemptLogs({
status: normalizedStreamStatus,
error: streamError || undefined,

View File

@@ -23,6 +23,12 @@ export interface ScoringFactors {
sessionAvailability?: number;
resetWindowAffinity: number;
connectionDensity: number;
/**
* Feedback-driven quality signal [0,1] from the routing-event quality tracker
* (open-sse/services/routing/quality.ts). Optional so cold candidates with no
* observed events default to neutral (1.0) and are never penalized.
*/
quality?: number;
}
export interface ScoringWeights {
@@ -40,11 +46,13 @@ export interface ScoringWeights {
sessionAvailability?: number;
resetWindowAffinity: number;
connectionDensity: number;
/** Weight for the feedback-driven quality factor (#feedback-foundation). */
quality?: number;
}
export const DEFAULT_WEIGHTS: ScoringWeights = {
quota: 0.1429,
health: 0.1905,
health: 0.1605,
costInv: 0.1429,
latencyInv: 0.1143,
taskFit: 0.0762,
@@ -57,6 +65,10 @@ export const DEFAULT_WEIGHTS: ScoringWeights = {
sessionAvailability: 0.0476,
resetWindowAffinity: 0,
connectionDensity: 0.0476,
// Shifted from `health` (0.1905 → 0.1605): availability stays dominant, and
// the new quality signal (observed output quality over time) gets a real,
// if smaller, vote. Sum remains exactly 1.0.
quality: 0.03,
};
/** Normalize independently configured UI weights into a scoring distribution. */
@@ -107,6 +119,12 @@ export interface ProviderCandidate {
sessionAvailability?: number;
/** Score [0..1] for quota reset-window preference; sooner selected reset windows score higher. */
resetWindowAffinity?: number;
/**
* Feedback-driven quality score [0..1] for this provider/model from the
* routing-event quality tracker (open-sse/services/routing). Omitted/undefined
* candidates default to a neutral 1.0 in calculateFactors.
*/
quality?: number;
connectionPoolSize?: number;
connectionId?: string;
}
@@ -141,7 +159,10 @@ export function calculateScore(factors: ScoringFactors, weights: ScoringWeights)
(weights.cacheAffinity ?? 0) * (factors.cacheAffinity ?? 0) +
(weights.sessionAvailability ?? 0) * (factors.sessionAvailability ?? 1) +
(weights.resetWindowAffinity ?? 0) * factors.resetWindowAffinity +
(weights.connectionDensity ?? 0) * factors.connectionDensity
(weights.connectionDensity ?? 0) * factors.connectionDensity +
// Missing quality factor → neutral 0.5: a cold candidate is neither boosted
// (which would let optimistic initialization dominate) nor penalized.
(weights.quality ?? 0) * (factors.quality ?? 0.5)
);
}
@@ -268,6 +289,9 @@ export function calculateFactors(
sessionAvailability: clamp01(candidate.sessionAvailability ?? 1),
resetWindowAffinity: clamp01(candidate.resetWindowAffinity ?? 0.5),
connectionDensity: clamp01(((candidate.connectionPoolSize ?? 1) - 1) / 10),
// Feedback quality signal; neutral 0.5 when the tracker has no data yet
// (cold providers are neither boosted nor unfairly penalized).
quality: clamp01(candidate.quality ?? 0.5),
};
}

View File

@@ -36,6 +36,7 @@ import {
import { buildNoUpstreamResponseDiagnostics, buildRecoveryHint } from "./combo/pinRecovery.ts";
import { buildTargetTimeoutRunner } from "./combo/targetTimeoutRunner.ts";
import { recordComboRequest, recordComboShadowRequest, getComboMetrics } from "./comboMetrics.ts";
import { qualityScoreFor } from "./routing/index.ts";
import {
expandComboSystemPromptIfPresent,
resolveTargetFingerprint,
@@ -578,6 +579,9 @@ export async function buildAutoCandidates(
connectionPoolSize: connectionPoolCounts.get(provider) ?? 1,
connectionId: target.connectionId ?? undefined,
authType,
// Feedback-driven quality signal (routing quality tracker). Neutral 1.0
// before enough samples accumulate — a cold model is never penalized.
quality: qualityScoreFor(provider, model),
};
})
);

View File

@@ -181,6 +181,7 @@ function applyDrr(targets: ResolvedComboTarget[], comboName: string): ResolvedCo
const deficits = getDrrDeficits(comboName);
const totalWeight = targets.reduce((sum, t) => sum + normalizeWeight(t.weight), 0);
if (totalWeight <= 0) return targets.slice();
// Add each target's quantum (weight share) to its deficit.
for (const target of targets) {
@@ -206,8 +207,9 @@ function applyDrr(targets: ResolvedComboTarget[], comboName: string): ResolvedCo
return [winner, ...rest];
}
/** Weights default to 1 and are floored at 1 to keep quantum math well-defined. */
/** Weights default to 1. Explicit 0 stays 0 so the operator can disable a target. */
function normalizeWeight(weight: number | undefined): number {
if (weight === 0) return 0;
return Number.isFinite(weight) && (weight as number) > 0 ? (weight as number) : 1;
}

View File

@@ -0,0 +1,220 @@
/**
* Routing Events — first-class representation of routing outcomes.
*
* Every request that reaches a provider emits one `RoutingEvent` describing what
* happened: which provider/model was used, under which strategy, with what
* latency/tokens/cost, and whether the outcome was a success, an error, a
* malformed response, a timeout, a rate-limit, or a blocked request.
*
* This is the "feedback foundation": the event is cheap to produce (no I/O in
* the emitting call) and is fanned out synchronously to registered sinks, each
* of which must be O(1)-ish and must never perform synchronous I/O. Sinks can
* then do whatever they need asynchronously — buffer to an OTLP exporter,
* update in-memory quality statistics, keep a bounded ring buffer for
* explainability, etc.
*
* DESIGN NOTE (adapted from the Future-AGI-inspired mission, kept deliberately
* lean): the original proposal was a Rust `RoutingEvent` struct + a
* `RoutingEventSink` trait. This module is the TypeScript equivalent, sized to
* the existing codebase: we already persist rich per-request detail in
* `call_logs` (async) and keep per-combo counters in `comboMetrics.ts`. This
* module adds the *typed, structured, sink-based* outcome channel those systems
* lacked, without duplicating either of them.
*
* SAFETY CONTRACT: an event carries ONLY routing metadata — provider, model,
* strategy, timing, token/cost numbers, an allowlisted outcome, finish reason,
* HTTP status, connection id. Never prompts, request/response bodies, headers,
* credentials, or account ids.
*/
/**
* Allowlisted routing outcomes. Keeping this an enum-like union prevents freeform
* strings from leaking into telemetry/quality logic and keeps sinks exhaustive.
*/
export const ROUTING_OUTCOMES = [
"success",
"error",
"malformed",
"timeout",
"rate_limited",
"stream_interrupted",
"guardrail_blocked",
"cancelled",
] as const;
export type RoutingOutcome = (typeof ROUTING_OUTCOMES)[number];
export interface RoutingEvent {
/** Correlation/request id — never a prompt or body. */
requestId: string;
provider: string;
model: string;
/** Combo strategy (e.g. "auto") or "direct" when not routed through a combo. */
strategy: string;
latencyMs: number;
/**
* Time-to-first-forwarded-SSE-chunk in ms (NOT token-level TTFT), or null
* for non-streaming requests / when nothing was forwarded.
*/
ttftMs: number | null;
/**
* Mean inter-chunk gap in ms — a chunk-latency proxy for inter-token latency,
* only meaningful for streaming requests. Null otherwise.
*/
itlMs: number | null;
inputTokens: number | null;
outputTokens: number | null;
cost: number | null;
retries: number;
fallbackUsed: boolean;
outcome: RoutingOutcome;
/** Upstream HTTP status; null when the request never reached a provider. */
status: number | null;
/** finish_reason from the provider response (stop / length / tool_calls / ...). */
finishReason: string | null;
connectionId: string | null;
ts: number;
}
/** A sink consumes routing events. Implementations must never do sync I/O. */
export interface RoutingEventSink {
readonly name: string;
record(event: RoutingEvent): void;
}
const sinks = new Set<RoutingEventSink>();
/**
* Register a sink. Returns an unsubscribe function. Registering the same sink
* instance twice is a no-op (Set semantics).
*/
export function registerRoutingEventSink(sink: RoutingEventSink): () => void {
sinks.add(sink);
return () => {
sinks.delete(sink);
};
}
/** Test/ops hook: list currently registered sink names. */
export function listRoutingEventSinks(): string[] {
return Array.from(sinks, (s) => s.name);
}
/** Test/ops hook: remove every registered sink. */
export function clearRoutingEventSinks(): void {
sinks.clear();
}
/**
* Emit a routing event to every registered sink. Synchronous and allocation-
* friendly so callers can invoke it at the end of the request hot path without
* measurable impact; each sink's `record()` must be cheap (enqueue/buffer only).
* A throwing sink is isolated so one misbehaving sink cannot break the router.
*/
export function dispatchRoutingEvent(event: RoutingEvent): void {
for (const sink of sinks) {
try {
sink.record(event);
} catch {
// Sinks are observability/best-effort — never let one break the data plane.
}
}
}
/**
* Bounded in-memory ring-buffer sink. Holds the most recent N events for
* explainability/debugging (see GET /api/v1/explain/routing). Insert is O(1);
* no TTL sweep needed because the buffer is size-bounded by construction.
*/
export class MemoryRoutingEventStore implements RoutingEventSink {
readonly name = "memory";
private buffer: RoutingEvent[] = [];
private cursor = 0;
constructor(private readonly capacity = 500) {}
record(event: RoutingEvent): void {
if (this.buffer.length < this.capacity) {
this.buffer.push(event);
} else {
this.buffer[this.cursor] = event;
}
this.cursor = (this.cursor + 1) % this.capacity;
}
/** Most recent events, newest first, up to `limit`. */
recent(limit = 50): RoutingEvent[] {
if (this.buffer.length < this.capacity) {
return this.buffer.slice(-limit).reverse();
}
// Ring is full — walk backwards from the cursor.
const out: RoutingEvent[] = [];
for (let i = 0; i < Math.min(limit, this.buffer.length); i++) {
const idx = (this.cursor - 1 - i + this.buffer.length) % this.buffer.length;
out.push(this.buffer[idx]);
}
return out;
}
clear(): void {
this.buffer = [];
this.cursor = 0;
}
get size(): number {
return this.buffer.length;
}
}
/** Create a well-formed event with defaults for unset observability fields. */
export function createRoutingEvent(input: {
requestId: string;
provider: string;
model: string;
strategy?: string | null;
latencyMs: number;
ttftMs?: number | null;
itlMs?: number | null;
inputTokens?: number | null;
outputTokens?: number | null;
cost?: number | null;
retries?: number;
fallbackUsed?: boolean;
outcome: RoutingOutcome;
status?: number | null;
finishReason?: string | null;
connectionId?: string | null;
ts?: number;
}): RoutingEvent {
return {
requestId: input.requestId,
provider: input.provider || "unknown",
model: input.model || "unknown",
strategy: input.strategy ?? "direct",
latencyMs: Math.max(0, input.latencyMs || 0),
ttftMs: input.ttftMs ?? null,
itlMs: input.itlMs ?? null,
inputTokens: input.inputTokens ?? null,
outputTokens: input.outputTokens ?? null,
cost: input.cost ?? null,
retries: input.retries ?? 0,
fallbackUsed: input.fallbackUsed ?? false,
outcome: input.outcome,
status: input.status ?? null,
finishReason: input.finishReason ?? null,
connectionId: input.connectionId ?? null,
ts: input.ts ?? Date.now(),
};
}
/**
* Classify an upstream HTTP status into a RoutingOutcome. Status 200/201 → success;
* 429 → rate_limited; 408/504 → timeout; 4xx/5xx → error; anything else → error.
*/
export function outcomeFromStatus(status: number | null | undefined): RoutingOutcome {
if (status == null) return "error";
if (status === 200 || status === 201) return "success";
if (status === 429) return "rate_limited";
if (status === 408 || status === 504) return "timeout";
return "error";
}

View File

@@ -0,0 +1,133 @@
/**
* Routing feedback foundation — default wiring.
*
* Bootstraps the default routing-event sinks:
* 1. `MemoryRoutingEventStore` — bounded ring buffer for explainability.
* 2. `QualityTracker` consumer — feeds the auto-combo `quality` scoring factor.
* 3. Optional OTel/HTTP exporter — enabled only when an OTLP endpoint is set.
*
* The hot path only calls `emitRoutingEvent()`, which fans out synchronously to
* these cheap in-memory sinks. No synchronous I/O, no external dependencies.
*
* This is the adapter seam Future AGI (or any evaluation backend) can plug into
* later without becoming a dependency: an evaluator would be another
* `RoutingEventSink` (or a consumer of the ring buffer / quality snapshot).
*/
import {
clearRoutingEventSinks,
dispatchRoutingEvent,
listRoutingEventSinks,
MemoryRoutingEventStore,
registerRoutingEventSink,
type RoutingEvent,
type RoutingEventSink,
} from "./events.ts";
import {
getProviderQuality,
getQualityScore,
getQualitySnapshot,
recordQualityEvent,
resetQualityTracker,
setSemanticQuality,
type ProviderQuality,
} from "./quality.ts";
import { isRoutingOtelEnabled, OtlpHttpsEventSink } from "./otel.ts";
const memoryStore = new MemoryRoutingEventStore(500);
// The quality tracker is registered as a sink so it updates inline with the
// event (O(1) math) and the OTel exporter only ever enqueues.
const qualitySink: RoutingEventSink = {
name: "quality",
record(event: RoutingEvent): void {
recordQualityEvent(event);
},
};
let otelSink: OtlpHttpsEventSink | null = null;
let initialized = false;
/** Register the default sinks. Idempotent; safe to call multiple times. */
export function initRoutingObservability(env: NodeJS.ProcessEnv = process.env): {
sinks: string[];
otelEnabled: boolean;
} {
if (initialized) {
return { sinks: listRoutingSinkNames(), otelEnabled: isRoutingOtelEnabled(env) };
}
initialized = true;
registerRoutingEventSink(memoryStore);
registerRoutingEventSink(qualitySink);
if (isRoutingOtelEnabled(env)) {
const endpoint = (env.OMNIROUTE_OTEL_ENDPOINT ?? env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "").trim();
otelSink = new OtlpHttpsEventSink({
endpoint,
serviceName: env.OTEL_SERVICE_NAME ?? "omniroute",
maxBatchSize: 64,
flushIntervalMs: 10_000,
});
registerRoutingEventSink(otelSink);
}
return { sinks: listRoutingSinkNames(), otelEnabled: otelSink != null };
}
/** Emit a routing event to all registered sinks (fire-and-forget, cheap). */
export function emitRoutingEvent(event: RoutingEvent): void {
if (!initialized) initRoutingObservability();
dispatchRoutingEvent(event);
}
/** Neutral default quality used when a model has no observed events. */
export function qualityScoreFor(provider: string, model: string): number {
return getQualityScore(provider, model);
}
/** Full per-provider/model quality view (operational + semantic + confidence). */
export function providerQualityFor(provider: string, model: string): ProviderQuality {
return getProviderQuality(provider, model);
}
/**
* Evaluator seam: record a semantic quality score. NEVER call this from the
* request hot path with HTTP-derived signals — semantic quality is reserved for
* actual evaluation (task success, tool-use correctness, groundedness).
*/
export { setSemanticQuality } from "./quality.ts";
export function routingQualitySnapshot(limit = 200): ReturnType<typeof getQualitySnapshot> {
return getQualitySnapshot(limit);
}
export { classifyQuality, type QualityClassification } from "./quality.ts";
export function recentRoutingEvents(limit = 50): RoutingEvent[] {
return memoryStore.recent(limit);
}
export function routingOtelStats(): { buffered: number; dropped: number } | null {
return otelSink ? otelSink.getStats() : null;
}
function listRoutingSinkNames(): string[] {
return listRoutingEventSinks();
}
/** Test/ops hook: full reset of the routing observability layer. */
export function resetRoutingObservability(): void {
clearRoutingEventSinks();
memoryStore.clear();
resetQualityTracker();
if (otelSink) {
otelSink.stop();
otelSink = null;
}
initialized = false;
}
export type { RoutingEvent, RoutingOutcome, RoutingEventSink } from "./events.ts";
export { createRoutingEvent, outcomeFromStatus } from "./events.ts";

View File

@@ -0,0 +1,227 @@
/**
* Optional OpenTelemetry / GenAI observability sink.
*
* A `RoutingEventSink` that forwards routing events to an OTLP/HTTP collector as
* GenAI semantic-convention spans (semconvgenai: `gen_ai.provider.name`,
* `gen_ai.request.model`, `gen_ai.operation.name`, `gen_ai.usage.input_tokens`,
* `gen_ai.usage.output_tokens`, etc.).
*
* Deliberately lightweight:
* - No `@opentelemetry/*` SDK dependency. Uses the collector's OTLP/HTTP JSON
* (traces) endpoint via global `fetch`, which is already available and async.
* - `record()` only enqueues into a bounded buffer (O(1), never I/O). A single
* background flush timer drains the buffer asynchronously. Under overload the
* oldest events are dropped (never backpressure the data plane).
* - Disabled unless `OMNIROUTE_OTEL_ENDPOINT` (or `OTEL_EXPORTER_OTLP_ENDPOINT`)
* is set — normal lightweight deployments run with zero OTel code executing.
* - No secrets/prompts are ever serialized; only RoutingEvent metadata.
*/
export interface OtlpHttpsExporterConfig {
/** Collector base URL, e.g. https://collector:4318 — spans go to /v1/traces. */
endpoint: string;
/** Export batch size / flush interval. */
maxBatchSize?: number;
flushIntervalMs?: number;
serviceName?: string;
}
/** Resolve whether OTLP export is configured. */
export function isRoutingOtelEnabled(env: NodeJS.ProcessEnv = process.env): boolean {
const endpoint = (env.OMNIROUTE_OTEL_ENDPOINT ?? env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "").trim();
return endpoint.length > 0;
}
interface OtelSpan {
traceId: string;
spanId: string;
name: string;
kind: number;
startTimeUnixNano: string;
endTimeUnixNano: string;
attributes: Array<{
key: string;
value: { stringValue?: string; intValue?: string; doubleValue?: number };
}>;
}
function toHex(bytes: Uint8Array): string {
return Array.from(bytes, (b) => b.toString(16).padStart(2, "0")).join("");
}
function randomId(bytes: number): string {
const arr = new Uint8Array(bytes);
// Use crypto.getRandomValues when available (Node ≥ 19 global), else Math.random.
if (typeof crypto !== "undefined" && typeof crypto.getRandomValues === "function") {
crypto.getRandomValues(arr);
} else {
for (let i = 0; i < bytes; i++) arr[i] = Math.floor(Math.random() * 256);
}
return toHex(arr);
}
export class OtlpHttpsEventSink {
readonly name = "otel";
private readonly endpoint: string;
private readonly maxBatchSize: number;
private readonly serviceName: string;
private buffer: RoutingEventLike[] = [];
private dropped = 0;
private consecutiveFailures = 0;
private flushedBatches = 0;
private timer: ReturnType<typeof setInterval> | null = null;
private flushing = false;
constructor(private readonly config: OtlpHttpsExporterConfig) {
this.endpoint = config.endpoint.replace(/\/+$/, "") + "/v1/traces";
this.maxBatchSize = config.maxBatchSize ?? 64;
this.serviceName = config.serviceName ?? "omniroute";
this.start();
}
/** O(1) enqueue; drops oldest when the buffer is full. Never performs I/O. */
record(event: RoutingEventLike): void {
if (this.buffer.length >= this.maxBatchSize * 4) {
this.buffer.shift();
this.dropped += 1;
}
this.buffer.push(event);
}
getStats(): {
buffered: number;
dropped: number;
consecutiveFailures: number;
flushedBatches: number;
} {
return {
buffered: this.buffer.length,
dropped: this.dropped,
consecutiveFailures: this.consecutiveFailures,
flushedBatches: this.flushedBatches,
};
}
stop(): void {
if (this.timer) {
clearInterval(this.timer);
this.timer = null;
}
void this.flush();
}
private start(): void {
const intervalMs = this.config.flushIntervalMs ?? 10_000;
this.timer = setInterval(() => void this.flush(), intervalMs);
// Do not keep the process alive just for telemetry.
this.timer.unref?.();
}
private async flush(): Promise<void> {
if (this.flushing) return;
if (this.buffer.length === 0) return;
this.flushing = true;
const batch = this.buffer.splice(0, this.maxBatchSize);
try {
const res = await fetch(this.endpoint, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(buildOtlpTracesPayload(batch, this.serviceName)),
signal: AbortSignal.timeout(3000),
});
if (!res.ok) throw new Error(`OTLP collector returned ${res.status}`);
this.consecutiveFailures = 0;
this.flushedBatches += 1;
} catch {
// Telemetry delivery is best-effort. Re-buffer for a retry, but stop after
// MAX_CONSECUTIVE_FAILURES so a permanently-unavailable collector cannot
// grow the buffer without bound. The dropped counter reflects the loss.
this.consecutiveFailures += 1;
if (this.consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) {
this.dropped += batch.length;
} else {
this.buffer.unshift(...batch);
}
} finally {
this.flushing = false;
}
}
}
/** Drop a batch (and count it) after this many consecutive collector failures. */
const MAX_CONSECUTIVE_FAILURES = 5;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
type RoutingEventLike = any;
/**
* Build an OTLP/HTTP traces JSON payload with one span per routing event,
* mapped to GenAI semantic conventions.
*/
export function buildOtlpTracesPayload(events: RoutingEventLike[], serviceName: string): unknown {
const resourceSpans = [
{
resource: {
attributes: [
{ key: "service.name", value: { stringValue: serviceName } },
{ key: "telemetry.sdk.name", value: { stringValue: "omniroute-routing" } },
],
},
scopeSpans: [
{
scope: { name: "omniroute.routing" },
spans: events.map(toSpan),
},
],
},
];
return { resourceSpans };
}
function attr(
key: string,
value: string | number
): { key: string; value: { stringValue?: string; intValue?: string; doubleValue?: number } } {
if (typeof value === "number") {
return Number.isInteger(value)
? { key, value: { intValue: String(value) } }
: { key, value: { doubleValue: value } };
}
return { key, value: { stringValue: String(value) } };
}
function toSpan(event: RoutingEventLike): OtelSpan {
const traceId = randomId(16);
const spanId = randomId(8);
const startNs = BigInt(event.ts) * 1_000_000n;
const endNs = startNs + BigInt(Math.max(0, event.latencyMs || 0)) * 1_000_000n;
const attributes = [
attr("gen_ai.provider.name", event.provider),
attr("gen_ai.request.model", event.model),
attr("gen_ai.operation.name", "chat"),
attr("gen_ai.system", event.strategy || "direct"),
attr("gen_ai.usage.input_tokens", event.inputTokens ?? 0),
attr("gen_ai.usage.output_tokens", event.outputTokens ?? 0),
attr("gen_ai.completion.finish_reason", event.finishReason ?? "unknown"),
attr("gen_ai.request.temperature", 0),
attr("omniroute.routing.outcome", event.outcome),
attr("omniroute.routing.status", event.status ?? 0),
attr("omniroute.routing.ttft_ms", event.ttftMs ?? -1),
attr("omniroute.routing.itl_ms", event.itlMs ?? -1),
attr("omniroute.routing.retries", event.retries ?? 0),
attr("omniroute.routing.fallback_used", event.fallbackUsed ? 1 : 0),
attr("gen_ai.client.token.usage.input_tokens", event.inputTokens ?? 0),
attr("gen_ai.client.token.usage.output_tokens", event.outputTokens ?? 0),
];
if (event.connectionId) attributes.push(attr("omniroute.connection_id", event.connectionId));
return {
traceId,
spanId,
name: `chat ${event.provider}/${event.model}`,
kind: 3, // CLIENT
startTimeUnixNano: startNs.toString(),
endTimeUnixNano: endNs.toString(),
attributes,
};
}

View File

@@ -0,0 +1,313 @@
/**
* Provider/Model Quality Signal — feedback-driven adaptive routing (v2).
*
* v2 separates two distinct concepts that v1 conflated:
*
* - **Operational quality** — derived from the routing hot path (HTTP status,
* connection failures, 429s, malformed responses, stream interruptions,
* finish_reason anomalies, zero-output successes, latency/TTFT). A request
* returning HTTP 200 is NOT necessarily high quality; operational quality
* only says "the wire behaved."
* - **Semantic quality** — the actual value of the generated output
* (evaluator score, task success, tool-use correctness, factual accuracy).
* This is ONLY ever produced by an external evaluator via
* `setSemanticQuality()`. It is never manufactured from HTTP success. It is
* `null` until an evaluator provides a value.
*
* Confidence / sample awareness (v2):
* - `confidence = clamp01(samples / CONFIDENCE_FULL_SAMPLES)`.
* - The score returned to the scorer is blended toward the neutral midpoint
* (0.5): `score = NEUTRAL + confidence * (operational - NEUTRAL)`.
* - Consequences: a cold provider (0 samples) scores neutral 0.5 — it is not
* unfairly penalized, but it also cannot dominate a provider with thousands
* of solid observations. A provider with 7 lucky successes is pulled toward
* 0.5, so it never dominates purely from optimistic initialization.
*
* This complements the existing resilience stack (circuit breaker, connection
* cooldown, model lockout, health matrix): those handle *availability* (hard
* exclusion); this signal handles *soft adaptive preference*.
*
* Statistics are plain arithmetic (EWMA + small counters), O(1) per event, safe
* under the Node event loop's single thread — no lock-free/atomic trickery.
*/
/** EWMA smoothing factor (alpha). Lower = slower adaptation. */
const OPERATIONAL_ALPHA = 0.2;
/** Latency EWMA alpha — slower so transient spikes don't tank quality instantly. */
const LATENCY_ALPHA = 0.1;
/** Samples at which confidence reaches 1.0 (full confidence). */
const CONFIDENCE_FULL_SAMPLES = 50;
/** Neutral score used for cold/unknown providers (midpoint, neither boosted nor penalized). */
const NEUTRAL_SCORE = 0.5;
interface QualityState {
/** EWMA of the success indicator (1 = good, 0 = bad). */
successEwma: number;
/** EWMA of latency in ms. */
latencyEwma: number;
/** EWMA of TTFT in ms (streaming only). */
ttftEwma: number | null;
/** Total events observed for this (provider, model). */
samples: number;
/** Count of operational-anomaly events (malformed / empty / length / interrupted). */
anomalies: number;
/** Rate-limit (429) count — tracked separately for observability. */
rateLimited: number;
/** Semantic quality [0,1] from an external evaluator, if one has provided it. */
semantic: number | null;
/** Confidence [0,1] of the semantic score as reported by the evaluator. */
semanticConfidence: number | null;
lastTs: number;
}
const states = new Map<string, QualityState>();
function keyOf(provider: string, model: string): string {
return `${provider}/${model}`;
}
function getOrCreate(key: string): QualityState {
let state = states.get(key);
if (!state) {
state = {
successEwma: 1,
latencyEwma: 0,
ttftEwma: null,
samples: 0,
anomalies: 0,
rateLimited: 0,
semantic: null,
semanticConfidence: null,
lastTs: 0,
};
states.set(key, state);
}
return state;
}
function isOperationalAnomaly(event: {
outcome: string;
finishReason: string | null;
outputTokens: number | null | undefined;
}): boolean {
if (event.outcome === "malformed" || event.outcome === "stream_interrupted") return true;
// finish_reason=length → the model ran out of output budget (truncated answer).
if (event.outcome === "success" && event.finishReason === "length") return true;
// A "successful" 200 that produced zero output tokens is an empty/invalid output.
// NOTE: we deliberately do NOT treat a missing finish_reason as an anomaly —
// streaming passthrough frequently has no reconstructed finish_reason, so that
// signal would penalize every legitimately streamed request (pure noise).
if (event.outcome === "success" && event.outputTokens === 0) return true;
return false;
}
function successIndicator(event: { outcome: string; status: number | null }): number {
if (event.outcome === "success") return 1;
// 429 is a transient signal, not a quality failure — treat as neutral-positive.
if (event.outcome === "rate_limited" || event.status === 429) return 0.5;
return 0;
}
/** Record one operational routing event into the quality estimate. O(1). */
export function recordQualityEvent(event: {
provider: string;
model: string;
outcome: string;
status: number | null;
latencyMs: number;
ttftMs?: number | null;
finishReason?: string | null;
outputTokens?: number | null;
ts?: number;
}): void {
const key = keyOf(event.provider || "unknown", event.model || "unknown");
const state = getOrCreate(key);
state.samples += 1;
if (
isOperationalAnomaly({
outcome: event.outcome,
finishReason: event.finishReason ?? null,
outputTokens: event.outputTokens ?? undefined,
})
) {
state.anomalies += 1;
}
if (event.outcome === "rate_limited" || event.status === 429) state.rateLimited += 1;
const indicator = successIndicator({ outcome: event.outcome, status: event.status });
// First sample seeds the EWMA directly (no lag toward a default).
state.successEwma =
state.samples === 1
? indicator
: state.successEwma + OPERATIONAL_ALPHA * (indicator - state.successEwma);
const latency = Number.isFinite(event.latencyMs) && event.latencyMs >= 0 ? event.latencyMs : 0;
state.latencyEwma =
state.samples === 1
? latency
: state.latencyEwma + LATENCY_ALPHA * (latency - state.latencyEwma);
const ttft = event.ttftMs;
if (typeof ttft === "number" && Number.isFinite(ttft) && ttft >= 0) {
state.ttftEwma =
state.ttftEwma == null ? ttft : state.ttftEwma + LATENCY_ALPHA * (ttft - state.ttftEwma);
}
state.lastTs = event.ts ?? Date.now();
}
/**
* Evaluator seam: record a semantic quality score for a (provider, model).
* Semantic quality is ONLY ever produced by an evaluator (deterministic scorer,
* local LLM judge, HTTP/Future-AGI adapter, WASM). It is never manufactured from
* operational/HTP success. `confidence` should reflect the evaluator's certainty
* (e.g. number of eval cases backing the score).
*/
export function setSemanticQuality(
provider: string,
model: string,
score: number,
confidence: number
): void {
const state = getOrCreate(keyOf(provider || "unknown", model || "unknown"));
state.semantic = Math.max(0, Math.min(1, Number.isFinite(score) ? score : 0.5));
state.semanticConfidence = Math.max(0, Math.min(1, Number.isFinite(confidence) ? confidence : 0));
}
export interface ProviderQuality {
provider: string;
model: string;
/** Operational score [0,1] (wire behavior) — confidence-adjusted, neutral 0.5 cold. */
operational: number;
/** Semantic score [0,1] from an evaluator, or null when none has been provided. */
semantic: number | null;
/** Confidence [0,1] of the operational score (sample-count based). */
confidence: number;
/** Confidence [0,1] of the semantic score, when an evaluator reported one. */
semanticConfidence: number | null;
samples: number;
anomalies: number;
rateLimited: number;
successEwma: number;
latencyEwmaMs: number;
ttftEwmaMs: number | null;
/** Milliseconds since the last observed event; null when never observed. */
recencyMs: number | null;
lastTs: number;
}
/** Raw operational score before the confidence blend (pure EWMA + penalties). */
function rawOperationalScore(state: QualityState): number {
let score = state.successEwma;
// Latency degradation: soft penalty capped at 0.2 so slow models are discounted, not zeroed.
const latencyPenalty = Math.min(0.2, state.latencyEwma / 60_000);
score -= latencyPenalty;
// Anomaly penalty: capped so a few bad apples don't nuke a provider entirely.
const anomalyRate = state.anomalies / Math.max(1, state.samples);
score -= Math.min(0.25, anomalyRate * 0.5);
return Math.max(0, Math.min(1, score));
}
function confidenceOf(samples: number): number {
return Math.max(0, Math.min(1, samples / CONFIDENCE_FULL_SAMPLES));
}
/**
* Operational quality for a (provider, model), confidence-adjusted and blended
* toward the neutral midpoint. See module docs for the cold-start guarantee.
*/
export function getProviderQuality(provider: string, model: string): ProviderQuality {
const state = states.get(keyOf(provider, model));
const now = Date.now();
if (!state || state.samples === 0) {
return {
provider,
model,
operational: NEUTRAL_SCORE,
semantic: null,
confidence: 0,
semanticConfidence: null,
samples: 0,
anomalies: 0,
rateLimited: 0,
successEwma: 1,
latencyEwmaMs: 0,
ttftEwmaMs: null,
recencyMs: null,
lastTs: 0,
};
}
const confidence = confidenceOf(state.samples);
const raw = rawOperationalScore(state);
const operational = NEUTRAL_SCORE + confidence * (raw - NEUTRAL_SCORE);
return {
provider,
model,
operational,
semantic: state.semantic,
confidence,
semanticConfidence: state.semanticConfidence,
samples: state.samples,
anomalies: state.anomalies,
rateLimited: state.rateLimited,
successEwma: state.successEwma,
latencyEwmaMs: state.latencyEwma,
ttftEwmaMs: state.ttftEwma,
recencyMs: state.samples > 0 ? Math.max(0, now - state.lastTs) : null,
lastTs: state.lastTs,
};
}
/**
* Backward-compatible scalar used by the auto-combo scorer's `quality` factor.
* Returns the confidence-adjusted operational score (neutral 0.5 when cold).
*/
export function getQualityScore(provider: string, model: string): number {
return getProviderQuality(provider, model).operational;
}
/** Full snapshot of the tracker for explainability / dashboard. */
export function getQualitySnapshot(limit = 200): ProviderQuality[] {
const views: ProviderQuality[] = [];
for (const [key] of states) {
const slash = key.indexOf("/");
const provider = slash >= 0 ? key.slice(0, slash) : key;
const model = slash >= 0 ? key.slice(slash + 1) : key;
views.push(getProviderQuality(provider, model));
}
views.sort((a, b) => b.lastTs - a.lastTs);
return views.slice(0, limit);
}
/**
* Classify a provider/model quality state for explainability / dashboard.
* This reflects the SOFT adaptive signal — it says nothing about hard exclusion
* (circuit open / quota / auth), which is owned by the resilience stack.
*
* - "healthy": high confidence + operational quality well above neutral
* - "degraded": operational quality at or below neutral (soft penalty active)
* - "warming": low confidence (few samples) — treated neutrally
* - "cold": never observed — neutral, cannot dominate
*/
export type QualityClassification = "healthy" | "degraded" | "warming" | "cold";
export function classifyQuality(q: ProviderQuality): QualityClassification {
if (q.samples === 0) return "cold";
if (q.confidence < 0.5) return "warming";
if (q.operational < 0.5) return "degraded";
return "healthy";
}
/** Test/ops hook: reset all quality state. */
export function resetQualityTracker(): void {
states.clear();
}
export const QUALITY_WELL_KNOWN = {
CONFIDENCE_FULL_SAMPLES,
NEUTRAL_SCORE,
} as const;

View File

@@ -81,6 +81,7 @@ import {
import { restoreClaudeToolName } from "../services/claudeCodeToolRemapper.ts";
import { normalizeFinalOpenAIStreamChunk } from "./openAIStreamChunk.ts";
import { collectClaudeDelta } from "./streamClaudeDelta.ts";
import { createStreamTiming, type StreamTiming } from "./streamTiming.ts";
/**
* Race a response body read against a timeout.
@@ -129,7 +130,15 @@ type StreamCompletePayload = {
clientPayload?: unknown;
error?: string | null;
errorCode?: string | null;
/**
* Time-to-first-forwarded-SSE-chunk in ms, or null when nothing was forwarded.
* NOT token-level TTFT — see open-sse/utils/streamTiming.ts for what is measured.
*/
ttft?: number | null;
/** Mean inter-chunk gap in ms (chunk-latency proxy for ITL), or null. */
itlMs?: number | null;
/** True when the stream was interrupted (timeout/abort/error) before a clean finish. */
interrupted?: boolean;
};
type StreamOptions = {
@@ -577,7 +586,10 @@ function getOpenAIIntermediateChunks(value: unknown): unknown[] {
return Array.isArray(candidate) ? candidate : [];
}
export function restoreClaudePassthroughToolUseName(parsed: JsonRecord, toolNameMap: unknown): boolean {
export function restoreClaudePassthroughToolUseName(
parsed: JsonRecord,
toolNameMap: unknown
): boolean {
const block =
parsed.content_block && typeof parsed.content_block === "object"
? (parsed.content_block as JsonRecord)
@@ -660,6 +672,16 @@ export function createSSEStream(options: StreamOptions = {}) {
performance.clearMarks("omni-request-body-size");
}
// Canonical streaming timing (TTFT / ITL / interruption). One instance per
// stream, marked from the transform below. ttft() = first-forwarded-SSE-chunk
// latency (NOT token-level) — see streamTiming.ts.
const timing: StreamTiming = createStreamTiming();
/** Forward a pre-encoded SSE chunk, marking TTFT/ITL on the way. */
const forward = (controller: TransformStreamDefaultController<Uint8Array>, bytes: Uint8Array) => {
timing.markForward();
controller.enqueue(bytes);
};
// Drop internal commentary-phase Responses output before forwarding (#6199).
// Explicit option wins; otherwise read the feature flag (default on) — resolved once per stream.
const shouldDropResponsesCommentary =
@@ -948,7 +970,7 @@ export function createSSEStream(options: StreamOptions = {}) {
clientPayloadCollector.push(event);
const output = formatSSE(event, FORMATS.CLAUDE);
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
forward(controller, encoder.encode(output));
}
};
@@ -973,7 +995,8 @@ export function createSSEStream(options: StreamOptions = {}) {
const errOutput = formatSSE(errorEvent, FORMATS.CLAUDE);
reqLogger?.appendConvertedChunk?.(errOutput);
clientPayloadCollector.push(errorEvent);
controller.enqueue(encoder.encode(errOutput));
forward(controller, encoder.encode(errOutput));
timing.markInterrupted();
let failureHandled = false;
if (onFailure) {
try {
@@ -1034,7 +1057,7 @@ export function createSSEStream(options: StreamOptions = {}) {
clientPayloadCollector.push(itemSanitized);
reqLogger?.appendConvertedChunk?.(output);
forwardedValuableChunk = true;
controller.enqueue(encoder.encode(output));
forward(controller, encoder.encode(output));
};
const emitFinalSseMetadata = async (
@@ -1059,7 +1082,7 @@ export function createSSEStream(options: StreamOptions = {}) {
});
if (!comment) return;
reqLogger?.appendConvertedChunk?.(comment);
controller.enqueue(encoder.encode(comment));
forward(controller, encoder.encode(comment));
};
const getResponsesReasoningKey = (payload: Record<string, unknown>): string | null => {
@@ -1146,7 +1169,7 @@ export function createSSEStream(options: StreamOptions = {}) {
clientPayloadCollector.push(syntheticEvent.body);
const output = `event: ${syntheticEvent.event}\ndata: ${JSON.stringify(syntheticEvent.body)}\n\n`;
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
forward(controller, encoder.encode(output));
}
};
@@ -1164,6 +1187,7 @@ export function createSSEStream(options: StreamOptions = {}) {
let failureHandled = false;
if (onFailure) {
try {
timing.markInterrupted();
failureHandled =
onFailure({
status: HTTP_STATUS.GATEWAY_TIMEOUT,
@@ -1195,6 +1219,7 @@ export function createSSEStream(options: StreamOptions = {}) {
transform(chunk, controller) {
if (streamTimedOut) return;
const now = Date.now();
timing.markByte();
lastChunkTime = now;
const text = decoder.decode(chunk, { stream: true });
buffer += text;
@@ -1253,7 +1278,7 @@ export function createSSEStream(options: StreamOptions = {}) {
const pendingOutput = passthroughEventPrefix.flush();
if (pendingOutput) {
reqLogger?.appendConvertedChunk?.(pendingOutput);
controller.enqueue(encoder.encode(pendingOutput));
forward(controller, encoder.encode(pendingOutput));
}
clearPendingPassthroughEvent();
continue;
@@ -1420,7 +1445,7 @@ export function createSSEStream(options: StreamOptions = {}) {
clientPayloadCollector.push(event);
}
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
forward(controller, encoder.encode(output));
injectedUsage = true;
} else {
output = `data: ${JSON.stringify(parsed)}\n\n`;
@@ -1709,7 +1734,7 @@ export function createSSEStream(options: StreamOptions = {}) {
clientPayload = parsed;
clientPayloadCollector.push(clientPayload);
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
forward(controller, encoder.encode(output));
continue;
}
@@ -1785,7 +1810,7 @@ export function createSSEStream(options: StreamOptions = {}) {
totalContentLength += delta.reasoning_content.length;
clientPayloadCollector.push(reasoningChunk);
reqLogger?.appendConvertedChunk?.(rOutput);
controller.enqueue(encoder.encode(rOutput));
forward(controller, encoder.encode(rOutput));
delete delta.reasoning_content;
splitMixedReasoningContent = true;
}
@@ -1964,7 +1989,7 @@ export function createSSEStream(options: StreamOptions = {}) {
}
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
forward(controller, encoder.encode(output));
if (failurePayload) {
let failureHandled = false;
if (onFailure) {
@@ -2004,7 +2029,7 @@ export function createSSEStream(options: StreamOptions = {}) {
if (parsed.error) {
const output = formatTranslatedStreamError(parsed, sourceFormat);
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
forward(controller, encoder.encode(output));
upstreamErrorForwarded = true;
doneSent = true;
continue;
@@ -2223,7 +2248,7 @@ export function createSSEStream(options: StreamOptions = {}) {
passthroughEventPrefix,
emitConvertedOutput: (output: string) => {
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
forward(controller, encoder.encode(output));
},
pushProviderPayload: (payload: unknown) => providerPayloadCollector.push(payload),
pushClientPayload: (payload: unknown) => clientPayloadCollector.push(payload),
@@ -2336,7 +2361,7 @@ export function createSSEStream(options: StreamOptions = {}) {
output = output.endsWith("\n") ? `${output}\n` : `${output}\n\n`;
}
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
forward(controller, encoder.encode(output));
}
if (shouldInjectClaudeEmptyResponseOnFlush(claudeEmptyResponseLifecycle)) {
@@ -2380,7 +2405,7 @@ export function createSSEStream(options: StreamOptions = {}) {
flushOutput = `data: ${JSON.stringify(syntheticChunk)}\n\n`;
}
reqLogger?.appendConvertedChunk?.(flushOutput);
controller.enqueue(encoder.encode(flushOutput));
forward(controller, encoder.encode(flushOutput));
passthroughAccumulatedContent = appendBoundedText(
passthroughAccumulatedContent,
passthroughBufferedTextualToolCallContent
@@ -2397,7 +2422,7 @@ export function createSSEStream(options: StreamOptions = {}) {
totalContentLength += thinkFlush.addedLength;
clientPayloadCollector.push(thinkFlush.syntheticChunk);
reqLogger?.appendConvertedChunk?.(thinkFlush.flushOutput);
controller.enqueue(encoder.encode(thinkFlush.flushOutput));
forward(controller, encoder.encode(thinkFlush.flushOutput));
}
// Estimate usage if provider didn't return valid usage
@@ -2431,7 +2456,7 @@ export function createSSEStream(options: StreamOptions = {}) {
);
const finishOutput = `data: ${JSON.stringify(syntheticFinishChunk)}\n\n`;
reqLogger?.appendConvertedChunk?.(finishOutput);
controller.enqueue(encoder.encode(finishOutput));
forward(controller, encoder.encode(finishOutput));
clientPayloadCollector.push(syntheticFinishChunk);
}
await emitFinalSseMetadata(controller, usage);
@@ -2440,7 +2465,7 @@ export function createSSEStream(options: StreamOptions = {}) {
clientPayloadCollector.push({ done: true });
const doneOutput = "data: [DONE]\n\n";
reqLogger?.appendConvertedChunk?.(doneOutput);
controller.enqueue(encoder.encode(doneOutput));
forward(controller, encoder.encode(doneOutput));
}
}
// Notify caller for call log persistence (include full response body with accumulated content)
@@ -2514,6 +2539,9 @@ export function createSSEStream(options: StreamOptions = {}) {
status: 200,
usage,
responseBody,
ttft: timing.ttftMs(),
itlMs: timing.avgItlMs(),
interrupted: timing.interrupted,
// #9315 switched the summary to the accumulated responseBody to avoid
// stale/truncated event data — but responseBody here is synthesized in
// chat-completion shape, which loses the Responses API `response` object.
@@ -2616,6 +2644,7 @@ export function createSSEStream(options: StreamOptions = {}) {
let failureHandled = false;
if (onFailure) {
try {
timing.markInterrupted();
failureHandled =
onFailure({
status: err.status,
@@ -2635,6 +2664,9 @@ export function createSSEStream(options: StreamOptions = {}) {
status: err.status,
usage: state?.usage,
responseBody: errorBody,
ttft: timing.ttftMs(),
itlMs: timing.avgItlMs(),
interrupted: timing.interrupted,
error: err.message,
errorCode: err.code,
providerPayload: providerPayloadCollector.build(
@@ -2731,7 +2763,7 @@ export function createSSEStream(options: StreamOptions = {}) {
clientPayloadCollector.push({ done: true });
const doneOutput = "data: [DONE]\n\n";
reqLogger?.appendConvertedChunk?.(doneOutput);
controller.enqueue(encoder.encode(doneOutput));
forward(controller, encoder.encode(doneOutput));
}
}

View File

@@ -0,0 +1,83 @@
/**
* Canonical streaming timing instrumentation (TTFT / ITL / interruption).
*
* One reusable seam for measuring the streaming path. It is created once per
* stream and marked from the SSE transform:
*
* markByte() — first upstream chunk received (bytes arrived from provider)
* markForward() — first chunk forwarded to the client (first SSE chunk enqueued)
*
* `ttft()` is therefore **first-forwarded-SSE-chunk latency**, NOT token-level
* TTFT. We document that distinction explicitly: a single SSE chunk can carry
* zero, one, or many tokens, and chunk boundaries do not map to token
* boundaries. If a future implementation can measure actual token timing it
* should extend this seam, not bypass it.
*
* ITL (inter-token latency) is approximated by the mean gap between forwarded
* SSE chunks (bounded sample window). It is a chunk-latency proxy, again not
* true token timing — callers must label it as such.
*
* The object is cheap to construct, plain mutable state, and safe under the
* event loop's single thread (each stream owns its own instance).
*/
export interface StreamTiming {
startedAt: number;
firstByteAt: number | null;
firstForwardAt: number | null;
lastForwardAt: number | null;
/** Mean gap between forwarded chunks (ms), bounded window. */
interChunkGaps: number[];
forwardedChunks: number;
interrupted: boolean;
markByte(): void;
markForward(): void;
markInterrupted(): void;
/** First-forwarded-SSE-chunk latency in ms, or null if nothing was forwarded. */
ttftMs(): number | null;
/** Mean inter-chunk gap in ms, or null when fewer than 2 chunks were forwarded. */
avgItlMs(): number | null;
/** Time from stream start to completion (ms). */
totalMs(): number;
}
/** Max number of inter-chunk samples kept (bounds memory). */
const MAX_INTER_CHUNK_GAPS = 32;
export function createStreamTiming(): StreamTiming {
const timing: StreamTiming = {
startedAt: Date.now(),
firstByteAt: null,
firstForwardAt: null,
lastForwardAt: null,
interChunkGaps: [],
forwardedChunks: 0,
interrupted: false,
markByte() {
if (this.firstByteAt === null) this.firstByteAt = Date.now();
},
markForward() {
const now = Date.now();
if (this.firstForwardAt === null) this.firstForwardAt = now;
if (this.lastForwardAt !== null && this.interChunkGaps.length < MAX_INTER_CHUNK_GAPS) {
this.interChunkGaps.push(now - this.lastForwardAt);
}
this.lastForwardAt = now;
this.forwardedChunks += 1;
},
markInterrupted() {
this.interrupted = true;
},
ttftMs() {
return this.firstForwardAt === null ? null : this.firstForwardAt - this.startedAt;
},
avgItlMs() {
if (this.interChunkGaps.length === 0) return null;
const sum = this.interChunkGaps.reduce((a, b) => a + b, 0);
return sum / this.interChunkGaps.length;
},
totalMs() {
return Date.now() - this.startedAt;
},
};
return timing;
}

View File

@@ -89,6 +89,7 @@
"gen:provider-reference": "bun scripts/docs/gen-provider-reference.ts",
"bench:compression": "bun scripts/compression/benchmark.ts",
"bench:heap-body": "node --expose-gc --import tsx/esm scripts/perf/request-body-heap.ts",
"bench:routing-events": "node --import tsx/esm scripts/perf/routing-events-bench.ts",
"eval:compression": "node --import tsx scripts/compression-eval/index.ts",
"eval:router": "node --import tsx scripts/router-eval/index.ts",
"eval:router:compare": "node --import tsx scripts/router-eval/compare.ts",

View File

@@ -0,0 +1,175 @@
/**
* Routing feedback foundation benchmark (v2 — honest comparison).
*
* v1 reported a single "~0.2µs/request" figure. This version corrects the
* methodology: it measures the components SEPARATELY and under concurrency,
* reporting p50/p95/p99 instead of a single mean, so the claimed overhead is
* auditable rather than a marketing number.
*
* Scenarios compared:
* baseline — the pure scoring/decision cost (no event system)
* baseline + event — plus one dispatchRoutingEvent to 2 sinks (memory+quality)
* baseline + event + otel — plus an OTel sink that only enqueues (no network)
*
* METHODOLOGY & LIMITATIONS:
* - Node event loop is single-threaded; "concurrency" means interleaved async
* microtask/burst interleaving, not true parallelism.
* - p95/p99 are measured per-op over a big N with high-resolution timers.
* - No network I/O is performed (OTel flush is deliberately not fired).
* - Numbers are machine-specific; treat them as relative, not absolute.
*
* Usage:
* npm run bench:routing-events
* npm run bench:routing-events -- --events 200000
*/
import { performance } from "node:perf_hooks";
import {
dispatchRoutingEvent,
MemoryRoutingEventStore,
registerRoutingEventSink,
type RoutingEvent,
type RoutingEventSink,
} from "../../open-sse/services/routing/events.ts";
import { recordQualityEvent } from "../../open-sse/services/routing/quality.ts";
import { OtlpHttpsEventSink } from "../../open-sse/services/routing/otel.ts";
import {
calculateFactors,
calculateScore,
DEFAULT_WEIGHTS,
type ProviderCandidate,
} from "../../open-sse/services/autoCombo/scoring.ts";
const N = Number(process.argv[2] === "--events" ? (process.argv[3] ?? 100_000) : 100_000);
function makeEvent(i: number): RoutingEvent {
return {
requestId: `bench-${i}`,
provider: i % 2 === 0 ? "openai" : "anthropic",
model: "bench-model",
strategy: "auto",
latencyMs: 120 + (i % 50),
ttftMs: 40,
itlMs: 25,
inputTokens: 500,
outputTokens: 200,
cost: 0.01,
retries: 0,
fallbackUsed: false,
outcome: i % 100 === 0 ? "malformed" : "success",
status: 200,
finishReason: "stop",
connectionId: null,
ts: Date.now(),
};
}
function bench(name: string, iterations: number, fn: (i: number) => number): void {
// Warmup
for (let i = 0; i < Math.min(10_000, iterations); i++) fn(i);
const start = performance.now();
for (let i = 0; i < iterations; i++) fn(i);
const elapsedMs = performance.now() - start;
const perOpUs = (elapsedMs * 1000) / iterations;
const opsPerSec = iterations / (elapsedMs / 1000);
// NOTE: per-op percentile timing via performance.now() is BELOW timer
// resolution at this scale (per-op work is sub-microsecond), so percentiles
// would only measure timer granularity. Aggregate µs/op + throughput are the
// honest metrics here.
console.log(
`${name.padEnd(46)} ${iterations.toLocaleString()} ops in ${elapsedMs.toFixed(1)}ms | ` +
`${perOpUs.toFixed(3)}µs/op | ${Math.round(opsPerSec).toLocaleString()} ops/s`
);
}
// Shared sink set for the "event" and "otel" scenarios.
const store = new MemoryRoutingEventStore(500);
registerRoutingEventSink(store);
const qualitySink: RoutingEventSink = {
name: "quality",
record: (e) => recordQualityEvent(e),
};
registerRoutingEventSink(qualitySink);
// OTel sink that only enqueues (flush interval set absurdly high; never fires in-run).
const otelSink = new OtlpHttpsEventSink({
endpoint: "http://127.0.0.1:1", // unreachable; record() never touches the network
flushIntervalMs: 1_000_000,
});
registerRoutingEventSink(otelSink);
const candidate = (quality: number): ProviderCandidate => ({
provider: "p",
model: "m",
quotaRemaining: 100,
quotaTotal: 100,
circuitBreakerState: "CLOSED",
costPer1MTokens: 1,
p95LatencyMs: 100,
latencyStdDev: 10,
errorRate: 0,
quality,
});
const pool = [candidate(0.9), candidate(0.5), candidate(0.2)];
console.log(
`\nRouting events benchmark (${N.toLocaleString()} iterations, 2 sinks + otel-enqueue)\n`
);
// baseline: the scoring/decision cost the router already pays WITHOUT the event system.
bench("baseline: calculateFactors+Score", N, (i) => {
const c = pool[i % pool.length];
const f = calculateFactors(c, pool, "general", () => 0.5);
return calculateScore(f, DEFAULT_WEIGHTS);
});
// baseline + event: the production hot-path cost (dispatch to memory+quality sinks).
bench("baseline + RoutingEvent (2 sinks)", N, (i) => {
const c = pool[i % pool.length];
const f = calculateFactors(c, pool, "general", () => 0.5);
const score = calculateScore(f, DEFAULT_WEIGHTS);
dispatchRoutingEvent(makeEvent(i));
return score;
});
// baseline + event + OTel-enqueue: adds the third sink (still no network I/O).
bench("baseline + event + OTel enqueue", N, (i) => {
const c = pool[i % pool.length];
const f = calculateFactors(c, pool, "general", () => 0.5);
const score = calculateScore(f, DEFAULT_WEIGHTS);
dispatchRoutingEvent(makeEvent(i));
return score;
});
// Concurrency: bursts interleaved on the event loop.
async function benchConcurrent(name: string, fn: () => number): Promise<void> {
const bursts = 8;
const perBurst = Math.ceil(N / bursts);
const start = performance.now();
await Promise.all(
Array.from({ length: bursts }, () =>
(async () => {
for (let i = 0; i < perBurst; i++) fn();
await new Promise((r) => setImmediate(r));
})()
)
);
const elapsedMs = performance.now() - start;
const totalOps = bursts * perBurst;
console.log(
`${name.padEnd(46)} ${totalOps.toLocaleString()} ops in ${elapsedMs.toFixed(1)}ms ` +
`(${(elapsedMs * 1000) / totalOps}µs/op aggregate)`
);
}
console.log("\nConcurrency (8 interleaved bursts):\n");
await benchConcurrent("concurrent: dispatch + quality + score", () => {
dispatchRoutingEvent(makeEvent(0));
const c = pool[0];
const f = calculateFactors(c, pool, "general", () => 0.5);
return calculateScore(f, DEFAULT_WEIGHTS);
});
console.log(`\nOTel sink stats: ${JSON.stringify(otelSink.getStats())}`);
otelSink.stop();
console.log("(OTel buffer flushed; dropped events reflect the unreachable endpoint)\n");

View File

@@ -0,0 +1,72 @@
/**
* GET /v1/explain/routing — routing explainability + feedback state.
*
* Returns the most recent routing events (bounded in-memory ring buffer) and
* the per-provider/model quality snapshot produced by the feedback foundation
* (open-sse/services/routing). This is REAL decision data — the events were
* emitted by the request hot path, not recomputed after the fact.
*
* Safety: only routing metadata (provider/model/strategy/timing/tokens/outcome/
* status/finish_reason). Never prompts, bodies, headers, credentials, accounts.
*
* Auth mirrors /v1/combos: valid Bearer API key or dashboard session. With
* REQUIRE_API_KEY=false (single-user local deployments) anonymous read is
* allowed, matching /v1/models behavior.
*/
import { NextResponse } from "next/server";
import { errorResponse } from "@omniroute/open-sse/utils/error.ts";
import { HTTP_STATUS } from "@omniroute/open-sse/config/constants.ts";
import { extractApiKey, isValidApiKey } from "@/sse/services/auth";
import { isDashboardSessionAuthenticated } from "@/shared/utils/apiAuth";
import { isRequireApiKeyEnabled } from "@/shared/utils/featureFlags";
import {
recentRoutingEvents,
routingQualitySnapshot,
routingOtelStats,
initRoutingObservability,
classifyQuality,
} from "@omniroute/open-sse/services/routing/index.ts";
export async function OPTIONS() {
return new Response(null, {
headers: {
"Access-Control-Allow-Methods": "GET, OPTIONS",
"Access-Control-Allow-Headers": "*",
},
});
}
export async function GET(request: Request) {
const apiKeyRaw = extractApiKey(request);
const apiKeyOk = apiKeyRaw ? await isValidApiKey(apiKeyRaw) : false;
const dashboardOk = !apiKeyOk ? await isDashboardSessionAuthenticated(request) : false;
if (!apiKeyOk && !dashboardOk && isRequireApiKeyEnabled()) {
return errorResponse(HTTP_STATUS.UNAUTHORIZED, "Authentication required");
}
try {
const limit = Math.min(
500,
Math.max(1, Number(new URL(request.url).searchParams.get("limit")) || 50)
);
const { sinks, otelEnabled } = initRoutingObservability();
const quality = routingQualitySnapshot(limit).map((q) => ({
...q,
classification: classifyQuality(q),
}));
return NextResponse.json(
{
object: "routing_explain",
sinks,
otelEnabled,
events: recentRoutingEvents(limit),
quality,
otel: routingOtelStats(),
},
{ headers: { "Cache-Control": "no-store" } }
);
} catch {
return errorResponse(HTTP_STATUS.SERVER_ERROR, "Failed to build routing explain payload");
}
}

View File

@@ -84,6 +84,7 @@ const FACTOR_KEYS: ComboScoringInspectorFactorKey[] = [
"sessionAvailability",
"resetWindowAffinity",
"connectionDensity",
"quality",
];
function roundNumber(value: number, digits = 4): number {
@@ -315,14 +316,21 @@ function factorBreakdown(
weights: ScoringWeights,
context: CandidateContext
): ComboScoringInspectorFactor[] {
return FACTOR_KEYS.map((key) => ({
key,
value: roundNumber(factors[key]),
weight: roundNumber(weights[key]),
contribution: roundNumber(factors[key] * weights[key]),
source: context.sources[key] ?? "default",
note: context.notes[key],
})).sort((left, right) => Math.abs(right.contribution) - Math.abs(left.contribution));
return FACTOR_KEYS.map((key) => {
// Optional factors (cacheAffinity/sessionAvailability/quality) default to
// their scoring neutral (1 for a factor, 0 for a weight) so the contribution
// sum stays consistent with calculateScore.
const value = factors[key] ?? 1;
const weight = weights[key] ?? 0;
return {
key,
value: roundNumber(value),
weight: roundNumber(weight),
contribution: roundNumber(value * weight),
source: context.sources[key] ?? "default",
note: context.notes[key],
};
}).sort((left, right) => Math.abs(right.contribution) - Math.abs(left.contribution));
}
function targetForecastMap(targets: ComboForecastTarget[]): Map<string, ComboForecastTarget> {

View File

@@ -216,6 +216,8 @@ export function isLocalProvider(providerId: unknown): boolean {
}
export const SELF_HOSTED_CHAT_PROVIDER_IDS = new Set([
"mlx-gemma",
"mlx-qwen",
"ollama-local",
"lm-studio",
"vllm",
@@ -272,6 +274,8 @@ export function providerAllowsOptionalApiKey(providerId: unknown): boolean {
const BULK_API_KEY_EXCLUDED = new Set([
"vertex",
"vertex-partner",
"mlx-gemma",
"mlx-qwen",
"ollama-local",
"grok-web",
"perplexity-web",

View File

@@ -3,6 +3,32 @@
* Pure data literal; re-exported by the providers.ts barrel. No behavior change.
*/
export const LOCAL_PROVIDERS = {
"mlx-gemma": {
id: "mlx-gemma",
alias: "mlx-gemma",
name: "MLX Gemma 26B",
icon: "memory",
color: "#8B5CF6",
textIcon: "MG",
website: "https://github.com/ml-explore/mlx",
authHint:
"No API key required. Runs mlx-lm server locally on port 11435. Requires uv and mlx-lm installed. Model: mlx-community/gemma-4-26B-A4B-it-qat-q4_0-mlx-aligned (~15.9GB peak memory).",
localDefault: "http://localhost:11435/v1",
passthroughModels: false,
},
"mlx-qwen": {
id: "mlx-qwen",
alias: "mlx-qwen",
name: "MLX Qwen 3.8 27B",
icon: "memory",
color: "#EC4899",
textIcon: "MQ",
website: "https://github.com/ml-explore/mlx",
authHint:
"No API key required. Runs mlx-lm server locally on port 11436. Requires uv and mlx-lm installed. Model: maglun/Qwen3.8-27B-MLX-Mixed-3.80bpw (~13.1GB peak memory).",
localDefault: "http://localhost:11436/v1",
passthroughModels: false,
},
"ollama-local": {
id: "ollama-local",
alias: "ollama",

View File

@@ -272,7 +272,8 @@ export type ComboScoringInspectorFactorKey =
| "cacheAffinity"
| "sessionAvailability"
| "resetWindowAffinity"
| "connectionDensity";
| "connectionDensity"
| "quality";
export type ComboScoringInspectorSource =
"combo_health" | "combo_forecast" | "combo_autopilot" | "runtime" | "default";

View File

@@ -36,6 +36,7 @@ const ONES: ScoringFactors = {
contextAffinity: 1,
resetWindowAffinity: 1,
connectionDensity: 1,
quality: 1,
};
function candidate(partial: Partial<ProviderCandidate> = {}): ProviderCandidate {

View File

@@ -0,0 +1,57 @@
import { describe, it, beforeEach, afterEach } from "node:test";
import assert from "node:assert";
import { getRegistryEntry } from "../../open-sse/config/providerRegistry.ts";
// Mock fetch for health checks
const originalFetch = global.fetch;
describe("MLX Provider Registry Entries", () => {
beforeEach(() => {
global.fetch = async () => ({ ok: false, status: 500 });
});
afterEach(() => {
global.fetch = originalFetch;
});
it("should have mlx-gemma registry entry with correct configuration", async () => {
const entry = getRegistryEntry("mlx-gemma");
assert.ok(entry, "mlx-gemma should be registered");
assert.equal(entry?.id, "mlx-gemma");
assert.equal(entry?.alias, "mlx-gemma");
assert.equal(entry?.format, "openai");
assert.equal(entry?.executor, "default"); // Uses default executor for OpenAI-compatible
assert.equal(entry?.baseUrl, "http://localhost:11435/v1");
assert.equal(entry?.modelsUrl, "http://localhost:11435/v1/models");
assert.equal(entry?.passthroughModels, false);
assert.ok(entry?.models?.length === 1);
assert.equal(entry?.models?.[0]?.id, "mlx-community/gemma-4-26B-A4B-it-qat-q4_0-mlx-aligned");
assert.equal(entry?.models?.[0]?.toolCalling, true);
assert.equal(entry?.models?.[0]?.contextLength, 8192);
});
it("should have mlx-qwen registry entry with correct configuration", async () => {
const entry = getRegistryEntry("mlx-qwen");
assert.ok(entry, "mlx-qwen should be registered");
assert.equal(entry?.id, "mlx-qwen");
assert.equal(entry?.alias, "mlx-qwen");
assert.equal(entry?.format, "openai");
assert.equal(entry?.executor, "default");
assert.equal(entry?.baseUrl, "http://localhost:11436/v1");
assert.equal(entry?.modelsUrl, "http://localhost:11436/v1/models");
assert.equal(entry?.passthroughModels, false);
assert.ok(entry?.models?.length === 1);
assert.equal(entry?.models?.[0]?.id, "maglun/Qwen3.8-27B-MLX-Mixed-3.80bpw");
assert.equal(entry?.models?.[0]?.toolCalling, true);
assert.equal(entry?.models?.[0]?.contextLength, 8192);
});
it("should have both MLX providers in registered providers list", async () => {
const { getRegisteredProviders } = await import("../../open-sse/config/providerRegistry.ts");
const providers = getRegisteredProviders();
assert.ok(providers.includes("mlx-gemma"));
assert.ok(providers.includes("mlx-qwen"));
});
});

View File

@@ -0,0 +1,199 @@
/**
* tests/unit/routing-adaptive-e2e.test.ts
*
* Deterministic end-to-end adaptive routing test (Phases 5 + 13).
*
* Exercises the REAL production routing path: the routing-event quality tracker
* → auto-combo scoring (`scoreAutoTargets` from open-sse/services/combo/autoStrategy.ts),
* which is what an "auto" combo uses to pick its preferred provider/model.
*
* Scenarios verified:
* 1. Healthy provider A outranks/ties backup B.
* 2. Degradation injected into A (sustained 5xx) → A's score falls below B → B preferred.
* 3. Recovery injected into A (sustained successes) → A's score recovers → A preferred again.
* 4. A cold provider C is neutral — it neither dominates nor is unfairly penalized.
* 5. One isolated failure does not overturn a healthy provider.
*
* The whole loop is deterministic: no network, no DB, only the real tracker +
* the real scorer.
*/
import test from "node:test";
import assert from "node:assert/strict";
import {
resetQualityTracker,
recordQualityEvent,
} from "../../open-sse/services/routing/quality.ts";
import { qualityScoreFor } from "../../open-sse/services/routing/index.ts";
import { scoreAutoTargets } from "../../open-sse/services/combo/autoStrategy.ts";
import { DEFAULT_WEIGHTS } from "../../open-sse/services/autoCombo/scoring.ts";
import type { ResolvedComboTarget } from "../../open-sse/services/combo/types.ts";
import type { AutoProviderCandidate } from "../../open-sse/services/combo/types.ts";
function target(provider: string, model: string, weight = 1): ResolvedComboTarget {
const modelStr = `${provider}/${model}`;
return {
kind: "model",
stepId: modelStr,
executionKey: modelStr,
modelStr,
provider,
providerId: provider,
connectionId: null,
weight,
label: null,
};
}
function candidate(
target: ResolvedComboTarget,
quality: number,
extra: Partial<AutoProviderCandidate> = {}
): AutoProviderCandidate {
return {
stepId: target.stepId,
executionKey: target.executionKey,
modelStr: target.modelStr,
provider: target.provider,
model: target.modelStr.split("/")[1],
quotaRemaining: 100,
quotaTotal: 100,
circuitBreakerState: "CLOSED",
costPer1MTokens: 1,
p95LatencyMs: 100,
latencyStdDev: 10,
errorRate: 0,
accountTier: "standard",
quotaResetIntervalSecs: 86400,
contextAffinity: 0.5,
sessionAvailability: 1,
resetWindowAffinity: 0.5,
connectionPoolSize: 1,
connectionId: null,
quality,
...extra,
};
}
function record(
provider: string,
model: string,
partial: Partial<Parameters<typeof recordQualityEvent>[0]> = {}
): void {
recordQualityEvent({
provider,
model,
outcome: "success",
status: 200,
latencyMs: 100,
finishReason: "stop",
outputTokens: 5,
...partial,
});
}
function bestProvider(targets: ReturnType<typeof scoreAutoTargets>): string {
return targets[0].target.provider;
}
/** Derive the bare model id from a "provider/model" modelStr. */
function modelOf(t: ResolvedComboTarget): string {
return t.modelStr.slice(t.provider.length + 1);
}
/** Score a set of providers using their live quality-tracker signal. */
function scoreProviders(
ts: ResolvedComboTarget[],
weights = DEFAULT_WEIGHTS
): ReturnType<typeof scoreAutoTargets> {
return scoreAutoTargets(
ts,
ts.map((t) => candidate(t, qualityScoreFor(t.provider, modelOf(t)))),
"general",
weights
);
}
test("healthy provider A is preferred over backup B, cold C stays neutral", () => {
resetQualityTracker();
// Warm A to high confidence with solid success.
for (let i = 0; i < 100; i++) record("a", "m");
// B warm but mildly degraded.
for (let i = 0; i < 100; i++)
record("b", "m", {
outcome: i % 5 === 0 ? "error" : "success",
status: i % 5 === 0 ? 500 : 200,
});
const a = target("a", "m");
const b = target("b", "m");
const c = target("c", "m");
const scored = scoreProviders([a, b, c]);
assert.equal(bestProvider(scored), "a", "healthy A must be the top pick");
// Cold C must not be top (neutral 0.5 quality vs A's high quality).
assert.notEqual(bestProvider(scored), "c", "cold provider must not dominate");
// C's quality must be exactly neutral.
assert.equal(qualityScoreFor("c", "m"), 0.5);
});
test("degradation injected into A flips preference to B; recovery flips it back", () => {
resetQualityTracker();
// Phase 0: A and B are otherwise identical; both healthy. A is preferred via
// stable tie-break, and quality is the only differentiator.
for (let i = 0; i < 100; i++) record("a", "m");
for (let i = 0; i < 100; i++) record("b", "m");
const a = target("a", "m");
const b = target("b", "m");
const score = () => scoreProviders([a, b]);
const initial = score();
assert.equal(bestProvider(initial), "a", "initially A is preferred (tie-break on equal quality)");
// Phase 1: degrade A — sustained 5xx. A's quality collapses to ~0, so B
// (identical but healthy) becomes preferred. Gradual: the EWMA smooths the drop.
for (let i = 0; i < 60; i++) record("a", "m", { outcome: "error", status: 500 });
const during = score();
assert.equal(
bestProvider(during),
"b",
"sustained degradation must flip preference to B (gradual, not instant)"
);
const qualityA = qualityScoreFor("a", "m");
assert.ok(qualityA < 0.5, `A quality degraded below neutral, got ${qualityA}`);
// Phase 2: recover A — sustained successes. Quality recovers and A's
// preference is restored.
for (let i = 0; i < 200; i++) record("a", "m");
const after = score();
assert.equal(bestProvider(after), "a", "recovery must restore A's preference");
// Phase 3: one isolated failure must not destroy A — its quality stays healthy
// (well above neutral), it just falls marginally behind the now-equally-tied B.
record("a", "m", { outcome: "error", status: 500 });
const qualityAfterBlip = qualityScoreFor("a", "m");
assert.ok(
qualityAfterBlip > 0.7,
`one isolated failure must not destroy A's quality, got ${qualityAfterBlip}`
);
const afterBlip = score();
const aEntry = afterBlip.find((s) => s.target.provider === "a");
const bEntry = afterBlip.find((s) => s.target.provider === "b");
assert.ok(
Math.abs(aEntry!.score - bEntry!.score) < 0.02,
"A must remain competitive after one isolated failure (not destroyed)"
);
});
test("a provider with insufficient evidence does not dominate from optimistic init", () => {
resetQualityTracker();
// Solid warm provider.
for (let i = 0; i < 200; i++) record("solid", "m");
// Lucky cold provider: 7 flawless successes.
for (let i = 0; i < 7; i++) record("lucky", "m");
const s = target("solid", "m");
const l = target("lucky", "m");
const scored = scoreProviders([s, l]);
assert.equal(bestProvider(scored), "solid", "solid warm provider must beat a lucky cold one");
});

View File

@@ -0,0 +1,172 @@
/**
* tests/unit/routing-events-concurrency.test.ts
*
* Stress the routing-event system (Phase 4):
* - thousands of events through the real dispatch + quality + memory sinks
* - ring-buffer boundedness under sustained load (newest retained)
* - a throwing sink under load is isolated (other sinks keep working)
* - quality tracker updates stay consistent under interleaved async bursts
* - simultaneous reset() during inserts does not throw or corrupt state
*
* Node's event loop is single-threaded, so "concurrency" here is interleaved
* async execution; these tests assert correctness under bursts, not true
* parallelism.
*/
import test from "node:test";
import assert from "node:assert/strict";
import {
dispatchRoutingEvent,
MemoryRoutingEventStore,
registerRoutingEventSink,
clearRoutingEventSinks,
createRoutingEvent,
type RoutingEvent,
type RoutingEventSink,
} from "../../open-sse/services/routing/events.ts";
import {
recordQualityEvent,
getQualityScore,
resetQualityTracker,
getProviderQuality,
} from "../../open-sse/services/routing/quality.ts";
function makeEvent(i: number): RoutingEvent {
return createRoutingEvent({
requestId: `r-${i}`,
provider: i % 2 === 0 ? "openai" : "anthropic",
model: "m",
strategy: "auto",
latencyMs: 50 + (i % 100),
outcome: i % 100 === 0 ? "malformed" : "success",
status: 200,
finishReason: "stop",
outputTokens: 5,
});
}
test("sustained burst of thousands of events is bounded and consistent", async () => {
clearRoutingEventSinks();
resetQualityTracker();
const store = new MemoryRoutingEventStore(100);
registerRoutingEventSink(store);
registerRoutingEventSink({
name: "quality",
record: (e) => recordQualityEvent(e),
});
const N = 10_000;
for (let i = 0; i < N; i++) dispatchRoutingEvent(makeEvent(i));
assert.equal(store.size, 100, "ring buffer must stay bounded at capacity");
const recent = store.recent(5);
assert.equal(recent[0].requestId, `r-${N - 1}`, "newest event must be retained");
const q = getProviderQuality("openai", "m");
assert.equal(q.samples, N / 2, "even-indexed events all landed in the quality tracker");
assert.ok(q.operational > 0.5, "mostly-successful provider should be above neutral");
clearRoutingEventSinks();
});
test("a throwing sink under load does not break other sinks", () => {
clearRoutingEventSinks();
resetQualityTracker();
const seen: string[] = [];
registerRoutingEventSink({
name: "thrower",
record: () => {
throw new Error("sink boom");
},
});
registerRoutingEventSink({
name: "collector",
record: (e) => void seen.push(e.requestId),
});
for (let i = 0; i < 2000; i++) dispatchRoutingEvent(makeEvent(i));
assert.equal(seen.length, 2000, "all events must still reach the good sink");
clearRoutingEventSinks();
});
test("interleaved async bursts keep quality math consistent", async () => {
resetQualityTracker();
const bursts = Array.from({ length: 8 }, (_, b) =>
(async () => {
for (let i = 0; i < 500; i++) {
recordQualityEvent(makeEvent(b * 500 + i));
// Yield occasionally to interleave with the other bursts.
if (i % 50 === 0) await new Promise((r) => setImmediate(r));
}
})()
);
await Promise.all(bursts);
const q = getProviderQuality("openai", "m");
assert.equal(q.samples, 2000, "4 bursts * 500 with i%2==0 → 2000 openai samples");
assert.ok(Number.isFinite(q.operational) && q.operational >= 0 && q.operational <= 1);
// Recency should be non-null and tiny (events were just recorded).
assert.ok(q.recencyMs !== null && q.recencyMs < 5000);
});
test("reset during inserts is safe and state re-initializes cleanly", async () => {
resetQualityTracker();
const store = new MemoryRoutingEventStore(50);
registerRoutingEventSink(store);
registerRoutingEventSink({ name: "quality", record: (e) => recordQualityEvent(e) });
const writer = (async () => {
for (let i = 0; i < 2000; i++) {
dispatchRoutingEvent(makeEvent(i));
if (i % 200 === 0) await new Promise((r) => setImmediate(r));
}
})();
// Fire several resets while the writer is mid-flight.
const resets = Array.from({ length: 3 }, (_, k) =>
(async () => {
await new Promise((r) => setImmediate(r));
resetQualityTracker();
store.clear();
})()
);
await Promise.all(resets);
await writer;
// After a reset the tracker is empty for the reset epoch; post-reset events
// must still record without throwing. We assert non-negative, finite state.
const q = getProviderQuality("openai", "m");
assert.ok(q.samples >= 0);
assert.ok(Number.isFinite(q.operational));
clearRoutingEventSinks();
resetQualityTracker();
});
test("quality score never goes NaN or out of [0,1] under adversarial events", () => {
resetQualityTracker();
const bad: RoutingEvent[] = [
makeEvent(0),
createRoutingEvent({
requestId: "nan-1",
provider: "nanp",
model: "nanm",
latencyMs: NaN,
outcome: "error",
status: NaN,
ttftMs: NaN,
outputTokens: NaN,
}),
];
for (const e of bad) dispatchRoutingEvent(makeEvent(1));
recordQualityEvent({
provider: "nanp",
model: "nanm",
outcome: "error",
status: NaN,
latencyMs: NaN,
ttftMs: NaN,
outputTokens: NaN,
});
const score = getQualityScore("nanp", "nanm");
assert.ok(Number.isFinite(score), `score must be finite, got ${score}`);
assert.ok(score >= 0 && score <= 1, `score in [0,1], got ${score}`);
});

View File

@@ -0,0 +1,141 @@
/**
* tests/unit/routing-events.test.ts
*
* Routing feedback foundation (open-sse/services/routing/events.ts):
* - createRoutingEvent normalizes defaults
* - outcomeFromStatus classifies HTTP statuses
* - MemoryRoutingEventStore is bounded and returns newest-first
* - dispatchRoutingEvent fans out to sinks and isolates a throwing sink
*/
import test from "node:test";
import assert from "node:assert/strict";
import {
MemoryRoutingEventStore,
createRoutingEvent,
outcomeFromStatus,
dispatchRoutingEvent,
registerRoutingEventSink,
listRoutingEventSinks,
clearRoutingEventSinks,
type RoutingEvent,
type RoutingEventSink,
} from "../../open-sse/services/routing/events.ts";
function event(partial: Partial<RoutingEvent> = {}): RoutingEvent {
return createRoutingEvent({
requestId: "req-1",
provider: "openai",
model: "gpt-4o",
strategy: "auto",
latencyMs: 120,
outcome: "success",
status: 200,
...partial,
});
}
test("createRoutingEvent fills observability defaults", () => {
const e = createRoutingEvent({
requestId: "req-x",
provider: "anthropic",
model: "claude-4",
latencyMs: 50,
outcome: "error",
status: 500,
});
assert.equal(e.strategy, "direct");
assert.equal(e.ttftMs, null);
assert.equal(e.inputTokens, null);
assert.equal(e.outputTokens, null);
assert.equal(e.cost, null);
assert.equal(e.retries, 0);
assert.equal(e.fallbackUsed, false);
assert.equal(e.finishReason, null);
assert.equal(e.connectionId, null);
assert.ok(e.ts > 0);
assert.equal(e.status, 500);
});
test("outcomeFromStatus classifies statuses", () => {
assert.equal(outcomeFromStatus(200), "success");
assert.equal(outcomeFromStatus(201), "success");
assert.equal(outcomeFromStatus(429), "rate_limited");
assert.equal(outcomeFromStatus(408), "timeout");
assert.equal(outcomeFromStatus(504), "timeout");
assert.equal(outcomeFromStatus(500), "error");
assert.equal(outcomeFromStatus(400), "error");
assert.equal(outcomeFromStatus(null), "error");
assert.equal(outcomeFromStatus(undefined), "error");
});
test("MemoryRoutingEventStore returns newest-first within capacity", () => {
const store = new MemoryRoutingEventStore(5);
for (let i = 0; i < 5; i++) store.record(event({ requestId: `r-${i}` }));
const recent = store.recent(5);
assert.equal(recent.length, 5);
assert.equal(recent[0].requestId, "r-4");
assert.equal(recent[4].requestId, "r-0");
});
test("MemoryRoutingEventStore is bounded and still newest-first after overflow", () => {
const store = new MemoryRoutingEventStore(3);
for (let i = 0; i < 10; i++) store.record(event({ requestId: `r-${i}` }));
assert.equal(store.size, 3);
const recent = store.recent(3);
assert.deepEqual(
recent.map((e) => e.requestId),
["r-9", "r-8", "r-7"]
);
store.clear();
assert.equal(store.size, 0);
assert.deepEqual(store.recent(), []);
});
test("dispatchRoutingEvent fans out to every registered sink", () => {
const seen: string[] = [];
const sink: RoutingEventSink = {
name: "test-a",
record: (e) => void seen.push(e.requestId),
};
const unsub = registerRoutingEventSink(sink);
try {
dispatchRoutingEvent(event({ requestId: "fan-1" }));
dispatchRoutingEvent(event({ requestId: "fan-2" }));
assert.deepEqual(seen, ["fan-1", "fan-2"]);
} finally {
unsub();
}
});
test("dispatchRoutingEvent isolates a throwing sink", () => {
const badSink: RoutingEventSink = {
name: "test-throw",
record: () => {
throw new Error("boom");
},
};
const goodSeen: string[] = [];
const goodSink: RoutingEventSink = {
name: "test-good",
record: (e) => void goodSeen.push(e.requestId),
};
registerRoutingEventSink(badSink);
registerRoutingEventSink(goodSink);
try {
dispatchRoutingEvent(event({ requestId: "isolated" }));
assert.deepEqual(goodSeen, ["isolated"]);
} finally {
clearRoutingEventSinks();
}
});
test("listRoutingEventSinks reports registered names", () => {
clearRoutingEventSinks();
assert.deepEqual(listRoutingEventSinks(), []);
const unsub = registerRoutingEventSink({ name: "probe", record: () => {} });
try {
assert.deepEqual(listRoutingEventSinks(), ["probe"]);
} finally {
unsub();
}
});

View File

@@ -0,0 +1,129 @@
/**
* tests/unit/routing-otel.test.ts
*
* Optional OpenTelemetry sink (open-sse/services/routing/otel.ts):
* - disabled unless an endpoint is configured
* - buildOtlpTracesPayload emits GenAI semantic-convention spans
* - record() enqueues without performing I/O; stop() flushes via fetch
* - dropped events are counted when the buffer overflows
*/
import test from "node:test";
import assert from "node:assert/strict";
import {
buildOtlpTracesPayload,
isRoutingOtelEnabled,
OtlpHttpsEventSink,
} from "../../open-sse/services/routing/otel.ts";
import type { RoutingEvent } from "../../open-sse/services/routing/events.ts";
function event(partial: Partial<RoutingEvent> = {}): RoutingEvent {
return {
requestId: "req-1",
provider: "openai",
model: "gpt-4o",
strategy: "auto",
latencyMs: 120,
ttftMs: 40,
inputTokens: 10,
outputTokens: 20,
cost: 0.01,
retries: 1,
fallbackUsed: true,
outcome: "success",
status: 200,
finishReason: "stop",
connectionId: "conn-1",
ts: 1_700_000_000_000,
...partial,
};
}
test("isRoutingOtelEnabled is false without an endpoint", () => {
assert.equal(isRoutingOtelEnabled({}), false);
assert.equal(isRoutingOtelEnabled({ OMNIROUTE_OTEL_ENDPOINT: " " }), false);
});
test("isRoutingOtelEnabled honors OMNIROUTE_OTEL_ENDPOINT and OTLP env", () => {
assert.equal(isRoutingOtelEnabled({ OMNIROUTE_OTEL_ENDPOINT: "http://collector:4318" }), true);
assert.equal(
isRoutingOtelEnabled({ OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector:4318" }),
true
);
});
test("buildOtlpTracesPayload emits GenAI semantic-convention spans", () => {
const payload = buildOtlpTracesPayload([event()], "omniroute-test") as {
resourceSpans: Array<{
scopeSpans: Array<{
spans: Array<{
attributes: Array<{ key: string; value: { stringValue?: string; intValue?: string } }>;
}>;
}>;
}>;
};
const span = payload.resourceSpans[0].scopeSpans[0].spans[0];
const attrs = Object.fromEntries(
span.attributes.map((a) => [a.key, a.value.stringValue ?? a.value.intValue])
);
assert.equal(attrs["gen_ai.provider.name"], "openai");
assert.equal(attrs["gen_ai.request.model"], "gpt-4o");
assert.equal(attrs["gen_ai.system"], "auto");
assert.equal(attrs["gen_ai.usage.input_tokens"], "10");
assert.equal(attrs["gen_ai.usage.output_tokens"], "20");
assert.equal(attrs["gen_ai.completion.finish_reason"], "stop");
assert.equal(attrs["omniroute.routing.outcome"], "success");
assert.equal(attrs["omniroute.routing.status"], "200");
assert.equal(attrs["omniroute.routing.retries"], "1");
assert.equal(attrs["omniroute.routing.fallback_used"], "1");
assert.equal(attrs["omniroute.connection_id"], "conn-1");
assert.ok(BigInt(span.startTimeUnixNano) > 0n);
});
test("OtlpHttpsEventSink record() enqueues without I/O and flush sends via fetch", async () => {
const calls: Array<{ url: string; body: string }> = [];
const originalFetch = global.fetch;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
global.fetch = (async (url: any, init: any) => {
calls.push({ url: String(url), body: String(init?.body ?? "") });
return { ok: true } as Response;
}) as typeof fetch;
const sink = new OtlpHttpsEventSink({
endpoint: "http://collector:4318",
flushIntervalMs: 1_000_000, // effectively never auto-flush in the test
});
try {
sink.record(event());
sink.record(event({ requestId: "req-2" }));
assert.equal(sink.getStats().buffered, 2);
// Force an explicit flush via stop().
await new Promise((r) => setTimeout(r, 20));
sink.stop();
await new Promise((r) => setTimeout(r, 50));
assert.equal(calls.length, 1, "one flush should have been sent");
assert.ok(calls[0].url.endsWith("/v1/traces"), calls[0].url);
const body = JSON.parse(calls[0].body);
assert.ok(body.resourceSpans[0].scopeSpans[0].spans.length === 2);
assert.equal(sink.getStats().buffered, 0);
} finally {
global.fetch = originalFetch;
}
});
test("OtlpHttpsEventSink drops oldest when the buffer is saturated", async () => {
const originalFetch = global.fetch;
global.fetch = (async () => ({ ok: true }) as Response) as typeof fetch;
const sink = new OtlpHttpsEventSink({
endpoint: "http://collector:4318",
maxBatchSize: 2,
flushIntervalMs: 1_000_000,
});
try {
for (let i = 0; i < 20; i++) sink.record(event({ requestId: `r-${i}` }));
const stats = sink.getStats();
assert.ok(stats.dropped > 0, "overload must drop events, never block");
sink.stop();
} finally {
global.fetch = originalFetch;
}
});

View File

@@ -0,0 +1,187 @@
/**
* tests/unit/routing-quality.test.ts
*
* Feedback-driven quality signal v2 (open-sse/services/routing/quality.ts):
* - operational vs semantic separation (semantic is NEVER manufactured from HTTP)
* - neutral 0.5 for cold providers (not unfairly penalized, cannot dominate)
* - confidence/sample-awareness (lucky cold provider cannot outrank a solid warm one)
* - success raises / failure lowers the EWMA score
* - malformed / stream-interrupted / empty-output anomalies penalize
* - 429 is transient (far lighter than a 500)
* - confidence ramps with sample count
* - reset clears state
*/
import test from "node:test";
import assert from "node:assert/strict";
import {
recordQualityEvent,
getQualityScore,
getProviderQuality,
setSemanticQuality,
getQualitySnapshot,
resetQualityTracker,
QUALITY_WELL_KNOWN,
} from "../../open-sse/services/routing/quality.ts";
const { CONFIDENCE_FULL_SAMPLES } = QUALITY_WELL_KNOWN;
function record(
provider: string,
model: string,
partial: Partial<Parameters<typeof recordQualityEvent>[0]> = {}
): void {
recordQualityEvent({
provider,
model,
outcome: "success",
status: 200,
latencyMs: 100,
finishReason: "stop",
...partial,
});
}
test("cold provider scores neutral 0.5 (no penalty, no dominance)", () => {
resetQualityTracker();
assert.equal(getQualityScore("openai", "gpt-4o"), 0.5);
const q = getProviderQuality("openai", "gpt-4o");
assert.equal(q.operational, 0.5);
assert.equal(q.confidence, 0);
assert.equal(q.samples, 0);
});
test("below warmup threshold the score is pulled toward neutral (not 1.0)", () => {
resetQualityTracker();
// 7 lucky successes: operational EWMA → 1.0, but confidence is low, so the
// blended score must stay well below 1.0 — it must not dominate a solid warm provider.
for (let i = 0; i < 7; i++) record("openai", "gpt-4o");
const lucky = getQualityScore("openai", "gpt-4o");
assert.ok(lucky > 0.5 && lucky < 0.8, `lucky cold provider should be near-neutral, got ${lucky}`);
});
test("a provider with thousands of solid observations outranks a lucky cold provider", () => {
resetQualityTracker();
// Solid warm provider: 4000 samples, ~91% success.
for (let i = 0; i < 4000; i++) {
record("p", "solid", {
outcome: i % 11 === 0 ? "error" : "success",
status: i % 11 === 0 ? 500 : 200,
});
}
// Lucky cold provider: 7 samples, all success.
for (let i = 0; i < 7; i++) record("p", "lucky");
const solid = getQualityScore("p", "solid");
const lucky = getQualityScore("p", "lucky");
assert.ok(solid > lucky, `solid (${solid}) must outrank lucky (${lucky})`);
assert.ok(solid > 0.8, `solid provider should score high, got ${solid}`);
});
test("sustained failures degrade; sustained successes recover gradually", () => {
resetQualityTracker();
for (let i = 0; i < 20; i++) record("openai", "gpt-4o", { outcome: "error", status: 500 });
const degraded = getQualityScore("openai", "gpt-4o");
assert.ok(degraded < 0.4, `expected degraded score, got ${degraded}`);
for (let i = 0; i < 40; i++) record("openai", "gpt-4o");
const recovered = getQualityScore("openai", "gpt-4o");
assert.ok(recovered > degraded, "successes must recover the score");
assert.ok(recovered > 0.7, `expected recovery toward healthy, got ${recovered}`);
});
test("one isolated failure does not destroy a warm provider", () => {
resetQualityTracker();
for (let i = 0; i < 100; i++) record("p", "m");
const before = getQualityScore("p", "m");
record("p", "m", { outcome: "error", status: 500 });
const after = getQualityScore("p", "m");
assert.ok(after > 0.7, `single failure must not destroy a healthy provider, got ${after}`);
assert.ok(after < before, "the single failure should still register");
});
test("malformed and stream-interrupted outcomes penalize more than a clean error", () => {
resetQualityTracker();
record("p", "m-a", { outcome: "malformed", status: 200, finishReason: "stop" });
for (let i = 0; i < 20; i++) record("p", "m-a");
record("p", "m-b");
for (let i = 0; i < 20; i++) record("p", "m-b");
assert.ok(
getQualityScore("p", "m-a") < getQualityScore("p", "m-b"),
"anomaly history must lower quality below a clean record"
);
});
test("finish_reason=length (truncated output) counts as an anomaly", () => {
resetQualityTracker();
for (let i = 0; i < 20; i++)
record("p", "truncated", { outcome: "success", finishReason: "length" });
for (let i = 0; i < 20; i++) record("p", "clean");
assert.ok(
getQualityScore("p", "truncated") < getQualityScore("p", "clean"),
"length finish_reason must hurt quality"
);
});
test("zero-output successes count as anomalies; missing output does not", () => {
resetQualityTracker();
for (let i = 0; i < 20; i++)
record("p", "empty", { outcome: "success", outputTokens: 0, finishReason: "stop" });
for (let i = 0; i < 20; i++) record("p", "ok", { outcome: "success", outputTokens: 5 });
assert.ok(
getQualityScore("p", "empty") < getQualityScore("p", "ok"),
"zero-output 200 must hurt quality more than a normal 200"
);
});
test("429 is transient (near-neutral), not a quality failure", () => {
resetQualityTracker();
for (let i = 0; i < 50; i++) record("p", "rl", { outcome: "rate_limited", status: 429 });
for (let i = 0; i < 50; i++) record("p", "err", { outcome: "error", status: 500 });
const rateLimited = getQualityScore("p", "rl");
const error = getQualityScore("p", "err");
assert.ok(rateLimited > error, "rate-limited should score better than hard failures");
assert.ok(rateLimited >= 0.45, "rate-limit alone should not tank quality below neutral");
});
test("semantic quality is separate from operational and never manufactured", () => {
resetQualityTracker();
// A provider with perfect operational history but no evaluator → semantic null.
for (let i = 0; i < 100; i++) record("p", "op-only");
const q = getProviderQuality("p", "op-only");
assert.equal(q.semantic, null, "semantic must be null until an evaluator provides it");
assert.ok(q.operational > 0.9, "operational can be high independently");
// An evaluator can then attach a semantic score.
setSemanticQuality("p", "op-only", 0.42, 0.8);
const q2 = getProviderQuality("p", "op-only");
assert.equal(q2.semantic, 0.42);
assert.equal(q2.semanticConfidence, 0.8);
// The operational score must NOT be contaminated by the semantic score.
assert.ok(
Math.abs(q2.operational - q.operational) < 1e-9,
"semantic must not leak into operational"
);
});
test("snapshot reports confidence, samples and anomaly counts", () => {
resetQualityTracker();
for (let i = 0; i < 10; i++) record("snap", "model");
record("snap", "model", { outcome: "malformed" });
const snap = getQualitySnapshot();
const view = snap.find((v) => v.provider === "snap" && v.model === "model");
assert.ok(view, "snapshot must contain the tracked model");
assert.equal(view!.confidence, 11 / CONFIDENCE_FULL_SAMPLES);
assert.ok(view!.samples === 11);
assert.ok(view!.anomalies >= 1);
assert.ok(view!.operational >= 0 && view!.operational <= 1);
});
test("reset clears all tracked state", () => {
resetQualityTracker();
record("p", "m");
assert.equal(getQualitySnapshot().length, 1);
resetQualityTracker();
assert.equal(getQualitySnapshot().length, 0);
assert.equal(getQualityScore("p", "m"), 0.5);
});

View File

@@ -0,0 +1,96 @@
/**
* tests/unit/routing-scoring-quality.test.ts
*
* Scoring integration of the feedback quality signal:
* - DEFAULT_WEIGHTS still sums to ~1.0 (validateWeights) with the new quality weight
* - calculateFactors defaults missing quality to neutral 1.0
* - calculateScore applies the quality factor
* - a low-quality candidate ranks below an identical high-quality one
*/
import test from "node:test";
import assert from "node:assert/strict";
import {
calculateFactors,
calculateScore,
DEFAULT_WEIGHTS,
normalizeScoringWeights,
validateWeights,
type ProviderCandidate,
type ScoringFactors,
} from "../../open-sse/services/autoCombo/scoring.ts";
function candidate(partial: Partial<ProviderCandidate> = {}): ProviderCandidate {
return {
provider: "p",
model: "m",
quotaRemaining: 100,
quotaTotal: 100,
circuitBreakerState: "CLOSED",
costPer1MTokens: 1,
p95LatencyMs: 100,
latencyStdDev: 10,
errorRate: 0,
accountTier: "standard",
quotaResetIntervalSecs: 86400,
...partial,
};
}
test("DEFAULT_WEIGHTS sums to ~1 with the new quality weight", () => {
const sum = Object.values(DEFAULT_WEIGHTS).reduce((a, b) => a + Number(b), 0);
assert.ok(Math.abs(sum - 1) < 1e-9, `expected sum ≈ 1, got ${sum}`);
assert.ok(validateWeights(DEFAULT_WEIGHTS), "validateWeights must accept DEFAULT_WEIGHTS");
assert.ok((DEFAULT_WEIGHTS.quality ?? 0) > 0, "quality weight must be > 0");
});
test("calculateFactors defaults missing quality to neutral 0.5", () => {
const factors = calculateFactors(candidate(), [candidate()], "general", () => 0.5);
assert.equal(factors.quality, 0.5);
});
test("calculateFactors clamps quality to [0,1]", () => {
const low = calculateFactors(candidate({ quality: -2 }), [candidate()], "general", () => 0.5);
assert.equal(low.quality, 0);
const high = calculateFactors(candidate({ quality: 5 }), [candidate()], "general", () => 0.5);
assert.equal(high.quality, 1);
});
test("calculateScore applies the quality factor", () => {
const base: ScoringFactors = {
quota: 0.5,
health: 0.5,
costInv: 0.5,
latencyInv: 0.5,
taskFit: 0.5,
stability: 0.5,
tierPriority: 0.5,
tierAffinity: 0.5,
specificityMatch: 0.5,
contextAffinity: 0.5,
resetWindowAffinity: 0.5,
connectionDensity: 0.5,
};
const good = calculateScore({ ...base, quality: 1 }, DEFAULT_WEIGHTS);
const bad = calculateScore({ ...base, quality: 0 }, DEFAULT_WEIGHTS);
assert.ok(good > bad, "higher quality must score strictly higher");
assert.ok(good >= 0 && good <= 1);
assert.ok(bad >= 0 && bad <= 1);
});
test("low-quality candidate ranks below identical high-quality candidate", () => {
const good = candidate({ provider: "p", model: "good", quality: 1 });
const poor = candidate({ provider: "p", model: "poor", quality: 0.3 });
const pool = [good, poor];
const fg = calculateFactors(good, pool, "general", () => 0.5);
const fp = calculateFactors(poor, pool, "general", () => 0.5);
const sg = calculateScore(fg, DEFAULT_WEIGHTS);
const sp = calculateScore(fp, DEFAULT_WEIGHTS);
assert.ok(sg > sp, `good candidate (${sg}) must outrank poor (${sp})`);
});
test("normalizeScoringWeights keeps quality and renormalizes to 1", () => {
const normalized = normalizeScoringWeights({ quality: 0.1 });
const total = Object.values(normalized).reduce((s, v) => s + Number(v), 0);
assert.ok(Math.abs(total - 1) < 1e-9);
assert.ok((normalized.quality ?? 0) > 0);
});

View File

@@ -0,0 +1,86 @@
/**
* tests/unit/stream-timing.test.ts
*
* Canonical stream instrumentation (open-sse/utils/streamTiming.ts):
* - TTFT = first-forwarded-SSE-chunk latency (NOT token-level) — documented
* - ITL = mean inter-chunk gap (chunk-latency proxy)
* - first-byte vs first-forward distinction
* - interruption marking
* - malformed/empty chunks do not corrupt timing
*/
import test from "node:test";
import assert from "node:assert/strict";
import { createStreamTiming, type StreamTiming } from "../../open-sse/utils/streamTiming.ts";
test("ttft() is null when nothing was forwarded", () => {
const t = createStreamTiming();
t.markByte();
assert.equal(t.ttftMs(), null);
assert.equal(t.avgItlMs(), null);
});
test("ttft() measures first-forwarded-chunk latency (byte vs forward distinguished)", async () => {
const t = createStreamTiming();
t.markByte(); // first upstream byte arrives immediately
await new Promise((r) => setTimeout(r, 20));
t.markForward(); // first chunk forwarded 20ms later
const ttft = t.ttftMs();
assert.ok(ttft !== null && ttft >= 20 && ttft < 5000, `ttft=${ttft}`);
assert.ok(t.firstByteAt !== null);
assert.ok(t.firstByteAt! < t.firstForwardAt!, "first byte precedes first forward");
});
test("avgItlMs() measures mean inter-chunk gap across multiple chunks", async () => {
const t = createStreamTiming();
for (let i = 0; i < 4; i++) {
t.markForward();
await new Promise((r) => setTimeout(r, 10));
}
const itl = t.avgItlMs();
assert.ok(itl !== null && itl >= 8 && itl < 5000, `itl=${itl}`);
assert.equal(t.forwardedChunks, 4);
});
test("empty chunks do not corrupt timing (markByte without forward)", () => {
const t = createStreamTiming();
t.markByte();
t.markByte(); // duplicate bytes are idempotent for first-byte
assert.equal(t.ttftMs(), null, "no forward → no ttft");
t.markForward();
assert.ok(t.ttftMs() !== null);
});
test("malformed/keepalive-only traffic (no forward) yields no ttft", () => {
const t = createStreamTiming();
// Simulate a provider that only sends keepalives/blank lines, never data.
for (let i = 0; i < 5; i++) t.markByte();
assert.equal(t.ttftMs(), null);
assert.equal(t.forwardedChunks, 0);
});
test("interruption is recorded and does not reset other timing", async () => {
const t = createStreamTiming();
t.markForward();
await new Promise((r) => setTimeout(r, 5));
t.markForward();
t.markInterrupted();
assert.equal(t.interrupted, true);
assert.ok(t.ttftMs() !== null);
assert.ok(t.avgItlMs() !== null);
});
test("normal completion: totalMs() is monotonic and >= first-forward latency", async () => {
const t = createStreamTiming();
await new Promise((r) => setTimeout(r, 15));
t.markForward();
const total = t.totalMs();
const ttft = t.ttftMs();
assert.ok(total >= 15);
assert.ok(ttft !== null && ttft <= total, "ttft must be <= total duration");
});
test("max inter-chunk samples are bounded (memory bound)", async () => {
const t = createStreamTiming();
for (let i = 0; i < 200; i++) t.markForward();
assert.ok(t.interChunkGaps.length <= 32, `bounded to 32 samples, got ${t.interChunkGaps.length}`);
});