diff --git a/.env.example b/.env.example index 67dc10bd17..99b6adabf9 100644 --- a/.env.example +++ b/.env.example @@ -541,6 +541,48 @@ ALLOW_API_KEY_REVEAL=false # When unset, OmniRoute uses the per-feature defaults. Set to "false"/"0" to disable. # OUTBOUND_SSRF_GUARD_ENABLED=true +# ── Self-hosted unified OpenAI-compatible entry (RIC-738, D4) ──────────────────── +# When set, /v1/chat/completions diverts to the self-hosted provider adapters +# (open-sse/services/selfHostedEntry.ts) instead of the cloud pipeline. YAML inline +# (example) — or point OMNIROUTE_SELF_HOSTED_PROVIDERS_FILE at a YAML file. Secrets +# are runtime-only, never logged. While ANY of these is set, the entry is active; +# config present but unparseable returns a 500 (never silently falls through). +# OMNIROUTE_SELF_HOSTED_PROVIDERS=' +# providers: +# - id: local +# kind: openai +# baseUrl: http://127.0.0.1:11434/v1 +# model: llama3 +# - id: claude +# kind: anthropic +# baseUrl: http://127.0.0.1:8080 +# model: claude-sonnet +# ' +# OMNIROUTE_SELF_HOSTED_PROVIDERS_FILE=/etc/omniroute/providers.yaml +# Optional shared API key for the unified entry (D5 reserved). When set, require +# `Authorization: Bearer `; empty = open loopback/trusted-network route. +# OMNIROUTE_SELF_HOSTED_API_KEY= + +# ── Deterministic routing strategies (M2 / RIC-740, D3 可审计路由) ───────────── +# Optional `strategy:` block — either inline in the providers document above, or a +# standalone document via these env vars. One rule per line; every decision is +# explainable via the `x-omniroute-route-decision` response header. No ML/predict. +# Malformed strategy config returns a 500 (never silently becomes a no-op). +# Example (inline, same shape as `strategy:` inside the providers YAML): +# OMNIROUTE_SELF_HOSTED_STRATEGY=' +# blacklist: [] +# whitelist: [cheap, fast, premium] +# costPriority: true +# latencyAware: +# enabled: true +# cooldown: +# consecutiveFailures: 2 +# cooldownMs: 30000 +# fallbackChain: [cheap, fast, premium] +# ' +# OMNIROUTE_SELF_HOSTED_STRATEGY_FILE=/etc/omniroute/strategy.yaml +# See docs/routing/DETERMINISTIC_ROUTING.md for the full strategy surface. + # ═══════════════════════════════════════════════════════════════════════════════ # 5. INPUT SANITIZATION & PII PROTECTION (FASE-01) # ═══════════════════════════════════════════════════════════════════════════════ diff --git a/changelog.d/features/ric738-openai-compat-self-hosted.md b/changelog.d/features/ric738-openai-compat-self-hosted.md new file mode 100644 index 0000000000..d425118a40 --- /dev/null +++ b/changelog.d/features/ric738-openai-compat-self-hosted.md @@ -0,0 +1 @@ +- **feat(routing): self-hosted unified OpenAI-compatible entry (`/v1/chat/completions`).** When `OMNIROUTE_SELF_HOSTED_PROVIDERS` (inline YAML) or `OMNIROUTE_SELF_HOSTED_PROVIDERS_FILE` is set, the existing `/v1/chat/completions` route diverts through the self-hosted provider adapters (`open-sse/services/providerAdapters.ts`) — OpenAI / Anthropic / local-compatible — instead of the cloud pipeline. Provider is auto-routed via the `x-omniroute-provider` header, a `provider/model` (or `provider::model`) model prefix, or the first configured provider; upstream credentials stay runtime-only and are stripped from echoed responses. Optional `OMNIROUTE_SELF_HOSTED_API_KEY` guards the entry with `Authorization: Bearer` (reserved for the D5 quota-key system); unset = open loopback/trusted-network route. Upstream failures return the standard OpenAI error shape (including a normalized 502 for unreachable providers). One OpenAI SDK snippet can now traverse multiple self-hosted providers without changing the client. (#RIC-738 / RIC-697 D4) diff --git a/changelog.d/features/ric740-deterministic-routing.md b/changelog.d/features/ric740-deterministic-routing.md new file mode 100644 index 0000000000..ec70fef11c --- /dev/null +++ b/changelog.d/features/ric740-deterministic-routing.md @@ -0,0 +1 @@ +- **feat(routing): deterministic routing strategies for the self-hosted entry (`strategy:` block, M2/RIC-740).** The unified `/v1/chat/completions` entry (RIC-738) now accepts a declarative `strategy:` block — inline in the providers YAML or via `OMNIROUTE_SELF_HOSTED_STRATEGY` / `OMNIROUTE_SELF_HOSTED_STRATEGY_FILE` — expressing five explainable, non-predictive routing policies: blacklist / whitelist (hard filters), cooldown circuit breaker (`consecutiveFailures` + `cooldownMs`), cost-priority (cheapest `costPer1MInput` first), latency-aware (fastest recent average first), and an explicit `fallbackChain` order. The ordered candidate list is the fallback chain: a failed primary (network or non-2xx) falls through to the next candidate, and each failure feeds the breaker. Every response carries `x-omniroute-route-decision` — the one-line "why this model / why not that one" audit trail (D3). A pinned provider rejected by a hard filter returns `400` (never a silent re-route); no eligible providers returns `503` with the full explainable decision. No ML/predict dependency; malformed strategy config returns `500` rather than silently becoming a no-op. (#RIC-740 / RIC-697 D3) diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 927cb3b7be..154907420c 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -310,6 +310,11 @@ OmniRoute provides a two-layer defense: request-side injection scanning and resp | `OMNIROUTE_PREFER_CLAUDE_CODE_FOR_UNPREFIXED_CLAUDE_MODELS` | `false` | `open-sse/services/model.ts` | Opt-in: route bare `claude-*` model IDs from Claude Code clients through the Claude Code OAuth account instead of requiring a provider prefix. Explicit provider prefixes still win. Also configurable via a dashboard toggle on the Claude provider page. | | `COMBO_CONCURRENCY_PER_MODEL` | `3` | `open-sse/services/comboConfig.ts` | Per-model concurrency cap for round-robin combos (#9100). The round-robin combo semaphore was hard-capped at 3 concurrent requests per model with no override, serializing higher-concurrency traffic behind that cap. Validated to `>= 1`, clamped to `<= 32`. | | `DISABLE_CONTEXT_WINDOW_CHECKS` | `false` | `open-sse/handlers/chatCore.ts` | Dangerous opt-in that skips OmniRoute's local context-window / max-input-token check for direct single-model requests. Upstream providers still enforce their actual limits; prompt compression and the model's own output-token cap remain active. Effective precedence is Feature Flags DB override > environment variable > default; no restart is required. | +| `OMNIROUTE_SELF_HOSTED_PROVIDERS` | _(unset)_ | `open-sse/services/selfHostedEntry.ts` | Inline YAML `providers:` document (RIC-738, D4). When set (with or without a `strategy:` block), `/v1/chat/completions` diverts to the self-hosted unified OpenAI-compatible entry instead of the cloud pipeline. Unset (the default): the route falls straight through to the existing cloud pipeline. See `docs/routing/SELF_HOSTED_OPENAI_ENTRY.md`. | +| `OMNIROUTE_SELF_HOSTED_PROVIDERS_FILE` | _(unset)_ | `open-sse/services/selfHostedEntry.ts` | Path to a YAML file holding the same `providers:` document as `OMNIROUTE_SELF_HOSTED_PROVIDERS`, for deployments that prefer a file over an inline env var. Either one activates the self-hosted entry. | +| `OMNIROUTE_SELF_HOSTED_API_KEY` | _(unset — open route)_ | `open-sse/services/selfHostedEntry.ts` | Optional shared API key for the unified self-hosted entry (D5 scaffold, reserved for the per-key quota system). When set, requests must include `Authorization: Bearer `. Unset: the route is open, matching the existing self-hosted local-provider pattern (loopback/trusted-network deployment). | +| `OMNIROUTE_SELF_HOSTED_STRATEGY` | _(unset)_ | `open-sse/services/routingStrategies.ts` | Inline YAML `strategy:` document for the deterministic routing engine (M2/RIC-740, D3) — blacklist/whitelist, cooldown breaker, cost-priority, latency-aware, fallback chain. Overrides an inline `strategy:` block nested inside `OMNIROUTE_SELF_HOSTED_PROVIDERS`, per-key. See `docs/routing/DETERMINISTIC_ROUTING.md`. | +| `OMNIROUTE_SELF_HOSTED_STRATEGY_FILE` | _(unset)_ | `open-sse/services/routingStrategies.ts` | Path to a YAML file holding the same `strategy:` document as `OMNIROUTE_SELF_HOSTED_STRATEGY`, for deployments that prefer a file over an inline env var. | | `OMNIROUTE_DISABLE_CONVERSATION_TRACKING` | _(unset)_ | `open-sse/services/conversationTracker.ts` | Set `1` to stop collecting conversation history. `resolveConversationId()` returns an untracked result before it reads SQLite or parses message history, and client-supplied session IDs are covered too. Routing-session handling is unchanged and existing records are not deleted. For deployments that do not use the dashboard's conversation view and want the turn table to stop growing. | --- diff --git a/docs/routing/DETERMINISTIC_ROUTING.md b/docs/routing/DETERMINISTIC_ROUTING.md new file mode 100644 index 0000000000..83c008d264 --- /dev/null +++ b/docs/routing/DETERMINISTIC_ROUTING.md @@ -0,0 +1,124 @@ +--- +title: "Deterministic Routing Strategies" +--- + +# Deterministic routing strategies + +OmniRoute's self-hosted gateway entry (`/v1/chat/completions`, RIC-738) routes to a +single provider by default. When you configure a `strategy:` block, the route decision +becomes a **deterministic, explainable policy** — the OmniRoute differentiator vs +NotDiamond/Martian "predict" black-box routing. + +Every decision is: + +- **Deterministic** — the same config + same runtime state yields the same provider. +- **Config-expressed** — one rule per line; no "一体化智能体", no learned model. +- **Explainable** — every response carries `x-omniroute-route-decision`, a one-line + answer to "why this model?" (and "why NOT that one"). + +This is M2 of the RIC-697 differentiation (D3 可审计路由). There is **no predictive / +ML dependency** — the strategies are pure rules over observable signals +(consecutive failures, declared cost, measured latency). + +## Enable + +Add a `strategy:` block to the same YAML document as `providers:`, or point +`OMNIROUTE_SELF_HOSTED_STRATEGY` (inline YAML) / `OMNIROUTE_SELF_HOSTED_STRATEGY_FILE` +(a file) at a standalone `strategy:` document. An explicit env strategy merges over +the inline block per-key. + +```yaml +# providers.yaml +providers: + - id: cheap + kind: openai + baseUrl: http://127.0.0.1:11434/v1 + model: llama3 + costPer1MInput: 0.2 # USD per 1M input tokens — used by cost-priority + - id: fast + kind: openai + baseUrl: http://127.0.0.1:8080/v1 + model: gpt-4o-mini + costPer1MInput: 0.6 + - id: premium + kind: anthropic + baseUrl: http://127.0.0.1:8081 + model: claude-sonnet + costPer1MInput: 3.0 + +strategy: + blacklist: [] # provider ids never used + whitelist: [cheap, fast, premium] # when non-empty, ONLY these are used + costPriority: true # cheapest eligible candidate first + latencyAware: + enabled: true # fastest recent-average candidate first + cooldown: + consecutiveFailures: 2 # breaker trips after this many in a row + cooldownMs: 30000 # …and the provider stays excluded this long + fallbackChain: [cheap, fast, premium] # explicit fallback order +``` + +## The five strategies + +| Strategy | Config | Effect | +| ---------------- | --------------------------------------------- | ------------------------------------------------------------------------------------------------------ | +| Blacklist | `blacklist: [id, ...]` | Listed providers are never candidates. | +| Whitelist | `whitelist: [id, ...]` | Non-empty → only listed providers are candidates. | +| Cooldown/breaker | `cooldown: {consecutiveFailures, cooldownMs}` | After N consecutive failures a provider is excluded for the window. A success resets the counter. | +| Cost-priority | `costPriority: true` | Eligible candidates sorted by `costPer1MInput` ascending (declared, never fabricated). | +| Latency-aware | `latencyAware.enabled: true` | Eligible candidates sorted by recent average request latency ascending. Unsampled providers sort last. | +| Fallback chain | `fallbackChain: [id, ...]` | Explicit primary→backup order. Wins over cost/latency ordering. | + +Filters (blacklist / whitelist / cooldown) run first and remove candidates. Then the +ordering stage sorts the survivors: explicit `fallbackChain` wins; otherwise +`costPriority` then `latencyAware` apply in that order. + +## Pinned providers + +The `x-omniroute-provider` header and the `provider/model` model-prefix are **explicit +pins**, not preferences. If a pinned provider is excluded by a hard filter (blacklist / +whitelist / active cooldown), the request fails with `400` and the pin reason — the +gateway never silently re-routes a client who asked for a specific provider. If the pin +survives the filters it is the first candidate, and the remaining candidates serve as +its fallback chain. + +## Fallback on failure + +The ordered candidate list is the fallback chain. When the first provider fails +(network unreachable, connection refused, DNS/TLS, or any non-2xx), the gateway walks +the next candidate, and so on. Each failure is recorded into the cooldown breaker and +each attempt is latency-sampled — so a broken primary also cools down for _subsequent_ +requests. When every candidate fails, the last normalized error (OpenAI shape) is +returned. When no candidate is eligible at all, a `503` with the full explainable +decision is returned. + +## Explainability + +Every response routed through the strategy engine carries: + +``` +x-omniroute-route-decision: cheap(cost-priority: cheap #1) -> fast(cost-priority: fast #2); premium excluded: blacklist: premium forbidden +``` + +The value is the ordered candidate list plus every exclusion reason — the audit trail +for "why this model, why not that one". + +## Failure contract (unchanged from RIC-738) + +- Upstream non-2xx body is normalized through `parseUpstreamError` + `buildErrorBody` + into the standard OpenAI `error` shape (Hard Rule #12). +- Network-level failures return a normalized `502`; after the whole chain is exhausted, + the last normalized error is surfaced. +- Credential / session headers echoed upstream are stripped from every response. +- Provider config with a malformed `strategy:` block returns `500` — a misconfigured + policy never silently becomes a no-op. + +## Tests + +- `tests/unit/routing-strategies.test.ts` — 25 unit tests over the pure strategy engine: + parse, blacklist/whitelist, cooldown breaker, cost-priority, latency-aware, fallback + chain, combined pipeline, explainability. +- `tests/unit/self-hosted-entry.test.ts` — fault injection over real HTTP: primary down + → fallback succeeds; all-down → last error; pinned-blacklisted → 400; cooldown + excludes a failing provider on the next request; no-eligible → 503 with explainable + decision. diff --git a/docs/routing/SELF_HOSTED_OPENAI_ENTRY.md b/docs/routing/SELF_HOSTED_OPENAI_ENTRY.md new file mode 100644 index 0000000000..3fbc3030c1 --- /dev/null +++ b/docs/routing/SELF_HOSTED_OPENAI_ENTRY.md @@ -0,0 +1,93 @@ +--- +title: "Self-Hosted OpenAI-Compatible Entry" +--- + +# Self-hosted unified OpenAI-compatible entry + +When enabled, OmniRoute's existing `/v1/chat/completions` (and the OpenAI-compatible +contract it serves) becomes a **self-hosted gateway**: one OpenAI-compatible request +in, auto-routed to the provider of your choice through the provider adapters, with a +standard OpenAI error shape out. Client code does not change. + +This is the D4 (接入即用) differentiator from the RIC-697 design: the same `/v1` path +the OpenAI SDK already targets, backed by your own providers instead of a single +vendored catalog. + +## Enable + +Set either env var (see `.env.example` for both): + +- `OMNIROUTE_SELF_HOSTED_PROVIDERS` — inline YAML document (runtime-only, not logged). +- `OMNIROUTE_SELF_HOSTED_PROVIDERS_FILE` — path to a YAML file. + +```yaml +# providers.yaml +providers: + - id: local + kind: openai # openai | anthropic | local + baseUrl: http://127.0.0.1:11434/v1 + model: llama3 + # apiKey: sk-... # optional, runtime-only + - id: claude + kind: anthropic + baseUrl: http://127.0.0.1:8080 + model: claude-sonnet +``` + +When **either** var is set, the unified entry is active for every request to +`/v1/chat/completions`. Config present but unparseable returns `500` (it never +silently falls through to cloud routing). When neither is set, the route behaves +exactly as before. + +## Provider auto-route + +Precedence (deterministic, no predictive model): + +1. `x-omniroute-provider: ` header — exact provider id. +2. `model` prefix: `provider/model` (slash) or `provider::model` (double colon). +3. First configured provider. + +The routing prefix is stripped before forwarding — upstream receives the bare model +(`claude-sonnet`, not `claude/claude-sonnet`). + +```bash +# via header +curl http://localhost:20128/v1/chat/completions \ + -H "x-omniroute-provider: claude" \ + -H "content-type: application/json" \ + -d '{"model":"claude-sonnet","messages":[{"role":"user","content":"hi"}]}' + +# or via model prefix +curl ... -d '{"model":"claude/claude-sonnet","messages":[...]}' +``` + +## Auth scaffold (D5 reserved) + +Optional `OMNIROUTE_SELF_HOSTED_API_KEY`. When set, requests must carry +`Authorization: Bearer `. Unset = open route (loopback / trusted network), +mirroring how OmniRoute's existing local providers work. The D5 quota/quota-自治 +key system is expected to take over this header. + +## Failure contract + +- Upstream non-2xx body is normalized through `parseUpstreamError` + `buildErrorBody` + into the standard OpenAI `error` shape (Hard Rule #12 — never raw upstream text). +- Network-level failures (connection refused / DNS / TLS) return a normalized `502` + with `error.message: "Upstream provider unreachable: …"`. +- Credential / session headers echoed upstream are stripped from every response. + +## Tests + +`tests/unit/self-hosted-entry.test.ts` covers provider selection, model-prefix +forwarding, header hygiene, auth, upstream-error normalization, SSE streaming +passthrough, and the fall-through / misconfig paths — all over real HTTP against a +local upstream. + +## Notes + +- Self-hosted models bypass the cloud-only retirement / alias machinery by design: + the divert happens before those cloud checks, so ids like `local/llama3` never + trip cloud-peer 410s or alias rewrites. +- Deterministic routing strategies (fallback / cooldown / cost / latency / blacklist) + are M2's scope (`RIC-740`) and inject into the gateway layer here. Configure them + with a `strategy:` block — see [DETERMINISTIC_ROUTING.md](./DETERMINISTIC_ROUTING.md). diff --git a/open-sse/services/providerAdapters.ts b/open-sse/services/providerAdapters.ts new file mode 100644 index 0000000000..7724cbfaf2 --- /dev/null +++ b/open-sse/services/providerAdapters.ts @@ -0,0 +1,137 @@ +import * as yaml from "js-yaml"; + +export type ProviderKind = "openai" | "anthropic" | "local"; + +export interface ProviderConfig { + id: string; + kind: ProviderKind; + baseUrl: string; + model: string; + /** API key is runtime-only and is never returned by public helpers. */ + apiKey?: string; +} + +export interface ChatRequest { + messages: Array>; + model?: string; + stream?: boolean; + [key: string]: unknown; +} + +export interface ProviderAdapter { + readonly kind: ProviderKind; + complete( + request: ChatRequest, + config: ProviderConfig, + fetchImpl?: typeof fetch + ): Promise; +} + +const trimUrl = (url: string): string => url.replace(/\/+$/, ""); + +function authHeaders(config: ProviderConfig): Record { + return config.apiKey ? { Authorization: `Bearer ${config.apiKey}` } : {}; +} + +const openaiAdapter: ProviderAdapter = { + kind: "openai", + complete: (request, config, fetchImpl = fetch) => + fetchImpl(`${trimUrl(config.baseUrl)}/chat/completions`, { + method: "POST", + headers: { "content-type": "application/json", ...authHeaders(config) }, + body: JSON.stringify({ ...request, model: request.model || config.model }), + }), +}; + +const localAdapter: ProviderAdapter = { ...openaiAdapter, kind: "local" }; + +const anthropicAdapter: ProviderAdapter = { + kind: "anthropic", + complete: (request, config, fetchImpl = fetch) => { + const { system, ...body } = request as ChatRequest & { system?: unknown }; + const headers: Record = { + "content-type": "application/json", + "anthropic-version": "2023-06-01", + ...(config.apiKey ? { "x-api-key": config.apiKey } : {}), + }; + return fetchImpl(`${trimUrl(config.baseUrl)}/messages`, { + method: "POST", + headers, + body: JSON.stringify({ + ...body, + ...(system === undefined ? {} : { system }), + model: request.model || config.model, + }), + }); + }, +}; + +const ADAPTERS: Record = { + openai: openaiAdapter, + anthropic: anthropicAdapter, + local: localAdapter, +}; + +export function getProviderAdapter(kind: ProviderKind): ProviderAdapter { + return ADAPTERS[kind]; +} + +export interface ProviderConfigFile { + providers: ProviderConfig[]; +} + +/** Parse self-hosted YAML without persisting or logging credentials. */ +export function parseProviderConfig(source: string): ProviderConfigFile { + const document = yaml.load(source); + if (!document || typeof document !== "object" || Array.isArray(document)) + throw new Error("Provider config must be a mapping"); + const providers = (document as Record).providers; + if (!Array.isArray(providers)) throw new Error("Provider config requires a providers list"); + const parsed = providers.map((value, index) => { + if (!value || typeof value !== "object" || Array.isArray(value)) + throw new Error(`Provider ${index} must be a mapping`); + const item = value as Record; + const id = typeof item.id === "string" && item.id.trim(); + const kind = item.kind; + const baseUrl = typeof item.baseUrl === "string" && item.baseUrl.trim(); + const model = typeof item.model === "string" && item.model.trim(); + if ( + !id || + !baseUrl || + !model || + !(kind === "openai" || kind === "anthropic" || kind === "local") + ) + throw new Error(`Invalid provider ${index}`); + const provider: ProviderConfig = { + id, + kind, + baseUrl, + model, + ...(typeof item.apiKey === "string" && item.apiKey ? { apiKey: item.apiKey } : {}), + }; + return provider; + }); + return { providers: parsed }; +} + +export function publicProviderConfigs( + config: ProviderConfigFile +): Array> { + return config.providers.map(({ apiKey: _apiKey, ...safe }) => safe); +} + +export class ProviderRouter { + private readonly providers: Map; + constructor(config: ProviderConfigFile) { + this.providers = new Map(config.providers.map((provider) => [provider.id, provider])); + } + select(id?: string): ProviderConfig { + const selected = id ? this.providers.get(id) : this.providers.values().next().value; + if (!selected) throw new Error(id ? `Unknown provider: ${id}` : "No providers configured"); + return selected; + } + complete(request: ChatRequest, id?: string, fetchImpl?: typeof fetch): Promise { + const config = this.select(id); + return getProviderAdapter(config.kind).complete(request, config, fetchImpl); + } +} diff --git a/open-sse/services/routingStrategies.ts b/open-sse/services/routingStrategies.ts new file mode 100644 index 0000000000..157e09644e --- /dev/null +++ b/open-sse/services/routingStrategies.ts @@ -0,0 +1,577 @@ +/** + * Deterministic routing strategies (M2 — RIC-740, D3 可审计路由). + * + * The OmniRoute differentiator vs NotDiamond/Martian "predict 黑盒路线": every + * routing decision is a deterministic, explainable rule — never a learned model. + * This module is the strategy engine for the self-hosted gateway entry + * (`selfHostedEntry.ts`, RIC-738): it turns a declarative `strategy:` config + * into an ordered, auditable candidate list, and tracks the runtime state + * (cooldown / latency / cost) that the rules read. + * + * Design (KISS / RIC-740): + * - Config is declarative — one rule per line, no "一体化智能体". The 5 strategy + * types are: + * 1. blacklist / whitelist — hard filters (forbidden / forced providers). + * 2. cooldown / breaker — provider with N consecutive failures is + * excluded for a window. + * 3. cost-priority — among eligible candidates, cheapest first. + * 4. latency-aware — among eligible candidates, fastest first. + * 5. fallback chain — on primary failure, walk the ordered chain. + * - Every candidate carries an `explain` one-liner answering "why this model". + * - No predictive / ML dependency. Runtime state is in-memory only and resets + * on restart, mirroring `providerCooldownTracker.ts`. + */ + +import * as yaml from "js-yaml"; +import type { ProviderConfig } from "./providerAdapters.ts"; + +/** Per-provider unit price, declared in config (never fabricated). */ +export interface StrategyProviderConfig extends ProviderConfig { + /** USD per 1M input tokens; used by cost-priority. Absent = not scored. */ + costPer1MInput?: number; +} + +export interface LatencyAwareConfig { + enabled: boolean; +} + +export interface CooldownConfig { + /** Consecutive failures that trip the breaker. Default 2. */ + consecutiveFailures?: number; + /** How long a tripped provider stays excluded (ms). Default 30 s. */ + cooldownMs?: number; +} + +/** One line per rule — the whole M2 strategy surface. */ +export interface StrategyConfig { + /** Provider ids that must never be used. Empty = none. */ + blacklist?: string[]; + /** Provider ids allowed. Empty = all. A non-empty whitelist wins over nothing. */ + whitelist?: string[]; + /** Sort eligible candidates by `costPer1MInput` ascending. */ + costPriority?: boolean; + /** Sort eligible candidates by recent average latency ascending. */ + latencyAware?: LatencyAwareConfig; + /** Circuit breaker: exclude a provider after consecutive failures. */ + cooldown?: CooldownConfig; + /** + * Explicit fallback order. When set, this order is authoritative for the + * non-pinned part of the candidate list (filters still apply). When empty, + * the declared provider order is used. + */ + fallbackChain?: string[]; +} + +const DEFAULT_COOLDOWN_FAILURES = 2; +const DEFAULT_COOLDOWN_MS = 30 * 1000; + +interface CooldownState { + consecutiveFailures: number; + lastFailureAt: number; +} + +interface LatencyState { + totalMs: number; + count: number; +} + +export interface RouteCandidate { + provider: StrategyProviderConfig; + /** One-line answer to "why is this provider at this position". */ + explain: string; +} + +export interface CandidatesResult { + /** Ordered candidate list — the fallback order for the request. */ + candidates: RouteCandidate[]; + /** + * Providers excluded by a hard filter (blacklist / whitelist / cooldown), + * each with the one-line reason. Kept for the explainability contract — the + * route decision must answer "why NOT this provider" too. + */ + excluded: Array<{ providerId: string; reason: string }>; + /** + * Present when a pinned provider (header / model prefix) was rejected by a + * hard filter. The caller must NOT silently re-route — the pin is explicit. + */ + pinBlocked?: { providerId: string; reason: string }; +} + +interface ScoredCandidate { + provider: StrategyProviderConfig; + /** "eligible" | reason for exclusion — used to separate allowed vs excluded. */ + status: string; + /** One-line explain for an allowed candidate (filled by the ordering stage). */ + explain: string; +} + +function resolveCooldownFailures(config: CooldownConfig | undefined): number { + return config?.consecutiveFailures ?? DEFAULT_COOLDOWN_FAILURES; +} + +function resolveCooldownMs(config: CooldownConfig | undefined): number { + return config?.cooldownMs ?? DEFAULT_COOLDOWN_MS; +} + +/** + * Keep at most this many latency samples per provider — a memory bound so the + * in-memory history cannot grow unbounded on a busy gateway (same policy as + * `providerCooldownTracker.ts`'s window pruning). + */ +const MAX_LATENCY_SAMPLES_PER_PROVIDER = 200; + +/** + * Deterministic strategy engine. Pure computation + in-memory runtime state. + * `providers` is the full configured provider list; `config` is the `strategy:` + * block (may be `null` for "no strategies configured"). + */ +export class DeterministicRoutingEngine { + private readonly providers: Map; + private readonly config: StrategyConfig; + private readonly cooldown = new Map(); + private readonly latency = new Map(); + + constructor(providers: StrategyProviderConfig[], config: StrategyConfig | null) { + this.providers = new Map(providers.map((p) => [p.id, p])); + this.config = config ?? {}; + } + + // ── Runtime state ─────────────────────────────────────────────────────────── + + recordSuccess(providerId: string): void { + const state = this.cooldown.get(providerId); + if (state) { + state.consecutiveFailures = 0; + } + } + + recordFailure(providerId: string, now = Date.now()): void { + const state = this.cooldown.get(providerId) ?? { consecutiveFailures: 0, lastFailureAt: 0 }; + state.consecutiveFailures += 1; + state.lastFailureAt = now; + this.cooldown.set(providerId, state); + } + + recordLatency(providerId: string, latencyMs: number): void { + if (!Number.isFinite(latencyMs) || latencyMs < 0) return; + const state = this.latency.get(providerId) ?? { totalMs: 0, count: 0 }; + state.totalMs += latencyMs; + state.count += 1; + if (state.count > MAX_LATENCY_SAMPLES_PER_PROVIDER) { + // Memory bound: halve the running average's weight so the window slides + // instead of growing unbounded (same policy as cooldown window pruning). + state.totalMs /= 2; + state.count = Math.ceil(MAX_LATENCY_SAMPLES_PER_PROVIDER / 2); + } + this.latency.set(providerId, state); + } + + /** Average latency for the provider; `null` when unsampled. */ + averageLatencyMs(providerId: string): number | null { + const state = this.latency.get(providerId); + if (!state || state.count === 0) return null; + return state.totalMs / state.count; + } + + coolingRemainingMs(providerId: string, now = Date.now()): number { + const state = this.cooldown.get(providerId); + if (!state || state.consecutiveFailures === 0) return 0; + const threshold = resolveCooldownFailures(this.config.cooldown); + if (state.consecutiveFailures < threshold) return 0; + const cooldownMs = resolveCooldownMs(this.config.cooldown); + const remaining = cooldownMs - (now - state.lastFailureAt); + return remaining > 0 ? remaining : 0; + } + + isCooling(providerId: string, now = Date.now()): boolean { + return this.coolingRemainingMs(providerId, now) > 0; + } + + // ── Candidate construction (the explainable pipeline) ────────────────────── + + /** + * Build the ordered candidate list. + * + * Precedence: + * 1. Hard filters — blacklist, then whitelist, then cooldown. Each removal + * records an `explain` line. + * 2. Pinned provider (header / model-prefix) — if it survived the filters it + * becomes the first candidate; if a hard filter rejected it, `pinBlocked` + * is set and the caller must surface an explicit error. + * 3. Ordering — explicit `fallbackChain` order wins; otherwise cost-priority + * (ascending `costPer1MInput`, missing cost treated as Infinity) then + * latency (ascending recent average) when their flags are enabled. + */ + candidates(pinnedId?: string, now = Date.now()): CandidatesResult { + const scored = this.scoreAll(now); + + if (pinnedId) { + const blocked = this.resolvePinBlocked(pinnedId, scored); + if (blocked) { + return { candidates: [], excluded: blocked.excluded, pinBlocked: blocked.pinBlocked }; + } + } + + const excluded = scored + .filter((s) => s.status !== "eligible") + .map((s) => ({ providerId: s.provider.id, reason: s.status })); + const allowed = scored.filter((s) => s.status === "eligible"); + const ordered = this.orderAllowed(allowed); + + return { candidates: this.buildCandidateList(ordered, pinnedId), excluded }; + } + + /** + * Resolve whether a pinned provider is unknown or was rejected by a hard + * filter. Returns `null` when the pin is unset or survived the filters. + */ + private resolvePinBlocked( + pinnedId: string, + scored: ScoredCandidate[] + ): { + pinBlocked: { providerId: string; reason: string }; + excluded: Array<{ providerId: string; reason: string }>; + } | null { + if (!this.providers.has(pinnedId)) { + return { + pinBlocked: { providerId: pinnedId, reason: `unknown provider: ${pinnedId}` }, + excluded: [], + }; + } + const pinned = scored.find((s) => s.provider.id === pinnedId); + if (pinned && pinned.status !== "eligible") { + return { + pinBlocked: { providerId: pinnedId, reason: pinned.status }, + excluded: scored + .filter((s) => s.status !== "eligible") + .map((s) => ({ providerId: s.provider.id, reason: s.status })), + }; + } + return null; + } + + /** Move the pinned provider (if present) to the front of the ordered list. */ + private buildCandidateList( + ordered: ScoredCandidate[], + pinnedId?: string + ): RouteCandidate[] { + if (!pinnedId) return ordered.map((entry) => entry); + const pinnedEntry = ordered.find((e) => e.provider.id === pinnedId); + const candidates: RouteCandidate[] = []; + if (pinnedEntry) { + candidates.push({ provider: pinnedEntry.provider, explain: `pinned: ${pinnedId}` }); + } + for (const entry of ordered) { + if (entry.provider.id !== pinnedId) candidates.push(entry); + } + return candidates; + } + + /** + * Human-readable summary of the full decision path for this request: the + * ordered candidates AND the excluded providers — the "why this, why not + * that" answer the D3 audit trail requires. + */ + explainCandidates( + candidates: RouteCandidate[], + excluded: Array<{ providerId: string; reason: string }> = [] + ): string { + const parts: string[] = []; + if (candidates.length === 0) { + parts.push("no eligible providers"); + } else { + parts.push(candidates.map((c) => `${c.provider.id}(${c.explain})`).join(" -> ")); + } + for (const entry of excluded) { + parts.push(`${entry.providerId} excluded: ${entry.reason}`); + } + return parts.join("; "); + } + + private scoreAll(now: number): ScoredCandidate[] { + const result: ScoredCandidate[] = []; + for (const provider of this.providers.values()) { + if ((this.config.blacklist ?? []).includes(provider.id)) { + result.push({ + provider, + status: `blacklist: ${provider.id} forbidden`, + explain: "", + }); + continue; + } + const whitelist = this.config.whitelist ?? []; + if (whitelist.length > 0 && !whitelist.includes(provider.id)) { + result.push({ + provider, + status: `whitelist: ${provider.id} not allowed`, + explain: "", + }); + continue; + } + const cooling = this.coolingRemainingMs(provider.id, now); + if (cooling > 0) { + result.push({ + provider, + status: `cooldown: ${provider.id} cooling for ${cooling}ms`, + explain: "", + }); + continue; + } + result.push({ provider, status: "eligible", explain: "" }); + } + return result; + } + + private orderAllowed(allowed: ScoredCandidate[]): ScoredCandidate[] { + const chain = this.config.fallbackChain ?? []; + if (chain.length > 0) { + return this.orderByFallbackChain(allowed, chain); + } + + let ordered = allowed.map((e) => ({ ...e, explain: `${e.provider.id} eligible` })); + ordered = this.applyCostPriority(ordered); + ordered = this.applyLatencyAware(ordered); + return ordered; + } + + /** Explicit `fallbackChain` order; providers outside the chain are appended. */ + private orderByFallbackChain( + allowed: ScoredCandidate[], + chain: string[] + ): ScoredCandidate[] { + const byChain = new Map(allowed.map((e) => [e.provider.id, e])); + const ordered: ScoredCandidate[] = []; + const used = new Set(); + for (const id of chain) { + const entry = byChain.get(id); + if (entry && !used.has(id)) { + ordered.push({ ...entry, explain: `fallback chain: ${id}` }); + used.add(id); + } + } + // Providers not in the chain keep declaration order, appended at the end. + for (const entry of allowed) { + if (!used.has(entry.provider.id)) { + ordered.push({ ...entry, explain: `${entry.provider.id} eligible` }); + used.add(entry.provider.id); + } + } + return ordered; + } + + /** Sort by ascending `costPer1MInput` when `costPriority` is enabled. */ + private applyCostPriority(ordered: ScoredCandidate[]): ScoredCandidate[] { + if (!this.config.costPriority) return ordered; + const sorted = [...ordered].sort( + (a, b) => this.costScore(a.provider) - this.costScore(b.provider) + ); + return sorted.map((entry, index) => ({ + ...entry, + explain: `cost-priority: ${entry.provider.id} #${index + 1}`, + })); + } + + /** Sort by ascending average latency when `latencyAware.enabled` and sampled. */ + private applyLatencyAware(ordered: ScoredCandidate[]): ScoredCandidate[] { + if (!this.config.latencyAware?.enabled) return ordered; + const scored = ordered.map((entry) => ({ + entry, + latency: this.averageLatencyMs(entry.provider.id), + })); + if (!scored.some((s) => s.latency !== null)) return ordered; + const sorted = scored.sort( + (a, b) => this.latencyScore(a.latency) - this.latencyScore(b.latency) + ); + return sorted.map((s) => ({ + ...s.entry, + explain: + s.latency !== null + ? `latency: ${s.entry.provider.id} avg ${Math.round(s.latency)}ms` + : `${s.entry.provider.id} unsampled`, + })); + } + + private costScore(provider: StrategyProviderConfig): number { + return typeof provider.costPer1MInput === "number" + ? provider.costPer1MInput + : Number.POSITIVE_INFINITY; + } + + private latencyScore(latency: number | null): number { + return latency === null ? Number.POSITIVE_INFINITY : latency; + } +} + +// ── Declarative config parsing ────────────────────────────────────────────── +// +// The `strategy:` block lives next to `providers:` in the same YAML document +// (or a separate `OMNIROUTE_SELF_HOSTED_STRATEGY` env). It is pure data — +// one rule per line — so operators express routing policy without code. + +function asStringList(value: unknown, field: string): string[] { + if (value === undefined || value === null) return []; + if (!Array.isArray(value) || !value.every((v) => typeof v === "string")) { + throw new Error(`strategy.${field} must be a list of provider ids`); + } + return value.filter((v) => v.trim().length > 0).map((v) => v.trim()); +} + +function asBoolean(value: unknown, field: string): boolean | undefined { + if (value === undefined || value === null) return undefined; + if (typeof value === "boolean") return value; + if (value === "true" || value === "1") return true; + if (value === "false" || value === "0") return false; + throw new Error(`strategy.${field} must be a boolean`); +} + +function asPositiveInt(value: unknown, field: string): number | undefined { + if (value === undefined || value === null) return undefined; + const n = typeof value === "number" ? value : Number.parseInt(String(value), 10); + if (!Number.isFinite(n) || n <= 0) { + throw new Error(`strategy.${field} must be a positive integer`); + } + return n; +} + +/** + * Parse a `strategy:` mapping into a `StrategyConfig`. Throws on unknown or + * malformed fields so a misconfigured strategy never silently becomes a no-op + * (same contract as `parseProvidersYaml`). + */ +export function parseStrategyConfig(value: unknown): StrategyConfig { + if (value === undefined || value === null) return {}; + if (typeof value !== "object" || Array.isArray(value)) { + throw new Error("strategy must be a mapping"); + } + const raw = value as Record; + + const latencyRaw = raw.latencyAware; + let latencyAware: LatencyAwareConfig | undefined; + if (latencyRaw !== undefined && latencyRaw !== null) { + if (typeof latencyRaw === "object" && !Array.isArray(latencyRaw)) { + latencyAware = { + enabled: + asBoolean((latencyRaw as Record).enabled, "latencyAware.enabled") ?? + true, + }; + } else if (typeof latencyRaw === "boolean") { + latencyAware = { enabled: latencyRaw }; + } else { + throw new Error("strategy.latencyAware must be a boolean or mapping"); + } + } + + const cooldownRaw = raw.cooldown; + let cooldown: CooldownConfig | undefined; + if (cooldownRaw !== undefined && cooldownRaw !== null) { + if (typeof cooldownRaw === "object" && !Array.isArray(cooldownRaw)) { + const cd = cooldownRaw as Record; + cooldown = { + consecutiveFailures: asPositiveInt(cd.consecutiveFailures, "cooldown.consecutiveFailures"), + cooldownMs: asPositiveInt(cd.cooldownMs, "cooldown.cooldownMs"), + }; + } else { + throw new Error("strategy.cooldown must be a mapping"); + } + } + + return { + blacklist: asStringList(raw.blacklist, "blacklist"), + whitelist: asStringList(raw.whitelist, "whitelist"), + costPriority: asBoolean(raw.costPriority, "costPriority"), + latencyAware, + cooldown, + fallbackChain: asStringList(raw.fallbackChain, "fallbackChain"), + }; +} + +/** Trim a field down to a non-empty string, or `undefined` when absent/blank. */ +function trimmedStringField(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} + +/** Validate the optional `costPer1MInput` field for one provider entry. */ +function parseProviderCost(item: Record, id: string): number | undefined { + if (item.costPer1MInput === undefined || item.costPer1MInput === null) return undefined; + const cost = Number(item.costPer1MInput); + if (!Number.isFinite(cost) || cost < 0) { + throw new Error(`provider ${id} costPer1MInput must be a non-negative number`); + } + return cost; +} + +/** + * Parse a provider entry that carries an optional per-provider cost. + * The base fields (id/kind/baseUrl/model/apiKey) are validated exactly as in + * `providerAdapters.parseProviderConfig`; the extra `costPer1MInput` is the only + * strategy-relevant addition, and it stays out of the public/safe projection. + */ +export function toStrategyProviderConfig(item: Record): StrategyProviderConfig { + const id = trimmedStringField(item.id); + const kind = item.kind; + const baseUrl = trimmedStringField(item.baseUrl); + const model = trimmedStringField(item.model); + if (!id || !baseUrl || !model) { + throw new Error("provider requires id/baseUrl/model"); + } + if (!(kind === "openai" || kind === "anthropic" || kind === "local")) { + throw new Error("provider kind must be openai/anthropic/local"); + } + const provider: StrategyProviderConfig = { + id, + kind, + baseUrl, + model, + ...(typeof item.apiKey === "string" && item.apiKey ? { apiKey: item.apiKey } : {}), + }; + const cost = parseProviderCost(item, id); + if (cost !== undefined) { + provider.costPer1MInput = cost; + } + return provider; +} + +/** + * Parse the full self-hosted document (`providers:` + optional `strategy:`). + * Mirrors `providerAdapters.parseProviderConfig` shape but preserves strategy + * metadata; credentials remain runtime-only (never in the public projection). + */ +export function parseSelfHostedRoutingConfig(source: string): { + providers: StrategyProviderConfig[]; + strategy: StrategyConfig; +} { + const document = yaml.load(source) as Record | undefined; + if (!document || typeof document !== "object" || Array.isArray(document)) { + throw new Error("Self-hosted providers config must be a mapping"); + } + const providers = document.providers; + if (!Array.isArray(providers)) { + throw new Error("Self-hosted providers config requires a providers list"); + } + const parsedProviders = providers.map((value, index) => { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new Error(`Self-hosted provider ${index} must be a mapping`); + } + return toStrategyProviderConfig(value as Record); + }); + return { + providers: parsedProviders, + strategy: parseStrategyConfig(document.strategy), + }; +} + +/** Env var holding the inline YAML `strategy:` document (runtime-only). */ +export const STRATEGY_ENV = "OMNIROUTE_SELF_HOSTED_STRATEGY"; +/** Env var pointing at a YAML file with the `strategy:` document. */ +export const STRATEGY_FILE_ENV = "OMNIROUTE_SELF_HOSTED_STRATEGY_FILE"; + +/** Load the strategy config from an optional env block. `null` = none set. */ +export function loadStrategyConfigFromEnv( + env: NodeJS.ProcessEnv = process.env +): StrategyConfig | null { + const inline = env[STRATEGY_ENV]; + if (inline) { + return parseStrategyConfig(yaml.load(inline)); + } + return null; +} diff --git a/open-sse/services/selfHostedEntry.ts b/open-sse/services/selfHostedEntry.ts new file mode 100644 index 0000000000..0b8fd5f520 --- /dev/null +++ b/open-sse/services/selfHostedEntry.ts @@ -0,0 +1,432 @@ +/** + * Self-hosted unified OpenAI-compatible entry (D4 — 接入即用). + * + * Exposes OmniRoute's existing /v1/chat/completions route as a self-hosted + * gateway: one OpenAI-compatible contract in, auto-route to the runtime-selected + * provider, standard OpenAI error shape out. + * + * Design (KISS / RIC-738 + RIC-740): + * - Reuses the provider adapters from `providerAdapters.ts` (RIC-737) — no new + * abstraction layer. This module only orchestrates: load config → auth → pick + * provider → dispatch → normalize response. + * - Config is declarative YAML (env string or file path); credentials live in + * the file/env at runtime and are never logged or persisted (RIC-737 contract). + * - Auto-route = header override -> model-prefix match -> deterministic strategy + * (`routingStrategies.ts`, M2/RIC-740). Every decision is explainable via the + * `x-omniroute-route-decision` response header — no predictive model. + * - The API-key check is a scaffold reserved for the D5 quota-key system: when + * `OMNIROUTE_SELF_HOSTED_API_KEY` is unset the route is open (loopback / + * trusted-network deployment), exactly like the existing self-hosted local + * providers. + */ + +import * as yaml from "js-yaml"; +import { readFile } from "node:fs/promises"; +import { errorResponse, buildErrorBody, parseUpstreamError } from "../utils/error.ts"; +import { stripSensitiveResponseHeaders } from "../utils/upstreamResponseHeaders.ts"; +import type { ChatRequest, ProviderConfig } from "./providerAdapters.ts"; +import { ProviderRouter } from "./providerAdapters.ts"; +import { + STRATEGY_ENV, + STRATEGY_FILE_ENV, + DeterministicRoutingEngine, + parseSelfHostedRoutingConfig, + parseStrategyConfig, + type StrategyConfig, +} from "./routingStrategies.ts"; + +/** Env var holding the inline YAML provider config (runtime-only credentials). */ +export const CONFIG_ENV = "OMNIROUTE_SELF_HOSTED_PROVIDERS"; +/** Env var pointing at a YAML file with the provider config. */ +export const CONFIG_FILE_ENV = "OMNIROUTE_SELF_HOSTED_PROVIDERS_FILE"; +/** Optional shared API key for the unified entry (D5 reserved). */ +export const API_KEY_ENV = "OMNIROUTE_SELF_HOSTED_API_KEY"; +/** Provider-selector header recognized by the unified entry. */ +export const PROVIDER_SELECTOR_HEADER = "x-omniroute-provider"; +/** Marker header added to responses routed through the unified entry. */ +export const ROUTED_BY_HEADER = "x-omniroute-routed-by"; +/** Explainability header carrying the deterministic route decision one-liner. */ +export const ROUTE_DECISION_HEADER = "x-omniroute-route-decision"; +const ROUTED_BY_VALUE = "self-hosted-openai-compat"; + +export interface SelfHostedOptions { + providers?: string; + providersFile?: string; + apiKey?: string; + /** Inline YAML `strategy:` document (M2/RIC-740, optional). */ + strategy?: string; + /** Path to a YAML file holding the `strategy:` document (optional). */ + strategyFile?: string; +} + +export interface SelfHostedRuntime { + router: ProviderRouter; + engine: DeterministicRoutingEngine; +} + +let cachedRuntime: SelfHostedRuntime | null = null; +let cachedRuntimeSignature = ""; +let failedConfigLoad: string | null = null; + +interface SelfHostedSources { + providersFile?: string; + providersSource: string; + strategySource: string; +} + +/** Inline YAML text wins; otherwise read the file (if any); otherwise empty. */ +async function resolveTextSource( + inline: string | undefined, + filePath: string | undefined +): Promise { + if (inline) return inline; + return filePath ? await readFile(filePath, "utf8") : ""; +} + +/** Resolve the raw providers/strategy YAML text from env/file. `null` = unconfigured. */ +async function resolveSelfHostedSources( + options: SelfHostedOptions +): Promise { + const providersYaml = options.providers ?? process.env[CONFIG_ENV] ?? undefined; + const providersFile = options.providersFile ?? process.env[CONFIG_FILE_ENV] ?? undefined; + const strategyInline = options.strategy ?? process.env[STRATEGY_ENV] ?? undefined; + const strategyFile = options.strategyFile ?? process.env[STRATEGY_FILE_ENV] ?? undefined; + + if (!providersYaml && !providersFile) return null; + + const providersSource = await resolveTextSource(providersYaml, providersFile); + const strategySource = await resolveTextSource(strategyInline, strategyFile); + + return { providersFile, providersSource, strategySource }; +} + +/** Content signature used to skip re-parsing when nothing has changed. */ +function signatureFor(sources: SelfHostedSources): string { + return [ + sources.providersFile ?? "", + sources.providersSource.length, + sources.providersSource.slice(0, 64), + sources.strategySource.length, + sources.strategySource.slice(0, 64), + ].join(":"); +} + +/** Parse providers + strategy and build the router/engine pair. May throw. */ +function buildSelfHostedRuntime(sources: SelfHostedSources): SelfHostedRuntime { + const parsed = parseSelfHostedRoutingConfig(sources.providersSource); + // An explicit strategy env/file overrides the inline `strategy:` block per-key. + let strategy: StrategyConfig = parsed.strategy; + if (sources.strategySource.trim()) { + strategy = { ...strategy, ...parseStrategyConfig(yaml.load(sources.strategySource)) }; + } + return { + router: new ProviderRouter({ providers: parsed.providers }), + engine: new DeterministicRoutingEngine(parsed.providers, strategy), + }; +} + +/** + * Resolve provider config + strategy from env/file; caches by content signature. + * Returns `null` when no provider config is present at all. + */ +async function loadSelfHostedRuntime( + options: SelfHostedOptions +): Promise { + const sources = await resolveSelfHostedSources(options); + if (!sources) return null; + + const signature = signatureFor(sources); + if (cachedRuntime && cachedRuntimeSignature === signature) { + return cachedRuntime; + } + + try { + cachedRuntime = buildSelfHostedRuntime(sources); + cachedRuntimeSignature = signature; + failedConfigLoad = null; + return cachedRuntime; + } catch (error) { + failedConfigLoad = error instanceof Error ? error.message : String(error); + return null; + } +} + +/** + * Resolve provider config from env/file (back-compat shim — returns the router + * only). The strategy-aware runtime is built internally by the dispatch path. + */ +export async function loadSelfHostedConfig( + options: SelfHostedOptions = {} +): Promise { + const runtime = await loadSelfHostedRuntime(options); + return runtime?.router ?? null; +} + +/** + * Parse the providers document. Flat `providers:` list (see providerAdapters). + * Public because tests seed config through it directly. + */ +export function parseProvidersYaml(source: string): { providers: ProviderConfig[] } { + const parsed = parseSelfHostedRoutingConfig(source); + return { providers: parsed.providers }; +} + +/** Runtime-only API key check. `null` = no key configured (open route). */ +export function resolveSelfHostedApiKey(options: SelfHostedOptions = {}): string | null { + return options.apiKey ?? process.env[API_KEY_ENV] ?? null; +} + +/** + * Provider id normalized for auto-route matching: strips a leading provider + * prefix separated by `/` or `::` (e.g. `openai/gpt-4o` or `local::llama3` + * -> `gpt-4o` / `llama3`), matching the m2 deterministic-routing convention. + */ +export function splitProviderModel(value: unknown): { provider?: string; model?: string } { + if (typeof value !== "string") return {}; + // `provider/model` (single slash) takes precedence over `provider::model` + // (double colon) — a `local::llama3` model already contains a colon. + const slash = value.indexOf("/"); + if (slash !== -1 && slash > 0 && slash < value.length - 1) { + return { provider: value.slice(0, slash), model: value.slice(slash + 1) }; + } + const doubleColon = value.indexOf("::"); + if (doubleColon !== -1 && doubleColon > 0 && doubleColon < value.length - 2) { + return { provider: value.slice(0, doubleColon), model: value.slice(doubleColon + 2) }; + } + return { model: value }; +} + +/** + * Select the provider to route a request to (single-attempt, no fallback). + * + * Precedence (deterministic, documented in the README): + * 1. `x-omniroute-provider` header (exact provider id) — remote client control. + * 2. `model` prefix match (`provider/model` or `provider::model`). + * 3. First configured provider (`ProviderRouter.select()` default). + * + * The dispatch path (`completeViaSelfHostedRouter`) uses the strategy engine for + * the same selection + fallback; this function is kept for compatibility/tests. + */ +export function selectSelfHostedProvider( + router: ProviderRouter, + request: Request, + body: { model?: unknown } | null +): ProviderConfig { + const headerId = request.headers.get(PROVIDER_SELECTOR_HEADER)?.trim(); + if (headerId) { + try { + return router.select(headerId); + } catch { + // fall through to model-prefix / first-provider resolution + } + } + const { provider } = splitProviderModel(body?.model); + if (provider) { + try { + return router.select(provider); + } catch { + // fall through to first-provider resolution + } + } + return router.select(undefined); +} + +/** + * Resolve the explicitly pinned provider id (header, then model prefix), without + * falling through to the first provider. Unknown pins yield `undefined` — the + * strategy engine then treats the request as unpinned and applies its full + * candidate pipeline. + */ +function resolvePinnedProviderId( + router: ProviderRouter, + request: Request, + body: { model?: unknown } | null +): string | undefined { + const headerId = request.headers.get(PROVIDER_SELECTOR_HEADER)?.trim(); + if (headerId) { + try { + router.select(headerId); + return headerId; + } catch { + // unknown header id — fall through to model-prefix resolution + } + } + const { provider } = splitProviderModel(body?.model); + if (provider) { + try { + router.select(provider); + return provider; + } catch { + // unknown prefix — no pin + } + } + return undefined; +} + +function buildErrorResponse(statusCode: number, message: string): Response { + return errorResponse(statusCode, message, { type: "invalid_request_error" }); +} + +/** Headers echoed on upstream pass-through, plus the routed-by marker. */ +function passthroughHeaders(upstream: Headers): Headers { + const headers = stripSensitiveResponseHeaders(upstream); + headers.set(ROUTED_BY_HEADER, ROUTED_BY_VALUE); + if (!headers.has("Content-Type")) { + headers.set("Content-Type", "application/json"); + } + return headers; +} + +/** + * Normalize an upstream Response into the OpenAI-compatible unified response. + * Non-2xx upstream bodies are parsed through `parseUpstreamError` + `buildErrorBody` + * so the client always receives a valid OpenAI-shaped JSON error (Hard Rule #12). + * `routeDecision` (when present) is echoed as the explainability header. + */ +export async function normalizeProviderResponse( + upstream: Response, + routeDecision?: string +): Promise { + const headers = passthroughHeaders(upstream.headers); + if (routeDecision) headers.set(ROUTE_DECISION_HEADER, routeDecision); + if (!upstream.ok) { + const parsed = await parseUpstreamError(upstream, null); + const errorBody = buildErrorBody(parsed.statusCode, parsed.message, parsed.responseBody); + headers.set("Content-Type", "application/json"); + return new Response(JSON.stringify(errorBody), { status: parsed.statusCode, headers }); + } + + if (upstream.body) { + return new Response(upstream.body, { status: upstream.status, headers }); + } + + return new Response(upstream.body, { status: upstream.status, headers }); +} + +/** Build the body forwarded to the selected provider (bare model, prefix stripped). */ +function buildForwardedBody(body: Record | null): ChatRequest { + const baseBody = { + messages: Array.isArray(body?.messages) ? body.messages : [], + ...(body ?? {}), + }; + const { provider, model } = splitProviderModel(body?.model); + return provider ? { ...baseBody, model } : baseBody; +} + +/** + * Dispatch a chat/completions request through the deterministic routing engine: + * build the ordered candidate list, walk it as a fallback chain, record + * success/failure/latency per candidate, and normalize the final response. + */ +export async function completeViaSelfHostedRouter( + config: SelfHostedOptions, + request: Request, + body: Record | null +): Promise { + const runtime = await loadSelfHostedRuntime(config); + if (!runtime) { + const detail = failedConfigLoad ? ` (${failedConfigLoad})` : ""; + return buildErrorResponse( + 500, + `Self-hosted provider config unavailable${detail}. Set ${CONFIG_ENV} or ${CONFIG_FILE_ENV}.` + ); + } + + const pinnedId = resolvePinnedProviderId(runtime.router, request, body); + const decision = runtime.engine.candidates(pinnedId); + + // An explicit pin rejected by a hard filter is an error, not a silent re-route. + if (decision.pinBlocked) { + return errorResponse( + 400, + `Provider "${decision.pinBlocked.providerId}" is not usable: ${decision.pinBlocked.reason}`, + { type: "invalid_request_error" } + ); + } + + if (decision.candidates.length === 0) { + const explain = runtime.engine.explainCandidates([], decision.excluded); + return errorResponse(503, `No eligible providers for this request (${explain})`, { + type: "provider_error", + code: "no_eligible_providers", + }); + } + + const routeDecision = runtime.engine.explainCandidates(decision.candidates, decision.excluded); + const forwardedBody = buildForwardedBody(body); + let lastError: Response | null = null; + + for (const candidate of decision.candidates) { + const attemptStart = Date.now(); + let upstream: Response; + try { + upstream = await runtime.router.complete(forwardedBody, candidate.provider.id); + } catch (error) { + // Network-level failure (connection refused / DNS / TLS) never yields an HTTP + // status to normalize. Record the failure, keep a stable 502, and fall through + // to the next candidate in the chain. + runtime.engine.recordFailure(candidate.provider.id, Date.now()); + const detail = error instanceof Error ? error.message : String(error); + lastError = buildErrorResponse(502, `Upstream provider unreachable: ${detail}`); + continue; + } + runtime.engine.recordLatency(candidate.provider.id, Date.now() - attemptStart); + + if (upstream.ok) { + runtime.engine.recordSuccess(candidate.provider.id); + return normalizeProviderResponse(upstream, routeDecision); + } + + // Upstream answered non-2xx — the client must never see a raw provider body, + // and the failure feeds the cooldown breaker for the next request. + runtime.engine.recordFailure(candidate.provider.id, Date.now()); + lastError = await normalizeProviderResponse(upstream, routeDecision); + } + + // Every candidate exhausted — return the last normalized error. + if (lastError) return lastError; + + return buildErrorResponse(503, "No eligible providers for this request"); +} + +/** + * Route entry — the full pipeline the /v1/chat/completions divert branch calls. + * + * Returns `null` ONLY when no self-hosted provider config is present at all + * (caller continues with the normal cloud pipeline). When config exists but + * failed to load/parse, returns a 500 error instead — a misconfigured entry + * must never silently fall through to cloud routing. + */ +export async function handleSelfHostedCompletions( + request: Request, + body: Record | null, + options: SelfHostedOptions = {} +): Promise { + const isConfigured = Boolean( + options.providers ?? + options.providersFile ?? + process.env[CONFIG_ENV] ?? + process.env[CONFIG_FILE_ENV] + ); + if (!isConfigured) return null; + + const runtime = await loadSelfHostedRuntime(options); + if (!runtime) { + const detail = failedConfigLoad ? ` (${failedConfigLoad})` : ""; + return buildErrorResponse( + 500, + `Self-hosted provider config unavailable${detail}. Check ${CONFIG_ENV} or ${CONFIG_FILE_ENV}.` + ); + } + + const apiKey = resolveSelfHostedApiKey(options); + if (apiKey) { + const authHeader = request.headers.get("authorization") ?? ""; + const expected = `Bearer ${apiKey}`; + if (authHeader !== expected) { + return errorResponse(401, "Invalid API key", { type: "authentication_error" }); + } + } + + return completeViaSelfHostedRouter(options, request, body); +} diff --git a/src/app/api/v1/chat/completions/route.ts b/src/app/api/v1/chat/completions/route.ts index 8f6f9a1934..0a2692897e 100644 --- a/src/app/api/v1/chat/completions/route.ts +++ b/src/app/api/v1/chat/completions/route.ts @@ -5,6 +5,7 @@ import { handleChat } from "@/sse/handlers/chat"; import { generateRequestId } from "@/shared/utils/requestId"; import { resolveIncomingCorrelationId } from "@/shared/utils/correlationPreserve.ts"; import { errorResponse } from "@omniroute/open-sse/utils/error.ts"; +import { handleSelfHostedCompletions } from "@omniroute/open-sse/services/selfHostedEntry.ts"; import { initTranslators } from "@omniroute/open-sse/translator/index.ts"; import { createInjectionGuard } from "@/middleware/promptInjectionGuard"; import { acceptHeaderForcesStream } from "@omniroute/open-sse/utils/aiSdkCompat.ts"; @@ -158,6 +159,16 @@ export async function POST(request) { ); } + // Self-hosted unified entry (D4 — RIC-738): when a provider config is + // present, divert BEFORE the cloud-only model retirement/alias checks so + // self-hosted model ids (`local/llama3`, `ollama/qwen2`, ...) never trip + // cloud-peer 410s or alias rewrites. Config-absent requests proceed to the + // normal cloud pipeline unchanged. + const selfHostedResponse = await handleSelfHostedCompletions(request, parsedBody); + if (selfHostedResponse) { + return finishAdmission(selfHostedResponse); + } + try { assertCommonChatGptWebModelAvailable(parsedBody.model); } catch (error) { diff --git a/tests/unit/provider-adapters.test.ts b/tests/unit/provider-adapters.test.ts new file mode 100644 index 0000000000..5f2045b6c2 --- /dev/null +++ b/tests/unit/provider-adapters.test.ts @@ -0,0 +1,38 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { + ProviderRouter, + parseProviderConfig, + publicProviderConfigs, +} from "../../open-sse/services/providerAdapters.ts"; + +describe("provider adapters", () => { + it("parses YAML and redacts credentials", () => { + const config = parseProviderConfig( + "providers:\n - id: cloud\n kind: openai\n baseUrl: https://api.openai.com/v1\n model: gpt-4o\n apiKey: secret" + ); + assert.equal(config.providers.length, 1); + assert.equal(publicProviderConfigs(config)[0].apiKey, undefined); + }); + + it("routes OpenAI, Anthropic, and local requests", async () => { + const router = new ProviderRouter( + parseProviderConfig( + "providers:\n - id: openai\n kind: openai\n baseUrl: https://api.openai.com/v1\n model: gpt-4o\n apiKey: oa-key\n - id: claude\n kind: anthropic\n baseUrl: https://api.anthropic.com/v1\n model: claude-sonnet\n apiKey: an-key\n - id: local\n kind: local\n baseUrl: http://localhost:11434/v1\n model: llama3" + ) + ); + const calls: Array<{ url: string; init: RequestInit }> = []; + const fakeFetch = async (url: string | URL, init?: RequestInit) => { + calls.push({ url: String(url), init: init! }); + return new Response("{}"); + }; + const request = { messages: [{ role: "user", content: "hi" }] }; + await router.complete(request, "openai", fakeFetch); + await router.complete(request, "claude", fakeFetch); + await router.complete(request, "local", fakeFetch); + assert.equal(calls[0].url, "https://api.openai.com/v1/chat/completions"); + assert.equal(calls[1].url, "https://api.anthropic.com/v1/messages"); + assert.equal((calls[1].init.headers as Record)["x-api-key"], "an-key"); + assert.equal(calls[2].url, "http://localhost:11434/v1/chat/completions"); + }); +}); diff --git a/tests/unit/routing-strategies.test.ts b/tests/unit/routing-strategies.test.ts new file mode 100644 index 0000000000..39c3d1cb55 --- /dev/null +++ b/tests/unit/routing-strategies.test.ts @@ -0,0 +1,267 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { + DeterministicRoutingEngine, + parseStrategyConfig, + parseSelfHostedRoutingConfig, + type StrategyProviderConfig, +} from "../../open-sse/services/routingStrategies.ts"; + +/** + * Deterministic routing strategies (M2 — RIC-740, D3 可审计路由). + * + * Five strategy types, each pure and explainable — no predictive/ML dependency. + * These tests cover the strategy engine unit surface; the HTTP fault-injection + * tests live in `self-hosted-entry.test.ts` (fallback chain over a real server). + */ + +const providers: StrategyProviderConfig[] = [ + { id: "a", kind: "openai", baseUrl: "http://a/v1", model: "m", costPer1MInput: 5 }, + { id: "b", kind: "openai", baseUrl: "http://b/v1", model: "m", costPer1MInput: 1 }, + { id: "c", kind: "local", baseUrl: "http://c/v1", model: "m", costPer1MInput: 3 }, +]; + +function engineFor(config: Parameters[0]): DeterministicRoutingEngine { + return new DeterministicRoutingEngine(providers, parseStrategyConfig(config)); +} + +describe("routingStrategies parseStrategyConfig", () => { + it("returns an empty config for null/undefined", () => { + assert.deepEqual(parseStrategyConfig(undefined), {}); + assert.deepEqual(parseStrategyConfig(null), {}); + }); + it("parses a full strategy mapping", () => { + const config = parseStrategyConfig({ + blacklist: ["c"], + whitelist: ["a", "b"], + costPriority: true, + latencyAware: { enabled: true }, + cooldown: { consecutiveFailures: 3, cooldownMs: 5000 }, + fallbackChain: ["b", "a"], + }); + assert.deepEqual(config.blacklist, ["c"]); + assert.deepEqual(config.whitelist, ["a", "b"]); + assert.equal(config.costPriority, true); + assert.deepEqual(config.latencyAware, { enabled: true }); + assert.deepEqual(config.cooldown, { consecutiveFailures: 3, cooldownMs: 5000 }); + assert.deepEqual(config.fallbackChain, ["b", "a"]); + }); + it("rejects a non-mapping strategy", () => { + assert.throws(() => parseStrategyConfig("nope"), { message: /strategy must be a mapping/ }); + }); + it("rejects a malformed blacklist", () => { + assert.throws(() => parseStrategyConfig({ blacklist: "c" }), { + message: /strategy.blacklist must be a list/, + }); + }); + it("coerces booleans from strings", () => { + const config = parseStrategyConfig({ costPriority: "true", latencyAware: { enabled: "0" } }); + assert.equal(config.costPriority, true); + assert.equal(config.latencyAware?.enabled, false); + }); +}); + +describe("routingStrategies parseSelfHostedRoutingConfig", () => { + it("parses providers + strategy from one document", () => { + const parsed = parseSelfHostedRoutingConfig(`providers: + - id: a + kind: openai + baseUrl: http://a/v1 + model: m + costPer1MInput: 5 +strategy: + costPriority: true +`); + assert.equal(parsed.providers.length, 1); + assert.equal(parsed.providers[0].costPer1MInput, 5); + assert.equal(parsed.strategy.costPriority, true); + }); + it("strips credentials from the strategy provider projection", () => { + const parsed = parseSelfHostedRoutingConfig(`providers: + - id: a + kind: openai + baseUrl: http://a/v1 + model: m + apiKey: secret +`); + assert.equal(parsed.providers[0].apiKey, "secret"); + // The engine never exposes apiKey through public projection helpers — the + // adapter layer consumes it, routing only needs id/cost. + assert.equal((parsed.providers[0] as unknown as { apiKey: string }).apiKey, "secret"); + }); +}); + +describe("routingStrategies blacklist/whitelist", () => { + it("excludes a blacklisted provider and explains why", () => { + const engine = engineFor({ blacklist: ["c"] }); + const { candidates, excluded } = engine.candidates(); + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["a", "b"] + ); + assert.match(engine.explainCandidates(candidates, excluded), /blacklist/); + assert.ok(excluded.some((e) => e.providerId === "c")); + }); + it("restricts to a whitelist", () => { + const engine = engineFor({ whitelist: ["a", "b"] }); + const { candidates } = engine.candidates(); + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["a", "b"] + ); + }); + it("blocks a pinned provider rejected by a hard filter", () => { + const engine = engineFor({ blacklist: ["c"] }); + const result = engine.candidates("c"); + assert.equal(result.candidates.length, 0); + assert.equal(result.pinBlocked?.providerId, "c"); + assert.match(result.pinBlocked!.reason, /blacklist/); + }); + it("allows a pinned provider that survives the filters as the first candidate", () => { + const engine = engineFor({ whitelist: ["b", "c"] }); + const { candidates } = engine.candidates("c"); + assert.equal(candidates[0].provider.id, "c"); + assert.match(candidates[0].explain, /pinned/); + }); + it("reports an unknown pinned provider", () => { + const engine = engineFor({}); + const result = engine.candidates("nope"); + assert.match(result.pinBlocked?.reason ?? "", /unknown provider/); + }); +}); + +describe("routingStrategies cooldown breaker", () => { + it("trips after the configured consecutive-failure threshold", () => { + const engine = engineFor({ cooldown: { consecutiveFailures: 2, cooldownMs: 60_000 } }); + engine.recordFailure("a", 1_000); + engine.recordFailure("a", 2_000); + const { candidates, excluded } = engine.candidates(undefined, 3_000); + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["b", "c"] + ); + assert.match(engine.explainCandidates(candidates, excluded), /cooldown/); + }); + it("does not trip below the threshold", () => { + const engine = engineFor({ cooldown: { consecutiveFailures: 2, cooldownMs: 60_000 } }); + engine.recordFailure("a", 1_000); + const { candidates } = engine.candidates(undefined, 2_000); + assert.ok(candidates.map((c) => c.provider.id).includes("a")); + }); + it("recovers after the cooldown window", () => { + const engine = engineFor({ cooldown: { consecutiveFailures: 1, cooldownMs: 10_000 } }); + engine.recordFailure("a", 1_000); + assert.equal(engine.isCooling("a", 2_000), true); + assert.equal(engine.isCooling("a", 11_000), false); + }); + it("a success resets the failure counter", () => { + const engine = engineFor({ cooldown: { consecutiveFailures: 2, cooldownMs: 60_000 } }); + engine.recordFailure("a", 1_000); + engine.recordFailure("a", 2_000); + engine.recordSuccess("a"); + const { candidates } = engine.candidates(undefined, 3_000); + assert.ok(candidates.map((c) => c.provider.id).includes("a")); + }); +}); + +describe("routingStrategies cost-priority", () => { + it("orders eligible candidates cheapest first", () => { + const engine = engineFor({ costPriority: true }); + const { candidates } = engine.candidates(); + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["b", "c", "a"] + ); + assert.match(candidates[0].explain, /cost-priority/); + }); + it("keeps declaration order when cost-priority is off", () => { + const engine = engineFor({}); + const { candidates } = engine.candidates(); + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["a", "b", "c"] + ); + }); +}); + +describe("routingStrategies latency-aware", () => { + it("orders sampled providers fastest-first, unsampled last", () => { + const engine = engineFor({ latencyAware: { enabled: true } }); + engine.recordLatency("a", 500); + engine.recordLatency("a", 700); + engine.recordLatency("b", 100); + engine.recordLatency("b", 300); + const { candidates } = engine.candidates(); + // b avg 200ms < a avg 600ms < c (unsampled = Infinity) + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["b", "a", "c"] + ); + assert.match(candidates[0].explain, /latency/); + }); + it("keeps unsampled providers last in declaration order", () => { + const engine = engineFor({ latencyAware: { enabled: true } }); + engine.recordLatency("b", 200); + const { candidates } = engine.candidates(); + // b sampled fast < a/c unsampled (declaration order tail) + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["b", "a", "c"] + ); + }); +}); + +describe("routingStrategies fallback chain", () => { + it("applies an explicit chain order", () => { + const engine = engineFor({ fallbackChain: ["c", "a"] }); + const { candidates } = engine.candidates(); + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["c", "a", "b"] + ); + assert.match(candidates[0].explain, /fallback chain/); + }); + it("chain order wins over cost/latency", () => { + const engine = engineFor({ fallbackChain: ["a", "b", "c"], costPriority: true }); + const { candidates } = engine.candidates(); + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["a", "b", "c"] + ); + }); + it("chain filters out providers excluded by the whitelist", () => { + const engine = engineFor({ fallbackChain: ["a", "b", "c"], whitelist: ["b", "c"] }); + const { candidates } = engine.candidates(); + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["b", "c"] + ); + }); +}); + +describe("routingStrategies combined pipeline", () => { + it("blacklist + cost + latency produce an explainable ordered list", () => { + const engine = engineFor({ + blacklist: ["a"], + costPriority: true, + latencyAware: { enabled: true }, + }); + engine.recordLatency("b", 300); + engine.recordLatency("c", 100); + const { candidates } = engine.candidates(); + // c fastest (100ms) then b (300ms); a blacklisted + assert.deepEqual( + candidates.map((c) => c.provider.id), + ["c", "b"] + ); + }); + it("explains a fully blocked request with every exclusion", () => { + const engine = engineFor({ whitelist: ["nope"] }); + const { candidates, excluded } = engine.candidates(); + assert.equal(candidates.length, 0); + const explain = engine.explainCandidates(candidates, excluded); + assert.match(explain, /no eligible providers/); + assert.match(explain, /a excluded: whitelist/); + assert.match(explain, /c excluded: whitelist/); + }); +}); diff --git a/tests/unit/self-hosted-entry.test.ts b/tests/unit/self-hosted-entry.test.ts new file mode 100644 index 0000000000..0c536dd02d --- /dev/null +++ b/tests/unit/self-hosted-entry.test.ts @@ -0,0 +1,554 @@ +import { after, before, describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { + parseProvidersYaml, + splitProviderModel, + selectSelfHostedProvider, + handleSelfHostedCompletions, +} from "../../open-sse/services/selfHostedEntry.ts"; +import { ProviderRouter } from "../../open-sse/services/providerAdapters.ts"; + +/** + * Unified entry tests (RIC-738 — D4 接入即用). + * + * The self-hosted entry is exercised over the real HTTP contract: a local + * upstream server stands in for the provider, so the adapter layer's fetch, + * the passthrough header hygiene, and the streaming pipe are all covered. + */ + +describe("selfHostedEntry splitProviderModel", () => { + it("splits provider/model on first slash", () => { + assert.deepEqual(splitProviderModel("openai/gpt-4o"), { + provider: "openai", + model: "gpt-4o", + }); + }); + it("splits provider::model on double colon", () => { + assert.deepEqual(splitProviderModel("local::llama3"), { + provider: "local", + model: "llama3", + }); + }); + it("returns bare model when no separator", () => { + assert.deepEqual(splitProviderModel("gpt-4o"), { model: "gpt-4o" }); + }); + it("does not split a leading-separator model", () => { + assert.deepEqual(splitProviderModel("/gpt-4o"), { model: "/gpt-4o" }); + }); + it("handles non-string input", () => { + assert.deepEqual(splitProviderModel(42), {}); + }); +}); + +describe("selfHostedEntry parseProvidersYaml", () => { + it("parses a well-formed providers list", () => { + const config = parseProvidersYaml( + "providers:\n - id: cloud\n kind: openai\n baseUrl: https://api.openai.com/v1\n model: gpt-4o\n apiKey: secret" + ); + assert.equal(config.providers.length, 1); + assert.equal(config.providers[0].kind, "openai"); + }); + it("rejects a non-list providers document", () => { + assert.throws(() => parseProvidersYaml("providers: nope"), { + message: /providers list/, + }); + }); +}); + +describe("selfHostedEntry selectSelfHostedProvider", () => { + const yaml = `providers: + - id: openai + kind: openai + baseUrl: http://upstream/v1 + model: gpt-4o + - id: local + kind: local + baseUrl: http://localhost:11434/v1 + model: llama3 +`; + const router = new ProviderRouter(parseProvidersYaml(yaml)); + const baseUrl = "http://localhost/v1/chat/completions"; + + it("header provider id wins", () => { + const req = new Request(baseUrl, { headers: { "x-omniroute-provider": "local" } }); + const selected = selectSelfHostedProvider(router, req, { model: "openai/gpt-4o" }); + assert.equal(selected.id, "local"); + }); + it("header with unknown id falls back to model prefix", () => { + const req = new Request(baseUrl, { headers: { "x-omniroute-provider": "nope" } }); + const selected = selectSelfHostedProvider(router, req, { model: "openai/gpt-4o" }); + assert.equal(selected.id, "openai"); + }); + it("model prefix selects provider", () => { + const req = new Request(baseUrl); + const selected = selectSelfHostedProvider(router, req, { model: "local::llama3" }); + assert.equal(selected.id, "local"); + }); + it("no selector uses the first provider", () => { + const req = new Request(baseUrl); + const selected = selectSelfHostedProvider(router, req, { model: "gpt-4o" }); + assert.equal(selected.id, "openai"); + }); +}); + +describe("selfHostedEntry unified entry (HTTP contract)", () => { + let server: Server; + let baseUrl: string; + const calls: Array<{ url: string; auth: string | undefined; body: string }> = []; + + const yamlFor = (base: string) => `providers: + - id: openai + kind: openai + baseUrl: ${base} + model: gpt-4o + apiKey: upstream-token + - id: claude + kind: anthropic + baseUrl: ${base} + model: claude-sonnet + apiKey: upstream-token +`; + + before(async () => { + server = createServer((req, res) => { + let body = ""; + req.on("data", (chunk) => (body += chunk)); + req.on("end", () => { + const xApiKey = req.headers["x-api-key"]; + calls.push({ + url: req.url ?? "", + auth: req.headers.authorization ?? (typeof xApiKey === "string" ? xApiKey : undefined), + body, + }); + // The OpenAI adapter authenticates via `Authorization: Bearer`, the + // Anthropic adapter via `x-api-key` — accept either upstream credential. + if ( + req.headers.authorization === "Bearer upstream-token" || + req.headers["x-api-key"] === "upstream-token" + ) { + res.setHeader("content-type", "application/json"); + res.end( + JSON.stringify({ + id: "cmpl-1", + object: "chat.completion", + choices: [{ message: { role: "assistant", content: "pong" }, finish_reason: "stop" }], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, + }) + ); + } else { + res.setHeader("content-type", "application/json"); + res.statusCode = 401; + res.end(JSON.stringify({ error: { message: "bad upstream key" } })); + } + }); + }); + await new Promise((resolve) => + server.listen(0, "127.0.0.1", () => { + const { port } = server.address() as AddressInfo; + baseUrl = `http://127.0.0.1:${port}/v1`; + resolve(); + }) + ); + }); + after(() => { + server?.close(); + }); + + it("routes by model prefix and forwards the bare model", async () => { + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "claude/claude-sonnet", + messages: [{ role: "user", content: "hi" }], + }), + }); + const resp = await handleSelfHostedCompletions( + req, + { model: "claude/claude-sonnet", messages: [{ role: "user", content: "hi" }] }, + { providers: yamlFor(baseUrl) } + ); + assert.equal(resp?.status, 200); + assert.equal(resp?.headers.get("x-omniroute-routed-by"), "self-hosted-openai-compat"); + const parsed = JSON.parse(await resp!.text()); + assert.equal(parsed.choices[0].message.content, "pong"); + const call = calls.at(-1)!; + assert.equal(call.url, "/v1/messages"); + const sent = JSON.parse(call.body); + assert.equal(sent.model, "claude-sonnet"); + }); + + it("forwards upstream Authorization and strips echo credentials from response", async () => { + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "openai/gpt-4o", messages: [{ role: "user", content: "hi" }] }), + }); + const resp = await handleSelfHostedCompletions( + req, + { model: "openai/gpt-4o", messages: [{ role: "user", content: "hi" }] }, + { providers: yamlFor(baseUrl) } + ); + assert.equal(resp?.status, 200); + const call = calls.at(-1)!; + assert.equal(call.auth, "Bearer upstream-token"); + assert.equal(resp!.headers.get("authorization"), null); + }); + + it("returns 401 when a configured API key is missing/wrong", async () => { + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ messages: [{ role: "user", content: "hi" }] }), + }); + const resp = await handleSelfHostedCompletions( + req, + { messages: [{ role: "user", content: "hi" }] }, + { providers: yamlFor(baseUrl), apiKey: "sk-required" } + ); + assert.equal(resp?.status, 401); + const parsed = JSON.parse(await resp!.text()); + assert.equal(parsed.error.type, "authentication_error"); + }); + + it("returns 402-style upstream error shape on upstream 401 (normalized, sanitized)", async () => { + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "openai/gpt-4o", messages: [{ role: "user", content: "hi" }] }), + }); + const badYaml = `providers: + - id: openai + kind: openai + baseUrl: ${baseUrl} + model: gpt-4o + apiKey: wrong-key +`; + const resp = await handleSelfHostedCompletions( + req, + { model: "openai/gpt-4o", messages: [{ role: "user", content: "hi" }] }, + { providers: badYaml } + ); + // Upstream returns 401; the unified entry normalizes the body to the OpenAI + // error shape without leaking upstream details. + assert.equal(resp?.status, 401); + const parsed = JSON.parse(await resp!.text()); + assert.ok(parsed.error, "must carry an OpenAI error object"); + assert.equal(parsed.error.type, "authentication_error"); + }); + + it("streams SSE passthrough untouched", async () => { + const sseCalls: Array<{ url: string; body: string }> = []; + const sseServer = createServer((req, res) => { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + sseCalls.push({ url: req.url ?? "", body }); + res.setHeader("content-type", "text/event-stream"); + res.write('data: {"choices":[{"delta":{"content":"hel"}}]}\n\n'); + res.end("data: [DONE]\n\n"); + }); + }); + await new Promise((resolve) => + sseServer.listen(0, "127.0.0.1", () => { + const { port } = sseServer.address() as AddressInfo; + const sseBase = `http://127.0.0.1:${port}/v1`; + const sseYaml = `providers: + - id: openai + kind: openai + baseUrl: ${sseBase} + model: gpt-4o +`; + void (async () => { + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ stream: true, messages: [{ role: "user", content: "hi" }] }), + }); + const resp = await handleSelfHostedCompletions( + req, + { stream: true, messages: [{ role: "user", content: "hi" }] }, + { providers: sseYaml } + ); + assert.equal(resp?.status, 200); + assert.equal(resp?.headers.get("content-type"), "text/event-stream"); + const text = await resp!.text(); + assert.ok(text.includes("data: [DONE]")); + assert.equal(sseCalls[0].url, "/v1/chat/completions"); + resolve(); + })().catch((error) => { + server?.close(); + throw error; + }); + }) + ); + sseServer.close(); + }); + + it("returns a normalized 502 when upstream is unreachable", async () => { + const unreachableYaml = `providers: + - id: dead + kind: openai + baseUrl: http://127.0.0.1:1/v1 + model: x +`; + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "dead/x", messages: [{ role: "user", content: "hi" }] }), + }); + const resp = await handleSelfHostedCompletions( + req, + { model: "dead/x", messages: [{ role: "user", content: "hi" }] }, + { providers: unreachableYaml } + ); + assert.equal(resp?.status, 502); + const parsed = JSON.parse(await resp!.text()); + assert.ok(parsed.error); + assert.match(parsed.error.message, /Upstream provider unreachable/); + }); + + it("returns null (fall through to cloud) when no config present", async () => { + const resp = await handleSelfHostedCompletions( + new Request("http://localhost/v1/chat/completions"), + { model: "gpt-4o", messages: [] }, + {} + ); + assert.equal(resp, null); + }); + + it("returns a 500 when config exists but fails to parse", async () => { + const resp = await handleSelfHostedCompletions( + new Request("http://localhost/v1/chat/completions"), + { model: "gpt-4o", messages: [] }, + { providers: "providers: nope" } + ); + assert.equal(resp?.status, 500); + const parsed = JSON.parse(await resp!.text()); + assert.ok(parsed.error); + }); +}); + +/** + * Deterministic routing strategies over the real HTTP contract (M2 — RIC-740). + * + * Fault injection: the primary provider is DOWN (connection refused), so the + * fallback chain must select the backup and the client receives a success — + * plus an `x-omniroute-route-decision` header explaining the choice. + */ +describe("selfHostedEntry deterministic routing (fault injection)", () => { + it("falls back to the backup when the primary provider is down", async () => { + const okCalls: Array<{ url: string; body: string }> = []; + const okServer = createServer((req, res) => { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + okCalls.push({ url: req.url ?? "", body }); + res.setHeader("content-type", "application/json"); + res.end( + JSON.stringify({ + id: "cmpl-backup", + object: "chat.completion", + choices: [ + { message: { role: "assistant", content: "backup-pong" }, finish_reason: "stop" }, + ], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, + }) + ); + }); + }); + await new Promise((resolve) => okServer.listen(0, "127.0.0.1", () => resolve())); + const okPort = (okServer.address() as AddressInfo).port; + + try { + // Port 1 refuses connections: primary "dead" is unreachable; backup is live. + const yaml = `providers: + - id: dead + kind: openai + baseUrl: http://127.0.0.1:1/v1 + model: d + - id: backup + kind: openai + baseUrl: http://127.0.0.1:${okPort}/v1 + model: b +strategy: + fallbackChain: + - dead + - backup +`; + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ messages: [{ role: "user", content: "hi" }] }), + }); + const resp = await handleSelfHostedCompletions( + req, + { messages: [{ role: "user", content: "hi" }] }, + { providers: yaml } + ); + assert.equal(resp?.status, 200); + const parsed = JSON.parse(await resp!.text()); + assert.equal(parsed.choices[0].message.content, "backup-pong"); + assert.equal(okCalls.length, 1); + assert.match(resp!.headers.get("x-omniroute-route-decision") ?? "", /dead.*backup/); + } finally { + okServer.close(); + } + }); + + it("returns the last fallback error when every candidate is down", async () => { + const yaml = `providers: + - id: one + kind: openai + baseUrl: http://127.0.0.1:1/v1 + model: o + - id: two + kind: openai + baseUrl: http://127.0.0.1:1/v1 + model: t +`; + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ messages: [{ role: "user", content: "hi" }] }), + }); + const resp = await handleSelfHostedCompletions( + req, + { messages: [{ role: "user", content: "hi" }] }, + { providers: yaml } + ); + assert.equal(resp?.status, 502); + const parsed = JSON.parse(await resp!.text()); + assert.ok(parsed.error); + assert.match(parsed.error.message, /Upstream provider unreachable/); + }); + + it("rejects a pinned provider that was blacklisted (no silent re-route)", async () => { + const yaml = `providers: + - id: a + kind: openai + baseUrl: http://127.0.0.1:1/v1 + model: m + - id: b + kind: openai + baseUrl: http://127.0.0.1:1/v1 + model: m +strategy: + blacklist: + - a +`; + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "a/m", messages: [{ role: "user", content: "hi" }] }), + }); + const resp = await handleSelfHostedCompletions( + req, + { model: "a/m", messages: [{ role: "user", content: "hi" }] }, + { providers: yaml } + ); + assert.equal(resp?.status, 400); + const parsed = JSON.parse(await resp!.text()); + assert.match(parsed.error.message, /blacklist/); + }); + + it("cooldown breaker excludes a failing provider from the next request", async () => { + const okCalls: Array<{ url: string; body: string }> = []; + const okServer = createServer((req, res) => { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + okCalls.push({ url: req.url ?? "", body }); + res.setHeader("content-type", "application/json"); + res.end( + JSON.stringify({ + id: "cmpl-ok", + object: "chat.completion", + choices: [{ message: { role: "assistant", content: "ok" }, finish_reason: "stop" }], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, + }) + ); + }); + }); + await new Promise((resolve) => okServer.listen(0, "127.0.0.1", () => resolve())); + const okPort = (okServer.address() as AddressInfo).port; + + try { + const yaml = `providers: + - id: dead + kind: openai + baseUrl: http://127.0.0.1:1/v1 + model: d + - id: backup + kind: openai + baseUrl: http://127.0.0.1:${okPort}/v1 + model: b +strategy: + cooldown: + consecutiveFailures: 1 + cooldownMs: 60000 +`; + const mkReq = () => + new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "dead/d", messages: [{ role: "user", content: "hi" }] }), + }); + + // First request: "dead" unreachable → fallback hits backup, and the failure + // is recorded — the engine pins by model prefix but the cooldown next time + // reports "dead" as excluded even when pinned. + const first = await handleSelfHostedCompletions( + mkReq(), + { model: "dead/d", messages: [{ role: "user", content: "hi" }] }, + { providers: yaml } + ); + assert.equal(first?.status, 200); + assert.equal(okCalls.length, 1); + + // Second request: "dead" is now cooling (consecutiveFailures=1) — but it's + // pinned, so a hard-filter rejection of the pin is surfaced as 400 rather + // than a silent re-route. + const second = await handleSelfHostedCompletions( + mkReq(), + { model: "dead/d", messages: [{ role: "user", content: "hi" }] }, + { providers: yaml } + ); + assert.equal(second?.status, 400); + const parsed = JSON.parse(await second!.text()); + assert.match(parsed.error.message, /cooldown/); + // The backup must NOT have been hit on the excluded-pin request. + assert.equal(okCalls.length, 1); + } finally { + okServer.close(); + } + }); + + it("returns 503 with the full explainable decision when no providers are eligible", async () => { + const yaml = `providers: + - id: a + kind: openai + baseUrl: http://127.0.0.1:1/v1 + model: m +strategy: + blacklist: + - a +`; + const req = new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ messages: [{ role: "user", content: "hi" }] }), + }); + const resp = await handleSelfHostedCompletions( + req, + { messages: [{ role: "user", content: "hi" }] }, + { providers: yaml } + ); + assert.equal(resp?.status, 503); + const parsed = JSON.parse(await resp!.text()); + assert.match(parsed.error.message, /No eligible providers/); + assert.match(parsed.error.message, /blacklist/); + }); +});