mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-09-19 05:12:16 +03:00
feat(routing): deterministic routing strategies for self-hosted entry (RIC-740) (#13611)
* feat(routing): self-hosted unified OpenAI-compatible entry (RIC-738) Divert /v1/chat/completions through the self-hosted provider adapters when OMNIROUTE_SELF_HOSTED_PROVIDERS / OMNIROUTE_SELF_HOSTED_PROVIDERS_FILE is set: one OpenAI-compatible contract in, auto-route to the selected provider (x-omniroute-provider header, provider/model prefix, or first provider), standard OpenAI error shape out. Optional OMNIROUTE_SELF_HOSTED_API_KEY guards the entry (D5 reserved); unset = open loopback route. Upstream credentials stay runtime-only and are stripped from echoed responses. Brings in the provider-adapters baseline from sibling branch (RIC-737) that this entry depends on. Includes 21 passing unit tests (provider selection, model-prefix forwarding, header hygiene, auth, error normalization, SSE passthrough, fall-through/misconfig), docs, env example, changelog fragment. * feat(routing): deterministic routing strategies for self-hosted entry (RIC-740) Add the M2 deterministic routing strategy engine (D3 可审计路由) to the self-hosted unified entry: a declarative `strategy:` block expressing five explainable, non-predictive policies — blacklist/whitelist hard filters, cooldown circuit breaker, cost-priority, latency-aware ordering, and an explicit fallback chain. 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 an x-omniroute-route-decision header answering "why this model / why not that one". 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. Covers the RIC-740 acceptance: 5 strategy types with unit tests + HTTP fault-injection tests (primary down -> fallback works), config matching docs, and no predict/ML deps. Adds docs, .env.example entries, and a changelog fragment. * refactor(routing): reduce complexity-ratchet violations in new self-hosted routing files Extract cost/id validation, pin-blocked resolution, ordering, and env/file source resolution into small helpers so routingStrategies.ts and selfHostedEntry.ts stay under the complexity-ratchets cap. No behavior change — the same 51 routing-strategies/self-hosted-entry/provider-adapters tests pass unmodified. Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> * docs(routing): document the 5 self-hosted env vars in ENVIRONMENT.md check:env-doc-sync failed because OMNIROUTE_SELF_HOSTED_PROVIDERS(_FILE), OMNIROUTE_SELF_HOSTED_API_KEY and OMNIROUTE_SELF_HOSTED_STRATEGY(_FILE) were present in .env.example but missing from docs/reference/ENVIRONMENT.md. Add them under "6. Tool & Routing Policies". Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --------- Co-authored-by: Ant Rich <ant@richants.com> Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Co-authored-by: luyuehm <luyuehm@users.noreply.github.com>
This commit is contained in:
luyuehm
42
.env.example
42
.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 <key>`; 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)
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
1
changelog.d/features/ric738-openai-compat-self-hosted.md
Normal file
1
changelog.d/features/ric738-openai-compat-self-hosted.md
Normal file
@@ -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)
|
||||
1
changelog.d/features/ric740-deterministic-routing.md
Normal file
1
changelog.d/features/ric740-deterministic-routing.md
Normal file
@@ -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)
|
||||
@@ -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 <key>`. 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. |
|
||||
|
||||
---
|
||||
|
||||
124
docs/routing/DETERMINISTIC_ROUTING.md
Normal file
124
docs/routing/DETERMINISTIC_ROUTING.md
Normal file
@@ -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.
|
||||
93
docs/routing/SELF_HOSTED_OPENAI_ENTRY.md
Normal file
93
docs/routing/SELF_HOSTED_OPENAI_ENTRY.md
Normal file
@@ -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: <id>` 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 <key>`. 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).
|
||||
137
open-sse/services/providerAdapters.ts
Normal file
137
open-sse/services/providerAdapters.ts
Normal file
@@ -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<Record<string, unknown>>;
|
||||
model?: string;
|
||||
stream?: boolean;
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
export interface ProviderAdapter {
|
||||
readonly kind: ProviderKind;
|
||||
complete(
|
||||
request: ChatRequest,
|
||||
config: ProviderConfig,
|
||||
fetchImpl?: typeof fetch
|
||||
): Promise<Response>;
|
||||
}
|
||||
|
||||
const trimUrl = (url: string): string => url.replace(/\/+$/, "");
|
||||
|
||||
function authHeaders(config: ProviderConfig): Record<string, string> {
|
||||
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<string, string> = {
|
||||
"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<ProviderKind, ProviderAdapter> = {
|
||||
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<string, unknown>).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<string, unknown>;
|
||||
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<Omit<ProviderConfig, "apiKey">> {
|
||||
return config.providers.map(({ apiKey: _apiKey, ...safe }) => safe);
|
||||
}
|
||||
|
||||
export class ProviderRouter {
|
||||
private readonly providers: Map<string, ProviderConfig>;
|
||||
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<Response> {
|
||||
const config = this.select(id);
|
||||
return getProviderAdapter(config.kind).complete(request, config, fetchImpl);
|
||||
}
|
||||
}
|
||||
577
open-sse/services/routingStrategies.ts
Normal file
577
open-sse/services/routingStrategies.ts
Normal file
@@ -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<string, StrategyProviderConfig>;
|
||||
private readonly config: StrategyConfig;
|
||||
private readonly cooldown = new Map<string, CooldownState>();
|
||||
private readonly latency = new Map<string, LatencyState>();
|
||||
|
||||
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<string>();
|
||||
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<string, unknown>;
|
||||
|
||||
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<string, unknown>).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<string, unknown>;
|
||||
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<string, unknown>, 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<string, unknown>): 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<string, unknown> | 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<string, unknown>);
|
||||
});
|
||||
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;
|
||||
}
|
||||
432
open-sse/services/selfHostedEntry.ts
Normal file
432
open-sse/services/selfHostedEntry.ts
Normal file
@@ -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<string> {
|
||||
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<SelfHostedSources | null> {
|
||||
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<SelfHostedRuntime | null> {
|
||||
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<ProviderRouter | null> {
|
||||
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<Response> {
|
||||
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<string, unknown> | 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<string, unknown> | null
|
||||
): Promise<Response> {
|
||||
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<string, unknown> | null,
|
||||
options: SelfHostedOptions = {}
|
||||
): Promise<Response | null> {
|
||||
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);
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
38
tests/unit/provider-adapters.test.ts
Normal file
38
tests/unit/provider-adapters.test.ts
Normal file
@@ -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<string, string>)["x-api-key"], "an-key");
|
||||
assert.equal(calls[2].url, "http://localhost:11434/v1/chat/completions");
|
||||
});
|
||||
});
|
||||
267
tests/unit/routing-strategies.test.ts
Normal file
267
tests/unit/routing-strategies.test.ts
Normal file
@@ -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<typeof parseStrategyConfig>[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/);
|
||||
});
|
||||
});
|
||||
554
tests/unit/self-hosted-entry.test.ts
Normal file
554
tests/unit/self-hosted-entry.test.ts
Normal file
@@ -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<void>((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<void>((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<void>((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<void>((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/);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user