feat: universal LLM gateway with embedded multilingual admin console
- OpenAI-compatible gateway with intelligent routing via cortiq-router (allaigate) - Embedded build-free admin SPA at /admin: 7 languages (en/ru/de/fr/es/zh/tr), light/dark themes, hot config reload (no restart) - Admin API, per-request statistics + Prometheus /metrics, secret store, playground - English documentation (primary) with Russian translations (*.ru.md) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
infosave2007
@@ -0,0 +1,58 @@
|
||||
//! Embedded static assets for the admin panel (SPA). The `web/` directory is compiled
|
||||
//! into the binary via `rust-embed`, so the gateway remains a single self-contained file.
|
||||
//!
|
||||
//! Served as the main router's fallback: everything under `/admin/*` not matched by
|
||||
//! an explicit route is served as a file; an unknown path under `/admin` falls back to
|
||||
//! `index.html` (the client-side hash router handles navigation).
|
||||
|
||||
use axum::{
|
||||
http::{header, StatusCode, Uri},
|
||||
response::{IntoResponse, Response},
|
||||
};
|
||||
use rust_embed::RustEmbed;
|
||||
|
||||
#[derive(RustEmbed)]
|
||||
#[folder = "web/"]
|
||||
struct WebAssets;
|
||||
|
||||
fn serve_file(path: &str) -> Response {
|
||||
match WebAssets::get(path) {
|
||||
Some(content) => {
|
||||
let mime = mime_guess::from_path(path).first_or_octet_stream();
|
||||
(
|
||||
[(header::CONTENT_TYPE, mime.as_ref())],
|
||||
content.data.into_owned(),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
None => {
|
||||
// SPA: unknown path under /admin → serve index.html
|
||||
match WebAssets::get("index.html") {
|
||||
Some(content) => (
|
||||
[(header::CONTENT_TYPE, "text/html; charset=utf-8")],
|
||||
content.data.into_owned(),
|
||||
)
|
||||
.into_response(),
|
||||
None => (StatusCode::NOT_FOUND, "not found").into_response(),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Application-wide fallback: serves `/admin/*`; everything else → 404 JSON.
|
||||
pub async fn fallback(uri: Uri) -> Response {
|
||||
let path = uri.path();
|
||||
if let Some(rest) = path.strip_prefix("/admin") {
|
||||
let rest = rest.trim_start_matches('/');
|
||||
let file = if rest.is_empty() { "index.html" } else { rest };
|
||||
serve_file(file)
|
||||
} else {
|
||||
(
|
||||
StatusCode::NOT_FOUND,
|
||||
axum::Json(serde_json::json!({
|
||||
"error": { "message": "not found", "type": "not_found", "code": "not_found" }
|
||||
})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,756 @@
|
||||
//! Admin API for the web management panel (`/admin/api/*`) + SPA serving (`/admin/*`).
|
||||
//!
|
||||
//! All endpoints under `/admin/api` are protected by a Bearer admin token. Config changes
|
||||
//! go through `AppState::reload` (validation → build `Live` → atomic TOML write
|
||||
//! → swap), so they are applied without a restart. Secrets are never returned —
|
||||
//! only their presence status.
|
||||
|
||||
pub mod assets;
|
||||
|
||||
use crate::config::{
|
||||
AdminCfg, ApiKeyCfg, BreakerCfg, Config, CortiqCfg, LogCfg, ModelCfg, ProtocolsCfg, RouteCfg,
|
||||
RouterCfg, RoutingCfg, RoutingPolicy, StatsCfg, TelemetryCfg, TierTargets,
|
||||
};
|
||||
use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective};
|
||||
use crate::state::SharedState;
|
||||
use crate::stats::parse_duration_secs;
|
||||
use axum::{
|
||||
extract::{Path, Query, Request, State},
|
||||
http::{header, StatusCode},
|
||||
middleware::{self, Next},
|
||||
response::{IntoResponse, Response},
|
||||
routing::{get, post},
|
||||
Json, Router,
|
||||
};
|
||||
use serde::Deserialize;
|
||||
use serde_json::{json, Value};
|
||||
use std::collections::HashMap;
|
||||
use std::time::Instant;
|
||||
|
||||
// ───────────────────────────── common types / helpers ───────────────────────
|
||||
|
||||
type ApiResult = Result<Json<Value>, ApiError>;
|
||||
|
||||
pub struct ApiError {
|
||||
status: StatusCode,
|
||||
message: String,
|
||||
}
|
||||
impl ApiError {
|
||||
fn bad(msg: impl Into<String>) -> Self {
|
||||
Self {
|
||||
status: StatusCode::BAD_REQUEST,
|
||||
message: msg.into(),
|
||||
}
|
||||
}
|
||||
fn not_found(msg: impl Into<String>) -> Self {
|
||||
Self {
|
||||
status: StatusCode::NOT_FOUND,
|
||||
message: msg.into(),
|
||||
}
|
||||
}
|
||||
}
|
||||
impl From<anyhow::Error> for ApiError {
|
||||
fn from(e: anyhow::Error) -> Self {
|
||||
Self::bad(e.to_string())
|
||||
}
|
||||
}
|
||||
impl IntoResponse for ApiError {
|
||||
fn into_response(self) -> Response {
|
||||
(
|
||||
self.status,
|
||||
Json(json!({"error": {"message": self.message, "type": "admin_error"}})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
|
||||
fn ok(v: Value) -> ApiResult {
|
||||
Ok(Json(v))
|
||||
}
|
||||
|
||||
fn current_cfg(state: &SharedState) -> Config {
|
||||
state.live().cfg.clone()
|
||||
}
|
||||
|
||||
/// Random hex token from `n_bytes` bytes of cryptographic randomness.
|
||||
pub fn random_token(n_bytes: usize) -> String {
|
||||
let mut buf = vec![0u8; n_bytes];
|
||||
if let Ok(mut f) = std::fs::File::open("/dev/urandom") {
|
||||
use std::io::Read;
|
||||
let _ = f.read_exact(&mut buf);
|
||||
} else {
|
||||
let t = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_nanos();
|
||||
for (i, b) in buf.iter_mut().enumerate() {
|
||||
*b = ((t >> ((i % 16) * 8)) as u8) ^ (i as u8).wrapping_mul(31);
|
||||
}
|
||||
}
|
||||
buf.iter().map(|b| format!("{b:02x}")).collect()
|
||||
}
|
||||
|
||||
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
|
||||
if a.len() != b.len() {
|
||||
return false;
|
||||
}
|
||||
let mut diff = 0u8;
|
||||
for (x, y) in a.iter().zip(b.iter()) {
|
||||
diff |= x ^ y;
|
||||
}
|
||||
diff == 0
|
||||
}
|
||||
|
||||
fn mask_key(key: &str) -> String {
|
||||
let n = key.chars().count();
|
||||
if n <= 8 {
|
||||
return "•".repeat(n.max(3));
|
||||
}
|
||||
let head: String = key.chars().take(6).collect();
|
||||
let tail: String = key.chars().skip(n - 4).collect();
|
||||
format!("{head}…{tail}")
|
||||
}
|
||||
|
||||
fn tiers_to_map(routing: &RoutingCfg) -> HashMap<String, Vec<String>> {
|
||||
routing
|
||||
.tiers
|
||||
.iter()
|
||||
.map(|(k, v)| {
|
||||
let list = match v {
|
||||
TierTargets::List(l) => l.clone(),
|
||||
TierTargets::One(s) => vec![s.clone()],
|
||||
};
|
||||
(k.clone(), list)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn parse_routing(model: &str) -> RoutingDirective {
|
||||
if let Some(rest) = model.strip_prefix("cortiq-auto") {
|
||||
let profile = rest.strip_prefix(':').map(|p| p.to_string());
|
||||
RoutingDirective::Auto { profile }
|
||||
} else {
|
||||
RoutingDirective::Pinned {
|
||||
model_id: model.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────────── router ───────────────────────────────────────
|
||||
|
||||
/// Admin API routes (token-protected). Mounted into the main router.
|
||||
/// The token is captured in a middleware closure so it does not depend on State.
|
||||
pub fn api_routes(admin_token: String) -> Router<SharedState> {
|
||||
Router::new()
|
||||
.route("/admin/api/health", get(health))
|
||||
.route("/admin/api/meta", get(meta))
|
||||
.route("/admin/api/config", get(get_config).put(put_config))
|
||||
.route("/admin/api/models", get(list_models).post(create_model))
|
||||
.route(
|
||||
"/admin/api/models/:id",
|
||||
axum::routing::put(update_model).delete(delete_model),
|
||||
)
|
||||
.route("/admin/api/models/:id/probe", post(probe_model))
|
||||
.route("/admin/api/routing", get(get_routing).put(put_routing))
|
||||
.route(
|
||||
"/admin/api/protocols",
|
||||
get(get_protocols).put(put_protocols),
|
||||
)
|
||||
.route("/admin/api/settings", get(get_settings).put(put_settings))
|
||||
.route("/admin/api/keys", get(list_keys).post(create_key))
|
||||
.route("/admin/api/keys/:key", axum::routing::delete(delete_key))
|
||||
.route(
|
||||
"/admin/api/secrets",
|
||||
get(list_secrets).put(set_secret).delete(clear_secret),
|
||||
)
|
||||
.route("/admin/api/stats", get(get_stats))
|
||||
.route("/admin/api/requests", get(get_requests))
|
||||
.route("/admin/api/test", post(run_test))
|
||||
.route_layer(middleware::from_fn(move |req: Request, next: Next| {
|
||||
let expected = admin_token.clone();
|
||||
async move { auth(expected, req, next).await }
|
||||
}))
|
||||
}
|
||||
|
||||
/// Extract Bearer admin token from the `Authorization` header.
|
||||
fn bearer_token(req: &Request) -> String {
|
||||
req.headers()
|
||||
.get(header::AUTHORIZATION)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(|v| {
|
||||
v.strip_prefix("Bearer ")
|
||||
.or_else(|| v.strip_prefix("bearer "))
|
||||
.unwrap_or(v)
|
||||
.trim()
|
||||
})
|
||||
.unwrap_or("")
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// Admin authentication middleware (constant-time comparison).
|
||||
async fn auth(expected: String, req: Request, next: Next) -> Response {
|
||||
let token = bearer_token(&req);
|
||||
if !expected.is_empty() && constant_time_eq(token.as_bytes(), expected.as_bytes()) {
|
||||
next.run(req).await
|
||||
} else {
|
||||
(
|
||||
StatusCode::UNAUTHORIZED,
|
||||
Json(json!({"error": {"message": "invalid or missing admin token", "type": "authentication_error", "code": "unauthorized"}})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────────── metrics ──────────────────────────────────────
|
||||
|
||||
/// `GET /metrics` (Prometheus). Exposed when `telemetry.metrics = true`.
|
||||
pub async fn metrics(State(state): State<SharedState>) -> Response {
|
||||
let live = state.live();
|
||||
if !live.cfg.telemetry.metrics {
|
||||
return (StatusCode::NOT_FOUND, "metrics disabled").into_response();
|
||||
}
|
||||
let body = state.stats.prometheus();
|
||||
([(header::CONTENT_TYPE, "text/plain; version=0.0.4")], body).into_response()
|
||||
}
|
||||
|
||||
// ───────────────────────────── health / meta ───────────────────────────────
|
||||
|
||||
async fn probe_router(url: &str) -> (bool, u64) {
|
||||
let client = match reqwest::Client::builder()
|
||||
.timeout(std::time::Duration::from_millis(1500))
|
||||
.danger_accept_invalid_certs(true)
|
||||
.build()
|
||||
{
|
||||
Ok(c) => c,
|
||||
Err(_) => return (false, 0),
|
||||
};
|
||||
let started = Instant::now();
|
||||
let target = format!("{}/healthz", url.trim_end_matches('/'));
|
||||
let ok = client
|
||||
.get(&target)
|
||||
.send()
|
||||
.await
|
||||
.map(|r| r.status().as_u16() < 500)
|
||||
.unwrap_or(false);
|
||||
(ok, started.elapsed().as_millis() as u64)
|
||||
}
|
||||
|
||||
async fn health(State(state): State<SharedState>) -> ApiResult {
|
||||
let live = state.live();
|
||||
let (router_ok, router_ms) = probe_router(&live.cfg.router.url).await;
|
||||
let models: Vec<Value> = live
|
||||
.cfg
|
||||
.models
|
||||
.iter()
|
||||
.map(|m| {
|
||||
let key = match &m.api_key_env {
|
||||
Some(env) => state.secrets.source(env),
|
||||
None => "none",
|
||||
};
|
||||
json!({
|
||||
"id": m.id,
|
||||
"provider": m.provider,
|
||||
"model": m.model,
|
||||
"kind": m.kind,
|
||||
"cost_tier": m.cost_tier,
|
||||
"key_source": key,
|
||||
"in_registry": live.registry.get(&m.id).is_some(),
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
ok(json!({
|
||||
"ok": true,
|
||||
"version": env!("CARGO_PKG_VERSION"),
|
||||
"listen": live.cfg.listen,
|
||||
"router": { "url": live.cfg.router.url, "reachable": router_ok, "latency_ms": router_ms },
|
||||
"models": models,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn meta() -> ApiResult {
|
||||
ok(json!({
|
||||
"version": env!("CARGO_PKG_VERSION"),
|
||||
"providers": ["openai", "anthropic", "ollama", "http"],
|
||||
"provider_impl": { "openai": true, "ollama": true, "http": true, "anthropic": false },
|
||||
"cost_tiers": ["cheap", "mid", "expensive"],
|
||||
"kinds": ["chat", "embedding"],
|
||||
"profiles": ["cost-saver", "balanced", "quality-first"],
|
||||
"text_strategies": ["last_user", "last_user_plus_system", "concat_all"],
|
||||
"policy_modes": ["fixed_table", "cost_aware"],
|
||||
"tiers": ["low", "medium", "high"],
|
||||
"caps": ["tools", "vision"],
|
||||
"protocols_impl": {
|
||||
"openai_chat": true, "openai_completions": false, "openai_embeddings": false,
|
||||
"openai_models": false, "anthropic_messages": false, "mcp": false, "native_passthrough": false
|
||||
},
|
||||
"languages": ["en", "ru", "de", "fr", "es", "zh", "tr"],
|
||||
}))
|
||||
}
|
||||
|
||||
// ───────────────────────────── config ──────────────────────────────────────
|
||||
|
||||
async fn get_config(State(state): State<SharedState>) -> ApiResult {
|
||||
let cfg = current_cfg(&state);
|
||||
ok(serde_json::to_value(&cfg).map_err(|e| ApiError::bad(e.to_string()))?)
|
||||
}
|
||||
|
||||
async fn put_config(State(state): State<SharedState>, Json(cfg): Json<Config>) -> ApiResult {
|
||||
state.reload(cfg)?;
|
||||
ok(json!({ "ok": true }))
|
||||
}
|
||||
|
||||
// ───────────────────────────── models ──────────────────────────────────────
|
||||
|
||||
async fn list_models(State(state): State<SharedState>) -> ApiResult {
|
||||
let live = state.live();
|
||||
let models: Vec<Value> = live
|
||||
.cfg
|
||||
.models
|
||||
.iter()
|
||||
.map(|m| {
|
||||
let mut v = serde_json::to_value(m).unwrap_or(json!({}));
|
||||
let key_source = match &m.api_key_env {
|
||||
Some(env) => state.secrets.source(env),
|
||||
None => "none",
|
||||
};
|
||||
v["key_source"] = json!(key_source);
|
||||
v
|
||||
})
|
||||
.collect();
|
||||
ok(json!({ "models": models }))
|
||||
}
|
||||
|
||||
async fn create_model(State(state): State<SharedState>, Json(m): Json<ModelCfg>) -> ApiResult {
|
||||
let mut cfg = current_cfg(&state);
|
||||
if cfg.models.iter().any(|x| x.id == m.id) {
|
||||
return Err(ApiError::bad(format!("model id '{}' already exists", m.id)));
|
||||
}
|
||||
cfg.models.push(m);
|
||||
state.reload(cfg)?;
|
||||
ok(json!({ "ok": true }))
|
||||
}
|
||||
|
||||
async fn update_model(
|
||||
State(state): State<SharedState>,
|
||||
Path(id): Path<String>,
|
||||
Json(m): Json<ModelCfg>,
|
||||
) -> ApiResult {
|
||||
let mut cfg = current_cfg(&state);
|
||||
let idx = cfg
|
||||
.models
|
||||
.iter()
|
||||
.position(|x| x.id == id)
|
||||
.ok_or_else(|| ApiError::not_found(format!("model '{id}' not found")))?;
|
||||
// if the id was renamed — must not conflict with an existing one
|
||||
if m.id != id && cfg.models.iter().any(|x| x.id == m.id) {
|
||||
return Err(ApiError::bad(format!("model id '{}' already exists", m.id)));
|
||||
}
|
||||
cfg.models[idx] = m;
|
||||
state.reload(cfg)?;
|
||||
ok(json!({ "ok": true }))
|
||||
}
|
||||
|
||||
async fn delete_model(State(state): State<SharedState>, Path(id): Path<String>) -> ApiResult {
|
||||
let mut cfg = current_cfg(&state);
|
||||
let before = cfg.models.len();
|
||||
cfg.models.retain(|x| x.id != id);
|
||||
if cfg.models.len() == before {
|
||||
return Err(ApiError::not_found(format!("model '{id}' not found")));
|
||||
}
|
||||
state.reload(cfg)?;
|
||||
ok(json!({ "ok": true }))
|
||||
}
|
||||
|
||||
async fn probe_model(State(state): State<SharedState>, Path(id): Path<String>) -> ApiResult {
|
||||
let live = state.live();
|
||||
let mcfg = live
|
||||
.cfg
|
||||
.models
|
||||
.iter()
|
||||
.find(|m| m.id == id)
|
||||
.ok_or_else(|| ApiError::not_found(format!("model '{id}' not found")))?
|
||||
.clone();
|
||||
|
||||
if mcfg.provider == "anthropic" {
|
||||
return ok(json!({
|
||||
"ok": false,
|
||||
"error": "anthropic provider not yet implemented (planned v0.2)",
|
||||
}));
|
||||
}
|
||||
let provider = live
|
||||
.registry
|
||||
.get(&id)
|
||||
.ok_or_else(|| ApiError::bad(format!("model '{id}' has no runtime provider")))?;
|
||||
|
||||
let req = ChatRequest {
|
||||
routing: RoutingDirective::Pinned {
|
||||
model_id: id.clone(),
|
||||
},
|
||||
messages: vec![Message {
|
||||
role: "user".into(),
|
||||
content: "ping".into(),
|
||||
tool_calls: vec![],
|
||||
}],
|
||||
tools: vec![],
|
||||
params: GenParams {
|
||||
temperature: Some(0.0),
|
||||
max_tokens: Some(1),
|
||||
..Default::default()
|
||||
},
|
||||
stream: false,
|
||||
meta: RequestMeta {
|
||||
protocol: "probe".into(),
|
||||
..Default::default()
|
||||
},
|
||||
};
|
||||
let started = Instant::now();
|
||||
let res = provider.chat(req).await;
|
||||
let latency_ms = started.elapsed().as_millis() as u64;
|
||||
match res {
|
||||
Ok(_) => ok(json!({ "ok": true, "latency_ms": latency_ms })),
|
||||
Err(e) => ok(json!({ "ok": false, "latency_ms": latency_ms, "error": e.to_string() })),
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────────── routing ─────────────────────────────────────
|
||||
|
||||
async fn get_routing(State(state): State<SharedState>) -> ApiResult {
|
||||
let live = state.live();
|
||||
ok(json!({
|
||||
"tiers": tiers_to_map(&live.cfg.routing),
|
||||
"default": live.cfg.routing.default,
|
||||
"policy": live.cfg.routing.policy,
|
||||
}))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RoutingBody {
|
||||
#[serde(default)]
|
||||
tiers: HashMap<String, Vec<String>>,
|
||||
default: String,
|
||||
#[serde(default)]
|
||||
policy: RoutingPolicy,
|
||||
}
|
||||
|
||||
async fn put_routing(State(state): State<SharedState>, Json(body): Json<RoutingBody>) -> ApiResult {
|
||||
let mut cfg = current_cfg(&state);
|
||||
cfg.routing = RoutingCfg {
|
||||
tiers: body
|
||||
.tiers
|
||||
.into_iter()
|
||||
.map(|(k, v)| (k, TierTargets::List(v)))
|
||||
.collect(),
|
||||
default: body.default,
|
||||
policy: body.policy,
|
||||
};
|
||||
state.reload(cfg)?;
|
||||
ok(json!({ "ok": true }))
|
||||
}
|
||||
|
||||
// ───────────────────────────── protocols ───────────────────────────────────
|
||||
|
||||
async fn get_protocols(State(state): State<SharedState>) -> ApiResult {
|
||||
let live = state.live();
|
||||
ok(serde_json::to_value(&live.cfg.protocols).unwrap_or(json!({})))
|
||||
}
|
||||
|
||||
async fn put_protocols(State(state): State<SharedState>, Json(p): Json<ProtocolsCfg>) -> ApiResult {
|
||||
let mut cfg = current_cfg(&state);
|
||||
cfg.protocols = p;
|
||||
state.reload(cfg)?;
|
||||
ok(json!({ "ok": true }))
|
||||
}
|
||||
|
||||
// ───────────────────────────── settings ────────────────────────────────────
|
||||
|
||||
async fn get_settings(State(state): State<SharedState>) -> ApiResult {
|
||||
let live = state.live();
|
||||
let c = &live.cfg;
|
||||
// admin section without token value (not stored here — only the env name)
|
||||
ok(json!({
|
||||
"listen": c.listen,
|
||||
"router": c.router,
|
||||
"route": c.route,
|
||||
"breaker": c.breaker,
|
||||
"log": c.log,
|
||||
"telemetry": c.telemetry,
|
||||
"cortiq": c.cortiq,
|
||||
"stats": c.stats,
|
||||
"admin": c.admin,
|
||||
}))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct SettingsBody {
|
||||
listen: Option<String>,
|
||||
router: Option<RouterCfg>,
|
||||
route: Option<RouteCfg>,
|
||||
breaker: Option<BreakerCfg>,
|
||||
log: Option<LogCfg>,
|
||||
telemetry: Option<TelemetryCfg>,
|
||||
cortiq: Option<CortiqCfg>,
|
||||
stats: Option<StatsCfg>,
|
||||
admin: Option<AdminCfg>,
|
||||
}
|
||||
|
||||
async fn put_settings(State(state): State<SharedState>, Json(b): Json<SettingsBody>) -> ApiResult {
|
||||
let mut cfg = current_cfg(&state);
|
||||
let mut needs_restart = false;
|
||||
if let Some(v) = b.listen {
|
||||
if v != cfg.listen {
|
||||
needs_restart = true;
|
||||
}
|
||||
cfg.listen = v;
|
||||
}
|
||||
if let Some(v) = b.router {
|
||||
cfg.router = v;
|
||||
}
|
||||
if let Some(v) = b.route {
|
||||
cfg.route = v;
|
||||
}
|
||||
if let Some(v) = b.breaker {
|
||||
cfg.breaker = v;
|
||||
}
|
||||
if let Some(v) = b.log {
|
||||
cfg.log = v;
|
||||
}
|
||||
if let Some(v) = b.telemetry {
|
||||
cfg.telemetry = v;
|
||||
}
|
||||
if let Some(v) = b.cortiq {
|
||||
cfg.cortiq = v;
|
||||
}
|
||||
if let Some(v) = b.stats {
|
||||
cfg.stats = v;
|
||||
}
|
||||
if let Some(v) = b.admin {
|
||||
if v.listen != cfg.admin.listen {
|
||||
needs_restart = true;
|
||||
}
|
||||
cfg.admin = v;
|
||||
}
|
||||
state.reload(cfg)?;
|
||||
ok(json!({ "ok": true, "needs_restart": needs_restart }))
|
||||
}
|
||||
|
||||
// ───────────────────────────── keys ────────────────────────────────────────
|
||||
|
||||
async fn list_keys(State(state): State<SharedState>) -> ApiResult {
|
||||
let live = state.live();
|
||||
let keys: Vec<Value> = live
|
||||
.cfg
|
||||
.api_keys
|
||||
.iter()
|
||||
.map(|k| {
|
||||
json!({
|
||||
"key": k.key,
|
||||
"key_masked": mask_key(&k.key),
|
||||
"account": k.account,
|
||||
"rate_per_min": k.rate_per_min,
|
||||
"allow_models": k.allow_models,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
ok(json!({ "keys": keys, "open_mode": live.cfg.api_keys.is_empty() }))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct NewKey {
|
||||
#[serde(default)]
|
||||
key: String,
|
||||
account: String,
|
||||
#[serde(default)]
|
||||
rate_per_min: u32,
|
||||
#[serde(default)]
|
||||
allow_models: Vec<String>,
|
||||
}
|
||||
|
||||
async fn create_key(State(state): State<SharedState>, Json(b): Json<NewKey>) -> ApiResult {
|
||||
let key = if b.key.trim().is_empty() {
|
||||
format!("sk-gw-{}", random_token(18))
|
||||
} else {
|
||||
b.key.trim().to_string()
|
||||
};
|
||||
let mut cfg = current_cfg(&state);
|
||||
if cfg.api_keys.iter().any(|k| k.key == key) {
|
||||
return Err(ApiError::bad("key already exists"));
|
||||
}
|
||||
cfg.api_keys.push(ApiKeyCfg {
|
||||
key: key.clone(),
|
||||
account: b.account,
|
||||
rate_per_min: b.rate_per_min,
|
||||
allow_models: b.allow_models,
|
||||
});
|
||||
state.reload(cfg)?;
|
||||
// show the full key only once
|
||||
ok(json!({ "ok": true, "key": key }))
|
||||
}
|
||||
|
||||
async fn delete_key(State(state): State<SharedState>, Path(key): Path<String>) -> ApiResult {
|
||||
let mut cfg = current_cfg(&state);
|
||||
let before = cfg.api_keys.len();
|
||||
cfg.api_keys.retain(|k| k.key != key);
|
||||
if cfg.api_keys.len() == before {
|
||||
return Err(ApiError::not_found("key not found"));
|
||||
}
|
||||
state.reload(cfg)?;
|
||||
ok(json!({ "ok": true }))
|
||||
}
|
||||
|
||||
// ───────────────────────────── secrets ─────────────────────────────────────
|
||||
|
||||
async fn list_secrets(State(state): State<SharedState>) -> ApiResult {
|
||||
let live = state.live();
|
||||
// collect all env names referenced in the config
|
||||
let mut names: Vec<String> = Vec::new();
|
||||
for m in &live.cfg.models {
|
||||
if let Some(env) = &m.api_key_env {
|
||||
if !names.contains(env) {
|
||||
names.push(env.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Some(env) = &live.cfg.router.api_key_env {
|
||||
if !names.contains(env) {
|
||||
names.push(env.clone());
|
||||
}
|
||||
}
|
||||
let secrets: Vec<Value> = names
|
||||
.iter()
|
||||
.map(|n| json!({ "name": n, "source": state.secrets.source(n) }))
|
||||
.collect();
|
||||
ok(json!({ "secrets": secrets }))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct SecretBody {
|
||||
name: String,
|
||||
value: String,
|
||||
}
|
||||
|
||||
async fn set_secret(State(state): State<SharedState>, Json(b): Json<SecretBody>) -> ApiResult {
|
||||
if b.name.trim().is_empty() {
|
||||
return Err(ApiError::bad("secret name is required"));
|
||||
}
|
||||
state.secrets.set(b.name.trim(), &b.value)?;
|
||||
// rebuild providers with the new key (without writing TOML)
|
||||
state.rebuild()?;
|
||||
ok(json!({ "ok": true }))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct NameQuery {
|
||||
name: String,
|
||||
}
|
||||
|
||||
async fn clear_secret(State(state): State<SharedState>, Query(q): Query<NameQuery>) -> ApiResult {
|
||||
state.secrets.clear(q.name.trim())?;
|
||||
state.rebuild()?;
|
||||
ok(json!({ "ok": true }))
|
||||
}
|
||||
|
||||
// ───────────────────────────── stats / requests ────────────────────────────
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct StatsQuery {
|
||||
#[serde(default)]
|
||||
range: Option<String>,
|
||||
#[serde(default)]
|
||||
groupby: Option<String>,
|
||||
}
|
||||
|
||||
async fn get_stats(State(state): State<SharedState>, Query(q): Query<StatsQuery>) -> ApiResult {
|
||||
let range_secs = q
|
||||
.range
|
||||
.as_deref()
|
||||
.and_then(parse_duration_secs)
|
||||
.unwrap_or(24 * 3600);
|
||||
let groupby = q.groupby.as_deref().unwrap_or("model");
|
||||
ok(state.stats.snapshot(range_secs, groupby))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RecentQuery {
|
||||
#[serde(default)]
|
||||
limit: Option<usize>,
|
||||
#[serde(default)]
|
||||
offset: Option<usize>,
|
||||
}
|
||||
|
||||
async fn get_requests(State(state): State<SharedState>, Query(q): Query<RecentQuery>) -> ApiResult {
|
||||
let limit = q.limit.unwrap_or(50).min(500);
|
||||
let offset = q.offset.unwrap_or(0);
|
||||
let items = state.stats.recent(limit, offset);
|
||||
ok(json!({ "requests": items }))
|
||||
}
|
||||
|
||||
// ───────────────────────────── playground (test) ───────────────────────────
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct TestBody {
|
||||
#[serde(default = "default_test_model")]
|
||||
model: String,
|
||||
messages: Vec<Message>,
|
||||
#[serde(default)]
|
||||
temperature: Option<f32>,
|
||||
#[serde(default)]
|
||||
max_tokens: Option<u32>,
|
||||
}
|
||||
fn default_test_model() -> String {
|
||||
"cortiq-auto".into()
|
||||
}
|
||||
|
||||
async fn run_test(State(state): State<SharedState>, Json(b): Json<TestBody>) -> ApiResult {
|
||||
if b.messages.is_empty() {
|
||||
return Err(ApiError::bad("messages must not be empty"));
|
||||
}
|
||||
let req = ChatRequest {
|
||||
routing: parse_routing(&b.model),
|
||||
messages: b.messages,
|
||||
tools: vec![],
|
||||
params: GenParams {
|
||||
temperature: b.temperature,
|
||||
max_tokens: b.max_tokens,
|
||||
..Default::default()
|
||||
},
|
||||
stream: false,
|
||||
meta: RequestMeta {
|
||||
account: "playground".into(),
|
||||
protocol: "playground".into(),
|
||||
..Default::default()
|
||||
},
|
||||
};
|
||||
let started = Instant::now();
|
||||
let res = state.pipeline.run(req, &state).await;
|
||||
let latency_ms = started.elapsed().as_millis() as u64;
|
||||
match res {
|
||||
Ok(resp) => {
|
||||
let answer = resp
|
||||
.choices
|
||||
.first()
|
||||
.map(|c| c.message.content.clone())
|
||||
.unwrap_or_default();
|
||||
ok(json!({
|
||||
"ok": true,
|
||||
"latency_ms": latency_ms,
|
||||
"answer": answer,
|
||||
"model_used": resp.model_used,
|
||||
"usage": {
|
||||
"prompt_tokens": resp.usage.prompt_tokens,
|
||||
"completion_tokens": resp.usage.completion_tokens,
|
||||
"total_tokens": resp.usage.total_tokens,
|
||||
},
|
||||
"cortiq": {
|
||||
"task_label": resp.cortiq.task_label,
|
||||
"complexity": { "score": resp.cortiq.complexity_score, "tier": resp.cortiq.complexity_tier },
|
||||
"selected_model": resp.cortiq.selected_model,
|
||||
"route_source": resp.cortiq.route_source,
|
||||
"router_request_id": resp.cortiq.router_request_id,
|
||||
"cost_usd": resp.cortiq.cost_usd,
|
||||
"failover": resp.cortiq.failover,
|
||||
},
|
||||
}))
|
||||
}
|
||||
Err(e) => ok(json!({ "ok": false, "latency_ms": latency_ms, "error": e.to_string() })),
|
||||
}
|
||||
}
|
||||
+474
@@ -0,0 +1,474 @@
|
||||
//! Configuration loading from TOML. Structs mirror config/gateway.example.toml.
|
||||
//! Secrets are read from environment variables by the names given in `*_env` fields.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct Config {
|
||||
pub listen: String,
|
||||
pub router: RouterCfg,
|
||||
#[serde(default)]
|
||||
pub route: RouteCfg,
|
||||
#[serde(default)]
|
||||
pub models: Vec<ModelCfg>,
|
||||
pub routing: RoutingCfg,
|
||||
#[serde(default)]
|
||||
pub breaker: BreakerCfg,
|
||||
#[serde(default)]
|
||||
pub protocols: ProtocolsCfg,
|
||||
#[serde(default)]
|
||||
pub cortiq: CortiqCfg,
|
||||
#[serde(default)]
|
||||
pub api_keys: Vec<ApiKeyCfg>,
|
||||
#[serde(default)]
|
||||
pub idempotency: IdempotencyCfg,
|
||||
#[serde(default)]
|
||||
pub log: LogCfg,
|
||||
#[serde(default)]
|
||||
pub telemetry: TelemetryCfg,
|
||||
#[serde(default)]
|
||||
pub admin: AdminCfg,
|
||||
#[serde(default)]
|
||||
pub stats: StatsCfg,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct RouterCfg {
|
||||
pub url: String,
|
||||
#[serde(default = "default_router_timeout")]
|
||||
pub timeout_ms: u64,
|
||||
#[serde(default = "default_verify_tls")]
|
||||
pub verify_tls: bool,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub api_key_env: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub taxonomy_id: Option<String>,
|
||||
}
|
||||
fn default_router_timeout() -> u64 {
|
||||
800
|
||||
}
|
||||
fn default_verify_tls() -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct RouteCfg {
|
||||
#[serde(default = "default_text_strategy")]
|
||||
pub text_strategy: String,
|
||||
#[serde(default = "default_max_chars")]
|
||||
pub max_chars: usize,
|
||||
#[serde(default = "default_cache_ttl")]
|
||||
pub cache_ttl: String,
|
||||
#[serde(default = "default_profile")]
|
||||
pub profile: String,
|
||||
}
|
||||
impl Default for RouteCfg {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
text_strategy: default_text_strategy(),
|
||||
max_chars: default_max_chars(),
|
||||
cache_ttl: default_cache_ttl(),
|
||||
profile: default_profile(),
|
||||
}
|
||||
}
|
||||
}
|
||||
fn default_text_strategy() -> String {
|
||||
"last_user".into()
|
||||
}
|
||||
fn default_max_chars() -> usize {
|
||||
4000
|
||||
}
|
||||
fn default_cache_ttl() -> String {
|
||||
"60s".into()
|
||||
}
|
||||
fn default_profile() -> String {
|
||||
"balanced".into()
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct ModelCfg {
|
||||
pub id: String,
|
||||
pub provider: String, // openai | anthropic | ollama | http
|
||||
pub base_url: String,
|
||||
pub model: String,
|
||||
#[serde(default)]
|
||||
pub cost_tier: String,
|
||||
#[serde(default)]
|
||||
pub price_in: f64,
|
||||
#[serde(default)]
|
||||
pub price_out: f64,
|
||||
#[serde(default = "default_kind")]
|
||||
pub kind: String, // chat | embedding
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub api_key_env: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub caps: Vec<String>,
|
||||
}
|
||||
fn default_kind() -> String {
|
||||
"chat".into()
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct RoutingCfg {
|
||||
#[serde(flatten)]
|
||||
pub tiers: HashMap<String, TierTargets>,
|
||||
pub default: String,
|
||||
#[serde(default)]
|
||||
pub policy: RoutingPolicy,
|
||||
}
|
||||
|
||||
/// A routing tier is either a list of model_ids or the string `default`.
|
||||
/// (`default` is handled by a separate field; only tier lists appear here.)
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
#[serde(untagged)]
|
||||
pub enum TierTargets {
|
||||
List(Vec<String>),
|
||||
One(String),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct RoutingPolicy {
|
||||
#[serde(default = "default_mode")]
|
||||
pub mode: String, // fixed_table | cost_aware
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max_cost_usd_per_request: Option<f64>,
|
||||
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
|
||||
pub min_class: HashMap<String, String>,
|
||||
}
|
||||
impl Default for RoutingPolicy {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
mode: default_mode(),
|
||||
min_class: HashMap::new(),
|
||||
max_cost_usd_per_request: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
fn default_mode() -> String {
|
||||
"fixed_table".into()
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct BreakerCfg {
|
||||
#[serde(default = "default_breaker_threshold")]
|
||||
pub threshold: u32,
|
||||
#[serde(default = "default_breaker_cooldown")]
|
||||
pub cooldown: String,
|
||||
}
|
||||
impl Default for BreakerCfg {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
threshold: default_breaker_threshold(),
|
||||
cooldown: default_breaker_cooldown(),
|
||||
}
|
||||
}
|
||||
}
|
||||
fn default_breaker_threshold() -> u32 {
|
||||
5
|
||||
}
|
||||
fn default_breaker_cooldown() -> String {
|
||||
"30s".into()
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
|
||||
pub struct ProtocolsCfg {
|
||||
#[serde(default)]
|
||||
pub openai_chat: bool,
|
||||
#[serde(default)]
|
||||
pub openai_completions: bool,
|
||||
#[serde(default)]
|
||||
pub openai_embeddings: bool,
|
||||
#[serde(default)]
|
||||
pub openai_models: bool,
|
||||
#[serde(default)]
|
||||
pub anthropic_messages: bool,
|
||||
#[serde(default)]
|
||||
pub mcp: bool,
|
||||
#[serde(default)]
|
||||
pub native_passthrough: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct CortiqCfg {
|
||||
#[serde(default)]
|
||||
pub echo: bool,
|
||||
}
|
||||
impl Default for CortiqCfg {
|
||||
fn default() -> Self {
|
||||
Self { echo: true }
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct ApiKeyCfg {
|
||||
pub key: String,
|
||||
pub account: String,
|
||||
#[serde(default)]
|
||||
pub rate_per_min: u32,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub allow_models: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct IdempotencyCfg {
|
||||
#[serde(default = "default_idem_ttl")]
|
||||
pub ttl: String,
|
||||
}
|
||||
impl Default for IdempotencyCfg {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
ttl: default_idem_ttl(),
|
||||
}
|
||||
}
|
||||
}
|
||||
fn default_idem_ttl() -> String {
|
||||
"10m".into()
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct LogCfg {
|
||||
#[serde(default)]
|
||||
pub bodies: bool,
|
||||
#[serde(default = "default_log_level")]
|
||||
pub level: String,
|
||||
}
|
||||
impl Default for LogCfg {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
bodies: false,
|
||||
level: default_log_level(),
|
||||
}
|
||||
}
|
||||
}
|
||||
fn default_log_level() -> String {
|
||||
"info".into()
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
|
||||
pub struct TelemetryCfg {
|
||||
#[serde(default)]
|
||||
pub metrics: bool,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub otlp_endpoint_env: Option<String>,
|
||||
}
|
||||
|
||||
/// Access to the web management panel and admin API.
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct AdminCfg {
|
||||
/// Whether the admin panel and admin API (`/admin`, `/admin/api/*`) are enabled.
|
||||
#[serde(default = "default_admin_enabled")]
|
||||
pub enabled: bool,
|
||||
/// Name of the env variable holding the admin token. If unset (or empty),
|
||||
/// a token is generated at startup and printed to the log.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub token_env: Option<String>,
|
||||
/// Optional separate bind address for the admin surface (e.g. `127.0.0.1:9001`).
|
||||
/// If unset, the admin interface lives on the main `listen` address under `/admin`.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub listen: Option<String>,
|
||||
}
|
||||
impl Default for AdminCfg {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
enabled: default_admin_enabled(),
|
||||
token_env: None,
|
||||
listen: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
fn default_admin_enabled() -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
/// Request statistics tracking (for the dashboard and `/metrics`).
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct StatsCfg {
|
||||
/// Whether to track request statistics.
|
||||
#[serde(default = "default_stats_enabled")]
|
||||
pub enabled: bool,
|
||||
/// Append-only request log file (JSONL). Empty means in-memory only.
|
||||
#[serde(default = "default_stats_file")]
|
||||
pub file: String,
|
||||
/// How many recent requests to keep in memory (the "recent" ring buffer).
|
||||
#[serde(default = "default_stats_ring")]
|
||||
pub ring_size: usize,
|
||||
/// Retention window for aggregates/time-series replayed on startup, e.g. `7d`.
|
||||
#[serde(default = "default_stats_retention")]
|
||||
pub retention: String,
|
||||
}
|
||||
impl Default for StatsCfg {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
enabled: default_stats_enabled(),
|
||||
file: default_stats_file(),
|
||||
ring_size: default_stats_ring(),
|
||||
retention: default_stats_retention(),
|
||||
}
|
||||
}
|
||||
}
|
||||
fn default_stats_enabled() -> bool {
|
||||
true
|
||||
}
|
||||
fn default_stats_file() -> String {
|
||||
"config/stats.jsonl".into()
|
||||
}
|
||||
fn default_stats_ring() -> usize {
|
||||
500
|
||||
}
|
||||
fn default_stats_retention() -> String {
|
||||
"7d".into()
|
||||
}
|
||||
|
||||
impl Config {
|
||||
/// Load and validate the config from a file.
|
||||
pub fn load(path: &str) -> anyhow::Result<Self> {
|
||||
let raw = std::fs::read_to_string(path)
|
||||
.map_err(|e| anyhow::anyhow!("cannot read config {path}: {e}"))?;
|
||||
let cfg: Config = toml::from_str(&raw)?;
|
||||
cfg.validate()?;
|
||||
Ok(cfg)
|
||||
}
|
||||
|
||||
/// Integrity validation: non-empty model pool, valid routing targets.
|
||||
pub fn validate(&self) -> anyhow::Result<()> {
|
||||
if self.models.is_empty() {
|
||||
anyhow::bail!("config: at least one [[models]] entry is required");
|
||||
}
|
||||
// uniqueness of model ids
|
||||
let mut seen = std::collections::HashSet::new();
|
||||
for m in &self.models {
|
||||
if !seen.insert(m.id.as_str()) {
|
||||
anyhow::bail!("config: duplicate model id '{}'", m.id);
|
||||
}
|
||||
}
|
||||
// every target in [routing] must exist in the model pool
|
||||
let ids: std::collections::HashSet<&str> =
|
||||
self.models.iter().map(|m| m.id.as_str()).collect();
|
||||
if !ids.contains(self.routing.default.as_str()) {
|
||||
anyhow::bail!(
|
||||
"config: routing.default '{}' is not a known model id",
|
||||
self.routing.default
|
||||
);
|
||||
}
|
||||
for (tier, targets) in &self.routing.tiers {
|
||||
let list = match targets {
|
||||
TierTargets::List(v) => v.clone(),
|
||||
TierTargets::One(s) => vec![s.clone()],
|
||||
};
|
||||
for id in &list {
|
||||
if !ids.contains(id.as_str()) {
|
||||
anyhow::bail!(
|
||||
"config: routing.{} references unknown model id '{}'",
|
||||
tier,
|
||||
id
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Serialize the config to a TOML string (no secrets — they are intentionally absent here).
|
||||
pub fn to_toml_string(&self) -> anyhow::Result<String> {
|
||||
let header = "# Cortiq Gateway — configuration (managed via the admin panel).\n\
|
||||
# This file is rewritten by the panel; manual edits are preserved\n\
|
||||
# only until the next change made through the UI.\n\n";
|
||||
let body = toml::to_string_pretty(self)?;
|
||||
Ok(format!("{header}{body}"))
|
||||
}
|
||||
|
||||
/// Atomically save the config to a file (write to a temporary file then rename).
|
||||
pub fn save(&self, path: &str) -> anyhow::Result<()> {
|
||||
self.validate()?;
|
||||
let data = self.to_toml_string()?;
|
||||
let p = std::path::Path::new(path);
|
||||
let dir = p.parent().unwrap_or_else(|| std::path::Path::new("."));
|
||||
std::fs::create_dir_all(dir).ok();
|
||||
let tmp = p.with_extension("toml.tmp");
|
||||
std::fs::write(&tmp, data.as_bytes())
|
||||
.map_err(|e| anyhow::anyhow!("cannot write config tmp {}: {e}", tmp.display()))?;
|
||||
std::fs::rename(&tmp, p)
|
||||
.map_err(|e| anyhow::anyhow!("cannot replace config {path}: {e}"))?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
const SAMPLE: &str = r#"
|
||||
listen = "0.0.0.0:9000"
|
||||
|
||||
[router]
|
||||
url = "http://localhost:8080"
|
||||
api_key_env = "CORTIQ_ROUTER_KEY"
|
||||
|
||||
[[models]]
|
||||
id = "local-qwen"
|
||||
provider = "openai"
|
||||
base_url = "http://localhost:8000/v1"
|
||||
model = "qwen2.5-7b-instruct"
|
||||
cost_tier = "cheap"
|
||||
caps = ["tools"]
|
||||
|
||||
[[models]]
|
||||
id = "gpt-4o-mini"
|
||||
provider = "openai"
|
||||
base_url = "https://api.openai.com/v1"
|
||||
model = "gpt-4o-mini"
|
||||
api_key_env = "OPENAI_API_KEY"
|
||||
cost_tier = "mid"
|
||||
price_in = 0.15
|
||||
price_out = 0.6
|
||||
|
||||
[routing]
|
||||
low = ["local-qwen"]
|
||||
medium = ["gpt-4o-mini", "local-qwen"]
|
||||
default = "gpt-4o-mini"
|
||||
|
||||
[routing.policy]
|
||||
mode = "cost_aware"
|
||||
max_cost_usd_per_request = 0.5
|
||||
min_class = { low = "cheap", medium = "mid" }
|
||||
"#;
|
||||
|
||||
#[test]
|
||||
fn roundtrip_preserves_routing_and_models() {
|
||||
let cfg: Config = toml::from_str(SAMPLE).expect("parse sample");
|
||||
cfg.validate().expect("sample valid");
|
||||
|
||||
// key risk: serialization of flattened [routing] + nested policy
|
||||
let serialized = cfg.to_toml_string().expect("serialize");
|
||||
let back: Config = toml::from_str(&serialized).expect("re-parse serialized");
|
||||
back.validate().expect("roundtrip valid");
|
||||
|
||||
assert_eq!(back.listen, "0.0.0.0:9000");
|
||||
assert_eq!(back.models.len(), 2);
|
||||
assert_eq!(back.routing.default, "gpt-4o-mini");
|
||||
assert_eq!(back.routing.policy.mode, "cost_aware");
|
||||
assert_eq!(back.routing.policy.max_cost_usd_per_request, Some(0.5));
|
||||
assert_eq!(
|
||||
back.routing.policy.min_class.get("low").map(String::as_str),
|
||||
Some("cheap")
|
||||
);
|
||||
|
||||
let low = back.routing.tiers.get("low").expect("low tier present");
|
||||
match low {
|
||||
TierTargets::List(v) => assert_eq!(v, &vec!["local-qwen".to_string()]),
|
||||
TierTargets::One(s) => assert_eq!(s, "local-qwen"),
|
||||
}
|
||||
|
||||
// admin/stats sections receive defaults
|
||||
assert!(back.admin.enabled);
|
||||
assert!(back.stats.enabled);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_rejects_unknown_routing_target() {
|
||||
let mut cfg: Config = toml::from_str(SAMPLE).unwrap();
|
||||
cfg.routing.default = "does-not-exist".into();
|
||||
assert!(cfg.validate().is_err());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
//! Unified gateway error type + rendering into an OpenAI-compatible envelope
|
||||
//! `{"error": {message, type, code, param}}` that SDK clients parse natively.
|
||||
//! See docs/PROTOCOLS.md §8.5.
|
||||
|
||||
use axum::{http::StatusCode, response::IntoResponse, Json};
|
||||
use serde_json::json;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum GatewayError {
|
||||
#[error("{0}")]
|
||||
InvalidRequest(String),
|
||||
#[error("authentication failed: {0}")]
|
||||
Unauthorized(String),
|
||||
#[error("rate limit exceeded")]
|
||||
RateLimited { retry_after_secs: u64 },
|
||||
#[error("upstream unavailable: {0}")]
|
||||
UpstreamUnavailable(String),
|
||||
#[error("internal error: {0}")]
|
||||
Internal(String),
|
||||
}
|
||||
|
||||
impl GatewayError {
|
||||
fn parts(&self) -> (StatusCode, &'static str, &'static str) {
|
||||
match self {
|
||||
Self::InvalidRequest(_) => (
|
||||
StatusCode::BAD_REQUEST,
|
||||
"invalid_request_error",
|
||||
"invalid_request",
|
||||
),
|
||||
Self::Unauthorized(_) => (
|
||||
StatusCode::UNAUTHORIZED,
|
||||
"authentication_error",
|
||||
"unauthorized",
|
||||
),
|
||||
Self::RateLimited { .. } => (
|
||||
StatusCode::TOO_MANY_REQUESTS,
|
||||
"rate_limit_error",
|
||||
"rate_limited",
|
||||
),
|
||||
Self::UpstreamUnavailable(_) => (
|
||||
StatusCode::BAD_GATEWAY,
|
||||
"upstream_unavailable",
|
||||
"no_healthy_model",
|
||||
),
|
||||
Self::Internal(_) => (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"internal_error",
|
||||
"internal",
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl IntoResponse for GatewayError {
|
||||
fn into_response(self) -> axum::response::Response {
|
||||
let (status, typ, code) = self.parts();
|
||||
let body = Json(json!({
|
||||
"error": {
|
||||
"message": self.to_string(),
|
||||
"type": typ,
|
||||
"code": code,
|
||||
"param": serde_json::Value::Null,
|
||||
}
|
||||
}));
|
||||
let mut resp = (status, body).into_response();
|
||||
if let Self::RateLimited { retry_after_secs } = self {
|
||||
resp.headers_mut()
|
||||
.insert("retry-after", retry_after_secs.to_string().parse().unwrap());
|
||||
}
|
||||
resp
|
||||
}
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, GatewayError>;
|
||||
+107
@@ -0,0 +1,107 @@
|
||||
//! Cortiq Gateway — entry point.
|
||||
//!
|
||||
//! Loads the config, builds the shared state, assembles routes from protocols and
|
||||
//! the admin panel, and starts the HTTP server. See docs/ARCHITECTURE.md.
|
||||
|
||||
// Part of the canonical model / error API surface (auth, rate-limit, idempotency,
|
||||
// traceparent, vision/streaming caps) is intentionally reserved for upcoming phases;
|
||||
// allow the unused items until those features land.
|
||||
#![allow(dead_code)]
|
||||
|
||||
mod admin;
|
||||
mod config;
|
||||
mod error;
|
||||
mod model;
|
||||
mod pipeline;
|
||||
mod protocols;
|
||||
mod providers;
|
||||
mod registry;
|
||||
mod router_client;
|
||||
mod routing;
|
||||
mod secrets;
|
||||
mod state;
|
||||
mod stats;
|
||||
|
||||
use axum::{routing::get, Router};
|
||||
use clap::Parser;
|
||||
use config::Config;
|
||||
use state::SharedState;
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(
|
||||
name = "cortiq-gateway",
|
||||
version,
|
||||
about = "Universal LLM gateway with intelligent routing"
|
||||
)]
|
||||
struct Args {
|
||||
/// Path to the TOML config file.
|
||||
#[arg(long, default_value = "config/gateway.toml")]
|
||||
config: String,
|
||||
/// Admin token for the web panel (overrides `[admin].token_env`).
|
||||
#[arg(long)]
|
||||
admin_token: Option<String>,
|
||||
}
|
||||
|
||||
/// Resolve the admin token: env from `[admin].token_env` → `--admin-token` →
|
||||
/// generate a random one. Returns `(token, generated)`.
|
||||
fn resolve_admin_token(cfg: &Config, args: &Args) -> (String, bool) {
|
||||
if let Some(env) = &cfg.admin.token_env {
|
||||
if let Ok(v) = std::env::var(env) {
|
||||
if !v.is_empty() {
|
||||
return (v, false);
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Some(t) = &args.admin_token {
|
||||
if !t.is_empty() {
|
||||
return (t.clone(), false);
|
||||
}
|
||||
}
|
||||
(admin::random_token(24), true)
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
let args = Args::parse();
|
||||
|
||||
let cfg = Config::load(&args.config)?;
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
tracing_subscriber::EnvFilter::try_from_default_env()
|
||||
.unwrap_or_else(|_| cfg.log.level.clone().into()),
|
||||
)
|
||||
.init();
|
||||
|
||||
let listen = cfg.listen.clone();
|
||||
let admin_enabled = cfg.admin.enabled;
|
||||
let (admin_token, generated) = resolve_admin_token(&cfg, &args);
|
||||
|
||||
let state = SharedState::build(cfg, args.config.clone())?;
|
||||
|
||||
let mut app = Router::new()
|
||||
.route("/healthz", get(|| async { "ok" }))
|
||||
.route("/readyz", get(|| async { "ready" }))
|
||||
.route("/metrics", get(admin::metrics))
|
||||
.merge(protocols::build_router());
|
||||
|
||||
if admin_enabled {
|
||||
app = app
|
||||
.merge(admin::api_routes(admin_token.clone()))
|
||||
.fallback(admin::assets::fallback);
|
||||
}
|
||||
|
||||
let app = app.with_state(state);
|
||||
|
||||
let listener = tokio::net::TcpListener::bind(&listen).await?;
|
||||
tracing::info!("cortiq-gateway listening on {listen}");
|
||||
if admin_enabled {
|
||||
tracing::info!("admin console: http://{listen}/admin");
|
||||
if generated {
|
||||
tracing::warn!(
|
||||
"admin token (auto-generated, set [admin].token_env to persist): {admin_token}"
|
||||
);
|
||||
}
|
||||
}
|
||||
axum::serve(listener, app).await?;
|
||||
Ok(())
|
||||
}
|
||||
+102
@@ -0,0 +1,102 @@
|
||||
//! Canonical (protocol-neutral) request/response model.
|
||||
//!
|
||||
//! All incoming protocols are translated into these types, and providers translate
|
||||
//! from them. This way N protocols × M providers do not become N×M translations:
|
||||
//! each adapter only knows its own protocol ↔ canonical mapping. See docs/ARCHITECTURE.md §3.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// How to handle routing for a specific request.
|
||||
#[derive(Clone, Debug)]
|
||||
pub enum RoutingDirective {
|
||||
/// `model = "cortiq-auto"` — ask cortiq-router and pick a model from the pool.
|
||||
Auto { profile: Option<String> },
|
||||
/// `model = "<real id>"` — call a specific model directly, bypassing routing.
|
||||
Pinned { model_id: String },
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct Message {
|
||||
pub role: String, // system | user | assistant | tool
|
||||
#[serde(default)]
|
||||
pub content: String,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub tool_calls: Vec<serde_json::Value>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct GenParams {
|
||||
pub temperature: Option<f32>,
|
||||
pub max_tokens: Option<u32>,
|
||||
pub top_p: Option<f32>,
|
||||
pub stop: Vec<String>,
|
||||
/// Fields we do not interpret but must proxy through to the provider.
|
||||
pub passthrough: serde_json::Map<String, serde_json::Value>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct RequestMeta {
|
||||
pub account: String,
|
||||
pub protocol: String,
|
||||
pub idempotency_key: Option<String>,
|
||||
pub traceparent: Option<String>,
|
||||
}
|
||||
|
||||
/// Canonical form of a generation request.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct ChatRequest {
|
||||
pub routing: RoutingDirective,
|
||||
pub messages: Vec<Message>,
|
||||
pub tools: Vec<serde_json::Value>,
|
||||
pub params: GenParams,
|
||||
pub stream: bool,
|
||||
pub meta: RequestMeta,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
pub struct Usage {
|
||||
pub prompt_tokens: u32,
|
||||
pub completion_tokens: u32,
|
||||
pub total_tokens: u32,
|
||||
}
|
||||
|
||||
/// Routing metadata attached to the response (X-Cortiq-* headers and the `cortiq` field).
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
pub struct RouteInfo {
|
||||
pub task_label: String,
|
||||
pub complexity_score: f32,
|
||||
pub complexity_tier: String,
|
||||
pub selected_model: String,
|
||||
pub route_source: String, // router | cache | fallback | pinned
|
||||
pub router_request_id: Option<String>,
|
||||
pub cost_usd: f64,
|
||||
#[serde(default)]
|
||||
pub failover: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
pub struct Choice {
|
||||
pub index: u32,
|
||||
pub message: Message,
|
||||
pub finish_reason: String,
|
||||
}
|
||||
|
||||
/// Canonical form of a response.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct ChatResponse {
|
||||
pub id: String,
|
||||
pub model_used: String,
|
||||
pub choices: Vec<Choice>,
|
||||
pub usage: Usage,
|
||||
pub cortiq: RouteInfo,
|
||||
}
|
||||
|
||||
/// Decision from cortiq-router (the subset of its contract needed by the gateway).
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct RouteDecision {
|
||||
pub task_label: String,
|
||||
pub complexity_score: f32,
|
||||
pub complexity_tier: String,
|
||||
pub router_request_id: Option<String>,
|
||||
pub source: String, // router | cache | fallback
|
||||
}
|
||||
+251
@@ -0,0 +1,251 @@
|
||||
//! Canonical request processing pipeline (Pipeline).
|
||||
//! Coordinates stages: Extract -> Route -> Select -> Translate & Call,
|
||||
//! and also measures latency and records per-request statistics.
|
||||
|
||||
use crate::error::{GatewayError, Result};
|
||||
use crate::model::{
|
||||
ChatRequest, ChatResponse, Message, RouteDecision, RouteInfo, RoutingDirective,
|
||||
};
|
||||
use crate::state::SharedState;
|
||||
use crate::stats::RequestRecord;
|
||||
use std::time::{Instant, SystemTime, UNIX_EPOCH};
|
||||
|
||||
pub struct Pipeline;
|
||||
|
||||
impl Pipeline {
|
||||
pub fn new() -> Self {
|
||||
Self
|
||||
}
|
||||
|
||||
/// Run the routing and generation pipeline.
|
||||
pub async fn run(&self, req: ChatRequest, state: &SharedState) -> Result<ChatResponse> {
|
||||
let started = Instant::now();
|
||||
let account = req.meta.account.clone();
|
||||
let protocol = req.meta.protocol.clone();
|
||||
let directive = match &req.routing {
|
||||
RoutingDirective::Pinned { .. } => "pinned",
|
||||
RoutingDirective::Auto { .. } => "auto",
|
||||
}
|
||||
.to_string();
|
||||
|
||||
let result = self.run_inner(req, state).await;
|
||||
|
||||
// record statistics (for both success and error)
|
||||
let latency_ms = started.elapsed().as_millis() as u64;
|
||||
let ts = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs();
|
||||
let rec = match &result {
|
||||
Ok(resp) => RequestRecord {
|
||||
ts,
|
||||
account,
|
||||
protocol,
|
||||
directive,
|
||||
task_label: resp.cortiq.task_label.clone(),
|
||||
tier: resp.cortiq.complexity_tier.clone(),
|
||||
score: resp.cortiq.complexity_score,
|
||||
model_id: resp.cortiq.selected_model.clone(),
|
||||
route_source: resp.cortiq.route_source.clone(),
|
||||
prompt_tokens: resp.usage.prompt_tokens,
|
||||
completion_tokens: resp.usage.completion_tokens,
|
||||
cost_usd: resp.cortiq.cost_usd,
|
||||
latency_ms,
|
||||
outcome: "ok".to_string(),
|
||||
failover: resp.cortiq.failover,
|
||||
error: None,
|
||||
},
|
||||
Err(err) => RequestRecord {
|
||||
ts,
|
||||
account,
|
||||
protocol,
|
||||
directive,
|
||||
task_label: String::new(),
|
||||
tier: String::new(),
|
||||
score: 0.0,
|
||||
model_id: String::new(),
|
||||
route_source: String::new(),
|
||||
prompt_tokens: 0,
|
||||
completion_tokens: 0,
|
||||
cost_usd: 0.0,
|
||||
latency_ms,
|
||||
outcome: "error".to_string(),
|
||||
failover: false,
|
||||
error: Some(err.to_string()),
|
||||
},
|
||||
};
|
||||
state.stats.record(rec);
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
async fn run_inner(&self, req: ChatRequest, state: &SharedState) -> Result<ChatResponse> {
|
||||
let live = state.live();
|
||||
|
||||
// 1. Extract text for classification
|
||||
let text = self.extract_text(
|
||||
&req.messages,
|
||||
&live.cfg.route.text_strategy,
|
||||
live.cfg.route.max_chars,
|
||||
);
|
||||
|
||||
// 2. Obtain the router decision (or use the pinned model)
|
||||
let (decision, route_source) = match &req.routing {
|
||||
RoutingDirective::Pinned { model_id: _ } => {
|
||||
let dec = RouteDecision {
|
||||
task_label: "pinned".to_string(),
|
||||
complexity_score: 0.0,
|
||||
complexity_tier: "pinned".to_string(),
|
||||
router_request_id: None,
|
||||
source: "pinned".to_string(),
|
||||
};
|
||||
(dec, "pinned".to_string())
|
||||
}
|
||||
RoutingDirective::Auto { profile } => {
|
||||
let prof = profile.as_deref().unwrap_or(&live.cfg.route.profile);
|
||||
match live.router.route(&text, prof).await {
|
||||
Ok(Some(dec)) => {
|
||||
let src = dec.source.clone();
|
||||
(dec, src)
|
||||
}
|
||||
Ok(None) | Err(_) => {
|
||||
tracing::warn!(
|
||||
"Router is unavailable. Gracefully degrading to default model."
|
||||
);
|
||||
let dec = RouteDecision {
|
||||
task_label: "degraded".to_string(),
|
||||
complexity_score: 0.5,
|
||||
complexity_tier: "degraded".to_string(),
|
||||
router_request_id: None,
|
||||
source: "fallback".to_string(),
|
||||
};
|
||||
(dec, "fallback".to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// 3. Select candidates by complexity tier / pinned model
|
||||
let candidates = match &req.routing {
|
||||
RoutingDirective::Pinned { model_id } => vec![model_id.clone()],
|
||||
RoutingDirective::Auto { .. } => {
|
||||
if route_source == "fallback" {
|
||||
vec![live.routing.default_model().to_string()]
|
||||
} else {
|
||||
live.routing.candidates(&decision.complexity_tier)
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Filter candidates by capabilities (e.g. tools support)
|
||||
let needs_tools = !req.tools.is_empty();
|
||||
let mut final_candidates = Vec::new();
|
||||
for model_id in candidates {
|
||||
if let Some(prov) = live.registry.get(&model_id) {
|
||||
if needs_tools && !prov.caps().tools {
|
||||
tracing::debug!(
|
||||
"Candidate '{}' skipped: lacks support for requested tools",
|
||||
model_id
|
||||
);
|
||||
continue;
|
||||
}
|
||||
final_candidates.push(model_id);
|
||||
}
|
||||
}
|
||||
|
||||
if final_candidates.is_empty() {
|
||||
return Err(GatewayError::UpstreamUnavailable(
|
||||
"No healthy model available that satisfies required capabilities (e.g. tools)"
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
// 4. Attempt provider call with failover (fall back to alternatives on error)
|
||||
let mut last_err = None;
|
||||
let mut attempt = 0usize;
|
||||
for model_id in final_candidates {
|
||||
let provider = match live.registry.get(&model_id) {
|
||||
Some(p) => p,
|
||||
None => continue,
|
||||
};
|
||||
|
||||
tracing::info!(
|
||||
"Routing request (label: {}, tier: {}) to model '{}' via source '{}'",
|
||||
decision.task_label,
|
||||
decision.complexity_tier,
|
||||
model_id,
|
||||
route_source
|
||||
);
|
||||
|
||||
match provider.chat(req.clone()).await {
|
||||
Ok(mut resp) => {
|
||||
let cost =
|
||||
provider.price(resp.usage.prompt_tokens, resp.usage.completion_tokens);
|
||||
resp.cortiq = RouteInfo {
|
||||
task_label: decision.task_label.clone(),
|
||||
complexity_score: decision.complexity_score,
|
||||
complexity_tier: decision.complexity_tier.clone(),
|
||||
selected_model: model_id,
|
||||
route_source: route_source.clone(),
|
||||
router_request_id: decision.router_request_id.clone(),
|
||||
cost_usd: cost,
|
||||
failover: attempt > 0,
|
||||
};
|
||||
return Ok(resp);
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::warn!(
|
||||
"Provider '{}' call failed: {:?}. Attempting failover...",
|
||||
model_id,
|
||||
err
|
||||
);
|
||||
last_err = Some(err);
|
||||
attempt += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Err(last_err.unwrap_or_else(|| {
|
||||
GatewayError::UpstreamUnavailable("All candidate models failed".to_string())
|
||||
}))
|
||||
}
|
||||
|
||||
fn extract_text(&self, messages: &[Message], strategy: &str, max_chars: usize) -> String {
|
||||
let raw_text = match strategy {
|
||||
"last_user" => messages
|
||||
.iter()
|
||||
.rfind(|m| m.role == "user")
|
||||
.map(|m| m.content.clone())
|
||||
.unwrap_or_default(),
|
||||
"last_user_plus_system" => {
|
||||
let system_msgs: Vec<String> = messages
|
||||
.iter()
|
||||
.filter(|m| m.role == "system")
|
||||
.map(|m| m.content.clone())
|
||||
.collect();
|
||||
let last_user = messages
|
||||
.iter()
|
||||
.rfind(|m| m.role == "user")
|
||||
.map(|m| m.content.as_str())
|
||||
.unwrap_or("");
|
||||
if system_msgs.is_empty() {
|
||||
last_user.to_string()
|
||||
} else {
|
||||
format!("{}\n{}", system_msgs.join("\n"), last_user)
|
||||
}
|
||||
}
|
||||
// "concat_all" and any unknown strategy fall back to concatenating all turns
|
||||
_ => messages
|
||||
.iter()
|
||||
.map(|m| format!("{}: {}", m.role, m.content))
|
||||
.collect::<Vec<String>>()
|
||||
.join("\n"),
|
||||
};
|
||||
|
||||
if raw_text.chars().count() > max_chars {
|
||||
raw_text.chars().take(max_chars).collect()
|
||||
} else {
|
||||
raw_text
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
//! Incoming protocol adapters. Each one builds its own axum routes and translates
|
||||
//! its protocol to/from the canonical model. Enabled via flags in `[protocols]`.
|
||||
|
||||
pub mod openai_chat;
|
||||
|
||||
use crate::state::SharedState;
|
||||
use axum::Router;
|
||||
|
||||
/// Build the protocols router. Implemented adapters are **always** mounted;
|
||||
/// each handler checks the live `protocols.*` flag from config itself, so
|
||||
/// toggles in the admin panel take effect without a restart (hot switching).
|
||||
pub fn build_router() -> Router<SharedState> {
|
||||
Router::new().merge(openai_chat::routes())
|
||||
// TODO: openai_completions, openai_embeddings, openai_models,
|
||||
// anthropic_messages, mcp, native_passthrough — add here as well,
|
||||
// with a check of the corresponding flag inside the handler.
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
//! Incoming adapter for OpenAI Chat Completions: `POST /v1/chat/completions`.
|
||||
//! Translates the OpenAI request body into a canonical [`ChatRequest`], passes it
|
||||
//! to the pipeline, then converts the canonical [`ChatResponse`] back to OpenAI format.
|
||||
|
||||
use crate::error::{GatewayError, Result};
|
||||
use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective};
|
||||
use crate::state::SharedState;
|
||||
use axum::{extract::State, routing::post, Json, Router};
|
||||
use serde::Deserialize;
|
||||
|
||||
pub fn routes() -> Router<SharedState> {
|
||||
Router::new().route("/v1/chat/completions", post(handler))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct OpenAiChatRequest {
|
||||
model: String,
|
||||
messages: Vec<Message>,
|
||||
#[serde(default)]
|
||||
temperature: Option<f32>,
|
||||
#[serde(default)]
|
||||
max_tokens: Option<u32>,
|
||||
#[serde(default)]
|
||||
top_p: Option<f32>,
|
||||
#[serde(default)]
|
||||
stream: bool,
|
||||
#[serde(default)]
|
||||
tools: Vec<serde_json::Value>,
|
||||
#[serde(flatten)]
|
||||
rest: serde_json::Map<String, serde_json::Value>,
|
||||
}
|
||||
|
||||
/// Parse the `model` field value into a routing directive.
|
||||
/// `cortiq-auto[:profile]` → Auto; otherwise → Pinned(real id).
|
||||
fn parse_routing(model: &str) -> RoutingDirective {
|
||||
if let Some(rest) = model.strip_prefix("cortiq-auto") {
|
||||
let profile = rest.strip_prefix(':').map(|p| p.to_string());
|
||||
RoutingDirective::Auto { profile }
|
||||
} else {
|
||||
RoutingDirective::Pinned {
|
||||
model_id: model.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handler(
|
||||
State(state): State<SharedState>,
|
||||
Json(req): Json<OpenAiChatRequest>,
|
||||
) -> Result<impl axum::response::IntoResponse> {
|
||||
// hot protocol toggle: if the adapter is disabled in config — return 404
|
||||
if !state.live().cfg.protocols.openai_chat {
|
||||
return Err(GatewayError::InvalidRequest(
|
||||
"openai_chat protocol is disabled".into(),
|
||||
));
|
||||
}
|
||||
if req.messages.is_empty() {
|
||||
return Err(GatewayError::InvalidRequest(
|
||||
"messages must not be empty".into(),
|
||||
));
|
||||
}
|
||||
|
||||
let canonical = ChatRequest {
|
||||
routing: parse_routing(&req.model),
|
||||
messages: req.messages,
|
||||
tools: req.tools,
|
||||
params: GenParams {
|
||||
temperature: req.temperature,
|
||||
max_tokens: req.max_tokens,
|
||||
top_p: req.top_p,
|
||||
stop: Vec::new(),
|
||||
passthrough: req.rest,
|
||||
},
|
||||
stream: req.stream,
|
||||
meta: RequestMeta {
|
||||
protocol: "openai_chat".into(),
|
||||
..Default::default()
|
||||
},
|
||||
};
|
||||
|
||||
if req.stream {
|
||||
return Err(GatewayError::InvalidRequest(
|
||||
"Streaming is not yet supported in this version".into(),
|
||||
));
|
||||
}
|
||||
|
||||
let resp = state.pipeline.run(canonical, &state).await?;
|
||||
|
||||
let mut headers = axum::http::HeaderMap::new();
|
||||
headers.insert(
|
||||
"X-Cortiq-Task-Label",
|
||||
axum::http::HeaderValue::from_str(&resp.cortiq.task_label)
|
||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
||||
);
|
||||
headers.insert(
|
||||
"X-Cortiq-Complexity-Score",
|
||||
axum::http::HeaderValue::from_str(&resp.cortiq.complexity_score.to_string())
|
||||
.unwrap_or(axum::http::HeaderValue::from_static("0.0")),
|
||||
);
|
||||
headers.insert(
|
||||
"X-Cortiq-Complexity-Tier",
|
||||
axum::http::HeaderValue::from_str(&resp.cortiq.complexity_tier)
|
||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
||||
);
|
||||
headers.insert(
|
||||
"X-Cortiq-Selected-Model",
|
||||
axum::http::HeaderValue::from_str(&resp.cortiq.selected_model)
|
||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
||||
);
|
||||
headers.insert(
|
||||
"X-Cortiq-Route-Source",
|
||||
axum::http::HeaderValue::from_str(&resp.cortiq.route_source)
|
||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
||||
);
|
||||
headers.insert(
|
||||
"X-Cortiq-Cost-Usd",
|
||||
axum::http::HeaderValue::from_str(&resp.cortiq.cost_usd.to_string())
|
||||
.unwrap_or(axum::http::HeaderValue::from_static("0.0")),
|
||||
);
|
||||
if let Some(req_id) = &resp.cortiq.router_request_id {
|
||||
headers.insert(
|
||||
"X-Cortiq-Request-Id",
|
||||
axum::http::HeaderValue::from_str(req_id)
|
||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
||||
);
|
||||
}
|
||||
|
||||
let mut body = serde_json::json!({
|
||||
"id": resp.id,
|
||||
"object": "chat.completion",
|
||||
"created": std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs(),
|
||||
"model": resp.model_used,
|
||||
"choices": resp.choices.iter().map(|c| {
|
||||
serde_json::json!({
|
||||
"index": c.index,
|
||||
"message": {
|
||||
"role": c.message.role,
|
||||
"content": c.message.content,
|
||||
"tool_calls": if c.message.tool_calls.is_empty() { serde_json::Value::Null } else { serde_json::Value::Array(c.message.tool_calls.clone()) },
|
||||
},
|
||||
"finish_reason": c.finish_reason,
|
||||
})
|
||||
}).collect::<Vec<_>>(),
|
||||
"usage": {
|
||||
"prompt_tokens": resp.usage.prompt_tokens,
|
||||
"completion_tokens": resp.usage.completion_tokens,
|
||||
"total_tokens": resp.usage.total_tokens,
|
||||
}
|
||||
});
|
||||
|
||||
if state.live().cfg.cortiq.echo {
|
||||
body["cortiq"] = serde_json::json!({
|
||||
"task_label": resp.cortiq.task_label,
|
||||
"complexity": {
|
||||
"score": resp.cortiq.complexity_score,
|
||||
"tier": resp.cortiq.complexity_tier,
|
||||
},
|
||||
"selected_model": resp.cortiq.selected_model,
|
||||
"route_source": resp.cortiq.route_source,
|
||||
"router_request_id": resp.cortiq.router_request_id,
|
||||
"cost_usd": resp.cortiq.cost_usd,
|
||||
});
|
||||
}
|
||||
|
||||
Ok((headers, Json(body)))
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
//! Anthropic Messages provider (v0.2). Translates the canonical ChatRequest to
|
||||
//! `POST /v1/messages` and back. The `system` message is extracted into a separate field.
|
||||
|
||||
use super::{Caps, Provider};
|
||||
use crate::config::ModelCfg;
|
||||
use crate::error::{GatewayError, Result};
|
||||
use crate::model::{ChatRequest, ChatResponse};
|
||||
use async_trait::async_trait;
|
||||
|
||||
pub struct AnthropicProvider {
|
||||
id: String,
|
||||
#[allow(dead_code)]
|
||||
base_url: String,
|
||||
#[allow(dead_code)]
|
||||
model: String,
|
||||
#[allow(dead_code)]
|
||||
api_key: Option<String>,
|
||||
caps: Caps,
|
||||
price_in: f64,
|
||||
price_out: f64,
|
||||
}
|
||||
|
||||
impl AnthropicProvider {
|
||||
pub fn new(cfg: &ModelCfg, api_key: Option<String>) -> anyhow::Result<Self> {
|
||||
Ok(Self {
|
||||
id: cfg.id.clone(),
|
||||
base_url: cfg.base_url.trim_end_matches('/').to_string(),
|
||||
model: cfg.model.clone(),
|
||||
api_key,
|
||||
caps: Caps::from_list(&cfg.caps),
|
||||
price_in: cfg.price_in,
|
||||
price_out: cfg.price_out,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Provider for AnthropicProvider {
|
||||
fn id(&self) -> &str {
|
||||
&self.id
|
||||
}
|
||||
fn caps(&self) -> &Caps {
|
||||
&self.caps
|
||||
}
|
||||
fn price(&self, prompt_tokens: u32, completion_tokens: u32) -> f64 {
|
||||
(prompt_tokens as f64 * self.price_in + completion_tokens as f64 * self.price_out)
|
||||
/ 1_000_000.0
|
||||
}
|
||||
async fn chat(&self, _req: ChatRequest) -> Result<ChatResponse> {
|
||||
// TODO(v0.2): extract system from messages → "system" field; messages → content blocks;
|
||||
// add x-api-key + anthropic-version headers; map response → canonical form.
|
||||
// Returns an error (not a panic) so that failover/playground continue to work normally.
|
||||
Err(GatewayError::UpstreamUnavailable(
|
||||
"anthropic provider not yet implemented (planned v0.2)".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
//! Outgoing provider adapters. The [`Provider`] trait is the single call site for LLMs;
|
||||
//! adapters translate the canonical [`ChatRequest`] to and from each provider's API format.
|
||||
//!
|
||||
//! Most local servers (vLLM, llama.cpp, LM Studio, Ollama `/v1`) are OpenAI-compatible
|
||||
//! → [`openai::OpenAiProvider`] is sufficient for all of them.
|
||||
|
||||
pub mod anthropic;
|
||||
pub mod openai;
|
||||
|
||||
use crate::model::{ChatRequest, ChatResponse};
|
||||
use async_trait::async_trait;
|
||||
|
||||
/// Model capabilities — checked at selection time (whether tools/vision are needed).
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct Caps {
|
||||
pub tools: bool,
|
||||
pub vision: bool,
|
||||
pub streaming: bool,
|
||||
}
|
||||
|
||||
impl Caps {
|
||||
pub fn from_list(list: &[String]) -> Self {
|
||||
Self {
|
||||
tools: list.iter().any(|c| c == "tools"),
|
||||
vision: list.iter().any(|c| c == "vision"),
|
||||
streaming: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait Provider: Send + Sync {
|
||||
/// The model_id from config (not the provider's model name).
|
||||
fn id(&self) -> &str;
|
||||
|
||||
fn caps(&self) -> &Caps;
|
||||
|
||||
/// Cost of a request in USD — computed from price_in/price_out × tokens (for billing).
|
||||
fn price(&self, prompt_tokens: u32, completion_tokens: u32) -> f64;
|
||||
|
||||
/// Non-streaming chat call.
|
||||
async fn chat(&self, req: ChatRequest) -> crate::error::Result<ChatResponse>;
|
||||
|
||||
// TODO(v0.1): async fn chat_stream(...) -> Result<ChatStream> (SSE).
|
||||
// TODO(v0.1): async fn embed(...) -> Result<EmbedResponse>.
|
||||
}
|
||||
@@ -0,0 +1,157 @@
|
||||
//! OpenAI-compatible provider. Covers OpenAI, OpenRouter, Together, Groq,
|
||||
//! and local servers such as vLLM / llama.cpp / LM Studio / Ollama (`/v1`).
|
||||
|
||||
use super::{Caps, Provider};
|
||||
use crate::config::ModelCfg;
|
||||
use crate::error::{GatewayError, Result};
|
||||
use crate::model::{ChatRequest, ChatResponse};
|
||||
use async_trait::async_trait;
|
||||
|
||||
pub struct OpenAiProvider {
|
||||
id: String,
|
||||
base_url: String,
|
||||
model: String,
|
||||
api_key: Option<String>,
|
||||
caps: Caps,
|
||||
price_in: f64,
|
||||
price_out: f64,
|
||||
http: reqwest::Client,
|
||||
}
|
||||
|
||||
impl OpenAiProvider {
|
||||
pub fn new(cfg: &ModelCfg, api_key: Option<String>) -> anyhow::Result<Self> {
|
||||
Ok(Self {
|
||||
id: cfg.id.clone(),
|
||||
base_url: cfg.base_url.trim_end_matches('/').to_string(),
|
||||
model: cfg.model.clone(),
|
||||
api_key,
|
||||
caps: Caps::from_list(&cfg.caps),
|
||||
price_in: cfg.price_in,
|
||||
price_out: cfg.price_out,
|
||||
http: reqwest::Client::new(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Canonical request → OpenAI Chat Completions request body.
|
||||
fn to_wire(&self, req: &ChatRequest) -> serde_json::Value {
|
||||
let mut body = serde_json::json!({
|
||||
"model": self.model,
|
||||
"messages": req.messages,
|
||||
"stream": req.stream,
|
||||
});
|
||||
if let Some(t) = req.params.temperature {
|
||||
body["temperature"] = t.into();
|
||||
}
|
||||
if let Some(m) = req.params.max_tokens {
|
||||
body["max_tokens"] = m.into();
|
||||
}
|
||||
if !req.tools.is_empty() {
|
||||
body["tools"] = serde_json::Value::Array(req.tools.clone());
|
||||
}
|
||||
// uninterpreted fields are proxied as-is
|
||||
for (k, v) in &req.params.passthrough {
|
||||
body[k] = v.clone();
|
||||
}
|
||||
body
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Provider for OpenAiProvider {
|
||||
fn id(&self) -> &str {
|
||||
&self.id
|
||||
}
|
||||
fn caps(&self) -> &Caps {
|
||||
&self.caps
|
||||
}
|
||||
fn price(&self, prompt_tokens: u32, completion_tokens: u32) -> f64 {
|
||||
(prompt_tokens as f64 * self.price_in + completion_tokens as f64 * self.price_out)
|
||||
/ 1_000_000.0
|
||||
}
|
||||
|
||||
async fn chat(&self, req: ChatRequest) -> Result<ChatResponse> {
|
||||
let mut r = self
|
||||
.http
|
||||
.post(format!("{}/chat/completions", self.base_url))
|
||||
.json(&self.to_wire(&req));
|
||||
if let Some(key) = &self.api_key {
|
||||
r = r.bearer_auth(key);
|
||||
}
|
||||
let resp = r
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?;
|
||||
if !resp.status().is_success() {
|
||||
let status = resp.status();
|
||||
let err_body = resp.text().await.unwrap_or_default();
|
||||
return Err(GatewayError::UpstreamUnavailable(format!(
|
||||
"provider {} returned HTTP {}: {}",
|
||||
self.id, status, err_body
|
||||
)));
|
||||
}
|
||||
let v: serde_json::Value = resp
|
||||
.json()
|
||||
.await
|
||||
.map_err(|e| GatewayError::Internal(format!("Failed to parse JSON response: {}", e)))?;
|
||||
|
||||
let id = v["id"].as_str().unwrap_or("").to_string();
|
||||
let model_used = v["model"].as_str().unwrap_or(&self.model).to_string();
|
||||
|
||||
let choices_arr = v["choices"].as_array().ok_or_else(|| {
|
||||
GatewayError::Internal(
|
||||
"invalid openai response: choices is missing or not an array".to_string(),
|
||||
)
|
||||
})?;
|
||||
|
||||
let mut choices = Vec::new();
|
||||
for (i, c) in choices_arr.iter().enumerate() {
|
||||
let msg_val = &c["message"];
|
||||
let role = msg_val["role"].as_str().unwrap_or("assistant").to_string();
|
||||
let content = msg_val["content"].as_str().unwrap_or("").to_string();
|
||||
|
||||
let tool_calls = msg_val["tool_calls"]
|
||||
.as_array()
|
||||
.cloned()
|
||||
.unwrap_or_default();
|
||||
|
||||
choices.push(crate::model::Choice {
|
||||
index: c["index"].as_u64().unwrap_or(i as u64) as u32,
|
||||
message: crate::model::Message {
|
||||
role,
|
||||
content,
|
||||
tool_calls,
|
||||
},
|
||||
finish_reason: c["finish_reason"].as_str().unwrap_or("stop").to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
let usage_val = &v["usage"];
|
||||
let prompt_tokens = usage_val["prompt_tokens"].as_u64().unwrap_or(0) as u32;
|
||||
let completion_tokens = usage_val["completion_tokens"].as_u64().unwrap_or(0) as u32;
|
||||
let total_tokens = usage_val["total_tokens"]
|
||||
.as_u64()
|
||||
.unwrap_or((prompt_tokens + completion_tokens) as u64)
|
||||
as u32;
|
||||
|
||||
Ok(ChatResponse {
|
||||
id,
|
||||
model_used,
|
||||
choices,
|
||||
usage: crate::model::Usage {
|
||||
prompt_tokens,
|
||||
completion_tokens,
|
||||
total_tokens,
|
||||
},
|
||||
cortiq: crate::model::RouteInfo {
|
||||
task_label: String::new(),
|
||||
complexity_score: 0.0,
|
||||
complexity_tier: String::new(),
|
||||
selected_model: self.id.clone(),
|
||||
route_source: String::new(),
|
||||
router_request_id: None,
|
||||
cost_usd: 0.0,
|
||||
failover: false,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
//! Model pool registry: builds providers from `[[models]]` and resolves
|
||||
//! `model_id → Arc<dyn Provider>`.
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::providers::{anthropic::AnthropicProvider, openai::OpenAiProvider, Provider};
|
||||
use crate::secrets::SecretStore;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
pub struct Registry {
|
||||
by_id: HashMap<String, Arc<dyn Provider>>,
|
||||
}
|
||||
|
||||
impl Registry {
|
||||
pub fn from_config(cfg: &Config, secrets: &SecretStore) -> anyhow::Result<Self> {
|
||||
let mut by_id: HashMap<String, Arc<dyn Provider>> = HashMap::new();
|
||||
for m in &cfg.models {
|
||||
if m.kind == "embedding" {
|
||||
continue; // embedding models are handled by a separate branch (v0.1)
|
||||
}
|
||||
// provider key: secret store → environment variable
|
||||
let api_key = m
|
||||
.api_key_env
|
||||
.as_ref()
|
||||
.and_then(|name| secrets.resolve(name));
|
||||
let provider: Arc<dyn Provider> = match m.provider.as_str() {
|
||||
"openai" | "ollama" | "http" => Arc::new(OpenAiProvider::new(m, api_key)?),
|
||||
"anthropic" => Arc::new(AnthropicProvider::new(m, api_key)?),
|
||||
other => anyhow::bail!("unknown provider '{}' for model '{}'", other, m.id),
|
||||
};
|
||||
by_id.insert(m.id.clone(), provider);
|
||||
}
|
||||
Ok(Self { by_id })
|
||||
}
|
||||
|
||||
pub fn get(&self, model_id: &str) -> Option<Arc<dyn Provider>> {
|
||||
self.by_id.get(model_id).cloned()
|
||||
}
|
||||
|
||||
pub fn ids(&self) -> impl Iterator<Item = &String> {
|
||||
self.by_id.keys()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
//! Client for cortiq-router. Issues `POST /v1/route` and (optionally) `POST /v1/feedback`.
|
||||
//! Returns the decision as a [`RouteDecision`]. When the router is unavailable the caller
|
||||
//! falls back to `routing.default` (graceful degradation).
|
||||
|
||||
use crate::config::RouterCfg;
|
||||
use crate::model::RouteDecision;
|
||||
|
||||
pub struct RouterClient {
|
||||
http: reqwest::Client,
|
||||
url: String,
|
||||
api_key: Option<String>,
|
||||
taxonomy_id: Option<String>,
|
||||
}
|
||||
|
||||
impl RouterClient {
|
||||
pub fn new(cfg: &RouterCfg) -> anyhow::Result<Self> {
|
||||
let mut builder =
|
||||
reqwest::Client::builder().timeout(std::time::Duration::from_millis(cfg.timeout_ms));
|
||||
|
||||
if !cfg.verify_tls {
|
||||
builder = builder
|
||||
.danger_accept_invalid_certs(true)
|
||||
.danger_accept_invalid_hostnames(true);
|
||||
}
|
||||
|
||||
let http = builder.build()?;
|
||||
let api_key = cfg
|
||||
.api_key_env
|
||||
.as_ref()
|
||||
.and_then(|name| std::env::var(name).ok());
|
||||
Ok(Self {
|
||||
http,
|
||||
url: cfg.url.trim_end_matches('/').to_string(),
|
||||
api_key,
|
||||
taxonomy_id: cfg.taxonomy_id.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Classify text. Returns `Ok(None)` on network error/timeout —
|
||||
/// the caller treats this as "router unavailable" and uses the default tier.
|
||||
pub async fn route(&self, text: &str, profile: &str) -> anyhow::Result<Option<RouteDecision>> {
|
||||
let mut body = serde_json::json!({
|
||||
"input": { "text": text },
|
||||
"options": { "policy_profile": profile, "allow_oracle": true }
|
||||
});
|
||||
if let Some(tax) = &self.taxonomy_id {
|
||||
body["taxonomy_id"] = serde_json::Value::String(tax.clone());
|
||||
}
|
||||
|
||||
let mut req = self.http.post(format!("{}/v1/route", self.url)).json(&body);
|
||||
if let Some(key) = &self.api_key {
|
||||
req = req.bearer_auth(key);
|
||||
}
|
||||
|
||||
let resp = match req.send().await {
|
||||
Ok(r) => r,
|
||||
Err(_) => return Ok(None), // degradation
|
||||
};
|
||||
if !resp.status().is_success() {
|
||||
return Ok(None);
|
||||
}
|
||||
let v: serde_json::Value = resp.json().await?;
|
||||
let d = &v["decision"];
|
||||
Ok(Some(RouteDecision {
|
||||
task_label: d["task_label"]
|
||||
.as_str()
|
||||
.unwrap_or("__unknown__")
|
||||
.to_string(),
|
||||
complexity_score: d["complexity"]["score"].as_f64().unwrap_or(0.5) as f32,
|
||||
complexity_tier: d["complexity"]["tier"]
|
||||
.as_str()
|
||||
.unwrap_or("medium")
|
||||
.to_string(),
|
||||
router_request_id: v["request_id"].as_str().map(|s| s.to_string()),
|
||||
source: "router".into(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
//! Model selection: `(task_label, complexity_tier) → ordered list of model_ids`.
|
||||
//! The list serves as both a priority order and a fallback chain. See docs/ROUTING.md.
|
||||
|
||||
use crate::config::{Config, TierTargets};
|
||||
|
||||
pub struct RoutingTable {
|
||||
tiers: std::collections::HashMap<String, Vec<String>>,
|
||||
default: String,
|
||||
}
|
||||
|
||||
impl RoutingTable {
|
||||
pub fn from_config(cfg: &Config) -> Self {
|
||||
let tiers = cfg
|
||||
.routing
|
||||
.tiers
|
||||
.iter()
|
||||
.map(|(tier, targets)| {
|
||||
let list = match targets {
|
||||
TierTargets::List(v) => v.clone(),
|
||||
TierTargets::One(s) => vec![s.clone()],
|
||||
};
|
||||
(tier.clone(), list)
|
||||
})
|
||||
.collect();
|
||||
Self {
|
||||
tiers,
|
||||
default: cfg.routing.default.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Candidates for the given complexity tier, in preference order.
|
||||
/// If the tier is not defined, the sole candidate is `default`.
|
||||
pub fn candidates(&self, tier: &str) -> Vec<String> {
|
||||
self.tiers
|
||||
.get(tier)
|
||||
.cloned()
|
||||
.unwrap_or_else(|| vec![self.default.clone()])
|
||||
}
|
||||
|
||||
/// Default model (used when the router is unavailable).
|
||||
pub fn default_model(&self) -> &str {
|
||||
&self.default
|
||||
}
|
||||
}
|
||||
|
||||
// TODO(v0.2): cost_aware mode and circuit breaker — pick the cheapest available
|
||||
// model no lower than min_class[tier], respecting max_cost_usd_per_request,
|
||||
// and skip models whose circuit breaker is open.
|
||||
@@ -0,0 +1,96 @@
|
||||
//! Secret store (provider API key values) kept outside the main config.
|
||||
//!
|
||||
//! The file `config/secrets.toml` (in `.gitignore`, mode `0600`) is a map of
|
||||
//! `env_name -> value`. Key resolution order: store entry > environment variable.
|
||||
//! This allows setting keys directly from the admin panel without a restart,
|
||||
//! without writing secrets into `gateway.toml`. Values are never exposed via the
|
||||
//! API — only their presence status is returned (see `source`).
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::RwLock;
|
||||
|
||||
pub struct SecretStore {
|
||||
path: String,
|
||||
values: RwLock<BTreeMap<String, String>>,
|
||||
}
|
||||
|
||||
impl SecretStore {
|
||||
/// Load the store from a file (missing file or parse error → empty store).
|
||||
pub fn load(path: &str) -> Self {
|
||||
let values = std::fs::read_to_string(path)
|
||||
.ok()
|
||||
.and_then(|raw| toml::from_str::<BTreeMap<String, String>>(&raw).ok())
|
||||
.unwrap_or_default();
|
||||
Self {
|
||||
path: path.to_string(),
|
||||
values: RwLock::new(values),
|
||||
}
|
||||
}
|
||||
|
||||
/// Secret value by env name: store first, then environment variable.
|
||||
pub fn resolve(&self, env_name: &str) -> Option<String> {
|
||||
if let Some(v) = self.values.read().unwrap().get(env_name) {
|
||||
if !v.is_empty() {
|
||||
return Some(v.clone());
|
||||
}
|
||||
}
|
||||
std::env::var(env_name).ok().filter(|v| !v.is_empty())
|
||||
}
|
||||
|
||||
/// Value source for display in the UI: `store` | `env` | `missing`.
|
||||
pub fn source(&self, env_name: &str) -> &'static str {
|
||||
let in_store = self
|
||||
.values
|
||||
.read()
|
||||
.unwrap()
|
||||
.get(env_name)
|
||||
.map(|v| !v.is_empty())
|
||||
.unwrap_or(false);
|
||||
if in_store {
|
||||
"store"
|
||||
} else if std::env::var(env_name)
|
||||
.ok()
|
||||
.filter(|v| !v.is_empty())
|
||||
.is_some()
|
||||
{
|
||||
"env"
|
||||
} else {
|
||||
"missing"
|
||||
}
|
||||
}
|
||||
|
||||
/// Set a secret (persisted to file with mode 0600).
|
||||
pub fn set(&self, env_name: &str, value: &str) -> anyhow::Result<()> {
|
||||
self.values
|
||||
.write()
|
||||
.unwrap()
|
||||
.insert(env_name.to_string(), value.to_string());
|
||||
self.persist()
|
||||
}
|
||||
|
||||
/// Remove a secret from the store (an environment variable value, if present, remains).
|
||||
pub fn clear(&self, env_name: &str) -> anyhow::Result<()> {
|
||||
self.values.write().unwrap().remove(env_name);
|
||||
self.persist()
|
||||
}
|
||||
|
||||
fn persist(&self) -> anyhow::Result<()> {
|
||||
let data = {
|
||||
let map = self.values.read().unwrap();
|
||||
toml::to_string(&*map)?
|
||||
};
|
||||
let p = std::path::Path::new(&self.path);
|
||||
if let Some(dir) = p.parent() {
|
||||
std::fs::create_dir_all(dir).ok();
|
||||
}
|
||||
let tmp = p.with_extension("toml.tmp");
|
||||
std::fs::write(&tmp, data.as_bytes())?;
|
||||
std::fs::rename(&tmp, p)?;
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
let _ = std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
+107
@@ -0,0 +1,107 @@
|
||||
//! Shared gateway state accessible across all request handlers.
|
||||
//!
|
||||
//! The config-derived part (`Live`) lives behind an `ArcSwap`, which allows
|
||||
//! hot-reloading the model pool, routing table, and router connection without
|
||||
//! a restart: the admin API validates the new config, builds a new `Live`,
|
||||
//! atomically writes the TOML, and swaps the pointer.
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::registry::Registry;
|
||||
use crate::router_client::RouterClient;
|
||||
use crate::routing::RoutingTable;
|
||||
use crate::secrets::SecretStore;
|
||||
use crate::stats::Stats;
|
||||
use arc_swap::ArcSwap;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct SharedState(pub Arc<AppState>);
|
||||
|
||||
/// The portion of state that is rebuilt whenever the config changes.
|
||||
pub struct Live {
|
||||
pub cfg: Config,
|
||||
pub registry: Registry,
|
||||
pub routing: RoutingTable,
|
||||
pub router: RouterClient,
|
||||
}
|
||||
|
||||
impl Live {
|
||||
pub fn build(cfg: Config, secrets: &SecretStore) -> anyhow::Result<Self> {
|
||||
let router = RouterClient::new(&cfg.router)?;
|
||||
let registry = Registry::from_config(&cfg, secrets)?;
|
||||
let routing = RoutingTable::from_config(&cfg);
|
||||
Ok(Self {
|
||||
cfg,
|
||||
registry,
|
||||
routing,
|
||||
router,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub struct AppState {
|
||||
pub live: ArcSwap<Live>,
|
||||
pub stats: Arc<Stats>,
|
||||
pub secrets: SecretStore,
|
||||
pub config_path: String,
|
||||
pub pipeline: crate::pipeline::Pipeline,
|
||||
}
|
||||
|
||||
impl std::ops::Deref for SharedState {
|
||||
type Target = AppState;
|
||||
fn deref(&self) -> &Self::Target {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
impl SharedState {
|
||||
pub fn build(cfg: Config, config_path: String) -> anyhow::Result<Self> {
|
||||
let secrets_path = sibling(&config_path, "secrets.toml");
|
||||
let secrets = SecretStore::load(&secrets_path);
|
||||
let stats = Stats::new(&cfg.stats);
|
||||
let live = Live::build(cfg, &secrets)?;
|
||||
Ok(SharedState(Arc::new(AppState {
|
||||
live: ArcSwap::from_pointee(live),
|
||||
stats,
|
||||
secrets,
|
||||
config_path,
|
||||
pipeline: crate::pipeline::Pipeline::new(),
|
||||
})))
|
||||
}
|
||||
|
||||
/// Current snapshot of the rebuildable state (safe to hold across await points).
|
||||
pub fn live(&self) -> Arc<Live> {
|
||||
self.0.live.load_full()
|
||||
}
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
/// Replace the config entirely: validate → build → write to disk → swap.
|
||||
/// If any step fails, the active state and file are left unchanged.
|
||||
pub fn reload(&self, new_cfg: Config) -> anyhow::Result<()> {
|
||||
new_cfg.validate()?;
|
||||
let live = Live::build(new_cfg.clone(), &self.secrets)?;
|
||||
new_cfg.save(&self.config_path)?;
|
||||
self.live.store(Arc::new(live));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Rebuild `Live` from the current config without writing to disk
|
||||
/// (e.g. after a secret change — only provider keys changed).
|
||||
pub fn rebuild(&self) -> anyhow::Result<()> {
|
||||
let cfg = self.live.load_full().cfg.clone();
|
||||
let live = Live::build(cfg, &self.secrets)?;
|
||||
self.live.store(Arc::new(live));
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Path to file `name` next to `path` (in the same directory).
|
||||
fn sibling(path: &str, name: &str) -> String {
|
||||
std::path::Path::new(path)
|
||||
.parent()
|
||||
.map(|d| d.join(name))
|
||||
.unwrap_or_else(|| std::path::PathBuf::from(name))
|
||||
.to_string_lossy()
|
||||
.into_owned()
|
||||
}
|
||||
+395
@@ -0,0 +1,395 @@
|
||||
//! Request statistics tracking for the dashboard and `GET /metrics`.
|
||||
//!
|
||||
//! Everything is held in memory (aggregates + per-minute buckets for the retention
|
||||
//! window + a ring buffer of recent requests) and optionally appended to a JSONL
|
||||
//! file that is replayed on startup — so statistics survive a restart with no
|
||||
//! database dependency.
|
||||
|
||||
use crate::config::StatsCfg;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::io::Write;
|
||||
use std::sync::Mutex;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
fn now_secs() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs()
|
||||
}
|
||||
|
||||
/// A single record for a completed request (one JSONL line and one "recent" ring entry).
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct RequestRecord {
|
||||
pub ts: u64,
|
||||
#[serde(default)]
|
||||
pub account: String,
|
||||
#[serde(default)]
|
||||
pub protocol: String,
|
||||
#[serde(default)]
|
||||
pub directive: String, // auto | pinned
|
||||
#[serde(default)]
|
||||
pub task_label: String,
|
||||
#[serde(default)]
|
||||
pub tier: String,
|
||||
#[serde(default)]
|
||||
pub score: f32,
|
||||
#[serde(default)]
|
||||
pub model_id: String,
|
||||
#[serde(default)]
|
||||
pub route_source: String,
|
||||
#[serde(default)]
|
||||
pub prompt_tokens: u32,
|
||||
#[serde(default)]
|
||||
pub completion_tokens: u32,
|
||||
#[serde(default)]
|
||||
pub cost_usd: f64,
|
||||
#[serde(default)]
|
||||
pub latency_ms: u64,
|
||||
#[serde(default)]
|
||||
pub outcome: String, // ok | error
|
||||
#[serde(default)]
|
||||
pub failover: bool,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub error: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, Serialize)]
|
||||
pub struct Totals {
|
||||
pub requests: u64,
|
||||
pub ok: u64,
|
||||
pub errors: u64,
|
||||
pub failovers: u64,
|
||||
pub prompt_tokens: u64,
|
||||
pub completion_tokens: u64,
|
||||
pub cost_usd: f64,
|
||||
pub latency_ms_sum: u64,
|
||||
}
|
||||
|
||||
impl Totals {
|
||||
fn apply(&mut self, r: &RequestRecord) {
|
||||
self.requests += 1;
|
||||
if r.outcome == "ok" {
|
||||
self.ok += 1;
|
||||
} else {
|
||||
self.errors += 1;
|
||||
}
|
||||
if r.failover {
|
||||
self.failovers += 1;
|
||||
}
|
||||
self.prompt_tokens += r.prompt_tokens as u64;
|
||||
self.completion_tokens += r.completion_tokens as u64;
|
||||
self.cost_usd += r.cost_usd;
|
||||
self.latency_ms_sum += r.latency_ms;
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-minute bucket for time-series charts.
|
||||
#[derive(Clone, Debug, Default, Serialize)]
|
||||
pub struct Bucket {
|
||||
pub minute: u64, // unix seconds, rounded down to the minute
|
||||
pub requests: u64,
|
||||
pub errors: u64,
|
||||
pub prompt_tokens: u64,
|
||||
pub completion_tokens: u64,
|
||||
pub cost_usd: f64,
|
||||
pub latency_ms_sum: u64,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct Inner {
|
||||
total: Totals,
|
||||
by_model: HashMap<String, Totals>,
|
||||
by_account: HashMap<String, Totals>,
|
||||
by_tier: HashMap<String, Totals>,
|
||||
by_label: HashMap<String, Totals>,
|
||||
buckets: VecDeque<Bucket>,
|
||||
recent: VecDeque<RequestRecord>,
|
||||
}
|
||||
|
||||
pub struct Stats {
|
||||
enabled: bool,
|
||||
file: Option<String>,
|
||||
ring_size: usize,
|
||||
retention_secs: u64,
|
||||
inner: Mutex<Inner>,
|
||||
}
|
||||
|
||||
impl Stats {
|
||||
pub fn new(cfg: &StatsCfg) -> std::sync::Arc<Self> {
|
||||
let file = if cfg.file.trim().is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(cfg.file.clone())
|
||||
};
|
||||
let s = Stats {
|
||||
enabled: cfg.enabled,
|
||||
file,
|
||||
ring_size: cfg.ring_size.max(1),
|
||||
retention_secs: parse_duration_secs(&cfg.retention).unwrap_or(7 * 86_400),
|
||||
inner: Mutex::new(Inner::default()),
|
||||
};
|
||||
s.replay();
|
||||
std::sync::Arc::new(s)
|
||||
}
|
||||
|
||||
/// Replay the JSONL file into aggregates (within the retention window only).
|
||||
fn replay(&self) {
|
||||
let Some(path) = &self.file else { return };
|
||||
let Ok(content) = std::fs::read_to_string(path) else {
|
||||
return;
|
||||
};
|
||||
let cutoff = now_secs().saturating_sub(self.retention_secs);
|
||||
let mut inner = self.inner.lock().unwrap();
|
||||
for line in content.lines() {
|
||||
let line = line.trim();
|
||||
if line.is_empty() {
|
||||
continue;
|
||||
}
|
||||
if let Ok(rec) = serde_json::from_str::<RequestRecord>(line) {
|
||||
// push everything to the "recent" ring (it self-limits by size),
|
||||
// but apply to aggregates/series only entries within the window.
|
||||
Self::push_recent(&mut inner, &rec, self.ring_size);
|
||||
if rec.ts >= cutoff {
|
||||
Self::apply_aggregates(&mut inner, &rec, self.retention_secs);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn push_recent(inner: &mut Inner, rec: &RequestRecord, ring_size: usize) {
|
||||
inner.recent.push_back(rec.clone());
|
||||
while inner.recent.len() > ring_size {
|
||||
inner.recent.pop_front();
|
||||
}
|
||||
}
|
||||
|
||||
fn apply_aggregates(inner: &mut Inner, rec: &RequestRecord, retention_secs: u64) {
|
||||
inner.total.apply(rec);
|
||||
inner
|
||||
.by_model
|
||||
.entry(rec.model_id.clone())
|
||||
.or_default()
|
||||
.apply(rec);
|
||||
inner
|
||||
.by_account
|
||||
.entry(if rec.account.is_empty() {
|
||||
"anonymous".to_string()
|
||||
} else {
|
||||
rec.account.clone()
|
||||
})
|
||||
.or_default()
|
||||
.apply(rec);
|
||||
inner
|
||||
.by_tier
|
||||
.entry(rec.tier.clone())
|
||||
.or_default()
|
||||
.apply(rec);
|
||||
inner
|
||||
.by_label
|
||||
.entry(rec.task_label.clone())
|
||||
.or_default()
|
||||
.apply(rec);
|
||||
|
||||
let minute = rec.ts - (rec.ts % 60);
|
||||
match inner.buckets.back_mut() {
|
||||
Some(b) if b.minute == minute => fill_bucket(b, rec),
|
||||
Some(b) if b.minute > minute => { /* arrived older than the tail — ignore for series */
|
||||
}
|
||||
_ => {
|
||||
let mut b = Bucket {
|
||||
minute,
|
||||
..Default::default()
|
||||
};
|
||||
fill_bucket(&mut b, rec);
|
||||
inner.buckets.push_back(b);
|
||||
}
|
||||
}
|
||||
// trim old buckets outside the retention window
|
||||
let cutoff = now_secs().saturating_sub(retention_secs);
|
||||
while let Some(front) = inner.buckets.front() {
|
||||
if front.minute < cutoff {
|
||||
inner.buckets.pop_front();
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Record a new request event.
|
||||
pub fn record(&self, rec: RequestRecord) {
|
||||
if !self.enabled {
|
||||
return;
|
||||
}
|
||||
{
|
||||
let mut inner = self.inner.lock().unwrap();
|
||||
Self::apply_aggregates(&mut inner, &rec, self.retention_secs);
|
||||
Self::push_recent(&mut inner, &rec, self.ring_size);
|
||||
}
|
||||
if let Some(path) = &self.file {
|
||||
if let Ok(line) = serde_json::to_string(&rec) {
|
||||
if let Ok(mut f) = std::fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.append(true)
|
||||
.open(path)
|
||||
{
|
||||
let _ = writeln!(f, "{line}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Snapshot for the admin API: aggregates + time series for `range_secs` + breakdown by `groupby`.
|
||||
pub fn snapshot(&self, range_secs: u64, groupby: &str) -> serde_json::Value {
|
||||
let inner = self.inner.lock().unwrap();
|
||||
let cutoff = now_secs().saturating_sub(range_secs);
|
||||
let series: Vec<&Bucket> = inner
|
||||
.buckets
|
||||
.iter()
|
||||
.filter(|b| b.minute >= cutoff)
|
||||
.collect();
|
||||
|
||||
let breakdown_src = match groupby {
|
||||
"account" => &inner.by_account,
|
||||
"tier" => &inner.by_tier,
|
||||
"label" => &inner.by_label,
|
||||
_ => &inner.by_model,
|
||||
};
|
||||
let mut breakdown: Vec<serde_json::Value> = breakdown_src
|
||||
.iter()
|
||||
.map(|(k, t)| {
|
||||
serde_json::json!({
|
||||
"key": k,
|
||||
"requests": t.requests,
|
||||
"ok": t.ok,
|
||||
"errors": t.errors,
|
||||
"prompt_tokens": t.prompt_tokens,
|
||||
"completion_tokens": t.completion_tokens,
|
||||
"cost_usd": t.cost_usd,
|
||||
"avg_latency_ms": avg(t.latency_ms_sum, t.requests),
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
breakdown.sort_by(|a, b| {
|
||||
b["requests"]
|
||||
.as_u64()
|
||||
.unwrap_or(0)
|
||||
.cmp(&a["requests"].as_u64().unwrap_or(0))
|
||||
});
|
||||
|
||||
serde_json::json!({
|
||||
"totals": {
|
||||
"requests": inner.total.requests,
|
||||
"ok": inner.total.ok,
|
||||
"errors": inner.total.errors,
|
||||
"failovers": inner.total.failovers,
|
||||
"prompt_tokens": inner.total.prompt_tokens,
|
||||
"completion_tokens": inner.total.completion_tokens,
|
||||
"total_tokens": inner.total.prompt_tokens + inner.total.completion_tokens,
|
||||
"cost_usd": inner.total.cost_usd,
|
||||
"avg_latency_ms": avg(inner.total.latency_ms_sum, inner.total.requests),
|
||||
"success_rate": if inner.total.requests > 0 {
|
||||
inner.total.ok as f64 / inner.total.requests as f64
|
||||
} else { 0.0 },
|
||||
},
|
||||
"groupby": groupby,
|
||||
"breakdown": breakdown,
|
||||
"series": series,
|
||||
})
|
||||
}
|
||||
|
||||
/// Recent requests (newest first), with pagination.
|
||||
pub fn recent(&self, limit: usize, offset: usize) -> Vec<RequestRecord> {
|
||||
let inner = self.inner.lock().unwrap();
|
||||
inner
|
||||
.recent
|
||||
.iter()
|
||||
.rev()
|
||||
.skip(offset)
|
||||
.take(limit)
|
||||
.cloned()
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Metrics text in Prometheus format.
|
||||
pub fn prometheus(&self) -> String {
|
||||
let inner = self.inner.lock().unwrap();
|
||||
let mut out = String::new();
|
||||
let t = &inner.total;
|
||||
out.push_str("# HELP gw_requests_total Total gateway requests.\n");
|
||||
out.push_str("# TYPE gw_requests_total counter\n");
|
||||
out.push_str(&format!("gw_requests_total {}\n", t.requests));
|
||||
out.push_str("# HELP gw_failovers_total Total provider failovers.\n");
|
||||
out.push_str("# TYPE gw_failovers_total counter\n");
|
||||
out.push_str(&format!("gw_failovers_total {}\n", t.failovers));
|
||||
out.push_str("# HELP gw_tokens_total Total tokens by direction.\n");
|
||||
out.push_str("# TYPE gw_tokens_total counter\n");
|
||||
out.push_str(&format!(
|
||||
"gw_tokens_total{{direction=\"in\"}} {}\n",
|
||||
t.prompt_tokens
|
||||
));
|
||||
out.push_str(&format!(
|
||||
"gw_tokens_total{{direction=\"out\"}} {}\n",
|
||||
t.completion_tokens
|
||||
));
|
||||
out.push_str("# HELP gw_cost_usd_total Total estimated cost in USD.\n");
|
||||
out.push_str("# TYPE gw_cost_usd_total counter\n");
|
||||
out.push_str(&format!("gw_cost_usd_total {:.6}\n", t.cost_usd));
|
||||
out.push_str("# HELP gw_provider_calls_total Calls by model and outcome.\n");
|
||||
out.push_str("# TYPE gw_provider_calls_total counter\n");
|
||||
for (model, mt) in &inner.by_model {
|
||||
let model = escape_label(model);
|
||||
out.push_str(&format!(
|
||||
"gw_provider_calls_total{{model_id=\"{model}\",outcome=\"ok\"}} {}\n",
|
||||
mt.ok
|
||||
));
|
||||
out.push_str(&format!(
|
||||
"gw_provider_calls_total{{model_id=\"{model}\",outcome=\"error\"}} {}\n",
|
||||
mt.errors
|
||||
));
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
fn fill_bucket(b: &mut Bucket, r: &RequestRecord) {
|
||||
b.requests += 1;
|
||||
if r.outcome != "ok" {
|
||||
b.errors += 1;
|
||||
}
|
||||
b.prompt_tokens += r.prompt_tokens as u64;
|
||||
b.completion_tokens += r.completion_tokens as u64;
|
||||
b.cost_usd += r.cost_usd;
|
||||
b.latency_ms_sum += r.latency_ms;
|
||||
}
|
||||
|
||||
fn avg(sum: u64, n: u64) -> u64 {
|
||||
sum.checked_div(n).unwrap_or(0)
|
||||
}
|
||||
|
||||
fn escape_label(s: &str) -> String {
|
||||
s.replace('\\', "\\\\")
|
||||
.replace('"', "\\\"")
|
||||
.replace('\n', " ")
|
||||
}
|
||||
|
||||
/// Parse a duration string of the form `7d` / `24h` / `90m` / `3600s` into seconds.
|
||||
pub fn parse_duration_secs(s: &str) -> Option<u64> {
|
||||
let s = s.trim();
|
||||
if s.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let (num, mult) = if let Some(n) = s.strip_suffix('d') {
|
||||
(n, 86_400)
|
||||
} else if let Some(n) = s.strip_suffix('h') {
|
||||
(n, 3_600)
|
||||
} else if let Some(n) = s.strip_suffix('m') {
|
||||
(n, 60)
|
||||
} else if let Some(n) = s.strip_suffix('s') {
|
||||
(n, 1)
|
||||
} else {
|
||||
(s, 1)
|
||||
};
|
||||
num.trim().parse::<u64>().ok().map(|v| v * mult)
|
||||
}
|
||||
Reference in New Issue
Block a user