From de55b64afe31c1cb63796c3a284b0669fb18e03e Mon Sep 17 00:00:00 2001
From: CMGS <ilskdw@gmail.com>
Date: Thu, 20 Aug 2026 03:14:05 +0800
Subject: [PATCH 1/7] live coverage: moderations, OpenAI audio, DashScope
 legacy wire, Gemini Live realtime
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Four mock-only surfaces go live. OpenAI moderations and TTS/STT get
matrix cases (characters and whole seconds land as billed units — the
whisper run exercised the vendor-duration path). The DashScope legacy
native wire runs against the real intl endpoint. The realtime bridge's
Gemini dialect connects for real: the Live socket is one bidi RPC authed
by key with the model in the setup frame, its frames are binary, and a
turn is admitted on clientContent.turnComplete — the dialect's own
generation signal — so the old cannot-gate 501 goes away; usageMetadata
settles the turn non-estimated (verified live: prompt 146 / completion
24 audio tokens, transcript through outputAudioTranscription).
---
 crates/config/src/lib.rs           |   2 +-
 crates/engines/src/realtime.rs     |  31 +++++++
 crates/server/tests/e2e.rs         |  18 ++--
 crates/views/src/lib.rs            |  27 +++---
 docs/api.md                        |   7 ++
 docs/development.md                |   8 +-
 docs/providers.md                  |   2 +-
 scripts/live-matrix/live.yaml      |  13 +++
 scripts/live-matrix/live_matrix.py | 128 +++++++++++++++++++++++++++++
 9 files changed, 212 insertions(+), 24 deletions(-)

diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs
index 620b103..8e64785 100644
--- a/crates/config/src/lib.rs
+++ b/crates/config/src/lib.rs
@@ -651,7 +651,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.
diff --git a/crates/engines/src/realtime.rs b/crates/engines/src/realtime.rs
index 4adc7b0..e4f74a7 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 {
@@ -332,6 +342,27 @@ 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}}}});
+        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/server/tests/e2e.rs b/crates/server/tests/e2e.rs
index ce4f3da..15a327a 100644
--- a/crates/server/tests/e2e.rs
+++ b/crates/server/tests/e2e.rs
@@ -2795,7 +2795,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 +2821,18 @@ 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"
-    );
+    // the dialect is governable (clientContent.turnComplete admits a turn), so
+    // the upgrade proceeds; the dead upstream then surfaces as an error frame
+    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/views/src/lib.rs b/crates/views/src/lib.rs
index cfccbbc..ec42630 100644
--- a/crates/views/src/lib.rs
+++ b/crates/views/src/lib.rs
@@ -22,7 +22,7 @@ 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,
+    is_client_turn, realtime_audio_tokens, realtime_output_delta, realtime_turn_started,
     realtime_usage,
 };
 use gw_handler::{BatchItem, OfflineHandler, OnlineHandler};
@@ -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))
     }
