search: Brave Search behind /v1/search

conf/gateway.yaml

@@ -143,9 +143,9 @@ models:
     protocol: video
     provider: kling
     unit_price_micros: 60000
-  - name: vidu-video # generic synchronous vendor shape (the video fallback wire)
+  - name: demo-video # generic synchronous vendor shape (the video fallback wire; no real vendor speaks it)
     protocol: video
-    provider: vidu
+    provider: demo-video
   - name: grok-imagine-video # async video: request_id + poll, billed per second on done
     protocol: video
     provider: xai
@@ -164,6 +164,8 @@ models:
     unit_price_micros: 300000
   - name: brave-search # search family
     protocol: search
+    provider: brave
+    unit_price_micros: 5000
   - name: hunyuan-lite # e2e: PTU failover (ptu account down -> paygo spillover)
     protocol: openai-chat
     provider: tencent
@@ -259,8 +261,8 @@ accounts:
     provider: kling
     priority: 1
     protocols: ["video"]
-  - name: mock-vidu-1
-    provider: vidu
+  - name: mock-generic-video-1
+    provider: demo-video
     priority: 1
     protocols: ["video"]
   - name: mock-xai-1 # e2e: the xAI host answers video with a request_id + poll

crates/config/src/lib.rs

@@ -30,7 +30,7 @@ pub enum ConfigError {
     UnknownProvider { model: String, provider: String },
     #[error("model `{model}` needs either protocol or provider")]
     ModelNeedsDispatch { model: String },
-    #[error("video model `{model}` must pin a provider (the video wire is provider-specific)")]
+    #[error("model `{model}` must pin a provider (its wire is provider-specific)")]
     VideoModelNeedsProvider { model: String },
     #[error("duplicate {kind} name `{name}`")]
     DuplicateName { kind: &'static str, name: String },
@@ -852,12 +852,12 @@ impl GatewayConfig {
     /// Structural and value invariants checked before use: wire types, prices,
     /// token-rate weights, retry_status range, quota/variant shape.
     fn validate(&self) -> Result<(), ConfigError> {
-        // the video wire is provider-specific (dialect by provider name): an
+        // video/search wires are provider-specific (dialect by provider name): an
         // unpinned model would round-robin one vendor's body into another's API
         if let Some(m) = self
             .models
             .iter()
-            .find(|m| m.protocol == "video" && m.provider.is_none())
+            .find(|m| matches!(m.protocol.as_str(), "video" | "search") && m.provider.is_none())
         {
             return Err(ConfigError::VideoModelNeedsProvider {
                 model: m.name.clone(),

crates/engines/src/families.rs

@@ -3,6 +3,8 @@
 //! that boundary. The mock protocol flags byte-level vendor differences as
 //! deferred to a later fidelity pass.
 
+use std::sync::Arc;
+
 use base64::Engine as _;
 use gw_models::{GResult, GatewayError, GatewayRequest, GatewayResponse, TypedParams, VideoParams};
 use gw_protocol::object;
@@ -821,7 +823,7 @@ fn video_outcome(model: &str, v: Value, status: u16) -> EngineOutcome {
 /// Poll an async video generation where it was submitted, in the account's dialect.
 pub async fn video_poll(
     transport: &dyn Transport,
-    account: &gw_models::Account,
+    account: &Arc<gw_models::Account>,
     id: &str,
 ) -> GResult<VideoPoll> {
     let base = account.base_url(VENDOR_SENTINEL);
@@ -908,7 +910,7 @@ fn normalize_video_poll(dialect: VideoDialect, status: u16, body: Value) -> Vide
 /// other dialects answer 404 because their poll carries a URL.
 pub async fn video_content(
     transport: &dyn Transport,
-    account: &gw_models::Account,
+    account: &Arc<gw_models::Account>,
     id: &str,
     poll: &Value,
 ) -> GResult<(u16, String, bytes::Bytes)> {
@@ -970,7 +972,7 @@ fn minimax_content_gap(message: &'static str) -> GatewayError {
 }
 
 fn video_request(
-    account: &gw_models::Account,
+    account: &Arc<gw_models::Account>,
     method: &'static str,
     url: String,
     body: Vec<u8>,
@@ -984,7 +986,7 @@ fn video_request(
 }
 
 fn video_upstream(
-    account: &gw_models::Account,
+    account: &Arc<gw_models::Account>,
     method: &'static str,
     url: String,
     headers: Headers,
@@ -998,13 +1000,13 @@ fn video_upstream(
         body,
         stream: false,
         account: account.name.clone(),
-        replay_account: None,
+        replay_account: Some(Arc::clone(account)),
     }
 }
 
 async fn video_send(
     transport: &dyn Transport,
-    account: &gw_models::Account,
+    account: &Arc<gw_models::Account>,
     method: &'static str,
     url: String,
     body: Vec<u8>,
@@ -1019,33 +1021,69 @@ base_engine!(SearchEngine);
 
 #[async_trait::async_trait]
 impl ModelEngine for SearchEngine {
-    /// Merges the bingsearch/brave/serp/google_custom_search engines.
+    /// Brave or Google CSE on their providers, else the generic mock shape.
     async fn run(&mut self) -> GResult<EngineOutcome> {
         let param = self.base.param()?;
         let (query, count) = match &param.typed {
             Some(TypedParams::Search(p)) => (p.query.as_str(), p.count),
             _ => (self.base.last_message_text(), 3),
         };
         require_non_empty(query, "search query")?;
-        let body = json!({"query": query, "count": count});
-        let (status, v) = self
-            .base
-            .round_trip(&self.base.vendor_url("search"), body)
-            .await?;
-        let titles: Vec<String> = v["results"]
-            .as_array()
+        let q = percent_encoding::utf8_percent_encode(query, percent_encoding::NON_ALPHANUMERIC);
+        let (status, v, results) = match self.base.provider() {
+            "brave" => {
+                let url = format!(
+                    "{}/res/v1/web/search?q={q}&count={count}",
+                    self.base.base_url(VENDOR_SENTINEL)
+                );
+                let headers = vec![
+                    ("accept", "application/json".into()),
+                    ("x-subscription-token", self.base.api_key()),
+                ];
+                let (status, v) = self.search_get(url, headers).await?;
+                (status, v, "/web/results")
+            }
+            _ => {
+                let body = json!({"query": query, "count": count});
+                let (status, v) = self
+                    .base
+                    .round_trip(&self.base.vendor_url("search"), body)
+                    .await?;
+                (status, v, "/results")
+            }
+        };
+        let titles: Vec<String> = v
+            .pointer(results)
+            .and_then(Value::as_array)
             .map(|rs| {
                 rs.iter()
                     .filter_map(|r| r["title"].as_str().map(str::to_owned))
                     .collect()
             })
             .unwrap_or_default();
-        Ok(family_outcome(
-            titles.join("; "),
-            &param.model_name,
-            v,
-            status,
-        ))
+        let mut out = family_outcome(titles.join("; "), &param.model_name, v, status);
+        out.response.billed_units = 1;
+        Ok(out)
+    }
+}
+
+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)
     }
 }
 
@@ -2144,7 +2182,7 @@ mod tests {
         assert_eq!(submit.http_status, 502);
         assert!(submit.message.contains("minimax base_resp 1008"));
 
-        let poll = match video_poll(transport.as_ref(), account.as_ref(), "task-1").await {
+        let poll = match video_poll(transport.as_ref(), &account, "task-1").await {
             Ok(_) => panic!("MiniMax business error accepted as a successful poll"),
             Err(e) => e,
         };

crates/engines/src/http_transport.rs

@@ -273,22 +273,35 @@ impl Transport for DispatchTransport {
     }
 }
 
-/// The vendor's `Retry-After` seconds, capped; unparseable waits at least a
-/// second, absent falls back to the connect path's linear backoff.
+/// The vendor's retry delay, capped; standard `Retry-After` takes precedence,
+/// then seconds-until-reset, else the connect path's linear backoff.
 fn status_backoff(headers: &reqwest::header::HeaderMap, attempt: u32) -> Duration {
     const MAX_RETRY_AFTER: Duration = Duration::from_secs(30);
     const MIN_HEADER_WAIT: Duration = Duration::from_secs(1);
-    match headers
+    if let Some(v) = headers
         .get(reqwest::header::RETRY_AFTER)
         .and_then(|v| v.to_str().ok())
     {
-        Some(v) => v
+        return v
             .trim()
             .parse::<u64>()
             .map(|secs| Duration::from_secs(secs).min(MAX_RETRY_AFTER))
-            .unwrap_or_else(|_| (RETRY_BACKOFF * attempt).max(MIN_HEADER_WAIT)),
-        None => RETRY_BACKOFF * attempt,
+            .unwrap_or_else(|_| (RETRY_BACKOFF * attempt).max(MIN_HEADER_WAIT));
     }
+    headers
+        .get("x-ratelimit-reset")
+        .and_then(|v| v.to_str().ok())
+        .and_then(|v| {
+            v.split(',')
+                .filter_map(|secs| secs.trim().parse::<u64>().ok())
+                .min()
+        })
+        .map(|secs| {
+            Duration::from_secs(secs)
+                .max(MIN_HEADER_WAIT)
+                .min(MAX_RETRY_AFTER)
+        })
+        .unwrap_or(RETRY_BACKOFF * attempt)
 }
 
 /// A live SSE byte stream that yields one terminal error when no chunk arrives
@@ -473,5 +486,8 @@ mod tests {
             Duration::from_secs(1),
             "an unparsed header still waits at least a second"
         );
+        h.remove(reqwest::header::RETRY_AFTER);
+        h.insert("x-ratelimit-reset", "1, 1419704".parse().unwrap());
+        assert_eq!(status_backoff(&h, 1), Duration::from_secs(1));
     }
 }

crates/engines/src/transport.rs

@@ -878,6 +878,35 @@ impl MockTransport {
     }
 
     fn search_reply(&self, req: &UpstreamRequest) -> GResult<UpstreamResponse> {
+        if req.url.contains("/res/v1/web/search") {
+            let params = req.url.split_once('?').map_or("", |(_, params)| params);
+            let param = |name| {
+                params.split('&').find_map(|pair| {
+                    let (key, value) = pair.split_once('=')?;
+                    (key == name).then_some(value)
+                })
+            };
+            let q = param("q")
+                .map(|q| {
+                    percent_encoding::percent_decode_str(q)
+                        .decode_utf8_lossy()
+                        .into_owned()
+                })
+                .unwrap_or_default();
+            let count = param("count")
+                .and_then(|count| count.parse::<i64>().ok())
+                .unwrap_or(3)
+                .clamp(1, 20);
+            let results: Vec<Value> = (0..count)
+                .map(|i| {
+                    json!({"title": format!("brave result {} for {q}", i + 1),
+                           "url": format!("mock://brave/{}", i + 1),
+                           "description": format!("[mock-brave] about {q}")})
+                })
+                .collect();
+            return Self::ok_json(json!({"query": {"original": q},
+                                        "web": {"results": results}}));
+        }
         let body = Self::parse(&req.body, "search")?;
         let q = body["query"].as_str().unwrap_or("");
         let n = body["count"].as_i64().unwrap_or(3).clamp(1, 10);
@@ -994,6 +1023,9 @@ impl Transport for MockTransport {
         if req.protocol == Protocol::Video {
             return self.video_reply(&req);
         }
+        if req.protocol == Protocol::Search {
+            return self.search_reply(&req);
+        }
         let u = req.url.as_str();
         if u.contains("/model/") {
             self.bedrock_reply(&req)
@@ -1021,8 +1053,6 @@ impl Transport for MockTransport {
             self.moderations_reply(&req)
         } else if u.contains("/rerank") {
             self.rerank_reply(&req)
-        } else if u.contains("/search") {
-            self.search_reply(&req)
         } else if u.contains("/responses") {
             self.responses_reply(&req)
         } else if u.contains("/v1/completions") {

crates/engines/tests/http_transport_wire.rs

@@ -321,10 +321,13 @@ async fn per_account_policy_and_connect_retry() {
         "an account dropped from the reload falls back to the default"
     );
 
-    let closed = {
+    // re-draw if another process steals the freed port before we dial it (CI runners do)
+    let closed = std::iter::repeat_with(|| {
         let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
         l.local_addr().unwrap()
-    };
+    })
+    .find(|addr| std::net::TcpStream::connect_timeout(addr, Duration::from_millis(50)).is_err())
+    .unwrap();
     let started = std::time::Instant::now();
     let err = transport
         .send(UpstreamRequest {

crates/engines/tests/request_construction.rs

@@ -855,7 +855,7 @@ async fn video_request_shape() {
     );
     let req = typed_req(
         Protocol::Video,
-        "vidu-video",
+        "demo-video",
         TypedParams::Video(VideoParams {
             prompt: "a dog surfing".into(),
             duration_seconds: Some(5),
@@ -866,7 +866,7 @@ async fn video_request_shape() {
     );
     let _ = VideoEngine::new(req, t.clone()).run().await.unwrap();
     let b = t.body_json();
-    assert_eq!(b["model"], "vidu-video");
+    assert_eq!(b["model"], "demo-video");
     assert_eq!(b["prompt"], "a dog surfing");
     assert_eq!(b["duration"], 5);
     assert_eq!(b["resolution"], "1080p");

crates/models/src/params.rs

@@ -19,9 +19,9 @@ pub enum TypedParams {
     AudioTts(TtsParams),
     /// whisper / azure / google / ... (speech-to-text)
     AudioStt(SttParams),
-    /// sora / veo / kling / runway / vidu / minimax (video generation)
+    /// Video generation (Sora / Wan / DashScope / Hailuo / Kling dialects + a generic shape).
     Video(VideoParams),
-    /// bing / brave / serp / google custom search
+    /// Web search (Brave dialect + a generic shape).
     Search(SearchParams),
     /// content moderation (openai moderations shape)
     Moderation(ModerationParams),