diff --git a/control-plane/web/src/App.tsx b/control-plane/web/src/App.tsx index d2d71c7..38950fb 100644 --- a/control-plane/web/src/App.tsx +++ b/control-plane/web/src/App.tsx @@ -27,6 +27,27 @@ export function useAuth(): AuthState { return value; } +interface NavItem { + to: string; + label: string; + icon: string; + minRole: Role; + page: ReactNode; + end?: boolean; +} + +const roleRank: Record = { member: 0, tenant_admin: 1, system_admin: 2 }; + +const navItems: NavItem[] = [ + { to: "/", label: "Overview", icon: "◫", end: true, minRole: "member", page: }, + { to: "/usage", label: "Usage & cost", icon: "⌁", minRole: "member", page: }, + { to: "/availability", label: "Availability", icon: "◉", minRole: "member", page: }, + { to: "/keys", label: "Access keys", icon: "⌘", minRole: "tenant_admin", page: }, + { to: "/audit", label: "Audit", icon: "≣", minRole: "tenant_admin", page: }, + { to: "/users", label: "Users & roles", icon: "♙", minRole: "system_admin", page: }, + { to: "/configuration", label: "Configuration", icon: "⌗", minRole: "system_admin", page: }, +]; + export default function App() { const [session, updateSession] = useState(null); const [loading, setLoading] = useState(true); @@ -69,27 +90,6 @@ export default function App() { ); } -interface NavItem { - to: string; - label: string; - icon: string; - minRole: Role; - page: ReactNode; - end?: boolean; -} - -const roleRank: Record = { member: 0, tenant_admin: 1, system_admin: 2 }; - -const navItems: NavItem[] = [ - { to: "/", label: "Overview", icon: "◫", end: true, minRole: "member", page: }, - { to: "/usage", label: "Usage & cost", icon: "⌁", minRole: "member", page: }, - { to: "/availability", label: "Availability", icon: "◉", minRole: "member", page: }, - { to: "/keys", label: "Access keys", icon: "⌘", minRole: "tenant_admin", page: }, - { to: "/audit", label: "Audit", icon: "≣", minRole: "tenant_admin", page: }, - { to: "/users", label: "Users & roles", icon: "♙", minRole: "system_admin", page: }, - { to: "/configuration", label: "Configuration", icon: "⌗", minRole: "system_admin", page: }, -]; - function Shell() { const { session, logout } = useAuth(); const role = session.user.role; diff --git a/control-plane/web/src/pages/ConfigPage.tsx b/control-plane/web/src/pages/ConfigPage.tsx index 1f7bcd5..94470a7 100644 --- a/control-plane/web/src/pages/ConfigPage.tsx +++ b/control-plane/web/src/pages/ConfigPage.tsx @@ -5,13 +5,9 @@ import { dateTime } from "../format"; import { useAPI, useAction } from "../hooks"; import type { ConfigDocument, ConfigVersion } from "../types"; -interface VersionList { - versions: ConfigVersion[]; -} - export default function ConfigPage() { const current = useAPI("/api/v1/admin/config"); - const history = useAPI("/api/v1/admin/config/versions"); + const history = useAPI<{ versions: ConfigVersion[] }>("/api/v1/admin/config/versions"); const [yaml, setYAML] = useState(""); const [message, setMessage] = useState(""); const { run, busy, error } = useAction(); diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index b2294c5..86e8f74 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -347,6 +347,19 @@ pub struct StabilityConf { pub availability_min_samples: u64, } +impl Default for StabilityConf { + fn default() -> Self { + Self { + failure_threshold: default_failure_threshold(), + cooldown_seconds: default_cooldown_seconds(), + availability_window_minutes: default_availability_window_minutes(), + unstable_error_rate: default_unstable_error_rate(), + unavailable_error_rate: default_unavailable_error_rate(), + availability_min_samples: default_availability_min_samples(), + } + } +} + fn default_failure_threshold() -> usize { 3 } @@ -366,19 +379,6 @@ fn default_cooldown_seconds() -> u64 { 30 } -impl Default for StabilityConf { - fn default() -> Self { - Self { - failure_threshold: default_failure_threshold(), - cooldown_seconds: default_cooldown_seconds(), - availability_window_minutes: default_availability_window_minutes(), - unstable_error_rate: default_unstable_error_rate(), - unavailable_error_rate: default_unavailable_error_rate(), - availability_min_samples: default_availability_min_samples(), - } - } -} - /// One automatic-suspension tier: this many admission rejections in a day /// suspend the key for `suspend_hours`. #[derive(Debug, Clone, Deserialize)] @@ -754,7 +754,6 @@ impl GatewayConfig { protocols: preset.wires.iter().map(|w| (*w).to_owned()).collect(), }); } - // normalize the global policy and every tenant override once at load compile_security(&mut self.security); for t in &mut self.tenants { if let Some(sec) = t.security.as_mut() { diff --git a/crates/dag/src/nodes.rs b/crates/dag/src/nodes.rs index 1e085c8..7c9d500 100644 --- a/crates/dag/src/nodes.rs +++ b/crates/dag/src/nodes.rs @@ -208,17 +208,18 @@ impl DagNode for CacheLookup { &["variant_select"] } async fn execute(&self, ctx: &mut DagContext) -> GResult<()> { - let param = match ctx.request.model_param_v2.as_ref() { - Some(p) => p, - None => return Ok(()), + let Some(param) = ctx.request.model_param_v2.as_ref() else { + return Ok(()); }; // online non-streaming cache_ttl models only; batch items bypass — // a hit is free (unbilled) and batches promise per-item billing - let ttl = ctx + let Some(ttl) = ctx .cfg .find_model(¶m.model_name) - .and_then(|m| m.cache_ttl_seconds); - let Some(ttl) = ttl else { return Ok(()) }; + .and_then(|m| m.cache_ttl_seconds) + else { + return Ok(()); + }; if ctx.request.stream || !ctx.request.is_online { return Ok(()); } diff --git a/crates/engines/src/bespoke.rs b/crates/engines/src/bespoke.rs index 1854767..eb49433 100644 --- a/crates/engines/src/bespoke.rs +++ b/crates/engines/src/bespoke.rs @@ -494,11 +494,13 @@ fn dashscope_raw_usage(resp: &GatewayResponse) -> Vec { #[cfg(test)] mod tests { - use super::*; - use crate::transport::{MockTransport, SharedTransport}; + use std::sync::Arc; + use gw_consts::Protocol; use gw_models::{ChatMsg, GatewayRequest, ModelParamV2}; - use std::sync::Arc; + + use super::*; + use crate::transport::{MockTransport, SharedTransport}; fn req(mt: Protocol, name: &str) -> GatewayRequest { GatewayRequest { diff --git a/crates/engines/src/families.rs b/crates/engines/src/families.rs index 8426af3..6711f8a 100644 --- a/crates/engines/src/families.rs +++ b/crates/engines/src/families.rs @@ -818,7 +818,10 @@ impl ResponsesEngine { } else { Some(Value::Array(tool_calls)) }, - model: v["model"].as_str().unwrap_or(&self.model_name()).to_owned(), + model: v["model"] + .as_str() + .map(str::to_owned) + .unwrap_or_else(|| self.model_name()), finish_reason: v["status"].as_str().unwrap_or("completed").to_owned(), prompt_tokens: input, completion_tokens: output, diff --git a/crates/engines/src/http_transport.rs b/crates/engines/src/http_transport.rs index 657380a..4dd21b3 100644 --- a/crates/engines/src/http_transport.rs +++ b/crates/engines/src/http_transport.rs @@ -83,13 +83,6 @@ impl HttpTransport { let p = self.policies.load(); p.per_account.get(account).copied().unwrap_or(p.default) } - - fn set_policies(&self, default: UpstreamPolicy, per_account: HashMap) { - self.policies.store(std::sync::Arc::new(Policies { - default, - per_account, - })); - } } #[async_trait::async_trait] @@ -99,7 +92,10 @@ impl Transport for HttpTransport { default: UpstreamPolicy, per_account: HashMap, ) { - self.set_policies(default, per_account); + self.policies.store(std::sync::Arc::new(Policies { + default, + per_account, + })); } async fn send(&self, req: UpstreamRequest) -> GResult { diff --git a/crates/handler/src/lib.rs b/crates/handler/src/lib.rs index d783cc3..d10cf88 100644 --- a/crates/handler/src/lib.rs +++ b/crates/handler/src/lib.rs @@ -268,11 +268,10 @@ impl OnlineHandler { Ok(ctx) } - /// Run the wired moderator over raw text; `Some(reason)` to deny, `None` to - /// allow. The seam the realtime surface uses (it has no `DagContext`); the - /// caller records the security event on its own surface. - /// [`Self::moderation`] narrowed to the realtime surface: a live session - /// can't switch models, so `Degrade` denies there. + /// Run the wired moderator over raw text; the seam the realtime surface + /// uses (it has no `DagContext`), so the caller records the security event + /// on its own surface. [`Self::moderation`] narrowed to this surface: a + /// live session can't switch models, so `Degrade` denies here instead. pub async fn moderate_rt(&self, sec: &gw_config::SecurityConf, text: &str) -> RtModeration { match self.moderation(sec, text).await { Moderation::Allow => RtModeration::Allow, @@ -349,10 +348,7 @@ pub enum RtModeration { /// A per-request correlation id: `req--`, time-sortable and /// unique within the process (the seq disambiguates same-millisecond requests). pub fn new_request_id() -> String { - let ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis()) - .unwrap_or(0); + let ms = gw_state::epoch_millis(); format!("req-{ms}-{}", REQ_SEQ.fetch_add(1, Ordering::Relaxed)) } diff --git a/crates/handler/src/offline.rs b/crates/handler/src/offline.rs index 25871da..1804a20 100644 --- a/crates/handler/src/offline.rs +++ b/crates/handler/src/offline.rs @@ -49,7 +49,7 @@ impl OfflineHandler { .batch_create(&ak.ak, &ak.tenant, &model, items.len()) .await?; let this = self.clone(); - let (id, model) = (job.id.clone(), model); + let id = job.id.clone(); // items are captured HERE: an erasure landing after this instant // must stop them, so the marker comparison point is submission, // not the spawned executor's first poll diff --git a/crates/handler/src/plugins.rs b/crates/handler/src/plugins.rs index d0b21c3..4febf4b 100644 --- a/crates/handler/src/plugins.rs +++ b/crates/handler/src/plugins.rs @@ -33,13 +33,7 @@ pub fn security_check(sec: &SecurityConf, request: &mut GatewayRequest) -> ScanO return ScanOutcome::default(); } let mut counts = ScanCounts::new(sec); - let mut visit = |s: &mut String| counts.visit(s); - for msg in &mut request.message { - for_each_message_text(msg, &mut visit); - } - if let Some(param) = request.model_param_v2.as_mut() { - for_each_param_text(param, &mut visit); - } + for_each_request_text(request, &mut |s| counts.visit(s)); counts.outcome() } @@ -98,22 +92,16 @@ impl<'a> ScanCounts<'a> { /// All inbound text a request carries — message content, multimodal text /// parts, assistant tool_calls, the Responses raw body, and the family typed -/// params — collected via the same traversals the blocklist scan and DLP run, +/// params — collected via the same traversal the blocklist scan and DLP run, /// so the field lists cannot drift apart. The one text view moderation and -/// content retention operate on. `&mut` only to share those traversals; -/// nothing is rewritten. +/// content retention operate on. `&mut` only to share that traversal; nothing +/// is rewritten. pub fn inbound_text(request: &mut GatewayRequest) -> String { let mut out = String::new(); - let mut collect = |s: &mut String| { + for_each_request_text(request, &mut |s| { push_text(&mut out, s); 0 - }; - for m in &mut request.message { - for_each_message_text(m, &mut collect); - } - if let Some(param) = request.model_param_v2.as_mut() { - for_each_param_text(param, &mut collect); - } + }); out } @@ -130,12 +118,7 @@ fn push_text(out: &mut String, s: &str) { /// the request's text slots. Returns the number of replaced ranges. pub fn apply_mask_spans(request: &mut GatewayRequest, spans: &[std::ops::Range]) -> usize { let mut masker = SpanMasker::new(spans); - for msg in &mut request.message { - for_each_message_text(msg, &mut |s| masker.apply(s)); - } - if let Some(param) = request.model_param_v2.as_mut() { - for_each_param_text(param, &mut |s| masker.apply(s)); - } + for_each_request_text(request, &mut |s| masker.apply(s)); masker.hits } @@ -252,6 +235,24 @@ fn walk_json_strings(v: &mut serde_json::Value, f: &mut impl FnMut(&mut String) } } +/// Every text-bearing field of a whole request — each message plus the tail +/// params — the ONE per-request traversal the blocklist scan, inbound-text +/// view, mask application, and DLP redaction all share, so they can't drift +/// apart field by field. +fn for_each_request_text( + request: &mut GatewayRequest, + f: &mut impl FnMut(&mut String) -> usize, +) -> usize { + let mut n = 0; + for msg in &mut request.message { + n += for_each_message_text(msg, f); + } + if let Some(param) = request.model_param_v2.as_mut() { + n += for_each_param_text(param, f); + } + n +} + /// The text-bearing fields of one chat turn — flat content, multimodal parts, /// and assistant tool_calls, all forwarded to the vendor by the engines — the /// ONE per-message field list the blocklist scan, DLP, and the moderation @@ -431,17 +432,8 @@ pub fn dlp_redact_request(sec: &SecurityConf, request: &mut GatewayRequest) -> u if !sec.dlp_redact && !sec.detect_secrets { return 0; } - let secrets = sec.detect_secrets; - let pii = sec.dlp_redact; - let mut redact_field = |s: &mut String| redact_in_place(s, pii, secrets); - let mut hits = 0; - for msg in &mut request.message { - hits += for_each_message_text(msg, &mut redact_field); - } - if let Some(param) = request.model_param_v2.as_mut() { - hits += for_each_param_text(param, &mut redact_field); - } - hits + let (pii, secrets) = (sec.dlp_redact, sec.detect_secrets); + for_each_request_text(request, &mut |s| redact_in_place(s, pii, secrets)) } /// A privacy-safe copy of `text` for content retention: PII and secrets are diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index eafc984..a53cf08 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -223,18 +223,18 @@ fn select_transport() -> anyhow::Result { Ok(match env::var("GW_TRANSPORT").as_deref() { Ok("mock") => { tracing::info!("transport = mock (zero egress)"); - std::sync::Arc::new(gw_engines::MockTransport) + Arc::new(gw_engines::MockTransport) } Ok("http") => { tracing::info!("transport = http (accounts without an endpoint fail)"); - std::sync::Arc::new(gw_engines::http_transport::HttpTransport::with_policies( + Arc::new(gw_engines::http_transport::HttpTransport::with_policies( Default::default(), Default::default(), )?) } _ => { tracing::info!("transport = auto (mock:// in-process, real URLs over HTTP)"); - std::sync::Arc::new( + Arc::new( gw_engines::http_transport::DispatchTransport::with_policies( Default::default(), Default::default(), diff --git a/crates/state/src/store.rs b/crates/state/src/store.rs index a9f8e8f..28f3bd0 100644 --- a/crates/state/src/store.rs +++ b/crates/state/src/store.rs @@ -661,21 +661,20 @@ impl MemoryStore { } } +/// Recovers a poisoned lock instead of panicking: every critical section here is infallible. +fn lock(m: &Mutex) -> std::sync::MutexGuard<'_, T> { + m.lock().unwrap_or_else(std::sync::PoisonError::into_inner) +} + #[async_trait::async_trait] impl Store for MemoryStore { async fn ledger_add(&self, r: &BillingRecord) -> GResult<()> { // watermark first: rollup-then-records is the lock order advance uses - let watermark = self - .rollup - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) + let watermark = lock(&self.rollup) .keys() .next_back() .map_or(0, |k| k.0 + ROLLUP_BUCKET_SECS); - let mut records = self - .records - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut records = lock(&self.records); records.push(r.clone()); if self.ledger_max_rows > 0 && records.len() > self.ledger_max_rows { // spare rows the rollup hasn't folded yet — lost here means lost @@ -696,20 +695,14 @@ impl Store for MemoryStore { } async fn ledger_snapshot(&self, limit: usize) -> GResult<(usize, Vec)> { - let records = self - .records - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let records = lock(&self.records); let total = records.len(); let page = records[total.saturating_sub(limit)..].to_vec(); Ok((total, page)) } async fn ledger_usage(&self, tenant: Option<&str>) -> GResult> { - let records = self - .records - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let records = lock(&self.records); let mut rollup: BTreeMap<(String, String), UsageRow> = BTreeMap::new(); for r in records.iter() { if tenant.is_some_and(|t| t != r.tenant) { @@ -750,10 +743,7 @@ impl Store for MemoryStore { let (since, until) = align_bounds(since, until); let mut map = BTreeMap::new(); let watermark = { - let rollup = self - .rollup - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let rollup = lock(&self.rollup); for ((minute, t, u, _), row) in rollup.iter() { if *minute >= bucket_floor(since) && *minute <= bucket_floor(until) @@ -770,10 +760,7 @@ impl Store for MemoryStore { .next_back() .map_or(0, |k| k.0 + ROLLUP_BUCKET_SECS) }; - let records = self - .records - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let records = lock(&self.records); let tail = records.iter().filter(|r| { r.created_at_epoch_secs >= since.max(watermark) && r.created_at_epoch_secs <= until @@ -796,10 +783,7 @@ impl Store for MemoryStore { let (since, until) = align_bounds(since, until); let mut map: BTreeMap = BTreeMap::new(); let watermark = { - let rollup = self - .rollup - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let rollup = lock(&self.rollup); for ((minute, t, u, _), row) in rollup.iter() { if *minute >= bucket_floor(since) && *minute <= bucket_floor(until) @@ -816,10 +800,7 @@ impl Store for MemoryStore { .next_back() .map_or(0, |k| k.0 + ROLLUP_BUCKET_SECS) }; - let records = self - .records - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let records = lock(&self.records); for r in records.iter().filter(|r| { r.created_at_epoch_secs >= since.max(watermark) && r.created_at_epoch_secs <= until @@ -835,10 +816,7 @@ impl Store for MemoryStore { async fn usage_rollup_advance(&self, now: i64) -> GResult { let hi = bucket_floor(now - ROLLUP_SETTLE_SECS); - let mut rollup = self - .rollup - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut rollup = lock(&self.rollup); let watermark = rollup .keys() .next_back() @@ -846,10 +824,7 @@ impl Store for MemoryStore { let lo = (hi - ROLLUP_BACKFILL_SECS).min(watermark); let mut fresh = BTreeMap::new(); { - let records = self - .records - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let records = lock(&self.records); for r in records .iter() .filter(|r| r.created_at_epoch_secs >= lo && r.created_at_epoch_secs < hi) @@ -873,10 +848,7 @@ impl Store for MemoryStore { } async fn security_event_add(&self, e: &SecurityEvent) -> GResult<()> { - self.sec_events - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .push(e.clone()); + lock(&self.sec_events).push(e.clone()); Ok(()) } @@ -885,10 +857,7 @@ impl Store for MemoryStore { tenant: Option<&str>, limit: usize, ) -> GResult> { - let events = self - .sec_events - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let events = lock(&self.sec_events); Ok(events .iter() .rev() @@ -899,34 +868,22 @@ impl Store for MemoryStore { } async fn admin_audit_add(&self, e: &AdminAudit) -> GResult<()> { - self.audit - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .push(e.clone()); + lock(&self.audit).push(e.clone()); Ok(()) } async fn admin_audit_list(&self, limit: usize) -> GResult> { - let audit = self - .audit - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let audit = lock(&self.audit); Ok(audit.iter().rev().take(limit).cloned().collect()) } async fn content_add(&self, r: &crate::ContentRecord) -> GResult<()> { - self.content - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .push(r.clone()); + lock(&self.content).push(r.clone()); Ok(()) } async fn content_purge(&self, now: i64) -> GResult { - let mut content = self - .content - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut content = lock(&self.content); let before = content.len(); content.retain(|r| r.expires_at_epoch_secs == 0 || r.expires_at_epoch_secs > now); Ok((before - content.len()) as u64) @@ -939,10 +896,7 @@ impl Store for MemoryStore { mut audit: AdminAudit, ) -> GResult { let mut erased = { - let mut content = self - .content - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut content = lock(&self.content); let before = content.len(); content.retain(|r| r.user_id != user || tenant.is_some_and(|t| t != r.tenant)); (before - content.len()) as u64 @@ -958,26 +912,17 @@ impl Store for MemoryStore { } } } - self.erasures - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert( - (tenant.unwrap_or_default().to_owned(), user.to_owned()), - crate::epoch_millis(), - ); + lock(&self.erasures).insert( + (tenant.unwrap_or_default().to_owned(), user.to_owned()), + crate::epoch_millis(), + ); audit.summary = format!("rows={erased}"); - self.audit - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .push(audit); + lock(&self.audit).push(audit); Ok(erased) } async fn user_erased_since(&self, tenant: &str, user: &str, since: i64) -> GResult { - let erasures = self - .erasures - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let erasures = lock(&self.erasures); let hit = |t: &str| { erasures .get(&(t.to_owned(), user.to_owned())) @@ -987,10 +932,7 @@ impl Store for MemoryStore { } async fn content_for(&self, request_id: &str) -> GResult> { - let content = self - .content - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let content = lock(&self.content); Ok(content .iter() .filter(|r| r.request_id == request_id) diff --git a/crates/task/src/lib.rs b/crates/task/src/lib.rs index ca42abd..392df84 100644 --- a/crates/task/src/lib.rs +++ b/crates/task/src/lib.rs @@ -237,8 +237,8 @@ mod tests { #[tokio::test] async fn avail_sweep_alerts_on_transition_only() { let yaml = "listen: {host: h, port: 1}\nmodels: [{name: m, protocol: openai-chat}, {name: rt, protocol: realtime}]\nstability: {availability_min_samples: 2}"; - let cfg = std::sync::Arc::new(gw_config::GatewayConfig::from_yaml(yaml).unwrap()); - let state = std::sync::Arc::new(GatewayState::from_config(&cfg)); + let cfg = Arc::new(gw_config::GatewayConfig::from_yaml(yaml).unwrap()); + let state = Arc::new(GatewayState::from_config(&cfg)); let shared = SharedConfig::new(cfg, state.clone()); let mut rx = state.alerts.take_receiver().expect("receiver"); let mut last = HashMap::new(); @@ -276,8 +276,8 @@ mod tests { std::env::set_var("GW_TEST_ALERT_URL", format!("http://{addr}/hook")); } let yaml = "listen: {host: h, port: 1}\nalerts: {webhook_url_env: GW_TEST_ALERT_URL}"; - let cfg = std::sync::Arc::new(gw_config::GatewayConfig::from_yaml(yaml).unwrap()); - let state = std::sync::Arc::new(GatewayState::from_config(&cfg)); + let cfg = Arc::new(gw_config::GatewayConfig::from_yaml(yaml).unwrap()); + let state = Arc::new(GatewayState::from_config(&cfg)); let shared = SharedConfig::new(cfg, state.clone()); let task = spawn_alert_dispatch(shared); state diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs index c94a77f..41a6692 100644 --- a/crates/views/src/lib.rs +++ b/crates/views/src/lib.rs @@ -2126,10 +2126,6 @@ fn anthropic_error(status: u16, message: impl Into) -> Response { .into_response() } -fn anthropic_gateway_error(e: GatewayError) -> Response { - anthropic_error(e.http_status, e.message) -} - /// Run the pipeline on its own task so a client disconnect can't cancel it /// mid-billing: once admitted, quota/ledger accounting runs to completion. async fn run_pipeline(s: &AppState, request: GatewayRequest, ak: AkInfo) -> GResult { @@ -2645,7 +2641,7 @@ async fn messages( let ctx = match run_pipeline(&s, request, ak).await { Ok(ctx) => ctx, - Err(e) => return anthropic_gateway_error(e), + Err(e) => return anthropic_error(e.http_status, e.message), }; log_access("messages", &ctx, started); let Some(mut outcome) = ctx.outcome else { @@ -2911,10 +2907,6 @@ async fn run_family( } } -/// The engine's native payload, or a 500 naming the engine that returned none. -/// A pre-stage content block answers 400 with the block message — these -/// surfaces have no in-band content_filter shape, and falling through would -/// misreport the block as an engine failure. /// An `input`-style field that may be a string or an array of strings /// (the OpenAI embeddings/moderations shape). fn string_or_string_array(v: Option) -> Vec { @@ -2931,6 +2923,10 @@ fn string_or_string_array(v: Option) -> Vec { } } +/// The engine's native payload, or a 500 naming the engine that returned none. +/// A pre-stage content block answers 400 with the block message — these +/// surfaces have no in-band content_filter shape, and falling through would +/// misreport the block as an engine failure. fn response_v2_or_500(outcome: Option, engine: &str) -> Response { match outcome { Some(o) if o.block.block => error_response(400, o.response.message),