live coverage: moderations, OpenAI audio, DashScope legacy, Gemini Live realtime; whole-repo quality round

crates/config/src/lib.rs

@@ -247,6 +247,9 @@ fn weight_one() -> f64 {
 pub struct AccountConf {
     pub name: String,
     pub provider: String,
+    /// Stamped from the provider preset in `normalize()`; never read from YAML.
+    #[serde(skip)]
+    pub kind: String,
     #[serde(default = "default_priority")]
     pub priority: i32,
     /// "ptu" (provisioned throughput, preferred) or "paygo" (default).
@@ -651,7 +654,7 @@ fn provider_preset(kind: &str) -> Option<ProviderPreset> {
         },
         "gemini" => ProviderPreset {
             endpoint: "https://generativelanguage.googleapis.com",
-            wires: &["gemini"],
+            wires: &["gemini", "realtime"],
             default_model_wire: "gemini",
         },
         // OpenAI-protocol vendors: same wire shape, different base URL.
@@ -800,6 +803,7 @@ impl GatewayConfig {
                 self.accounts.push(AccountConf {
                     name: p.name.clone(),
                     provider: p.name.clone(),
+                    kind: String::new(),
                     priority: 1,
                     tier: String::new(),
                     cost_input_price_per_1k_micros: 0,
@@ -816,6 +820,7 @@ impl GatewayConfig {
             }
             // an empty endpoint would answer from the mock transport with fabricated successes
             for a in self.accounts.iter_mut().filter(|a| a.provider == p.name) {
+                a.kind = p.kind.clone();
                 if a.endpoint.is_empty() {
                     a.endpoint = if p.endpoint.is_empty() {
                         preset.endpoint.to_owned()
@@ -1361,6 +1366,15 @@ models:
         );
     }
 
+    #[test]
+    fn provider_specific_wires_must_pin_a_provider() {
+        let yaml = "listen: {host: h, port: 1}\nmodels: [{name: v, protocol: video}]\naccounts: [{name: a, provider: p, protocols: ['video']}]";
+        assert!(matches!(
+            GatewayConfig::from_yaml(yaml),
+            Err(ConfigError::VideoModelNeedsProvider { model }) if model == "v"
+        ));
+    }
+
     #[test]
     fn provider_presets_expand_to_accounts_and_model_defaults() {
         let cfg = GatewayConfig::from_yaml(PROVIDER_YAML).unwrap();

crates/engines/src/base.rs

@@ -30,7 +30,7 @@ impl Base {
         self.request
             .account
             .as_ref()
-            .map(|a| a.provider.as_str())
+            .map(|a| a.wire_kind())
             .unwrap_or_default()
     }
 

crates/engines/src/families.rs

@@ -827,7 +827,7 @@ pub async fn video_poll(
     id: &str,
 ) -> GResult<VideoPoll> {
     let base = account.base_url(VENDOR_SENTINEL);
-    let dialect = video_dialect(&account.provider);
+    let dialect = video_dialect(account.wire_kind());
     let (method, url, req_body) = match dialect {
         VideoDialect::SiliconFlow => (
             "POST",
@@ -915,7 +915,7 @@ pub async fn video_content(
     poll: &Value,
 ) -> GResult<(u16, String, bytes::Bytes)> {
     let base = account.base_url(VENDOR_SENTINEL);
-    let request = match video_dialect(&account.provider) {
+    let request = match video_dialect(account.wire_kind()) {
         VideoDialect::Sora => video_request(
             account,
             "GET",
@@ -1040,7 +1040,21 @@ impl ModelEngine for SearchEngine {
                     ("accept", "application/json".into()),
                     ("x-subscription-token", self.base.api_key()),
                 ];
-                let (status, v) = self.search_get(url, headers).await?;
+                let reply = self
+                    .base
+                    .transport
+                    .send(UpstreamRequest {
+                        protocol: gw_consts::Protocol::Search,
+                        method: "GET",
+                        url,
+                        headers,
+                        body: Vec::new(),
+                        stream: false,
+                        account: self.base.account(),
+                        replay_account: self.base.replay_account(),
+                    })
+                    .await?;
+                let (status, v) = parse_json_reply(reply)?;
                 (status, v, "/web/results")
             }
             _ => {
@@ -1067,26 +1081,6 @@ impl ModelEngine for SearchEngine {
     }
 }
 
-impl SearchEngine {
-    async fn search_get(&self, url: String, headers: Headers) -> GResult<(u16, Value)> {
-        let reply = self
-            .base
-            .transport
-            .send(UpstreamRequest {
-                protocol: gw_consts::Protocol::Search,
-                method: "GET",
-                url,
-                headers,
-                body: Vec::new(),
-                stream: false,
-                account: self.base.account(),
-                replay_account: self.base.replay_account(),
-            })
-            .await?;
-        parse_json_reply(reply)
-    }
-}
-
 base_engine!(ModerationsEngine);
 
 #[async_trait::async_trait]
@@ -2155,6 +2149,43 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn kling_envelope_errors_surface_as_502() {
+        let account = Arc::new(gw_models::Account {
+            name: "kling-1".into(),
+            provider: "kling".into(),
+            endpoint: "https://api-singapore.klingai.com".into(),
+            ..Default::default()
+        });
+        let transport = Arc::new(BytesReply(
+            br#"{"code":1102,"message":"Account balance not enough","request_id":"trace"}"#,
+        ));
+        let mut request = req(
+            Protocol::Video,
+            "kling-v1-6",
+            Some(TypedParams::Video(VideoParams {
+                prompt: "a paper boat".into(),
+                ..Default::default()
+            })),
+        );
+        request.account = Some(Arc::clone(&account));
+        let submit = VideoEngine::new(request, transport.clone())
+            .run()
+            .await
+            .unwrap_err();
+        assert_eq!(submit.http_status, 502);
+        assert!(
+            submit.message.contains("kling code 1102"),
+            "{}",
+            submit.message
+        );
+        let poll = match video_poll(transport.as_ref(), &account, "task-1").await {
+            Ok(_) => panic!("kling envelope error accepted as a successful poll"),
+            Err(e) => e,
+        };
+        assert_eq!(poll.http_status, 502);
+    }
+
     #[tokio::test]
     async fn minimax_video_business_errors_surface_on_submit_and_poll() {
         let account = Arc::new(gw_models::Account {

crates/engines/src/realtime.rs

@@ -9,6 +9,16 @@ pub fn is_response_create(frame: &Value) -> bool {
     frame["type"] == "response.create"
 }
 
+/// The client frame that starts a generation — the admission point. OpenAI
+/// signals it as `response.create`, Gemini Live as a completed client turn.
+pub fn is_client_turn(provider: &str, frame: &Value) -> bool {
+    if is_gemini_realtime(provider) {
+        return frame["clientContent"]["turnComplete"] == Value::Bool(true)
+            || frame["client_content"]["turn_complete"] == Value::Bool(true);
+    }
+    is_response_create(frame)
+}
+
 /// String values that never carry human text (base64 media, protocol ids);
 /// everything else is scanned, fail closed, and config objects still recurse.
 fn skip_scalar(k: &str, text_delta: bool) -> bool {
@@ -70,14 +80,68 @@ pub fn is_gemini_realtime(provider: &str) -> bool {
     matches!(provider, "google" | "gemini" | "vertex")
 }
 
+/// Gemini may send `usageMetadata` on any server frame, not only the turn
+/// boundary: the latest snapshot settles a turnComplete that arrives bare.
+pub fn gemini_usage_update(provider: &str, frame: &Value) -> Option<Value> {
+    if !is_gemini_realtime(provider) {
+        return None;
+    }
+    frame
+        .get("usageMetadata")
+        .filter(|u| u.is_object())
+        .cloned()
+}
+
+/// (input, output) token counts of a Gemini `usageMetadata` object.
+/// The AUDIO-modality (input, output) token counts of a Gemini `usageMetadata` object.
+pub fn gemini_audio_tokens(u: &Value) -> (i64, i64) {
+    let modality = |details: &Value| {
+        details
+            .as_array()
+            .map(|ds| {
+                ds.iter()
+                    .filter(|d| d["modality"] == "AUDIO")
+                    .map(|d| d["tokenCount"].as_i64().unwrap_or(0).max(0))
+                    .sum()
+            })
+            .unwrap_or(0)
+    };
+    (
+        modality(&u["promptTokensDetails"]),
+        modality(&u["responseTokensDetails"]),
+    )
+}
+
+pub fn gemini_tokens(u: &Value) -> (i64, i64) {
+    let it = u["promptTokenCount"].as_i64().unwrap_or(0);
+    let ot = u["responseTokenCount"]
+        .as_i64()
+        .or_else(|| u["candidatesTokenCount"].as_i64())
+        .unwrap_or(0);
+    (it, ot)
+}
+
 /// Whether `frame` is a server-initiated (VAD) turn start the gateway must gate.
 pub fn realtime_turn_started(provider: &str, frame: &Value) -> bool {
     !is_gemini_realtime(provider) && frame["type"] == "response.created"
 }
 
-/// Text and opaque payload units carried by one OpenAI-dialect output-delta
-/// frame. Opaque units count base64 quanta.
+/// Delivered output in one frame: OpenAI deltas yield text (or audio quanta),
+/// Gemini `modelTurn` parts count as byte-estimated opaque units.
 pub fn realtime_output_delta(frame: &Value) -> (Option<&str>, usize) {
+    // Gemini Live: delivered output rides serverContent.modelTurn parts
+    if let Some(parts) = frame["serverContent"]["modelTurn"]["parts"].as_array() {
+        let opaque = parts
+            .iter()
+            .map(|p| {
+                p["inlineData"]["data"]
+                    .as_str()
+                    .map_or(0, |d| d.len().div_ceil(4))
+                    + p["text"].as_str().map_or(0, |t| t.len().div_ceil(4))
+            })
+            .sum();
+        return (None, opaque);
+    }
     let frame_type = frame["type"].as_str().unwrap_or_default();
     let Some(delta) = frame["delta"].as_str() else {
         return (None, 0);
@@ -99,13 +163,7 @@ pub fn realtime_usage(provider: &str, frame: &Value) -> Option<(i64, i64)> {
         if frame["serverContent"]["turnComplete"] != Value::Bool(true) {
             return None;
         }
-        let u = &frame["usageMetadata"];
-        let it = u["promptTokenCount"].as_i64().unwrap_or(0);
-        let ot = u["responseTokenCount"]
-            .as_i64()
-            .or_else(|| u["candidatesTokenCount"].as_i64())
-            .unwrap_or(0);
-        (it, ot)
+        gemini_tokens(&frame["usageMetadata"])
     } else {
         // a turn ends on response.done, any status, possibly with zero usage
         if frame["type"] != "response.done" {
@@ -125,22 +183,7 @@ pub fn realtime_usage(provider: &str, frame: &Value) -> Option<(i64, i64)> {
 /// audio apart from text; zero when the vendor reports no modality split.
 pub fn realtime_audio_tokens(provider: &str, frame: &Value) -> (i64, i64) {
     if is_gemini_realtime(provider) {
-        let modality = |details: &Value| {
-            details
-                .as_array()
-                .map(|ds| {
-                    ds.iter()
-                        .filter(|d| d["modality"] == "AUDIO")
-                        .map(|d| d["tokenCount"].as_i64().unwrap_or(0).max(0))
-                        .sum()
-                })
-                .unwrap_or(0)
-        };
-        let u = &frame["usageMetadata"];
-        return (
-            modality(&u["promptTokensDetails"]),
-            modality(&u["responseTokensDetails"]),
-        );
+        return gemini_audio_tokens(&frame["usageMetadata"]);
     }
     let u = &frame["response"]["usage"];
     (
@@ -332,6 +375,45 @@ mod tests {
         let done = json!({"type":"response.done","response":{"usage":{"input_tokens":12,"output_tokens":34,
             "input_token_details":{"text_tokens":2,"audio_tokens":10},
             "output_token_details":{"text_tokens":4,"audio_tokens":30}}}});
+        let periodic = json!({"serverContent": {"modelTurn": {"parts": [{"text": "x"}]}},
+            "usageMetadata": {"promptTokenCount": 9, "responseTokenCount": 3}});
+        assert_eq!(
+            realtime_usage("gemini", &periodic),
+            None,
+            "off-boundary usage never settles"
+        );
+        assert_eq!(
+            gemini_usage_update("gemini", &periodic).map(|u| gemini_tokens(&u)),
+            Some((9, 3))
+        );
+        assert_eq!(gemini_usage_update("openai", &periodic), None);
+        let bare_done = json!({"serverContent": {"turnComplete": true}});
+        assert_eq!(realtime_usage("gemini", &bare_done), Some((0, 0)));
+        let gem_out = json!({"serverContent": {"modelTurn": {"parts": [
+            {"text": "pong"}, {"inlineData": {"mimeType": "audio/pcm", "data": "AAAAAAAAAAAA"}}
+        ]}}});
+        assert_eq!(realtime_output_delta(&gem_out), (None, 4));
+        assert!(is_client_turn(
+            "openai",
+            &json!({"type": "response.create"})
+        ));
+        assert!(!is_client_turn(
+            "openai",
+            &json!({"clientContent": {"turnComplete": true}})
+        ));
+        assert!(is_client_turn(
+            "gemini",
+            &json!({"clientContent": {"turnComplete": true}})
+        ));
+        assert!(is_client_turn(
+            "google",
+            &json!({"client_content": {"turn_complete": true}})
+        ));
+        assert!(!is_client_turn(
+            "gemini",
+            &json!({"clientContent": {"turnComplete": false}})
+        ));
+        assert!(!is_client_turn("gemini", &json!({"setup": {"model": "m"}})));
         assert_eq!(realtime_usage("openai", &done), Some((12, 34)));
         assert_eq!(realtime_usage("azure", &done), Some((12, 34)));
         assert_eq!(realtime_audio_tokens("openai", &done), (10, 30));

crates/models/src/request.rs

@@ -222,6 +222,8 @@ pub mod domain {
     pub struct Account {
         pub name: String,
         pub provider: String,
+        /// The provider preset's kind when synthesized from `providers:`; empty on raw accounts.
+        pub kind: String,
         pub priority: i32,
         /// consts::account_tier::{PTU, PAYGO}; empty = paygo.
         pub tier: String,
@@ -248,6 +250,16 @@ pub mod domain {
             self.tier == gw_consts::account_tier::PTU
         }
 
+        /// The wire-dialect key: the preset kind when the account came from a
+        /// `providers:` entry, else the operator's provider label.
+        pub fn wire_kind(&self) -> &str {
+            if self.kind.is_empty() {
+                &self.provider
+            } else {
+                &self.kind
+            }
+        }
+
         /// Base URL for building the upstream request: the account's endpoint if
         /// set, else the given `mock://…` sentinel.
         pub fn base_url<'a>(&'a self, mock_sentinel: &'a str) -> &'a str {