@@ -770,7 +761,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.provider);
+    // 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 +784,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 {
@@ -847,7 +846,7 @@ async fn realtime_bridge(
                     // —
                     // upstream rejects the duplicate and a raced accept is caught by the
                     // response.created gate
-                    if is_response_create(&frame) && pending.is_none() {
+                    if is_client_turn(&account.provider, &frame) && pending.is_none() {
                         match realtime_gate(&s, &ak, &rtm, &hint).await {
                             Ok(admit) => {
                                 pending = Some(RealtimeTurn::new(admit));
diff --git a/docs/api.md b/docs/api.md
index b41d1f0..7fa77b7 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -229,6 +229,13 @@ 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 provider: a `gemini` provider bridges Google's
+Live API socket (binary frames relayed as-is, the key on the query string) and
+admits each turn on `clientContent.turnComplete` — the dialect's own
+generation signal — settling the `usageMetadata` that rides the completed
+turn; every other provider 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..0ff1fcc 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`) |
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)

From ca58229ba6e4898b09ed97fbd5f3718378973676 Mon Sep 17 00:00:00 2001
From: CMGS <ilskdw@gmail.com>
Date: Thu, 20 Aug 2026 03:41:33 +0800
Subject: [PATCH 2/7] =?UTF-8?q?review:=20loc=20round=20=E2=80=94=20inline?=
 =?UTF-8?q?=20the=20one-caller=20search=20send,=20one=20head=20for=20the?=
 =?UTF-8?q?=20video=20reads?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The whole-repo audit (541 pub items swept, 11 mechanisms adjudicated)
returned two cuts: SearchEngine::search_get lost its second caller with
the CSE removal and folds back into the Brave arm; the two video read
routes shared a verbatim admit+resolve+poll head that now lives whole in
admit_video_job. Three audit-flagged coverage gaps get their tests:
unpinned video/search models fail config load, a Kling business-error
envelope is a 502 on submit and poll, and a delivered sync video reply
must not register its task_id as an async job.
---
 crates/config/src/lib.rs       |  9 +++++
 crates/engines/src/families.rs | 73 ++++++++++++++++++++++++----------
 crates/server/tests/e2e.rs     | 18 +++++++--
 crates/views/src/lib.rs        | 59 ++++++++++++---------------
 4 files changed, 102 insertions(+), 57 deletions(-)

diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs
index 8e64785..bc8f264 100644
--- a/crates/config/src/lib.rs
+++ b/crates/config/src/lib.rs
@@ -1361,6 +1361,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/families.rs b/crates/engines/src/families.rs
index 255e816..aa431a5 100644
--- a/crates/engines/src/families.rs
+++ b/crates/engines/src/families.rs
@@ -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/server/tests/e2e.rs b/crates/server/tests/e2e.rs
index 15a327a..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 [
@@ -2821,8 +2835,6 @@ models:
         .unwrap();
     req.headers_mut()
         .insert("authorization", "Bearer ak-rt".parse().unwrap());
-    // the dialect is governable (clientContent.turnComplete admits a turn), so
-    // the upgrade proceeds; the dead upstream then surfaces as an error frame
     let (mut ws, resp) = tokio_tungstenite::connect_async(req).await.unwrap();
     assert_eq!(resp.status().as_u16(), 101);
     use futures::StreamExt;
diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs
index ec42630..25ba2e0 100644
--- a/crates/views/src/lib.rs
+++ b/crates/views/src/lib.rs
@@ -4005,12 +4005,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();
@@ -4024,7 +4031,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(
@@ -4092,25 +4106,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
@@ -4120,20 +4123,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");
     }

From fc89987419f51dff481e4a0a212ceade6b3b8c74 Mon Sep 17 00:00:00 2001
From: CMGS <ilskdw@gmail.com>
Date: Thu, 20 Aug 2026 04:20:13 +0800
Subject: [PATCH 3/7] realtime: the dialect key survives preset naming; gemini
 pins, refuses, estimates
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Codex round: the wire dialect now keys on the account's preset kind
(Account.kind, stamped at providers: expansion) falling back to the raw
provider label, so a provider named gemini-prod still speaks the Live
socket — the same fix covers the video and search dialects. The bridge
rewrites a gemini setup.model to the entitled served model (the socket
carries no model, so the setup frame was a billing/entitlement bypass),
refuses realtimeInput loudly until audio turns have an admission point,
and the abort estimate counts gemini's serverContent parts so a
disconnect after delivery bills instead of refunding.
---
 crates/config/src/lib.rs       |  4 ++++
 crates/engines/src/base.rs     |  2 +-
 crates/engines/src/families.rs |  4 ++--
 crates/engines/src/realtime.rs | 17 +++++++++++++++++
 crates/models/src/request.rs   | 12 ++++++++++++
 crates/state/src/lib.rs        |  1 +
 crates/views/src/lib.rs        | 35 +++++++++++++++++++++++++++-------
 7 files changed, 65 insertions(+), 10 deletions(-)

diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs
index bc8f264..aaaf942 100644
--- a/crates/config/src/lib.rs
+++ b/crates/config/src/lib.rs
@@ -247,6 +247,8 @@ fn weight_one() -> f64 {
 pub struct AccountConf {
     pub name: String,
     pub provider: String,
+    #[serde(skip)]
+    pub kind: String,
     #[serde(default = "default_priority")]
     pub priority: i32,
     /// "ptu" (provisioned throughput, preferred) or "paygo" (default).
@@ -800,6 +802,7 @@ impl GatewayConfig {
                 self.accounts.push(AccountConf {
                     name: p.name.clone(),
                     provider: p.name.clone(),
+                    kind: p.kind.clone(),
                     priority: 1,
                     tier: String::new(),
                     cost_input_price_per_1k_micros: 0,
@@ -816,6 +819,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()
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 aa431a5..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",
diff --git a/crates/engines/src/realtime.rs b/crates/engines/src/realtime.rs
index e4f74a7..c7adc2e 100644
--- a/crates/engines/src/realtime.rs
+++ b/crates/engines/src/realtime.rs
@@ -88,6 +88,19 @@ pub fn realtime_turn_started(provider: &str, frame: &Value) -> bool {
 /// Text and opaque payload units carried by one OpenAI-dialect output-delta
 /// frame. Opaque units count base64 quanta.
 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);
@@ -342,6 +355,10 @@ 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 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"})
diff --git a/crates/models/src/request.rs b/crates/models/src/request.rs
index 023e29d..f336d44 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,
@@ -250,6 +252,16 @@ pub mod domain {
 
         /// Base URL for building the upstream request: the account's endpoint if
         /// set, else the given `mock://…` sentinel.
+        /// 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
+            }
+        }
+
         pub fn base_url<'a>(&'a self, mock_sentinel: &'a str) -> &'a str {
             if self.endpoint.is_empty() {
                 mock_sentinel
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 25ba2e0..f0d0bb1 100644
--- a/crates/views/src/lib.rs
+++ b/crates/views/src/lib.rs
@@ -762,7 +762,7 @@ async fn realtime_bridge(
         .replacen("https://", "wss://", 1)
         .replacen("http://", "ws://", 1);
     let key = account.api_key().unwrap_or_else(|| "mock".to_owned());
-    let gemini = gw_engines::realtime::is_gemini_realtime(&account.provider);
+    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!(
@@ -825,6 +825,27 @@ 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 have no admission point yet: refuse loudly, keep the session
+                        if frame.get("realtimeInput").is_some() || frame.get("realtime_input").is_some() {
+                            if cl_tx
+                                .send(rt_error_frame(
+                                    ErrClass::Validation,
+                                    "realtimeInput is not wired; send clientContent turns",
+                                ))
+                                .await
+                                .is_err()
+                            {
+                                break;
+                            }
+                            continue;
+                        }
+                    }
                     match rt_inbound_policy(&s, &ak, &hint, &mut frame).await {
                         Err(reason) => {
                             if cl_tx
@@ -846,7 +867,7 @@ async fn realtime_bridge(
                     // —
                     // upstream rejects the duplicate and a raced accept is caught by the
                     // response.created gate
-                    if is_client_turn(&account.provider, &frame) && pending.is_none() {
+                    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));
@@ -894,13 +915,13 @@ async fn realtime_bridge(
                     Some(mut v) => {
                         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)) => {
@@ -912,7 +933,7 @@ async fn realtime_bridge(
                                     relay = false;
                                 }
                             }
-                        } else if let Some((it, ot)) = realtime_usage(&account.provider, &v) {
+                        } else if let Some((it, ot)) = realtime_usage(account.wire_kind(), &v) {
                             // turn boundary: settle the admitted turn; one with no gated turn bills
                             // unreserved
                             match pending.take() {
@@ -925,7 +946,7 @@ async fn realtime_bridge(
                                         gw_models::TokenInput {
                                             prompt: it,
                                             completion: ot,
-                                            ..turn_audio(&account.provider, &v)
+                                            ..turn_audio(account.wire_kind(), &v)
                                         },
                                         false,
                                     )
@@ -961,7 +982,7 @@ async fn realtime_bridge(
                                         gw_models::TokenInput {
                                             prompt: it,
                                             completion: ot,
-                                            ..turn_audio(&account.provider, &v)
+                                            ..turn_audio(account.wire_kind(), &v)
                                         },
                                         false,
                                     )

From 729f983fec5789d5e55e638635bad4ec9c2548c3 Mon Sep 17 00:00:00 2001
From: CMGS <ilskdw@gmail.com>
Date: Thu, 20 Aug 2026 04:34:00 +0800
Subject: [PATCH 4/7] realtime: one governed gemini turn at a time

Codex round 2: a clientContent.turnComplete arriving while a turn is
pending would relay ungated and bill unreserved once the interrupted
turn consumed the reservation; it now answers the same in-band error as
realtimeInput. Docs carry the dialect contract.
---
 crates/views/src/lib.rs | 10 +++++++---
 docs/api.md             | 15 +++++++++------
 docs/providers.md       |  3 ++-
 3 files changed, 18 insertions(+), 10 deletions(-)

diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs
index f0d0bb1..29df3ca 100644
--- a/crates/views/src/lib.rs
+++ b/crates/views/src/lib.rs
@@ -831,12 +831,16 @@ async fn realtime_bridge(
                             setup.insert("model".into(), format!("models/{}", rtm.served).into());
                             forward = UMsg::text(frame.to_string());
                         }
-                        // audio-driven turns have no admission point yet: refuse loudly, keep the session
-                        if frame.get("realtimeInput").is_some() || frame.get("realtime_input").is_some() {
+                        // 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,
-                                    "realtimeInput is not wired; send clientContent turns",
+                                    "one governed clientContent turn at a time; realtimeInput is not wired",
                                 ))
                                 .await
                                 .is_err()
diff --git a/docs/api.md b/docs/api.md
index 7fa77b7..66f33f1 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -229,12 +229,15 @@ 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 provider: a `gemini` provider bridges Google's
-Live API socket (binary frames relayed as-is, the key on the query string) and
-admits each turn on `clientContent.turnComplete` — the dialect's own
-generation signal — settling the `usageMetadata` that rides the completed
-turn; every other provider speaks the OpenAI Realtime shape and admits on
-`response.create`.
+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
 
diff --git a/docs/providers.md b/docs/providers.md
index 0ff1fcc..de73b37 100644
--- a/docs/providers.md
+++ b/docs/providers.md
@@ -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`/`azure` → 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

From 8d3c382599cb6cc5aaf1e5c97a5113815210b869 Mon Sep 17 00:00:00 2001
From: CMGS <ilskdw@gmail.com>
Date: Thu, 20 Aug 2026 10:04:54 +0800
Subject: [PATCH 5/7] =?UTF-8?q?review:=20final=20sweep=20=E2=80=94=20docs?=
 =?UTF-8?q?=20reattach,=20the=20push-time=20kind=20stays=20empty?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The wire_kind insertion had orphaned base_url's doc onto the wrong item;
the synthesized account's push-time kind was dead (the provider loop
stamps every matching account two lines later); the output-delta doc
names both dialects; the stale azure→Sora doc claim goes.
---
 crates/config/src/lib.rs       | 3 ++-
 crates/engines/src/realtime.rs | 4 ++--
 crates/models/src/request.rs   | 4 ++--
 docs/api.md                    | 2 +-
 docs/providers.md              | 2 +-
 5 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs
index aaaf942..b395a29 100644
--- a/crates/config/src/lib.rs
+++ b/crates/config/src/lib.rs
@@ -247,6 +247,7 @@ 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")]
@@ -802,7 +803,7 @@ impl GatewayConfig {
                 self.accounts.push(AccountConf {
                     name: p.name.clone(),
                     provider: p.name.clone(),
-                    kind: p.kind.clone(),
+                    kind: String::new(),
                     priority: 1,
                     tier: String::new(),
                     cost_input_price_per_1k_micros: 0,
diff --git a/crates/engines/src/realtime.rs b/crates/engines/src/realtime.rs
index c7adc2e..ed24280 100644
--- a/crates/engines/src/realtime.rs
+++ b/crates/engines/src/realtime.rs
@@ -85,8 +85,8 @@ 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() {
diff --git a/crates/models/src/request.rs b/crates/models/src/request.rs
index f336d44..ce12d5b 100644
--- a/crates/models/src/request.rs
+++ b/crates/models/src/request.rs
@@ -250,8 +250,6 @@ pub mod domain {
             self.tier == gw_consts::account_tier::PTU
         }
 
-        /// Base URL for building the upstream request: the account's endpoint if
-        /// set, else the given `mock://…` sentinel.
         /// 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 {
@@ -262,6 +260,8 @@ pub mod domain {
             }
         }
 
+        /// 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 {
             if self.endpoint.is_empty() {
                 mock_sentinel
diff --git a/docs/api.md b/docs/api.md
index 66f33f1..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,
diff --git a/docs/providers.md b/docs/providers.md
index de73b37..02c7856 100644
--- a/docs/providers.md
+++ b/docs/providers.md
@@ -89,7 +89,7 @@ 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 preset `kind` (falling back to the raw
-provider label for hand-written accounts) — `openai`/`azure` → Sora,
+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

From e74bec4f772123e33a48e196a9395fccda7c444b Mon Sep 17 00:00:00 2001
From: CMGS <ilskdw@gmail.com>
Date: Thu, 20 Aug 2026 10:28:18 +0800
Subject: [PATCH 6/7] realtime: a bare gemini turnComplete settles from the
 latest usage snapshot
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The Live API may send cumulative usageMetadata on any server frame; only
reading it off the turnComplete frame degraded a split delivery to the
prompt=0 output estimate. Each usage-bearing frame overwrites one
snapshot, every boundary consumes it — used only when the boundary
itself is bare, so same-frame usage and the estimate/refund fallbacks
are unchanged.
---
 crates/engines/src/realtime.rs | 44 ++++++++++++++++++++++++++++------
 crates/views/src/lib.rs        | 16 ++++++++++---
 2 files changed, 50 insertions(+), 10 deletions(-)

diff --git a/crates/engines/src/realtime.rs b/crates/engines/src/realtime.rs
index ed24280..897ab42 100644
--- a/crates/engines/src/realtime.rs
+++ b/crates/engines/src/realtime.rs
@@ -80,6 +80,28 @@ 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.
+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"
@@ -122,13 +144,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" {
@@ -355,6 +371,20 @@ 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"}}
         ]}}});
diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs
index 29df3ca..ae1066a 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_client_turn, realtime_audio_tokens, realtime_output_delta, realtime_turn_started,
-    realtime_usage,
+    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::{
@@ -811,6 +811,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 {
@@ -917,6 +919,9 @@ 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.wire_kind(), &v).is_some() {
@@ -937,7 +942,12 @@ async fn realtime_bridge(
                                     relay = false;
                                 }
                             }
-                        } else if let Some((it, ot)) = realtime_usage(account.wire_kind(), &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() {

From 0276e84faed9ba5d2206b68ed44a2124efa71667 Mon Sep 17 00:00:00 2001
From: CMGS <ilskdw@gmail.com>
Date: Thu, 20 Aug 2026 11:11:58 +0800
Subject: [PATCH 7/7] realtime: an aborted gemini turn settles from the usage
 snapshot

A session dropped before turnComplete used to bill the byte estimate
with prompt=0 even when a periodic usageMetadata frame had already
reported real counts. A leftover snapshot now bills real prompt, audio
modality shares, and max(reported, delivered-estimate) completion; the
estimate/refund path is unchanged when no snapshot exists. The audio
extraction takes the usage object directly (gemini_audio_tokens) instead
of wrapping it in a synthetic frame.
---
 crates/engines/src/realtime.rs | 36 +++++++++++++++++++---------------
 crates/views/src/lib.rs        | 31 +++++++++++++++++++++--------
 2 files changed, 43 insertions(+), 24 deletions(-)

diff --git a/crates/engines/src/realtime.rs b/crates/engines/src/realtime.rs
index 897ab42..b4252e1 100644
--- a/crates/engines/src/realtime.rs
+++ b/crates/engines/src/realtime.rs
@@ -93,6 +93,25 @@ pub fn gemini_usage_update(provider: &str, frame: &Value) -> Option<Value> {
 }
 
 /// (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"]
@@ -164,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"];
     (
diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs
index ae1066a..99bb9ae 100644
--- a/crates/views/src/lib.rs
+++ b/crates/views/src/lib.rs
@@ -22,7 +22,7 @@ use gw_consts::ErrClass;
 use gw_dag::DagContext;
 use gw_engines::SharedTransport;
 use gw_engines::realtime::{
-    gemini_tokens, gemini_usage_update, is_client_turn, realtime_audio_tokens,
+    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};
@@ -741,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,
@@ -869,10 +868,7 @@ 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
+                    // 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) => {
@@ -1069,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;