Nemo provider bypasses sub request connector admission control and circuit breaking 755

apis/src/lib.rs

@@ -17,7 +17,7 @@ pub mod openai;
 pub mod promotion;
 #[cfg(feature = "store")]
 pub mod store;
-pub(crate) mod subrequest;
+pub mod subrequest;
 pub(crate) mod web_search;
 
 /// Whether a `Content-Type` header value indicates `text/event-stream`,

apis/src/subrequest.rs

@@ -13,7 +13,7 @@
 use std::{future::Future, net::SocketAddr, time::Duration};
 
 use pingora_core::upstreams::peer::HttpPeer;
-pub(crate) use praxis_core::subrequest::{SubRequest, SubRequestClient, SubRequestError, SubResponse};
+pub use praxis_core::subrequest::{SubRequest, SubRequestClient, SubRequestError, SubResponse};
 use tracing::debug;
 
 /// Parsed URL components needed to resolve and execute a request.
@@ -102,8 +102,16 @@ async fn with_deadline<T>(
 ///
 /// The configured timeout covers URL resolution and the complete HTTP
 /// exchange. All resolved addresses are tried in order when connecting,
-/// while the original URL authority is preserved in `Host`.
-pub(crate) async fn execute_url(
+/// while the original URL authority is preserved in `Host`. Admission
+/// control and per-peer circuit breaking are inherited from `client`.
+///
+/// # Errors
+///
+/// Returns [`SubRequestError`] when the URL cannot be parsed, DNS
+/// resolution or connect fails, the deadline is exceeded, admission
+/// or circuit breaking rejects the call, the response exceeds
+/// `max_response_bytes`, or I/O fails during the exchange.
+pub async fn execute_url(
     client: &SubRequestClient,
     url: &str,
     request: SubRequest,

filters/Cargo.toml

@@ -52,7 +52,6 @@ base64 = { workspace = true }
 bytes = { workspace = true }
 chrono = { workspace = true }
 dashmap = { workspace = true }
-futures = { workspace = true }
 http = { workspace = true }
 metrics = { workspace = true }
 notify = { workspace = true }

filters/src/guardrails/filter.rs

@@ -5,6 +5,7 @@
 
 use async_trait::async_trait;
 use bytes::Bytes;
+use praxis_core::subrequest::{SubRequestClient, SubRequestConnector};
 use praxis_filter::{
     BodyAccess, BodyMode, FilterAction, FilterError, HttpFilter, HttpFilterContext, Rejection, parse_filter_config,
 };
@@ -62,15 +63,45 @@ pub struct AiGuardrailsFilter {
 }
 
 impl AiGuardrailsFilter {
-    /// Create from parsed YAML config.
+    /// Create a filter from parsed YAML config.
     ///
-    /// Parses the shared config, then delegates provider-specific
-    /// config parsing and validation to the provider's `from_config`.
+    /// Uses an isolated [`SubRequestClient`] with a default pool
+    /// size of 4. Prefer [`from_config_with_client`] when a shared
+    /// client is available.
     ///
     /// # Errors
     ///
     /// Returns [`FilterError`] if config parsing or validation fails.
+    ///
+    /// [`FilterError`]: praxis_filter::FilterError
+    /// [`SubRequestClient`]: praxis_core::subrequest::SubRequestClient
+    /// [`from_config_with_client`]: Self::from_config_with_client
     pub fn from_config(config: &serde_yaml::Value) -> Result<Box<dyn HttpFilter>, FilterError> {
+        let client = SubRequestClient::new(SubRequestConnector::new(4, None));
+        Self::build(config, client)
+    }
+
+    /// Create a filter using the shared [`SubRequestClient`].
+    ///
+    /// The shared client inherits the server-level pool size and
+    /// connection limits from the runtime configuration.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`FilterError`] if config parsing or validation fails.
+    ///
+    /// [`FilterError`]: praxis_filter::FilterError
+    /// [`SubRequestClient`]: praxis_core::subrequest::SubRequestClient
+    pub fn from_config_with_client(
+        config: &serde_yaml::Value,
+        client: SubRequestClient,
+    ) -> Result<Box<dyn HttpFilter>, FilterError> {
+        Self::build(config, client)
+    }
+
+    /// Shared constructor body for [`from_config`](Self::from_config) and
+    /// [`from_config_with_client`](Self::from_config_with_client).
+    fn build(config: &serde_yaml::Value, client: SubRequestClient) -> Result<Box<dyn HttpFilter>, FilterError> {
         let cfg: AiGuardrailsConfig = parse_filter_config("ai_guardrails", config)?;
 
         if cfg.phase.response {
@@ -82,7 +113,7 @@ impl AiGuardrailsFilter {
         }
 
         let provider: Box<dyn GuardProvider> = match cfg.provider.provider_type {
-            ProviderType::Nemo => Box::new(NemoProvider::from_config(&cfg.provider.config)?),
+            ProviderType::Nemo => Box::new(NemoProvider::from_config(&cfg.provider.config, client)?),
         };
 
         Ok(Box::new(Self {

filters/src/guardrails/providers/nemo.rs

@@ -7,8 +7,9 @@
 use std::time::Duration;
 
 use async_trait::async_trait;
-use bytes::{Bytes, BytesMut};
-use futures::StreamExt as _;
+use bytes::Bytes;
+use http::{HeaderMap, HeaderValue, Method, Uri};
+use praxis_ai_apis::subrequest::{self, SubRequest, SubRequestClient, SubRequestError, SubResponse};
 use praxis_filter::FilterError;
 use serde::{Deserialize, Serialize};
 
@@ -68,25 +69,31 @@ struct NemoResponse {
 
 /// `NeMo` Guardrails provider.
 pub(in crate::guardrails) struct NemoProvider {
-    /// Pre-configured HTTP client.
-    client: reqwest::Client,
+    /// Bounded HTTP client with admission control and circuit breaking.
+    client: SubRequestClient,
 
     /// `NeMo` endpoint URL.
     endpoint: String,
 
     /// Model name included in every request. Empty string when not configured.
     model: String,
+
+    /// Per-request deadline covering admission, connect, and I/O.
+    timeout: Duration,
 }
 
 impl NemoProvider {
     /// Parse and validate `NeMo`-specific config from the provider settings.
     ///
-    /// Builds a new `NeMo` provider with a pre-configured HTTP client.
+    /// Uses the provided [`SubRequestClient`] so callouts inherit the
+    /// runtime's admission control, circuit breaking, and deadline.
     ///
     /// # Errors
     ///
     /// Returns `FilterError` if the configuration is invalid.
-    pub fn from_config(config: &serde_yaml::Value) -> Result<Self, FilterError> {
+    ///
+    /// [`SubRequestClient`]: praxis_core::subrequest::SubRequestClient
+    pub fn from_config(config: &serde_yaml::Value, client: SubRequestClient) -> Result<Self, FilterError> {
         let cfg: NemoConfig = serde_yaml::from_value(config.clone())
             .map_err(|e| -> FilterError { format!("ai_guardrails (nemo): {e}").into() })?;
         if cfg.endpoint.is_empty() {
@@ -96,37 +103,24 @@ impl NemoProvider {
             return Err("ai_guardrails (nemo): 'timeout_ms' must be greater than zero".into());
         }
 
-        let client = reqwest::Client::builder()
-            .timeout(Duration::from_millis(cfg.timeout_ms))
-            .build()
-            .map_err(|e| -> FilterError { format!("ai_guardrails (nemo): failed to build HTTP client: {e}").into() })?;
-
         Ok(Self {
             client,
             endpoint: cfg.endpoint,
             model: cfg.model,
+            timeout: Duration::from_millis(cfg.timeout_ms),
         })
     }
 }
 
 #[async_trait]
 impl GuardProvider for NemoProvider {
     async fn evaluate(&self, messages: Vec<serde_json::Value>, _phase: GuardPhase) -> Result<GuardResult, FilterError> {
-        let payload = NemoRequest {
-            model: self.model.clone(),
-            messages,
-        };
-        let response = self
-            .client
-            .post(&self.endpoint)
-            .json(&payload)
-            .send()
+        let request = build_request(&self.model, messages)?;
+        let response = subrequest::execute_url(&self.client, &self.endpoint, request, MAX_RESPONSE_SIZE, self.timeout)
             .await
-            .map_err(|e| -> FilterError { format!("ai_guardrails (nemo): failed to send request: {e}").into() })?;
-        check_content_length(&response)?;
+            .map_err(|error| map_subrequest_error(&error))?;
         ensure_success_status(&response)?;
-        let response_body = read_response_body(response).await?;
-        let nemo_response: NemoResponse = serde_json::from_slice(&response_body)
+        let nemo_response: NemoResponse = serde_json::from_slice(&response.body)
             .map_err(|e| -> FilterError { format!("ai_guardrails (nemo): failed to parse response: {e}").into() })?;
         map_nemo_response(&nemo_response)
     }
@@ -136,49 +130,45 @@ impl GuardProvider for NemoProvider {
 // Private Utilities
 // -----------------------------------------------------------------------------
 
-/// Reject responses whose declared `Content-Length` exceeds [`MAX_RESPONSE_SIZE`].
-fn check_content_length(response: &reqwest::Response) -> Result<(), FilterError> {
-    let Some(len) = response.content_length() else {
-        return Ok(());
+/// Build the outbound `NeMo` JSON callout.
+fn build_request(model: &str, messages: Vec<serde_json::Value>) -> Result<SubRequest, FilterError> {
+    let payload = NemoRequest {
+        model: model.to_owned(),
+        messages,
     };
-    if usize::try_from(len).map_or(true, |l| l > MAX_RESPONSE_SIZE) {
-        return Err(format!(
-            "ai_guardrails (nemo): response Content-Length too large \
-             ({len} bytes, limit {MAX_RESPONSE_SIZE})"
-        )
-        .into());
-    }
-    Ok(())
+    let body =
+        Bytes::from(serde_json::to_vec(&payload).map_err(|e| -> FilterError {
+            format!("ai_guardrails (nemo): failed to serialize request: {e}").into()
+        })?);
+
+    let mut headers = HeaderMap::new();
+    headers.insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/json"));
+    headers.insert(http::header::ACCEPT, HeaderValue::from_static("application/json"));
+
+    Ok(SubRequest {
+        method: Method::POST,
+        uri: Uri::default(),
+        headers,
+        body,
+    })
 }
 
-/// Reject non-2xx HTTP responses from the provider.
-fn ensure_success_status(response: &reqwest::Response) -> Result<(), FilterError> {
-    let status = response.status();
-    if !status.is_success() {
-        return Err(format!("ai_guardrails (nemo): provider returned HTTP status code {status}").into());
-    }
-    Ok(())
+/// Map a sub-request failure to a filter error, preserving the
+/// distinct admission / circuit-open / I/O variants in the message.
+fn map_subrequest_error(error: &SubRequestError) -> FilterError {
+    format!("ai_guardrails (nemo): failed to send request: {error}").into()
 }
 
-/// Read the response body incrementally, aborting as soon as the
-/// running total exceeds [`MAX_RESPONSE_SIZE`].
-async fn read_response_body(response: reqwest::Response) -> Result<Bytes, FilterError> {
-    let mut stream = response.bytes_stream();
-    let mut body = BytesMut::new();
-    while let Some(chunk) = stream.next().await {
-        let chunk = chunk.map_err(|e| -> FilterError {
-            format!("ai_guardrails (nemo): failed to read response body: {e}").into()
-        })?;
-        if body.len() + chunk.len() > MAX_RESPONSE_SIZE {
-            return Err(format!(
-                "ai_guardrails (nemo): response body too large \
-                 (limit {MAX_RESPONSE_SIZE} bytes)"
-            )
-            .into());
-        }
-        body.extend_from_slice(&chunk);
+/// Reject non-2xx HTTP responses from the provider.
+fn ensure_success_status(response: &SubResponse) -> Result<(), FilterError> {
+    if !(200..300).contains(&response.status) {
+        return Err(format!(
+            "ai_guardrails (nemo): provider returned HTTP status code {}",
+            response.status
+        )
+        .into());
     }
-    Ok(body.freeze())
+    Ok(())
 }
 
 /// Map a deserialized [`NemoResponse`] to a [`GuardResult`].

filters/src/register.rs

@@ -23,7 +23,7 @@ use crate::{
 /// Register all in-tree AI HTTP filters into `registry`.
 ///
 /// When `subrequest_client` is provided, filters that make HTTP
-/// callouts (`openai_file_resolve`, `openai_web_search`,
+/// callouts (`ai_guardrails`, `openai_file_resolve`, `openai_web_search`,
 /// `anthropic_web_search`) capture the
 /// shared client instead of creating isolated per-filter connectors.
 ///
@@ -45,6 +45,7 @@ pub fn register_ai_filters(registry: &mut FilterRegistry, subrequest_client: Opt
     #[cfg(feature = "gcp-adc-filter")]
     register_gcp_filters(registry);
     register_general_ai_filters(registry);
+    register_ai_guardrails(registry, subrequest_client);
     register_anthropic_filters(registry, subrequest_client);
     register_openai_filters(registry, subrequest_client);
     register_routing_filters(registry);
@@ -100,10 +101,6 @@ fn register_gcp_filters(registry: &mut FilterRegistry) {
 
 /// Register general-purpose AI filters.
 fn register_general_ai_filters(registry: &mut FilterRegistry) {
-    praxis_filter::register_filters!(
-        @register registry,
-        http "ai_guardrails" => AiGuardrailsFilter::from_config
-    );
     #[cfg(feature = "http-callout-filter")]
     praxis_filter::register_filters!(
         @register registry,
@@ -279,6 +276,28 @@ fn register_openai_agentic_filters(registry: &mut FilterRegistry) {
 // Sub-request-aware registration
 // -----------------------------------------------------------------------------
 
+/// Register `ai_guardrails` with the shared client when
+/// available, otherwise fall back to an isolated per-filter connector.
+#[expect(clippy::panic, reason = "matches register_filters! macro convention")]
+fn register_ai_guardrails(registry: &mut FilterRegistry, subrequest_client: Option<&SubRequestClient>) {
+    if let Some(client) = subrequest_client {
+        let client = client.clone();
+        registry
+            .register(
+                "ai_guardrails",
+                praxis_filter::FilterFactory::Http(std::sync::Arc::new(move |config| {
+                    AiGuardrailsFilter::from_config_with_client(config, client.clone())
+                })),
+            )
+            .unwrap_or_else(|_| panic!("duplicate filter name: 'ai_guardrails'"));
+    } else {
+        praxis_filter::register_filters!(
+            @register registry,
+            http "ai_guardrails" => AiGuardrailsFilter::from_config
+        );
+    }
+}
+
 /// Register `anthropic_web_search` with the shared client when
 /// available, otherwise fall back to an isolated per-filter connector.
 #[expect(clippy::panic, reason = "matches register_filters! macro convention")]