Get rid of the unneeded Arc<Mutex> around the gateway client

crates/harness-gateway-client/src/catalog.rs

@@ -7,6 +7,7 @@ use runvil::model::{ModelCatalog, ModelId, ModelInfo, ThinkingMode};
 use serde::Deserialize;
 
 use crate::failure::{malformed, transport_failure, with_detail};
+use crate::transport::shared_client;
 use crate::wire::classify::classify_http_failure;
 use crate::wire::stream::escape_controls;
 
@@ -115,17 +116,6 @@ async fn read_error_body_bounded(
     Ok(escaped)
 }
 
-/// Returns the process-wide catalog HTTP client, building it once on first use.
-///
-/// A single reusable client (MODEL-018) lets catalog fetches share one
-/// connection pool and transport configuration rather than each constructing a
-/// throwaway client with its own pool. The returned handle is a cheap clone of
-/// the shared client (its state is reference-counted internally).
-fn catalog_client() -> reqwest::Client {
-    static CATALOG_CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
-    CATALOG_CLIENT.get_or_init(reqwest::Client::new).clone()
-}
-
 /// Sends a bearer-authed GET through the shared client (MODEL-018) and
 /// returns the success response, classifying every failure the same way for
 /// each gateway endpoint: `Transport` (or `Timeout`) when the send fails, the
