diff --git a/docs/architecture/ADAPTIVE_ROUTING.md b/docs/architecture/ADAPTIVE_ROUTING.md new file mode 100644 index 0000000000..73bacfec3a --- /dev/null +++ b/docs/architecture/ADAPTIVE_ROUTING.md @@ -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. diff --git a/docs/architecture/meta.json b/docs/architecture/meta.json index dba5872324..c5b10b7923 100644 --- a/docs/architecture/meta.json +++ b/docs/architecture/meta.json @@ -12,6 +12,7 @@ "ROUTER_BACKENDS", "admission-lanes", "cluster-decisions", - "persistence-backend-boundary" + "persistence-backend-boundary", + "ADAPTIVE_ROUTING" ] } diff --git a/docs/getting-started/PROVIDERS-GUIDE.md b/docs/getting-started/PROVIDERS-GUIDE.md index 65de3c63ea..d6d20906c5 100644 --- a/docs/getting-started/PROVIDERS-GUIDE.md +++ b/docs/getting-started/PROVIDERS-GUIDE.md @@ -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 + + # 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 " \ + -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 diff --git a/docs/reference/PROVIDER_REFERENCE.md b/docs/reference/PROVIDER_REFERENCE.md index f6b2568af5..e4f3c4ad58 100644 --- a/docs/reference/PROVIDER_REFERENCE.md +++ b/docs/reference/PROVIDER_REFERENCE.md @@ -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). | diff --git a/open-sse/config/providers/index.ts b/open-sse/config/providers/index.ts index 34189e0d00..c098dbf29f 100644 --- a/open-sse/config/providers/index.ts +++ b/open-sse/config/providers/index.ts @@ -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 = { aimlapi: aimlapiProvider, + "mlx-gemma": mlxGemmaProvider, + "mlx-qwen": mlxQwenProvider, "ollama-cloud": ollama_cloudProvider, synthetic: syntheticProvider, ideogram: ideogramProvider, diff --git a/open-sse/config/providers/registry/mlx/index.ts b/open-sse/config/providers/registry/mlx/index.ts new file mode 100644 index 0000000000..24d3e8b232 --- /dev/null +++ b/open-sse/config/providers/registry/mlx/index.ts @@ -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 +}); diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index e955ffd5d3..fa6085ad19 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -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; + const choices = record.choices; + if (Array.isArray(choices)) { + const first = choices[0]; + if (first && typeof first === "object") { + const fr = (first as Record).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).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).prompt_tokens; + return typeof promptTokens === "number" && Number.isFinite(promptTokens) + ? promptTokens + : null; + })() + : null, + outputTokens: + usage && typeof usage === "object" + ? (() => { + const completionTokens = (usage as Record).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).prompt_tokens; + return typeof promptTokens === "number" && Number.isFinite(promptTokens) + ? promptTokens + : null; + })() + : null, + outputTokens: + streamUsage && typeof streamUsage === "object" + ? (() => { + const completionTokens = (streamUsage as Record).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, diff --git a/open-sse/services/autoCombo/scoring.ts b/open-sse/services/autoCombo/scoring.ts index b8f66a797e..4c939501c7 100644 --- a/open-sse/services/autoCombo/scoring.ts +++ b/open-sse/services/autoCombo/scoring.ts @@ -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), }; } diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index b6ae5d534a..cf602cf98d 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -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), }; }) ); diff --git a/open-sse/services/combo/quotaShareStrategy.ts b/open-sse/services/combo/quotaShareStrategy.ts index e9e4293543..e12958042c 100644 --- a/open-sse/services/combo/quotaShareStrategy.ts +++ b/open-sse/services/combo/quotaShareStrategy.ts @@ -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; } diff --git a/open-sse/services/routing/events.ts b/open-sse/services/routing/events.ts new file mode 100644 index 0000000000..955bbe28e2 --- /dev/null +++ b/open-sse/services/routing/events.ts @@ -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(); + +/** + * 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"; +} diff --git a/open-sse/services/routing/index.ts b/open-sse/services/routing/index.ts new file mode 100644 index 0000000000..1cb3077036 --- /dev/null +++ b/open-sse/services/routing/index.ts @@ -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 { + 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"; diff --git a/open-sse/services/routing/otel.ts b/open-sse/services/routing/otel.ts new file mode 100644 index 0000000000..23578011c0 --- /dev/null +++ b/open-sse/services/routing/otel.ts @@ -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 | 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 { + 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, + }; +} diff --git a/open-sse/services/routing/quality.ts b/open-sse/services/routing/quality.ts new file mode 100644 index 0000000000..b4e06a9ed9 --- /dev/null +++ b/open-sse/services/routing/quality.ts @@ -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(); + +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; diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index e6547bb3e5..275669e983 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -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, 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 | 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)); } } diff --git a/open-sse/utils/streamTiming.ts b/open-sse/utils/streamTiming.ts new file mode 100644 index 0000000000..9f26f5d5b3 --- /dev/null +++ b/open-sse/utils/streamTiming.ts @@ -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; +} diff --git a/package.json b/package.json index a199504707..fdacc42ef4 100644 --- a/package.json +++ b/package.json @@ -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", diff --git a/scripts/perf/routing-events-bench.ts b/scripts/perf/routing-events-bench.ts new file mode 100644 index 0000000000..bf89c4a3fc --- /dev/null +++ b/scripts/perf/routing-events-bench.ts @@ -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 { + 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"); diff --git a/src/app/api/v1/explain/routing/route.ts b/src/app/api/v1/explain/routing/route.ts new file mode 100644 index 0000000000..313bcf27b5 --- /dev/null +++ b/src/app/api/v1/explain/routing/route.ts @@ -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"); + } +} diff --git a/src/lib/usage/comboScoringInspector.ts b/src/lib/usage/comboScoringInspector.ts index 764f7ed0fb..54f97cd214 100644 --- a/src/lib/usage/comboScoringInspector.ts +++ b/src/lib/usage/comboScoringInspector.ts @@ -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 { diff --git a/src/shared/constants/providers.ts b/src/shared/constants/providers.ts index 0ae41d9532..37174092b2 100644 --- a/src/shared/constants/providers.ts +++ b/src/shared/constants/providers.ts @@ -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", diff --git a/src/shared/constants/providers/local.ts b/src/shared/constants/providers/local.ts index a16939fbb9..a3e455d64f 100644 --- a/src/shared/constants/providers/local.ts +++ b/src/shared/constants/providers/local.ts @@ -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", diff --git a/src/shared/types/utilization.ts b/src/shared/types/utilization.ts index 20ec1850d1..4b637e7385 100644 --- a/src/shared/types/utilization.ts +++ b/src/shared/types/utilization.ts @@ -272,7 +272,8 @@ export type ComboScoringInspectorFactorKey = | "cacheAffinity" | "sessionAvailability" | "resetWindowAffinity" - | "connectionDensity"; + | "connectionDensity" + | "quality"; export type ComboScoringInspectorSource = "combo_health" | "combo_forecast" | "combo_autopilot" | "runtime" | "default"; diff --git a/tests/unit/auto-combo-scoring-clamp.test.ts b/tests/unit/auto-combo-scoring-clamp.test.ts index 1e00aca021..a73d24cf8b 100644 --- a/tests/unit/auto-combo-scoring-clamp.test.ts +++ b/tests/unit/auto-combo-scoring-clamp.test.ts @@ -36,6 +36,7 @@ const ONES: ScoringFactors = { contextAffinity: 1, resetWindowAffinity: 1, connectionDensity: 1, + quality: 1, }; function candidate(partial: Partial = {}): ProviderCandidate { diff --git a/tests/unit/mlx-provider.test.ts b/tests/unit/mlx-provider.test.ts new file mode 100644 index 0000000000..341717e46d --- /dev/null +++ b/tests/unit/mlx-provider.test.ts @@ -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")); + }); +}); diff --git a/tests/unit/routing-adaptive-e2e.test.ts b/tests/unit/routing-adaptive-e2e.test.ts new file mode 100644 index 0000000000..2d71bb289e --- /dev/null +++ b/tests/unit/routing-adaptive-e2e.test.ts @@ -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 { + 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[0]> = {} +): void { + recordQualityEvent({ + provider, + model, + outcome: "success", + status: 200, + latencyMs: 100, + finishReason: "stop", + outputTokens: 5, + ...partial, + }); +} + +function bestProvider(targets: ReturnType): 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 { + 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"); +}); diff --git a/tests/unit/routing-events-concurrency.test.ts b/tests/unit/routing-events-concurrency.test.ts new file mode 100644 index 0000000000..1eff7a1779 --- /dev/null +++ b/tests/unit/routing-events-concurrency.test.ts @@ -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}`); +}); diff --git a/tests/unit/routing-events.test.ts b/tests/unit/routing-events.test.ts new file mode 100644 index 0000000000..672e566664 --- /dev/null +++ b/tests/unit/routing-events.test.ts @@ -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 { + 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(); + } +}); diff --git a/tests/unit/routing-otel.test.ts b/tests/unit/routing-otel.test.ts new file mode 100644 index 0000000000..1a68484f81 --- /dev/null +++ b/tests/unit/routing-otel.test.ts @@ -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 { + 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; + } +}); diff --git a/tests/unit/routing-quality.test.ts b/tests/unit/routing-quality.test.ts new file mode 100644 index 0000000000..56197401f1 --- /dev/null +++ b/tests/unit/routing-quality.test.ts @@ -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[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); +}); diff --git a/tests/unit/routing-scoring-quality.test.ts b/tests/unit/routing-scoring-quality.test.ts new file mode 100644 index 0000000000..974ede1c51 --- /dev/null +++ b/tests/unit/routing-scoring-quality.test.ts @@ -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 { + 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); +}); diff --git a/tests/unit/stream-timing.test.ts b/tests/unit/stream-timing.test.ts new file mode 100644 index 0000000000..b6c5782579 --- /dev/null +++ b/tests/unit/stream-timing.test.ts @@ -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}`); +});