diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs
index 620b103..b395a29 100644
--- a/crates/config/src/lib.rs
+++ b/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();
diff --git a/crates/engines/src/base.rs b/crates/engines/src/base.rs
index 5dcf5a5..9af4f60 100644
--- a/crates/engines/src/base.rs
+++ b/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()
     }
 
diff --git a/crates/engines/src/families.rs b/crates/engines/src/families.rs
index 255e816..f789148 100644
--- a/crates/engines/src/families.rs
+++ b/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 {
diff --git a/crates/engines/src/realtime.rs b/crates/engines/src/realtime.rs
index 4adc7b0..b4252e1 100644
--- a/crates/engines/src/realtime.rs
+++ b/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));
diff --git a/crates/models/src/request.rs b/crates/models/src/request.rs
index 023e29d..ce12d5b 100644
--- a/crates/models/src/request.rs
+++ b/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 {
diff --git a/crates/server/tests/e2e.rs b/crates/server/tests/e2e.rs
index ce4f3da..27418c9 100644
--- a/crates/server/tests/e2e.rs
+++ b/crates/server/tests/e2e.rs
@@ -1463,7 +1463,21 @@ async fn async_video_bills_once_on_the_first_done_poll() {
         .await
         .unwrap();
     assert_eq!(resp.status(), StatusCode::OK);
-    assert_eq!(body_json(resp).await["video_url"], "mock://videos/out.mp4");
+    let sync = body_json(resp).await;
+    assert_eq!(sync["video_url"], "mock://videos/out.mp4");
+    let resp = app
+        .clone()
+        .oneshot(get_authed(&format!(
+            "/v1/videos/{}",
+            sync["task_id"].as_str().unwrap()
+        )))
+        .await
+        .unwrap();
+    assert_eq!(
+        resp.status(),
+        StatusCode::NOT_FOUND,
+        "a delivered sync reply must not register its task_id as an async job"
+    );
 
     let mut ids = Vec::new();
     for prompt in [
@@ -2795,7 +2809,7 @@ models:
 }
 
 #[tokio::test]
-async fn realtime_refuses_ungovernable_provider() {
+async fn realtime_upgrades_the_gemini_dialect_and_gates_on_the_client_turn() {
     use tokio_tungstenite::tungstenite::client::IntoClientRequest;
 
     let yaml = r#"
@@ -2821,10 +2835,16 @@ models:
         .unwrap();
     req.headers_mut()
         .insert("authorization", "Bearer ak-rt".parse().unwrap());
-    assert!(
-        tokio_tungstenite::connect_async(req).await.is_err(),
-        "realtime must refuse a provider it cannot gate before generation"
-    );
+    let (mut ws, resp) = tokio_tungstenite::connect_async(req).await.unwrap();
+    assert_eq!(resp.status().as_u16(), 101);
+    use futures::StreamExt;
+    let frame = tokio::time::timeout(std::time::Duration::from_secs(5), ws.next())
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    let v: Value = serde_json::from_str(frame.to_text().unwrap()).unwrap();
+    assert_eq!(v["type"], "error", "{v}");
 }
 
 #[tokio::test]
diff --git a/crates/state/src/lib.rs b/crates/state/src/lib.rs
index 0d215da..9d83335 100644
--- a/crates/state/src/lib.rs
+++ b/crates/state/src/lib.rs
@@ -377,6 +377,7 @@ impl AccountPool {
                 Arc::new(Account {
                     name: a.name.clone(),
                     provider: a.provider.clone(),
+                    kind: a.kind.clone(),
                     priority: a.priority,
                     tier: a.tier.clone(),
                     endpoint: a.endpoint.clone(),
diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs
index cfccbbc..99bb9ae 100644
--- a/crates/views/src/lib.rs
+++ b/crates/views/src/lib.rs
@@ -22,8 +22,8 @@ use gw_consts::ErrClass;
 use gw_dag::DagContext;
 use gw_engines::SharedTransport;
 use gw_engines::realtime::{
-    is_response_create, realtime_audio_tokens, realtime_output_delta, realtime_turn_started,
-    realtime_usage,
+    gemini_audio_tokens, gemini_tokens, gemini_usage_update, is_client_turn, realtime_audio_tokens,
+    realtime_output_delta, realtime_turn_started, realtime_usage,
 };
 use gw_handler::{BatchItem, OfflineHandler, OnlineHandler};
 use gw_models::{
@@ -338,15 +338,6 @@ async fn realtime_ws(
         ws.on_upgrade(move |socket| {
             realtime_session(socket, s, ak, m, mt, account.name.clone(), hint)
         })
-    } else if gw_engines::realtime::is_gemini_realtime(&account.provider) {
-        // no pre-generation gate signal in this dialect — refuse rather than bill after the fact
-        error_response(
-            501,
-            format!(
-                "realtime is not supported for provider `{}`",
-                account.provider
-            ),
-        )
     } else {
         ws.on_upgrade(move |socket| realtime_bridge(socket, s, ak, m, mt, account, hint))
     }
@@ -750,8 +741,7 @@ fn upstream_text_to_client(
 }
 
 /// Bridge one realtime session to a real upstream: transparent relay plus auth,
-/// per-generation gates and per-turn billing; only the OpenAI dialect reaches
-/// here ([`realtime_ws`] refuses providers it cannot gate).
+/// per-generation gates and per-turn billing.
 async fn realtime_bridge(
     mut client: axum::extract::ws::WebSocket,
     s: AppState,
@@ -770,7 +760,16 @@ async fn realtime_bridge(
     let ws_base = base
         .replacen("https://", "wss://", 1)
         .replacen("http://", "ws://", 1);
-    let url = format!("{ws_base}/v1/realtime?model={}", rtm.served);
+    let key = account.api_key().unwrap_or_else(|| "mock".to_owned());
+    let gemini = gw_engines::realtime::is_gemini_realtime(account.wire_kind());
+    // Gemini's Live socket is one bidi RPC authed by key; the model rides the setup frame
+    let url = if gemini {
+        format!(
+            "{ws_base}/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key={key}"
+        )
+    } else {
+        format!("{ws_base}/v1/realtime?model={}", rtm.served)
+    };
     let mut req = match url.into_client_request() {
         Ok(r) => r,
         Err(e) => {
@@ -784,8 +783,7 @@ async fn realtime_bridge(
             return;
         }
     };
-    let key = account.api_key().unwrap_or_else(|| "mock".to_owned());
-    if let Ok(v) = format!("Bearer {key}").parse() {
+    if !gemini && let Ok(v) = format!("Bearer {key}").parse() {
         req.headers_mut().insert("authorization", v);
     }
     let upstream = match tokio_tungstenite::connect_async(req).await {
@@ -812,6 +810,8 @@ async fn realtime_bridge(
     let mut pending: Option<RealtimeTurn> = None;
     // denied server-VAD turn: swallow its upstream frames until its terminal frame
     let mut suppress = false;
+    // Gemini sends cumulative usage on any server frame; the latest settles a bare turnComplete
+    let mut usage_snapshot: Option<Value> = None;
     // outbound DLP redactions summed within a turn, recorded once at its boundary
     let mut out_redacted = 0i64;
     loop {
@@ -826,6 +826,31 @@ async fn realtime_bridge(
                     Some(Ok(_)) => continue, // ping/pong handled by the ws stacks
                 };
                 if let Some(mut frame) = frame {
+                    if gemini {
+                        // pin the entitled model: the socket carries no model, the setup frame does
+                        if let Some(setup) = frame.get_mut("setup").and_then(Value::as_object_mut) {
+                            setup.insert("model".into(), format!("models/{}", rtm.served).into());
+                            forward = UMsg::text(frame.to_string());
+                        }
+                        // audio-driven turns and barge-in have no admission point yet: refuse
+                        // loudly, keep the session
+                        let ungoverned = frame.get("realtimeInput").is_some()
+                            || frame.get("realtime_input").is_some()
+                            || (pending.is_some() && is_client_turn(account.wire_kind(), &frame));
+                        if ungoverned {
+                            if cl_tx
+                                .send(rt_error_frame(
+                                    ErrClass::Validation,
+                                    "one governed clientContent turn at a time; realtimeInput is not wired",
+                                ))
+                                .await
+                                .is_err()
+                            {
+                                break;
+                            }
+                            continue;
+                        }
+                    }
                     match rt_inbound_policy(&s, &ak, &hint, &mut frame).await {
                         Err(reason) => {
                             if cl_tx
@@ -843,11 +868,8 @@ async fn realtime_bridge(
                             }
                         }
                     }
-                    // gate each generation trigger; with a turn already admitted it relays ungated
-                    // —
-                    // upstream rejects the duplicate and a raced accept is caught by the
-                    // response.created gate
-                    if is_response_create(&frame) && pending.is_none() {
+                    // dup triggers relay ungated: upstream rejects them, response.created gates a raced accept
+                    if is_client_turn(account.wire_kind(), &frame) && pending.is_none() {
                         match realtime_gate(&s, &ak, &rtm, &hint).await {
                             Ok(admit) => {
                                 pending = Some(RealtimeTurn::new(admit));
@@ -893,15 +915,18 @@ async fn realtime_bridge(
                 let mut output_units = 0;
                 match frame {
                     Some(mut v) => {
+                        if let Some(u) = gemini_usage_update(account.wire_kind(), &v) {
+                            usage_snapshot = Some(u);
+                        }
                         if suppress {
                             relay = false;
-                            if realtime_usage(&account.provider, &v).is_some() {
+                            if realtime_usage(account.wire_kind(), &v).is_some() {
                                 suppress = false;
                             }
                         }
                         // server-VAD auto-starts a turn with no client response.create — gate it
                         // like a manual one
-                        else if realtime_turn_started(&account.provider, &v) && pending.is_none() {
+                        else if realtime_turn_started(account.wire_kind(), &v) && pending.is_none() {
                             match realtime_gate(&s, &ak, &rtm, &hint).await {
                                 Ok(admit) => pending = Some(RealtimeTurn::new(admit)),
                                 Err((class, denied)) => {
@@ -913,7 +938,12 @@ async fn realtime_bridge(
                                     relay = false;
                                 }
                             }
-                        } else if let Some((it, ot)) = realtime_usage(&account.provider, &v) {
+                        } else if let Some((mut it, mut ot)) = realtime_usage(account.wire_kind(), &v) {
+                            // every boundary consumes the snapshot so none leaks into the next turn
+                            if let (Some(u), true) = (usage_snapshot.take(), it + ot == 0) {
+                                (it, ot) = gemini_tokens(&u);
+                                v["usageMetadata"] = u;
+                            }
                             // turn boundary: settle the admitted turn; one with no gated turn bills
                             // unreserved
                             match pending.take() {
@@ -926,7 +956,7 @@ async fn realtime_bridge(
                                         gw_models::TokenInput {
                                             prompt: it,
                                             completion: ot,
-                                            ..turn_audio(&account.provider, &v)
+                                            ..turn_audio(account.wire_kind(), &v)
                                         },
                                         false,
                                     )
@@ -962,7 +992,7 @@ async fn realtime_bridge(
                                         gw_models::TokenInput {
                                             prompt: it,
                                             completion: ot,
-                                            ..turn_audio(&account.provider, &v)
+                                            ..turn_audio(account.wire_kind(), &v)
                                         },
                                         false,
                                     )
@@ -1035,7 +1065,26 @@ async fn realtime_bridge(
         }
     }
     if let Some(turn) = pending {
-        settle_realtime_abort(turn, &rtm, mt, &account.name).await;
+        if let Some(usage) = usage_snapshot {
+            let (prompt, reported_completion) = gemini_tokens(&usage);
+            // the snapshot may predate the last delivered parts; the byte estimate floors it
+            let completion = turn
+                .estimated_output_tokens()
+                .map_or(reported_completion, |delivered| {
+                    delivered.max(reported_completion)
+                });
+            let (audio_prompt, audio_completion) = gemini_audio_tokens(&usage);
+            let tokens = gw_models::TokenInput {
+                prompt,
+                completion,
+                audio_prompt,
+                audio_completion,
+                ..Default::default()
+            };
+            bill_realtime_turn(&turn.admit, &rtm, mt, &account.name, tokens, true).await;
+        } else {
+            settle_realtime_abort(turn, &rtm, mt, &account.name).await;
+        }
     }
     // a turn aborted before its boundary still redacted per frame — flush the count for the audit
     flush_rt_out_dlp(&s, &ak, &hint, out_redacted).await;
@@ -4006,12 +4055,19 @@ async fn videos_generations(
 }
 
 /// The shared head of both video read routes: spend the caller's rate limits,
-/// load the job under its tenant gate.
+/// load the job under its tenant gate, poll the vendor and settle a first done.
 async fn admit_video_job(
     s: &AppState,
     ak: &AkInfo,
     id: &str,
-) -> Result<(Arc<GatewayState>, VideoJob), Response> {
+) -> Result<
+    (
+        VideoJob,
+        Arc<gw_models::Account>,
+        gw_engines::families::VideoPoll,
+    ),
+    Response,
+> {
     let state = s.handler.state();
     let cfg = s.handler.cfg();
     let gov = state.governance.as_ref();
@@ -4025,7 +4081,14 @@ async fn admit_video_job(
     }
     let found = state.store.video_job_get(id).await;
     let job = tenant_owned(found, |j| &j.tenant, &ak.tenant, "video", id)?;
-    Ok((state, job))
+    let Some(account) = state.pool.named(&job.account) else {
+        return Err(error_response(
+            503,
+            format!("account {} is no longer configured", job.account),
+        ));
+    };
+    let poll = poll_and_settle_video(s, &job, &account).await?;
+    Ok((job, account, poll))
 }
 
 async fn poll_and_settle_video(
@@ -4093,25 +4156,14 @@ async fn videos_get(
     Authed(ak): Authed,
     Path(id): Path<String>,
 ) -> Response {
-    let (state, job) = match admit_video_job(&s, &ak, &id).await {
-        Ok(admitted) => admitted,
-        Err(resp) => return resp,
-    };
-    let Some(account) = state.pool.named(&job.account) else {
-        return error_response(
-            503,
-            format!("account {} is no longer configured", job.account),
-        );
-    };
-    let poll = match poll_and_settle_video(&s, &job, &account).await {
-        Ok(poll) => poll,
-        Err(resp) => return resp,
-    };
-    (
-        StatusCode::from_u16(poll.status).unwrap_or(StatusCode::OK),
-        Json(poll.body),
-    )
-        .into_response()
+    match admit_video_job(&s, &ak, &id).await {
+        Ok((_, _, poll)) => (
+            StatusCode::from_u16(poll.status).unwrap_or(StatusCode::OK),
+            Json(poll.body),
+        )
+            .into_response(),
+        Err(resp) => resp,
+    }
 }
 
 /// GET /v1/videos/{id}/content — the finished clip's bytes, proxied for vendors
@@ -4121,20 +4173,10 @@ async fn videos_content(
     Authed(ak): Authed,
     Path(id): Path<String>,
 ) -> Response {
-    let (state, job) = match admit_video_job(&s, &ak, &id).await {
+    let (job, account, poll) = match admit_video_job(&s, &ak, &id).await {
         Ok(admitted) => admitted,
         Err(resp) => return resp,
     };
-    let Some(account) = state.pool.named(&job.account) else {
-        return error_response(
-            503,
-            format!("account {} is no longer configured", job.account),
-        );
-    };
-    let poll = match poll_and_settle_video(&s, &job, &account).await {
-        Ok(poll) => poll,
-        Err(resp) => return resp,
-    };
     if !poll.done {
         return error_response(409, "video is not completed");
     }
diff --git a/docs/api.md b/docs/api.md
index b41d1f0..c48606b 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -156,7 +156,7 @@ stripped from the response; the visible turn still serves.
 
 `POST /v1/videos/generations` runs the pipeline like any family (auth, limits,
 routing, a ledger row) and returns the vendor's reply as is. The wire follows
-the serving account's provider name: `openai`/`azure` speak Sora's `/v1/videos`
+the serving account's preset kind (else its provider label): `openai` speaks Sora's `/v1/videos`
 (`seconds`, `size`, a video object back, the finished clip via
 `GET /v1/videos/{id}/content`), `siliconflow` Wan's `video/submit` +
 `video/status`, `alibaba`/`dashscope` the DashScope task API (async header,
@@ -229,6 +229,16 @@ boundary, the delivered text or audio is billed from an estimate; a turn that
 delivered nothing is refunded. An endpoint-less account serves a local mock
 session (OpenAI Realtime event shape) for offline development.
 
+The wire follows the account's preset kind (else its provider label): a
+`gemini` account bridges Google's Live API socket (binary frames relayed
+as-is, the key on the query string, `setup.model` rewritten to the entitled
+served model) and admits each turn on `clientContent.turnComplete` — the
+dialect's own generation signal — settling the `usageMetadata` that rides the
+completed turn; `realtimeInput` audio turns and a second
+`clientContent` during an active generation answer an in-band error until they
+have an admission point. Every other account speaks the OpenAI Realtime shape
+and admits on `response.create`.
+
 ## Introspection
 
 | Method | Path | Notes |
diff --git a/docs/development.md b/docs/development.md
index 85bbe83..29d5cb7 100644
--- a/docs/development.md
+++ b/docs/development.md
@@ -100,6 +100,8 @@ Last full run (2026-08-19): all cases across Anthropic, OpenAI (incl.
 gpt-realtime-mini through `/v1/realtime` and sora-2 video), Gemini, DeepSeek,
 MiniMax (incl. Hailuo video), Qwen/DashScope (incl. wan2.2-t2v-plus video),
 Qianfan, Moonshot, SiliconFlow (incl. Wan2.2 video), OpenRouter, Cohere/Jina
-rerank, xAI Grok (chat, Responses, image, video), Bedrock (InvokeModel,
-Converse, Llama) and a local Ollama through the generic OpenAI-compatible path
-— every ledger row matched the oracle exactly.
+rerank, xAI Grok (chat, Responses, image, video), Kling video, Brave search,
+Bedrock (InvokeModel, Converse, Llama), a local Ollama through the generic
+OpenAI-compatible path, OpenAI moderations/TTS/STT, the DashScope legacy wire
+and the Gemini Live realtime dialect — every ledger row matched the oracle
+exactly.
diff --git a/docs/providers.md b/docs/providers.md
index 6cddcf2..02c7856 100644
--- a/docs/providers.md
+++ b/docs/providers.md
@@ -24,7 +24,7 @@ models:
 |------|----------|-----------|------|
 | `openai` | `https://api.openai.com` | openai-chat, embeddings, image, tts, stt, responses, completions, realtime, moderations, video | `Bearer` (realtime live-verified on gpt-realtime-mini — the bridge settles the vendor's `response.done` usage, audio output at its `token_rate.audio_completion` weight, `estimated: false`; video = Sora, see below) |
 | `anthropic` | `https://api.anthropic.com` | anthropic-messages | `x-api-key` + `anthropic-version` |
-| `gemini` | `https://generativelanguage.googleapis.com` | gemini | `x-goog-api-key` |
+| `gemini` | `https://generativelanguage.googleapis.com` | gemini, realtime | `x-goog-api-key` (realtime = the Live API socket, live-verified: the bridge admits on `clientContent.turnComplete`, relays the binary frames, and settles `usageMetadata` — audio output tokens at their own weight) |
 | `deepseek` | `https://api.deepseek.com` | openai-chat | `Bearer` |
 | `openrouter` | `https://openrouter.ai/api` | openai-chat | `Bearer` (its `reasoning_details` shape is the one this gateway emits, so signed Anthropic reasoning round-trips through tool loops; verified live on free and paid models) |
 | `moonshot` | `https://api.moonshot.cn` | openai-chat | `Bearer` (Kimi K2 thinking: `reasoning_content` in and out, `thinking: {type: disabled}` passes through; the vendor's `/anthropic` base also works as `kind: anthropic` + `endpoint`) |
@@ -88,7 +88,8 @@ keep access until 2027-01-01), so no reachable configuration exists. The factory
 generic `audio`, and `passthrough` protocols (kling-v1-6, grok-imagine-video,
 sora-2 and brave-search ship example accounts in the default config).
 `protocol: video`
-picks its wire from the account's provider name — `openai`/`azure` → Sora,
+picks its wire from the account's preset `kind` (falling back to the raw
+provider label for hand-written accounts) — `openai` → Sora,
 `siliconflow` → Wan submit/status, `alibaba`/`dashscope` → the DashScope task
 API (the account endpoint is the bare `https://dashscope-intl.aliyuncs.com`;
 some Wan models reject submit `parameters`, so the gateway forwards only what
diff --git a/scripts/live-matrix/live.yaml b/scripts/live-matrix/live.yaml
index f7cf5bc..f1aed63 100644
--- a/scripts/live-matrix/live.yaml
+++ b/scripts/live-matrix/live.yaml
@@ -87,6 +87,14 @@ models:
   # local ollama through the generic openai-compatible path
   - {name: "qwen2.5:0.5b", provider: ollama, input_price_per_1k_micros: 100, output_price_per_1k_micros: 100}
   - {name: grok-voice-latest, provider: xai, protocol: realtime, input_price_per_1k_micros: 1000, output_price_per_1k_micros: 4000}
+  # moderations + audio unit pricing (openai)
+  - {name: omni-moderation-latest, provider: openai, protocol: moderations}
+  - {name: gpt-4o-mini-tts, provider: openai, protocol: tts, unit_price_micros: 20}
+  - {name: whisper-1, provider: openai, protocol: stt, unit_price_micros: 100}
+  # dashscope legacy native wire
+  - {name: qwen-turbo, provider: dashscope-native, protocol: dashscope, input_price_per_1k_micros: 50, output_price_per_1k_micros: 200}
+  # gemini live (realtime bridge, gemini dialect)
+  - {name: gemini-3.1-flash-live-preview, provider: gemini, protocol: realtime, input_price_per_1k_micros: 300, output_price_per_1k_micros: 2500}
   # web search
   - {name: brave-search, provider: brave, protocol: search, unit_price_micros: 5000}
   # rerank vendors
@@ -100,6 +108,11 @@ models:
   - {name: "us.meta.llama3-1-8b-instruct-v1:0", protocol: aws-llama, input_price_per_1k_micros: 220, output_price_per_1k_micros: 220}
 
 accounts:
+  - name: dashscope-native
+    provider: dashscope-native
+    endpoint: https://dashscope-intl.aliyuncs.com
+    api_key_env: DASHSCOPE_API_KEY
+    protocols: ["dashscope"]
   - name: kling
     provider: kling
     endpoint: https://api-singapore.klingai.com
diff --git a/scripts/live-matrix/live_matrix.py b/scripts/live-matrix/live_matrix.py
index 11cd356..b8a4a4f 100644
--- a/scripts/live-matrix/live_matrix.py
+++ b/scripts/live-matrix/live_matrix.py
@@ -40,6 +40,9 @@
     "xai",
     "video",
     "search",
+    "openai-aux",
+    "dashscope-native",
+    "gemini-rt",
     "openai-rt",
     "ollama",
 ]
@@ -455,6 +458,123 @@ def case_video(
     record(name, ok, detail)
 
 
+def case_moderations(gw: Gateway, model: str) -> None:
+    """Moderation verdicts pass through; the call lands one ledger row."""
+    name = f"{model} moderations"
+    before, _ = gw.ledger()
+    st, txt = gw.call("/v1/moderations", {"model": model, "input": ["I want to hurt them badly", "good morning"]})
+    if st != 200:
+        record(name, False, f"HTTP {st}: {txt[:200]}")
+        return
+    j = json.loads(txt)
+    results = j.get("results", [])
+    count, _ = gw.ledger()
+    ok = count == before + 1 and len(results) == 2 and results[0].get("flagged") is True
+    record(name, ok, f"results={len(results)} flagged0={results and results[0].get('flagged')}")
+
+
+def case_tts(gw: Gateway, model: str, text: str = "Hello from the gateway, this is a voice check.") -> None:
+    """TTS bills one unit per input character."""
+    name = f"{model} tts"
+    before, _ = gw.ledger()
+    st, txt = gw.call("/v1/audio/speech", {"model": model, "input": text, "voice": "alloy"})
+    if st != 200:
+        record(name, False, f"HTTP {st}: {txt[:200]}")
+        return
+    count, row = gw.ledger()
+    chars = len(text)
+    ok = (
+        count == before + 1
+        and row["billed_units"] == chars
+        and row["cost_micros"] == chars * MODELS[model]["unit"]
+        and len(txt) > 1000
+    )
+    record(name, ok, f"audio={len(txt)}B ledger units/cost={row['billed_units']}/{row['cost_micros']} expected {chars}/{chars * MODELS[model]['unit']}")
+
+
+def case_stt(gw: Gateway, model: str) -> None:
+    """STT bills whole seconds: the vendor's duration, else the upload's own play length."""
+    import base64
+    import io
+    import math as m
+    import struct
+    import wave
+
+    name = f"{model} stt"
+    seconds, rate = 2, 8000
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as w:
+        w.setnchannels(1)
+        w.setsampwidth(2)
+        w.setframerate(rate)
+        w.writeframes(b"".join(struct.pack("<h", int(12000 * m.sin(2 * m.pi * 440 * i / rate))) for i in range(rate * seconds)))
+    before, _ = gw.ledger()
+    st, txt = gw.call("/v1/audio/transcriptions", {"model": model, "audio_b64": base64.b64encode(buf.getvalue()).decode()})
+    if st != 200:
+        record(name, False, f"HTTP {st}: {txt[:200]}")
+        return
+    count, row = gw.ledger()
+    ok = (
+        count == before + 1
+        and row["billed_units"] == seconds
+        and row["cost_micros"] == seconds * MODELS[model]["unit"]
+    )
+    record(name, ok, f"ledger units/cost={row['billed_units']}/{row['cost_micros']} expected {seconds}/{seconds * MODELS[model]['unit']} text={txt[:60]}")
+
+
+def case_realtime_gemini(gw: Gateway, model: str) -> None:
+    """One Live-API text turn through the bridge: usageMetadata settles the ledger row."""
+    name = f"{model} realtime"
+    try:
+        import websocket
+    except ImportError:
+        record(name, False, "websocket-client not installed")
+        return
+    before, _ = gw.ledger()
+    ws = websocket.create_connection(
+        gw.base.replace("http", "ws", 1) + f"/v1/realtime?model={model}",
+        header={"Authorization": f"Bearer {gw.ak}"},
+        timeout=90,
+    )
+    ws.send(json.dumps({"setup": {"model": f"models/{model}",
+                                   "generationConfig": {"responseModalities": ["AUDIO"]},
+                                   "outputAudioTranscription": {}}}))
+    setup = json.loads(ws.recv())
+    if "setupComplete" not in setup:
+        record(name, False, f"no setupComplete: {json.dumps(setup)[:200]}")
+        ws.close()
+        return
+    ws.send(json.dumps({"clientContent": {"turns": [{"role": "user", "parts": [{"text": "Reply with the single word: pong"}]}], "turnComplete": True}}))
+    usage: dict[str, Any] = {}
+    text = ""
+    deadline = time.time() + 90
+    while time.time() < deadline:
+        frame = json.loads(ws.recv())
+        sc = frame.get("serverContent") or {}
+        text += (sc.get("outputTranscription") or {}).get("text", "")
+        if sc.get("turnComplete"):
+            usage = frame.get("usageMetadata") or {}
+            break
+    ws.close()
+    it = usage.get("promptTokenCount", 0)
+    ot = usage.get("responseTokenCount") or usage.get("candidatesTokenCount") or 0
+    cost = (MODELS[model]["in"] * it) // 1000 + (MODELS[model]["out"] * ot) // 1000
+    count, row = gw.ledger()
+    ok = (
+        count == before + 1
+        and (row["prompt_tokens"], row["completion_tokens"]) == (it, ot)
+        and row["cost_micros"] == cost
+        and not row["estimated"]
+        and "pong" in text.lower()
+    )
+    record(
+        name,
+        ok,
+        f"wire it/ot={it}/{ot} text='{text[:24]}' ledger p/c/cost/est=({row['prompt_tokens']}, {row['completion_tokens']}, "
+        f"{row['cost_micros']}, {row['estimated']}) oracle=({it}, {ot}, {cost})",
+    )
+
+
 def case_search(gw: Gateway, model: str) -> None:
     """One web search bills one unit at the model's unit price; results pass through."""
     name = f"{model} search"
@@ -996,6 +1116,14 @@ def run_group(gw: Gateway, group: str) -> None:
         case_search(gw, "brave-search")
     elif group == "openai-rt":
         case_realtime(gw, "gpt-realtime-mini")
+    elif group == "openai-aux":
+        case_moderations(gw, "omni-moderation-latest")
+        case_tts(gw, "gpt-4o-mini-tts")
+        case_stt(gw, "whisper-1")
+    elif group == "dashscope-native":
+        case_chat(gw, "qwen-turbo")
+    elif group == "gemini-rt":
+        case_realtime_gemini(gw, "gemini-3.1-flash-live-preview")
     elif group == "ollama":
         case_chat(gw, "qwen2.5:0.5b")
         case_chat(gw, "qwen2.5:0.5b", stream=True)