@@ -135,7 +125,7 @@ fn catalog_client() -> reqwest::Client {
 /// the [`reqwest::Error`] to the message under the same timeout marking as a
 /// send failure.
 async fn get_authed(url: String, token: &str) -> std::result::Result<reqwest::Response, Error> {
-    let response = catalog_client()
+    let response = shared_client()
         .get(url)
         .bearer_auth(token)
         .send()

crates/harness-gateway-client/src/search.rs

@@ -14,6 +14,7 @@ use runvil::error::{Error, ErrorCode};
 use runvil::tools::tool_error;
 
 use crate::config::{GatewayEndpoint, SecretString};
+use crate::transport::shared_client;
 
 /// The largest error body kept for diagnostics, in bytes.
 const MAX_ERROR_BODY: usize = 2000;
@@ -142,7 +143,7 @@ impl GatewaySearch {
         timeout: Duration,
     ) -> GatewaySearch {
         GatewaySearch {
-            http: reqwest::Client::new(),
+            http: shared_client(),
             base_url: endpoint.url,
             key,
             timeout,

crates/harness-gateway-client/src/transport.rs

@@ -42,6 +42,18 @@ pub struct GatewayChat {
     max_response_bytes: u64,
 }
 
+/// Returns the process-wide HTTP client, building it once on first use.
+///
+/// Every request this crate sends goes through it (MODEL-018), so every
+/// [`GatewayChat`], `GatewaySearch`, and catalog fetch shares one connection
+/// pool however often the caller rebuilds them. The pool is keyed by scheme,
+/// host, and port, so a gateway at a new address gets its own connections.
+/// The handle is a reference-counted clone.
+pub(crate) fn shared_client() -> reqwest::Client {
+    static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
+    CLIENT.get_or_init(reqwest::Client::new).clone()
+}
+
 /// Default longest wait for the next receive.
 const DEFAULT_RECEIVE_TIMEOUT: Duration = Duration::from_secs(120);
 /// Default response-body ceiling.
@@ -93,7 +105,7 @@ impl GatewayChat {
     #[must_use]
     pub fn new(endpoint: GatewayEndpoint, key: SecretString) -> GatewayChat {
         GatewayChat {
-            transport: GatewayTransport::Http(reqwest::Client::new()),
+            transport: GatewayTransport::Http(shared_client()),
             base_url: endpoint.url,
             key: Some(key),
             receive_timeout: DEFAULT_RECEIVE_TIMEOUT,
@@ -116,7 +128,7 @@ impl GatewayChat {
     #[must_use]
     pub fn keyless(endpoint: GatewayEndpoint) -> GatewayChat {
         GatewayChat {
-            transport: GatewayTransport::Http(reqwest::Client::new()),
+            transport: GatewayTransport::Http(shared_client()),
             base_url: endpoint.url,
             key: None,
             receive_timeout: DEFAULT_RECEIVE_TIMEOUT,

crates/harness-gateway-client/src/transport/tests.rs

@@ -8,6 +8,7 @@ use super::*;
 
 mod env;
 mod limits;
+mod pool;
 mod streaming;
 
 /// Serves `app` on a loopback port and returns its `/v1` root.

crates/harness-gateway-client/src/transport/tests/pool.rs

@@ -0,0 +1,59 @@
+//! The shared connection pool: clients built separately reuse one
+//! connection to the same gateway.
+
+use std::net::SocketAddr;
+use std::sync::{Arc, Mutex, PoisonError};
+
+use axum::Router;
+use axum::extract::ConnectInfo;
+use axum::routing::post;
+use runvil::model::Message;
+
+use super::*;
+
+#[tokio::test]
+async fn two_clients_built_separately_send_their_rounds_over_one_connection() {
+    let peers: Arc<Mutex<Vec<SocketAddr>>> = Arc::default();
+    let seen = Arc::clone(&peers);
+    let app = Router::new().route(
+        "/v1/chat/completions",
+        post(move |ConnectInfo(peer): ConnectInfo<SocketAddr>| {
+            let seen = Arc::clone(&seen);
+            async move {
+                seen.lock()
+                    .unwrap_or_else(PoisonError::into_inner)
+                    .push(peer);
+                (
+                    [(axum::http::header::CONTENT_TYPE, "text/event-stream")],
+                    ok_stream(),
+                )
+            }
+        }),
+    );
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        axum::serve(
+            listener,
+            app.into_make_service_with_connect_info::<SocketAddr>(),
+        )
+        .await
+        .unwrap();
+    });
+    let endpoint = GatewayEndpoint::new(&format!("http://{addr}/v1")).expect("valid test endpoint");
+    let key = SecretString::new("tok").expect("non-empty test key");
+
+    for _ in 0..2 {
+        GatewayChat::new(endpoint.clone(), key.clone())
+            .complete(&[Message::user("hi")], None, &openai_options(), |_| {})
+            .await
+            .expect("the mock gateway answers the round");
+    }
+
+    let peers = peers.lock().unwrap_or_else(PoisonError::into_inner).clone();
+    assert_eq!(peers.len(), 2, "both rounds reach the gateway");
+    assert_eq!(
+        peers[0], peers[1],
+        "the second client reuses the first client's connection"
+    );
+}

crates/workshop/server/src/agents/broker.rs

@@ -5,18 +5,19 @@
 //!
 //! The Workshop broker holds the server's [`Registry`] and reads the
 //! gateway through it on every call, as the search provider does, so a
-//! replaced gateway serves the next round. It keeps the Gateway broker of
-//! the last generation it served under. A model list first waits until
-//! the menu's catalog holds a chat-capable model, so a conversation
-//! launched before the Gateway's catalog arrives starts its run only once
-//! it can bind one.
+//! replaced gateway serves the next round. It builds a Gateway broker per
+//! call; the Gateway client's shared connection pool makes that a few
+//! clones and keeps connections alive across rounds. A model list first
+//! waits until the menu's catalog holds a chat-capable model, so a
+//! conversation launched before the Gateway's catalog arrives starts its
+//! run only once it can bind one.
 //!
 //! Each round reads the dropdown's pick through the registry as it is
 //! sent, so a model picked between rounds serves the next one; a round
 //! already in flight finishes on its model.
 
 use std::num::NonZeroU32;
-use std::sync::{Arc, Mutex, PoisonError};
+use std::sync::Arc;
 
 use harness::{BoxFuture, InferenceBroker};
 use harness_gateway_client::{GatewayBroker, StreamDelta};
@@ -40,37 +41,22 @@ pub(super) struct WorkshopBroker {
     /// The subsystem registry the gateway and menu handles are read
     /// through.
     registry: Registry,
-    /// The broker built for the last generation served under, shared with
-    /// the model lists still waiting.
-    cached: Arc<Mutex<Option<(u64, GatewayBroker)>>>,
 }
 
 impl WorkshopBroker {
     /// Builds the broker over the server's subsystem registry.
     pub(super) fn new(registry: Registry) -> Self {
-        Self {
-            registry,
-            cached: Arc::new(Mutex::new(None)),
-        }
+        Self { registry }
     }
 
-    /// The Gateway broker for the current generation, built on the first
-    /// call under it. No usable gateway fails as `Unavailable` with the
-    /// code's fixed phrase, so no URL or key reaches the message.
+    /// A Gateway broker for the current generation. No usable gateway
+    /// fails as `Unavailable` with the code's fixed phrase, so no URL or
+    /// key reaches the message.
     fn current(&self) -> Result<GatewayBroker, Error> {
         let unavailable = ErrorCode::Unavailable;
         let gateway = usable_gateway(&self.registry)
             .ok_or_else(|| completion_error(unavailable, phrase(unavailable)))?;
-        // A poisoned lock holds a pair written whole by one store.
-        let mut cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner);
-        if let Some((generation, broker)) = cached.as_ref()
-            && *generation == gateway.generation
-        {
-            return Ok(broker.clone());
-        }
-        let broker = GatewayBroker::new(gateway.endpoint, gateway.key);
-        *cached = Some((gateway.generation, broker.clone()));
-        Ok(broker)
+        Ok(GatewayBroker::new(gateway.endpoint, gateway.key))
     }
 
     /// Resolves once the menu's catalog holds a chat-capable model, at

crates/workshop/server/src/agents/gateway.rs

@@ -1,6 +1,6 @@
 //! The one read of a usable gateway that the agent sessions' broker,
-//! search provider, and launch refusal share: the generation, the API
-//! root, and the bearer of the gateway the registry currently holds.
+//! search provider, and launch refusal share: the API root and the bearer
+//! of the gateway the registry currently holds.
 //!
 //! It reads no heartbeat health; a caller that needs reachability checks
 //! it itself.
@@ -11,8 +11,6 @@ use workshop_registry::Registry;
 
 /// One gateway generation whose endpoint and key both build.
 pub(super) struct UsableGateway {
-    /// The generation the server assigned before publishing it.
-    pub(super) generation: u64,
     /// The gateway's OpenAI-compatible API root: its base URL with `/v1`.
     pub(super) endpoint: GatewayEndpoint,
     /// The bearer paired with the base URL.
@@ -26,7 +24,6 @@ impl UsableGateway {
         let snapshot = handles.binding().snapshot();
         let api_root = format!("{}/v1", snapshot.base_url().trim_end_matches('/'));
         Some(Self {
-            generation: snapshot.generation(),
             endpoint: GatewayEndpoint::new(&api_root).ok()?,
             key: SecretString::new(snapshot.api_key()).ok()?,
         })

crates/workshop/server/src/agents/search.rs

@@ -3,10 +3,9 @@
 //!
 //! The provider holds the server's [`Registry`] and reads the gateway
 //! handles through it on every search, as the inference broker does, so a
-//! replaced gateway serves the next search. It keeps the [`GatewaySearch`]
-//! of the last generation it searched under and hands each search to it.
-
-use std::sync::{Mutex, PoisonError};
+//! replaced gateway serves the next search. It builds a [`GatewaySearch`]
+//! per search; the Gateway client's shared connection pool makes that a
+//! few clones and keeps connections alive across searches.
 
 use harness_gateway_client::GatewaySearch;
 use plugin_web::{SearchError, SearchErrorKind, SearchProvider, SearchQuery, SearchResults};
@@ -23,23 +22,17 @@ const NO_GATEWAY: &str = "request failed";
 pub(super) struct GatewaySearchProvider {
     /// The subsystem registry the gateway handles are read through.
     registry: Registry,
-    /// The client built for the last generation searched under.
-    cached: Mutex<Option<(u64, GatewaySearch)>>,
 }
 
 impl GatewaySearchProvider {
     /// Builds the provider over the server's subsystem registry.
     pub(super) fn new(registry: Registry) -> Self {
-        Self {
-            registry,
-            cached: Mutex::new(None),
-        }
+        Self { registry }
     }
 
-    /// The client for the current gateway generation, built on the first
-    /// search under it. A missing gateway registration, a gateway the
-    /// heartbeat reports down, or an endpoint or key that cannot be built
-    /// fails as transport with [`NO_GATEWAY`].
+    /// A client for the current gateway generation. A missing gateway
+    /// registration, a gateway the heartbeat reports down, or an endpoint
+    /// or key that cannot be built fails as transport with [`NO_GATEWAY`].
     fn client(&self) -> Result<GatewaySearch, SearchError> {
         let no_gateway = || SearchError::new(SearchErrorKind::Transport, NO_GATEWAY);
         let handles = self
@@ -50,16 +43,7 @@ impl GatewaySearchProvider {
             return Err(no_gateway());
         }
         let gateway = UsableGateway::read(&handles).ok_or_else(no_gateway)?;
-        // A poisoned lock holds a pair written whole by one store.
-        let mut cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner);
-        if let Some((generation, client)) = cached.as_ref()
-            && *generation == gateway.generation
-        {
-            return Ok(client.clone());
-        }
-        let client = GatewaySearch::new(gateway.endpoint, gateway.key);
-        *cached = Some((gateway.generation, client.clone()));
-        Ok(client)
+        Ok(GatewaySearch::new(gateway.endpoint, gateway.key))
     }
 }