praxis-proxy-filter 0.5.6

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 Praxis Contributors

//! Reusable execution of a filtered HTTP sub-request.
//!
//! [`FilteredSubrequestExecutor`] runs a single named filter pipeline against a
//! sub-request and returns owned continuation state: it reconstructs a nested
//! [`HttpFilterContext`](crate::HttpFilterContext), runs the request, request-body,
//! response, and (for buffered responses) response-body phases, applies the same
//! forwarding boundary as the normal upstream path, dispatches the transport in
//! either buffered or streaming mode, and captures a
//! [`FilteredSubrequestContinuation`] the caller drives to completion.
//!
//! The executor owns three transient extension mechanisms end-to-end —
//! [`RetainedFilterResults`], [`PendingStreamChunks`], and
//! [`StreamTermination`] — and recognizes two framework-defined caller-staged
//! channels: [`PendingCredentials`], which it drains to materialize each
//! authority-bound secret only after resolving the destination (see
//! [`DeferredCredential`](crate::DeferredCredential)), and [`StagedUpstream`],
//! which it drains to seed the sub-request's upstream before the request phase so
//! a callout can dial a known destination without an upstream-selecting filter.
//! It otherwise never
//! inspects caller-injected extension types. Callers that stash their own state
//! in the request extensions recover it from
//! [`FilteredSubrequestError::into_parts`],
//! [`FilteredSubrequestContinuation::into_parent_extensions`], or
//! [`FilteredSubrequestContinuation::into_completion`], and strip it themselves.
//!
//! Retained-state accounting is delegated to the caller through the
//! [`RetainedStateAccounting`] hook so the executor stays independent of any
//! particular caller's state layout while still enforcing a ceiling at every
//! phase boundary.

mod context;
mod continuation;
mod sanitize;
mod streaming;
#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::too_many_lines,
    reason = "tests"
)]
mod tests;
mod transport;

use std::{
    net::SocketAddr,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};

use bytes::Bytes;
use http::HeaderMap;
use praxis_core::{
    config::{CachedClusterTls, ClusterTls},
    connectivity::{ConnectionOptions, PreparedTarget, Upstream},
    subrequest::{FrameworkHeaders, StreamLimits, SubRequestError, SubResponseBody},
};
use tracing::{Instrument as _, warn};

use self::{
    context::{SubrequestRuntimeResources, build_sub_filter_context},
    sanitize::{
        apply_pre_read_header_mutations, apply_request_header_mutations, body_exceeds_limit, ensure_destination_host,
        response_body_overflow_limit, sanitize_subrequest_headers, sanitize_subresponse_headers, set_authority_host,
        streaming_transport_limit, strip_reserved_headers, subresponse_from_rejection,
    },
    transport::{build_peer, classify_transport_failure, stream_termination_cause},
};
pub(crate) use self::{
    continuation::{FilteredSubrequestContinuation, SubrequestCompletion},
    sanitize::normalize_response_status,
    streaming::{CalloutStreamingBody, FilteredStreamingBody},
};
use crate::{
    FilterAction, FilterError, FilterPipeline, StreamTermination, StreamTerminationCause, SubRequest,
    SubRequestResponseMode, SubResponse,
    actions::Rejection,
    context::PendingStreamChunks,
    credentials::{PendingCredentials, ResolvedDestination},
    extensions::RequestExtensions,
    results::RetainedFilterResults,
};

/// Idle timeout applied to a streaming sub-request transport.
///
/// The absolute per-step deadline still bounds total duration; this only
/// caps the gap between upstream chunks so a stalled source is abandoned.
pub(crate) const STREAMING_IDLE_TIMEOUT: Duration = Duration::from_secs(30);

/// Caller-supplied accounting for retained sub-request state.
///
/// The executor enforces a ceiling on caller-owned retained state at every
/// phase boundary without knowing how that state is represented. Callers
/// implement this to report, from the live request extensions, whether their
/// retained footprint currently exceeds the limit.
pub(crate) trait RetainedStateAccounting {
    /// Whether the caller-owned retained state currently exceeds its ceiling.
    fn exceeds_limit(&self, extensions: &RequestExtensions) -> bool;
}

/// Where a captured sub-response originated.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ResponseOrigin {
    /// A real upstream produced the response.
    Upstream,
    /// A nested filter produced the response locally.
    Local,
    /// A transport failure was synthesized into a response.
    Transport,
}

/// Transport failure classification used to synthesize a gateway response.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum TransportFailure {
    /// Admission control timed out before dispatch.
    AdmissionTimeout,
    /// The circuit breaker was open.
    CircuitOpen,
    /// The connection could not be established.
    Connect,
    /// A generic I/O failure occurred.
    Io,
    /// The transport deadline was exceeded.
    DeadlineExceeded,
    /// The response exceeded the configured size limit.
    ResponseTooLarge {
        /// Observed cumulative body size that tripped the limit.
        actual: usize,
        /// The effective limit that was exceeded.
        limit: usize,
    },
}

/// Typed detail for a response that exceeded its configured size ceiling.
///
/// Carried across the executor's public boundary so a caller can classify an
/// oversized response and map it to its own status (for example HTTP 413)
/// instead of the opaque gateway response [`run`](FilteredSubrequestExecutor::run)
/// synthesizes.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) struct ResponseTooLargeInfo {
    /// Observed size that tripped the limit, when known.
    pub(crate) actual: Option<usize>,
    /// The effective limit that was exceeded.
    pub(crate) limit: usize,
}

/// Owned downstream request attributes carried into a filtered sub-request.
///
/// A chain-binding callout builds this from its request context so the nested
/// pipeline sees the real client identity, transport security, and request
/// clock — the inputs security and observability filters in an outbound chain
/// depend on.
#[derive(Clone)]
pub struct SubrequestRuntime {
    /// Original downstream client address.
    pub(crate) client_addr: Option<std::net::IpAddr>,
    /// Whether the original downstream uses TLS.
    pub(crate) downstream_tls: bool,
    /// Verified downstream peer identity.
    pub(crate) peer_identity: Option<Arc<praxis_tls::TlsPeerIdentity>>,
    /// Start time of the logical client request.
    pub(crate) request_start: Instant,
}

impl SubrequestRuntime {
    /// Capture the downstream attributes to forward into a filtered sub-request.
    ///
    /// `client_addr`, `downstream_tls`, and `peer_identity` come from the
    /// originating client connection; `request_start` is the instant the logical
    /// client request began, used for consistent duration accounting across the
    /// sub-request.
    #[must_use]
    pub fn new(
        client_addr: Option<std::net::IpAddr>,
        downstream_tls: bool,
        peer_identity: Option<Arc<praxis_tls::TlsPeerIdentity>>,
        request_start: Instant,
    ) -> Self {
        Self {
            client_addr,
            downstream_tls,
            peer_identity,
            request_start,
        }
    }
}

/// A captured sub-response together with its origin classification.
pub(crate) struct SubrequestOutcome {
    /// The buffered or header-only response.
    pub(crate) response: SubResponse,
    /// Where the response came from.
    pub(crate) origin: ResponseOrigin,
    /// Transport failure detail, when the response was synthesized.
    pub(crate) transport_error: Option<TransportFailure>,
}

/// Transport/body shape selected by the sub-request's filters.
pub(crate) enum OpenedResponse {
    /// The complete response was collected and filtered.
    Complete(SubrequestOutcome),
    /// Response headers are filtered; body remains pull-based.
    Streaming {
        /// Live upstream response body.
        body: Box<SubResponseBody>,
        /// Header-time transition metadata.
        outcome: SubrequestOutcome,
    },
}

/// One opened sub-request, including state needed for body/completion processing.
pub(crate) struct OpenedSubrequest {
    /// Owned filter lifecycle state.
    pub(crate) continuation: FilteredSubrequestContinuation,
    /// Buffered or pull-based response source.
    pub(crate) kind: OpenedResponse,
}

/// A sub-request error together with the caller extensions it borrowed.
pub(crate) struct FilteredSubrequestError {
    /// Underlying filter or lifecycle error.
    error: FilterError,
    /// Caller-owned extensions recovered from the nested filter context.
    extensions: RequestExtensions,
    /// Typed response-size-overflow detail, when this error is an overflow.
    too_large: Option<ResponseTooLargeInfo>,
}

impl FilteredSubrequestError {
    /// Build an error before a nested filter context exists.
    fn new(error: FilterError, extensions: RequestExtensions) -> Self {
        Self {
            error,
            extensions,
            too_large: None,
        }
    }

    /// Recover caller-owned extensions from a failed nested context.
    ///
    /// Only the executor-owned mechanisms are stripped; caller-injected
    /// extension types remain for the caller to remove before returning them
    /// to the parent request context.
    fn capture(error: FilterError, ctx: &mut crate::HttpFilterContext<'_>) -> Self {
        ctx.extensions.remove::<PendingStreamChunks>();
        ctx.extensions.remove::<RetainedFilterResults>();
        ctx.extensions.remove::<StreamTermination>();
        Self::new(error, std::mem::take(&mut ctx.extensions))
    }

    /// Attach the typed overflow detail carried by a response-size-limit breach.
    #[must_use]
    fn with_too_large(mut self, info: ResponseTooLargeInfo) -> Self {
        self.too_large = Some(info);
        self
    }

    /// The typed overflow detail, when this error is a response-size breach.
    pub(crate) fn too_large(&self) -> Option<ResponseTooLargeInfo> {
        self.too_large
    }

    /// Split the error from the extensions its caller must restore.
    ///
    /// The typed overflow detail is intentionally dropped here: string-based
    /// callers (the iterative request router's `IrrStepRunner`) keep their
    /// existing behavior, while callers that need the classification read it
    /// through [`too_large`](Self::too_large) first.
    pub(crate) fn into_parts(self) -> (FilterError, RequestExtensions) {
        (self.error, self.extensions)
    }
}

/// Internal result before owned continuation state is captured.
enum RawResponse {
    /// Complete buffered or synthetic response.
    Complete(SubrequestOutcome),
    /// Local filter rejection.
    Rejected(Rejection),
    /// Open pull-based upstream response.
    Streaming {
        /// Live upstream response body.
        body: Box<SubResponseBody>,
        /// Header-time transition metadata.
        outcome: SubrequestOutcome,
    },
    /// An executor-side body-limit check tripped after the response transition.
    ///
    /// Carried out of the timed block (rather than returned as an error inline)
    /// so the shared post-processing path attaches the typed overflow detail to
    /// [`FilteredSubrequestError`] uniformly.
    ResponseTooLarge {
        /// Observed body size that tripped the limit.
        actual: usize,
        /// The effective limit that was exceeded.
        limit: usize,
        /// Human-readable message preserved for string-based callers.
        message: &'static str,
    },
}

/// One sub-request to execute against a named step pipeline.
pub(crate) struct FilteredSubrequestInput<'a> {
    /// Pre-built pipeline for this step.
    pub(crate) pipeline: &'a Arc<FilterPipeline>,
    /// The sub-request to dispatch.
    pub(crate) request: &'a SubRequest,
    /// Human-readable step label for tracing and error messages.
    pub(crate) label: &'a str,
    /// Zero-based iteration index for tracing.
    pub(crate) iteration: u32,
    /// Absolute overall deadline shared across the logical request.
    pub(crate) deadline: Instant,
    /// Caller-provided request extensions, moved into the nested context.
    pub(crate) extensions: RequestExtensions,
}

/// The response an outbound chain produced for a callout.
///
/// The outbound chain — not the calling method — decides whether the response
/// is delivered whole or streamed, by whether a filter selects a streaming
/// sub-request response. [`run`](FilteredSubrequestExecutor::run) surfaces that
/// choice so the caller handles each shape explicitly, mirroring the epic's
/// "one buffered or streaming subrequest" contract.
pub enum CalloutResponse {
    /// The complete response, fully filtered through the response-body phase.
    Buffered(SubResponse),
    /// Transition-time headers now, with the body pulled through the outbound
    /// chain's response-body filters as it flows.
    Streaming {
        /// Status and headers after the response-header phase; the body field
        /// is empty because the payload is delivered through `body`.
        response: SubResponse,
        /// Pull-based response body. Each [`next_chunk`] applies the outbound
        /// chain's response-body filters and, after upstream EOF, flushes any
        /// completion output the chain emits before yielding `None`. An
        /// upstream failure the chain did not convert into a valid terminal
        /// sequence surfaces as an error after buffered chunks drain.
        ///
        /// [`next_chunk`]: crate::StreamingResponseBody::next_chunk
        body: Box<dyn crate::StreamingResponseBody>,
    },
}

/// The classified outcome of a callout run.
///
/// Additive companion to [`CalloutResponse`], returned by
/// [`run_classified`](FilteredSubrequestExecutor::run_classified). It preserves
/// the typed response-size-overflow classification that
/// [`run`](FilteredSubrequestExecutor::run) collapses into a generic gateway
/// response, so a caller can map an oversized response to its own status (for
/// example HTTP 413) instead of an opaque 502. The overflow classification is
/// exposed directly — never inferred from a `502` status.
///
/// Marked `#[non_exhaustive]` so future classified outcomes can be added
/// without breaking downstream `match` arms.
#[non_exhaustive]
pub enum CalloutOutcome {
    /// The outbound chain produced a response — buffered or streaming.
    Response(CalloutResponse),
    /// The response exceeded the configured size limit before delivery.
    ///
    /// Covers both a transport-level overflow (the upstream body exceeded the
    /// per-response ceiling mid-download) and an executor-side body-limit breach
    /// (a response-body filter grew the body past the ceiling). The
    /// response-filter lifecycle runs in both cases before this is surfaced.
    ResponseTooLarge {
        /// Observed size that tripped the limit, when known.
        actual: Option<usize>,
        /// The effective limit that was exceeded.
        limit: usize,
    },
}

/// A pre-resolved upstream a callout stages so the executor dials a specific
/// destination without the outbound chain needing an upstream-selecting filter.
///
/// A chain-binding callout that already knows its destination — for example a
/// provider URL prepared via [`prepare_url_target`] — stages one of these in the
/// request extensions it passes to [`run`](FilteredSubrequestExecutor::run). The
/// executor seeds [`HttpFilterContext::upstream`](crate::HttpFilterContext) from
/// it before the request phase, so the outbound chain carries only cross-cutting
/// filters (observability, security, credentials) and never has to resolve a
/// cluster. Central SSRF, TLS/SNI, and Host enforcement still apply at transport
/// time exactly as for a chain-resolved upstream.
///
/// [`prepare_url_target`]: praxis_core::connectivity::prepare_url_target
pub struct StagedUpstream(pub Upstream);

impl StagedUpstream {
    /// Build a staged upstream from a [`PreparedTarget`].
    ///
    /// The transport address is pinned to the first address the target resolved
    /// (so the executor dials the same endpoint the SSRF validation hook saw,
    /// closing the resolve-then-dial race), the HTTP `Host` authority is the
    /// URL's authority, and TLS/SNI are derived from the URL scheme and host.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if the target resolved no addresses or its TLS
    /// material cannot be prepared.
    pub fn from_prepared_target(target: &PreparedTarget) -> Result<Self, FilterError> {
        let address = target
            .addresses()
            .first()
            .ok_or_else(|| -> FilterError { "filtered_subrequest: prepared target resolved no addresses".into() })?;
        let tls = if target.is_tls() {
            let cluster_tls = ClusterTls {
                sni: Some(target.sni().to_owned()),
                ..ClusterTls::default()
            };
            Some(
                CachedClusterTls::try_from_config(&cluster_tls)
                    .map_err(|error| -> FilterError { format!("filtered_subrequest: invalid TLS: {error}").into() })?,
            )
        } else {
            None
        };
        Ok(Self(Upstream {
            address: Arc::from(address.to_string().as_str()),
            authority: Some(target.host_authority().clone()),
            connection: Arc::new(ConnectionOptions::default()),
            tls,
        }))
    }
}

/// Caller-staged multi-address fallback set for a [`StagedUpstream`].
///
/// [`StagedUpstream`] pins the sub-request's primary transport address (the
/// first address its [`PreparedTarget`] resolved), which closes the
/// resolve-then-dial SSRF race. When a hostname resolved to several addresses,
/// staging this alongside it lets the executor advance past a connection
/// refusal to the next validated address — preserving the DNS fallback the
/// low-level transport performs — without ever re-resolving DNS. Every address
/// was SSRF-validated together by the same preparation hook, and each literal
/// is re-checked at connect time, so dialing any of them is safe. Absent this
/// extension (or with a single address), the executor dials the single staged
/// upstream exactly as before.
pub struct StagedUpstreamFallback(Vec<SocketAddr>);

impl StagedUpstreamFallback {
    /// Capture every address a [`PreparedTarget`] resolved, in resolver order.
    #[must_use]
    pub fn from_prepared_target(target: &PreparedTarget) -> Self {
        Self(target.addresses().to_vec())
    }

    /// The validated fallback addresses, in resolver order.
    #[must_use]
    pub fn addresses(&self) -> &[SocketAddr] {
        &self.0
    }
}

/// No-op retained-state accounting for callers that keep no cross-sub-request
/// state, so the callout entry point never has to expose the
/// [`RetainedStateAccounting`] hook.
struct NoRetainedState;

impl RetainedStateAccounting for NoRetainedState {
    fn exceeds_limit(&self, _extensions: &RequestExtensions) -> bool {
        false
    }
}

/// Executes exactly one filtered sub-request and returns owned continuation state.
pub struct FilteredSubrequestExecutor {
    /// Caller-supplied retained-state ceiling accounting.
    accounting: Box<dyn RetainedStateAccounting + Send + Sync>,
    /// Shared transport client.
    client: praxis_core::subrequest::SubRequestClient,
    /// Nested depth forwarded to sub-requests.
    depth: u8,
    /// Owned downstream request attributes.
    downstream: SubrequestRuntime,
    /// Per-step buffered response ceiling.
    max_response_bytes: usize,
    /// Retained-state raw byte ceiling for stream chunk emission.
    max_state_bytes: usize,
    /// Per-step duration ceiling.
    step_timeout: Duration,
}

impl FilteredSubrequestExecutor {
    /// Build an executor for one logical filtered-sub-request sequence.
    #[expect(
        clippy::too_many_arguments,
        reason = "executor owns explicit subrequest limits and resources"
    )]
    pub(crate) fn new(
        accounting: Box<dyn RetainedStateAccounting + Send + Sync>,
        client: praxis_core::subrequest::SubRequestClient,
        depth: u8,
        downstream: SubrequestRuntime,
        max_response_bytes: usize,
        max_state_bytes: usize,
        step_timeout: Duration,
    ) -> Self {
        Self {
            accounting,
            client,
            depth,
            downstream,
            max_response_bytes,
            max_state_bytes,
            step_timeout,
        }
    }

    /// Build an executor for an application callout that runs a single bound
    /// outbound chain and keeps no cross-sub-request retained state.
    ///
    /// This is the small constructor a chain-binding filter (for example an AI
    /// provider callout) uses together with [`run`]. `client` is the
    /// shared sub-request transport (available from
    /// `HttpFilterContext::subrequest_client`); `downstream` carries the
    /// originating client attributes; `depth` is the current sub-request nesting
    /// depth; `max_response_bytes` caps the response (the buffered body, or the
    /// cumulative bytes emitted by a streaming body); `step_timeout` bounds the
    /// sub-request's own duration within the caller's deadline.
    ///
    /// [`run`]: Self::run
    #[must_use]
    pub fn for_callout(
        client: praxis_core::subrequest::SubRequestClient,
        downstream: SubrequestRuntime,
        depth: u8,
        max_response_bytes: usize,
        step_timeout: Duration,
    ) -> Self {
        // A callout retains no state across sub-requests, so the response ceiling
        // doubles as the stream-chunk emission ceiling.
        Self::new(
            Box::new(NoRetainedState),
            client,
            depth,
            downstream,
            max_response_bytes,
            max_response_bytes,
            step_timeout,
        )
    }

    /// Run `request` through `pipeline` as a filtered sub-request and return the
    /// response the outbound chain produced — buffered or streaming.
    ///
    /// This is the deliberately small entry point for application callout
    /// filters that bound an outbound chain via
    /// [`ChainBindingContext::bind_chain`]. It runs the request, request-body,
    /// response, and (for buffered responses) response-body phases; enforces the
    /// resolved destination authority, DNS/SSRF, TLS/SNI, Host, deadline, and
    /// transport limits; and materializes any staged [`PendingCredentials`] only
    /// after the destination is resolved. `extensions` is the caller-staged
    /// projection moved into the isolated child context.
    ///
    /// The outbound chain decides the response shape: a filter that selects a
    /// streaming sub-request response yields [`CalloutResponse::Streaming`] with
    /// the transition-time headers and a pull-based body that keeps applying the
    /// chain's response-body filters as chunks flow; otherwise the fully
    /// filtered response is returned as [`CalloutResponse::Buffered`].
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if a filter phase errors or the deadline is
    /// exceeded before the response transition. For a streaming response, errors
    /// that occur while pulling the body surface from
    /// [`StreamingResponseBody::next_chunk`](crate::StreamingResponseBody::next_chunk)
    /// instead.
    ///
    /// # Examples
    ///
    /// A callout drives its prebuilt outbound chain to a response. In production
    /// the pipeline comes from [`ChainBindingContext::bind_chain`] at
    /// construction and the shared client from the filter's
    /// [`HttpFilterContext`]; here both are built directly so the example runs,
    /// and a `static_response` filter answers locally so no upstream is needed.
    ///
    /// ```
    /// use std::{
    ///     sync::Arc,
    ///     time::{Duration, Instant},
    /// };
    ///
    /// use praxis_core::subrequest::{SubRequestClient, SubRequestConnector};
    /// use praxis_filter::{
    ///     CalloutResponse, FilterEntry, FilterPipeline, FilterRegistry, FilteredSubrequestExecutor,
    ///     RequestExtensions, SubRequest, SubrequestRuntime,
    /// };
    ///
    /// let rt = tokio::runtime::Builder::new_current_thread()
    ///     .enable_all()
    ///     .build()
    ///     .unwrap();
    /// rt.block_on(async {
    ///     // The outbound chain a callout binds once at construction.
    ///     let registry = FilterRegistry::with_builtins();
    ///     let mut chain: Vec<FilterEntry> = serde_yaml::from_str(
    ///         "- filter: static_response\n  status: 200\n  body: hello from the outbound chain\n",
    ///     )
    ///     .unwrap();
    ///     let outbound = Arc::new(FilterPipeline::build(&mut chain, &registry).unwrap());
    ///
    ///     // At request time the callout builds an executor from the shared client
    ///     // and the downstream attributes it reads off its `HttpFilterContext`.
    ///     let client = SubRequestClient::new(SubRequestConnector::new(1, None));
    ///     let downstream = SubrequestRuntime::new(None, false, None, Instant::now());
    ///     let executor = FilteredSubrequestExecutor::for_callout(
    ///         client,
    ///         downstream,
    ///         0,                      // sub-request nesting depth
    ///         1 << 20,                // 1 MiB response ceiling
    ///         Duration::from_secs(5), // per-sub-request step timeout
    ///     );
    ///
    ///     let request = SubRequest {
    ///         method: http::Method::GET,
    ///         uri: http::Uri::from_static("/"),
    ///         headers: http::HeaderMap::new(),
    ///         body: bytes::Bytes::new(),
    ///     };
    ///     let deadline = Instant::now() + Duration::from_secs(5);
    ///     let response = match executor
    ///         .run(&outbound, &request, RequestExtensions::default(), deadline)
    ///         .await
    ///         .expect("the outbound chain produces a response")
    ///     {
    ///         CalloutResponse::Buffered(response) => response,
    ///         CalloutResponse::Streaming { .. } => unreachable!("static_response is buffered"),
    ///     };
    ///
    ///     assert_eq!(response.status, 200);
    ///     assert_eq!(&response.body[..], b"hello from the outbound chain");
    /// });
    /// ```
    ///
    /// [`ChainBindingContext::bind_chain`]: crate::ChainBindingContext::bind_chain
    /// [`HttpFilterContext`]: crate::HttpFilterContext
    #[expect(clippy::large_futures, reason = "delegates to the full step future")]
    #[expect(
        clippy::large_stack_frames,
        reason = "delegates to execute, which reconstructs a full filter context"
    )]
    pub async fn run(
        &self,
        pipeline: &Arc<FilterPipeline>,
        request: &SubRequest,
        extensions: RequestExtensions,
        deadline: Instant,
    ) -> Result<CalloutResponse, FilterError> {
        let input = FilteredSubrequestInput {
            pipeline,
            request,
            label: "callout",
            iteration: 0,
            deadline,
            extensions,
        };
        let opened = self.execute(input).await.map_err(|error| error.into_parts().0)?;
        Ok(self.callout_response_from(opened))
    }

    /// Run `request` through `pipeline` and return a *classified* outcome.
    ///
    /// Additive companion to [`run`](Self::run) for callers that must
    /// distinguish an oversized response from other results. It behaves exactly
    /// like `run` for every response the outbound chain produces, but instead of
    /// collapsing a response-size overflow into a synthesized gateway response it
    /// returns [`CalloutOutcome::ResponseTooLarge`] carrying the observed size
    /// (when known) and the tripped limit. Both the transport-level overflow
    /// (upstream body over the per-response ceiling) and the executor-side
    /// body-limit breach (a response-body filter grew the body past the ceiling)
    /// are covered, and the response-filter lifecycle runs in both cases.
    ///
    /// A genuine upstream `502` is delivered as
    /// [`CalloutOutcome::Response`] — overflow is never inferred from status.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] under the same conditions as [`run`](Self::run),
    /// except that a response-size overflow is reported as
    /// [`CalloutOutcome::ResponseTooLarge`] rather than an error.
    ///
    /// # Examples
    ///
    /// Like [`run`](Self::run), but the caller matches a [`CalloutOutcome`].
    /// Because the enum is `#[non_exhaustive]` a wildcard arm is required — here
    /// it also absorbs the streaming response shape. As in the `run` example the
    /// pipeline and client are built directly and a `static_response` filter
    /// answers locally, so the example runs without an upstream.
    ///
    /// ```
    /// use std::{
    ///     sync::Arc,
    ///     time::{Duration, Instant},
    /// };
    ///
    /// use praxis_core::subrequest::{SubRequestClient, SubRequestConnector};
    /// use praxis_filter::{
    ///     CalloutOutcome, CalloutResponse, FilterEntry, FilterPipeline, FilterRegistry,
    ///     FilteredSubrequestExecutor, RequestExtensions, SubRequest, SubrequestRuntime,
    /// };
    ///
    /// let rt = tokio::runtime::Builder::new_current_thread()
    ///     .enable_all()
    ///     .build()
    ///     .unwrap();
    /// rt.block_on(async {
    ///     let registry = FilterRegistry::with_builtins();
    ///     let mut chain: Vec<FilterEntry> = serde_yaml::from_str(
    ///         "- filter: static_response\n  status: 200\n  body: hello from the outbound chain\n",
    ///     )
    ///     .unwrap();
    ///     let outbound = Arc::new(FilterPipeline::build(&mut chain, &registry).unwrap());
    ///
    ///     let client = SubRequestClient::new(SubRequestConnector::new(1, None));
    ///     let downstream = SubrequestRuntime::new(None, false, None, Instant::now());
    ///     let executor = FilteredSubrequestExecutor::for_callout(
    ///         client,
    ///         downstream,
    ///         0,                      // sub-request nesting depth
    ///         1 << 20,                // 1 MiB response ceiling
    ///         Duration::from_secs(5), // per-sub-request step timeout
    ///     );
    ///
    ///     let request = SubRequest {
    ///         method: http::Method::GET,
    ///         uri: http::Uri::from_static("/"),
    ///         headers: http::HeaderMap::new(),
    ///         body: bytes::Bytes::new(),
    ///     };
    ///     let deadline = Instant::now() + Duration::from_secs(5);
    ///     match executor
    ///         .run_classified(&outbound, &request, RequestExtensions::default(), deadline)
    ///         .await
    ///         .expect("the outbound chain produces a response")
    ///     {
    ///         CalloutOutcome::Response(CalloutResponse::Buffered(response)) => {
    ///             assert_eq!(response.status, 200);
    ///             assert_eq!(&response.body[..], b"hello from the outbound chain");
    ///         },
    ///         CalloutOutcome::ResponseTooLarge { actual, limit } => {
    ///             unreachable!("within the {limit}-byte ceiling (saw {actual:?})")
    ///         },
    ///         // `CalloutOutcome` is `#[non_exhaustive]`: downstream callers must
    ///         // keep a wildcard arm so added outcomes — and the streaming
    ///         // response shape — stay forward-compatible.
    ///         _ => unreachable!("static_response yields a buffered response"),
    ///     }
    /// });
    /// ```
    #[expect(clippy::large_futures, reason = "delegates to the full step future")]
    #[expect(
        clippy::large_stack_frames,
        reason = "delegates to execute, which reconstructs a full filter context"
    )]
    pub async fn run_classified(
        &self,
        pipeline: &Arc<FilterPipeline>,
        request: &SubRequest,
        extensions: RequestExtensions,
        deadline: Instant,
    ) -> Result<CalloutOutcome, FilterError> {
        let input = FilteredSubrequestInput {
            pipeline,
            request,
            label: "callout",
            iteration: 0,
            deadline,
            extensions,
        };
        match self.execute(input).await {
            Ok(opened) => {
                // A transport-level overflow rides in the completed outcome's
                // transport classification; surface it as the typed outcome
                // rather than the synthesized gateway response `run` returns.
                if let OpenedResponse::Complete(outcome) = &opened.kind
                    && let Some(TransportFailure::ResponseTooLarge { actual, limit }) = outcome.transport_error
                {
                    return Ok(CalloutOutcome::ResponseTooLarge {
                        actual: Some(actual),
                        limit,
                    });
                }
                Ok(CalloutOutcome::Response(self.callout_response_from(opened)))
            },
            // An executor-side body-limit breach carries the typed detail on the
            // error; every other error stays a plain `FilterError`.
            Err(error) => match error.too_large() {
                Some(info) => Ok(CalloutOutcome::ResponseTooLarge {
                    actual: info.actual,
                    limit: info.limit,
                }),
                None => Err(error.into_parts().0),
            },
        }
    }

    /// Map an opened sub-request onto the buffered/streaming callout response,
    /// shared by [`run`](Self::run) and [`run_classified`](Self::run_classified).
    fn callout_response_from(&self, opened: OpenedSubrequest) -> CalloutResponse {
        let OpenedSubrequest { continuation, kind } = opened;
        match kind {
            OpenedResponse::Complete(outcome) => CalloutResponse::Buffered(outcome.response),
            OpenedResponse::Streaming { body, outcome } => CalloutResponse::Streaming {
                response: outcome.response,
                body: Box::new(CalloutStreamingBody::new(
                    FilteredStreamingBody::new(body, continuation),
                    self.max_response_bytes,
                )),
            },
        }
    }

    /// Execute one sub-request under the remaining overall deadline.
    #[expect(
        clippy::too_many_lines,
        reason = "one sub-request owns the complete filter and transport lifecycle"
    )]
    #[expect(clippy::large_futures, reason = "step execution spans filter and transport futures")]
    #[expect(
        clippy::large_stack_frames,
        reason = "step execution reconstructs a full filter context"
    )]
    pub(crate) async fn execute(
        &self,
        input: FilteredSubrequestInput<'_>,
    ) -> Result<OpenedSubrequest, FilteredSubrequestError> {
        let FilteredSubrequestInput {
            pipeline,
            request: current_request,
            label,
            iteration,
            deadline,
            mut extensions,
        } = input;

        let remaining = deadline
            .checked_duration_since(Instant::now())
            .unwrap_or(Duration::ZERO);
        if remaining.is_zero() {
            return Err(FilteredSubrequestError::new(
                "filtered_subrequest: overall deadline exceeded".to_owned().into(),
                extensions,
            ));
        }

        let mut sub_headers = current_request.headers.clone();
        strip_reserved_headers(&mut sub_headers);
        let sub_req = crate::Request {
            method: current_request.method.clone(),
            uri: current_request.uri.clone(),
            headers: sub_headers.clone(),
        };
        let mut routed_req = sub_req.clone();
        let mut response_header = crate::Response {
            headers: HeaderMap::new(),
            status: http::StatusCode::OK,
        };
        let resources = SubrequestRuntimeResources {
            client_addr: self.downstream.client_addr,
            downstream_tls: self.downstream.downstream_tls,
            health_registry: pipeline.health_registry(),
            id_generator: pipeline.id_generator(),
            kv_stores: pipeline.kv_stores(),
            session_stores: pipeline.session_stores(),
            peer_identity: self.downstream.peer_identity.as_ref(),
            request_start: self.downstream.request_start,
            subrequest_client: Some(&self.client),
            time_source: pipeline.time_source(),
        };
        let mut filter_ctx = build_sub_filter_context(pipeline, &sub_req, resources);
        filter_ctx.extensions = std::mem::take(&mut extensions);
        filter_ctx.extensions.insert(RetainedFilterResults::default());
        filter_ctx.enable_stream_chunk_emission(self.max_state_bytes);
        // A callout may stage a pre-resolved upstream (for example a URL prepared
        // via `prepare_url_target`) so the executor dials a specific destination
        // without the outbound chain needing an upstream-selecting filter. Seed it
        // before the request phase so chain filters observe the resolved upstream
        // and the central SSRF/TLS/Host enforcement at `build_peer` still applies.
        //
        // Keep a copy so the destination can be re-pinned after the request phase:
        // a chain filter may observe the seeded upstream but must not be able to
        // retarget a callout that already resolved and validated its destination.
        // Without the re-pin a filter could redirect the dial — and any body-borne
        // credential (e.g. Tavily's key) — to an authority the callout never
        // prepared. `Upstream` is `Arc`-backed, so the clone is a refcount bump.
        let pinned_upstream = filter_ctx.extensions.remove::<StagedUpstream>().map(|staged| staged.0);
        if let Some(upstream) = &pinned_upstream {
            filter_ctx.upstream = Some(upstream.clone());
        }
        // A callout may additionally stage the full validated address set so a
        // connection refusal on the pinned primary address falls back to the
        // next resolved address (matching the low-level transport's DNS
        // behavior) without re-resolving. Every address was SSRF-validated
        // together by the same preparation hook.
        let fallback_addresses = filter_ctx
            .extensions
            .remove::<StagedUpstreamFallback>()
            .map(|fallback| fallback.0)
            .unwrap_or_default();

        let step_budget = remaining.min(self.step_timeout);
        let step_started = Instant::now();
        let step_deadline = step_started.checked_add(step_budget).unwrap_or(deadline);
        let in_transport = Arc::new(AtomicBool::new(false));
        let in_transport_inner = Arc::clone(&in_transport);

        let step_span = tracing::info_span!("filtered_subrequest", step = label, iteration = iteration);

        let timed: Result<Result<RawResponse, FilterError>, tokio::time::error::Elapsed> =
            tokio::time::timeout(step_budget, async {
            let mut request_body = Some(current_request.body.clone());
            if body_exceeds_limit(
                pipeline.body_capabilities().request_body_mode,
                request_body.as_ref().map_or(0, Bytes::len),
            ) {
                return Ok(RawResponse::Rejected(Rejection::status(413)));
            }

            let pre_read_body = matches!(
                pipeline.body_capabilities().request_body_mode,
                crate::BodyMode::StreamBuffer { .. }
            );
            if pre_read_body {
                let action = pipeline
                    .execute_http_request_body(&mut filter_ctx, &mut request_body, true)
                    .await?;
                if let FilterAction::Reject(rejection) = action {
                    return Ok(RawResponse::Rejected(rejection));
                }
                if self.accounting.exceeds_limit(&filter_ctx.extensions) {
                    return Ok(RawResponse::Rejected(Rejection::status(413)));
                }
                apply_pre_read_header_mutations(&mut routed_req.headers, &filter_ctx);
                filter_ctx.extra_request_headers.clear();
                filter_ctx.request_headers_to_remove.clear();
                filter_ctx.request_headers_to_set.clear();
                filter_ctx.pre_read_mutations.clear();
                sub_headers.clone_from(&routed_req.headers);
                filter_ctx.request = &routed_req;
            }

            let action = pipeline.execute_http_request(&mut filter_ctx).await?;
            if let FilterAction::Reject(rejection) = action {
                return Ok(RawResponse::Rejected(rejection));
            }
            if self.accounting.exceeds_limit(&filter_ctx.extensions) {
                return Ok(RawResponse::Rejected(Rejection::status(413)));
            }
            if !pre_read_body {
                let action = pipeline
                    .execute_http_request_body(&mut filter_ctx, &mut request_body, true)
                    .await?;
                if let FilterAction::Reject(rejection) = action {
                    return Ok(RawResponse::Rejected(rejection));
                }
                if self.accounting.exceeds_limit(&filter_ctx.extensions) {
                    return Ok(RawResponse::Rejected(Rejection::status(413)));
                }
            }

            // Re-pin the staged upstream: the request phase may have let a chain
            // filter observe (and try to rewrite) `ctx.upstream`, but a resolved
            // callout must dial only the destination it prepared. Reasserting here
            // overrides any mid-chain rewrite so a staged credential — a header
            // credential, or for a body-authenticated provider the request body
            // itself — can only ever reach the authority it was prepared for.
            if let Some(upstream) = &pinned_upstream {
                filter_ctx.upstream = Some(upstream.clone());
            }
            let upstream = filter_ctx.upstream.as_ref().ok_or_else(|| -> FilterError {
                format!("filtered_subrequest: step '{label}' did not resolve an upstream").into()
            })?;
            let destination_authority = Arc::clone(&upstream.address);
            // The credential-matching key and upstream Host are the *logical*
            // authority the operator configured for this upstream (`authority`
            // override), not the transport endpoint bytes travel to. Derived
            // from trusted config, never the client-influenced Host header.
            let authority_override: Option<Arc<str>> = upstream
                .authority
                .as_ref()
                .and_then(|value| value.to_str().ok())
                .map(Arc::from);
            // Fall back to the transport only when no override is set.
            let logical_authority: Arc<str> = authority_override
                .clone()
                .unwrap_or_else(|| Arc::clone(&destination_authority));
            in_transport_inner.store(true, Ordering::Release);
            // Build one transport peer per validated address so a connection
            // refusal can fall back to the next. With no staged fallback set (or
            // a single address) this is exactly the prior single-peer dial: the
            // seeded upstream address — a pinned literal on the staged path, or a
            // cluster-resolved hostname on the non-staged path — resolved once.
            // Each fallback peer reuses the seeded upstream's authority, TLS, and
            // connection options; only the transport socket address varies.
            let peers = if fallback_addresses.len() > 1 {
                let mut built = Vec::with_capacity(fallback_addresses.len());
                for address in &fallback_addresses {
                    let mut per_address = upstream.clone();
                    per_address.address = Arc::from(address.to_string().as_str());
                    built.push(build_peer(&per_address, pipeline.allow_private_upstreams()).await);
                }
                built
            } else {
                vec![build_peer(upstream, pipeline.allow_private_upstreams()).await]
            };
            apply_request_header_mutations(&mut sub_headers, &filter_ctx);
            // Mirror the normal proxy path (`apply_authority_override`): a
            // configured authority override becomes the upstream Host,
            // replacing any prior step's or caller's Host. Without an override,
            // default the Host to the transport endpoint only when absent.
            match authority_override.as_deref() {
                Some(authority) => set_authority_host(&mut sub_headers, authority)?,
                None => ensure_destination_host(&mut sub_headers, &destination_authority)?,
            }
            sanitize_subrequest_headers(&mut sub_headers);
            // Destination-bound credential injection: only now that the upstream
            // authority is resolved do we materialize staged secrets, and only
            // into a request bound for the authority each credential was issued
            // for. Injecting after sanitization keeps the credential header from
            // being stripped as hop-by-hop/framing.
            if let Some(pending) = filter_ctx.extensions.remove::<PendingCredentials>() {
                let destination = ResolvedDestination {
                    authority: &logical_authority,
                    transport: &destination_authority,
                };
                let injected = pending.inject_authorized(&destination, &mut sub_headers);
                if injected > 0 {
                    // A credential was authorized against `logical_authority`, so
                    // the upstream must see exactly that authority as its Host. Pin
                    // it here — replacing any step- or caller-set Host the
                    // no-override path preserved — so a secret can never be
                    // delivered under a divergent Host to a shared-vhost endpoint.
                    // (With an override the Host already equals `logical_authority`;
                    // this keeps that true.) A staged credential that matched
                    // nothing injects no secret, so it must not retarget the Host —
                    // doing so would silently change virtual-host routing.
                    set_authority_host(&mut sub_headers, &logical_authority)?;
                    tracing::debug!(
                        authority = %logical_authority,
                        transport = %destination_authority,
                        injected,
                        "materialized destination-bound credentials"
                    );
                }
            }
            let request = SubRequest {
                method: current_request.method.clone(),
                uri: filter_ctx.rewritten_path.as_ref().map_or_else(
                    || current_request.uri.clone(),
                    |path| http::Uri::try_from(path.as_str()).unwrap_or_else(|_| current_request.uri.clone()),
                ),
                headers: sub_headers,
                body: request_body.unwrap_or_default(),
            };
            let mut framework_headers = FrameworkHeaders::new();
            framework_headers.set_depth(self.depth + 1);
            let transport_budget = step_budget
                .checked_sub(step_started.elapsed())
                .unwrap_or(Duration::ZERO);
            if transport_budget.is_zero() {
                return Ok(RawResponse::Rejected(Rejection::status(504)));
            }

            match filter_ctx.subrequest_response_mode {
                SubRequestResponseMode::Streaming => {
                    if matches!(
                        pipeline.body_capabilities().response_body_mode,
                        crate::BodyMode::StreamBuffer { .. }
                    ) {
                        return Err(format!(
                            "filtered_subrequest: step '{label}' selected streaming despite a StreamBuffer response mode"
                        )
                        .into());
                    }
                    let limits = StreamLimits {
                        idle_timeout: STREAMING_IDLE_TIMEOUT,
                        // FilteredStreamingBody enforces the original absolute step
                        // deadline so header time cannot be granted twice.
                        max_stream_duration: None,
                        max_total_bytes: streaming_transport_limit(
                            pipeline.body_capabilities().response_body_mode,
                        ),
                    };
                    let mut response =
                        Err(SubRequestError::Connect("filtered_subrequest: no addresses to dial".to_owned()));
                    for peer in &peers {
                        // Recompute the remaining budget before every fallback
                        // attempt so a slow-but-not-refused earlier address cannot
                        // hand each later attempt the full step budget again; the
                        // absolute step deadline bounds all attempts together (the
                        // outer `tokio::time::timeout` is the hard backstop).
                        let attempt_budget = step_deadline
                            .checked_duration_since(Instant::now())
                            .unwrap_or(Duration::ZERO);
                        if attempt_budget.is_zero() {
                            break;
                        }
                        response = match peer {
                            Ok(peer) => self
                                .client
                                .send_streaming(peer, &request, attempt_budget, limits.clone(), Some(&framework_headers))
                                .await,
                            Err(error) => Err(SubRequestError::Connect(error.to_string())),
                        };
                        // Advance to the next validated address only on a
                        // connection refusal; any other outcome (a response, or a
                        // substantive transport error) is final.
                        if matches!(response, Err(SubRequestError::Connect(_))) {
                            continue;
                        }
                        break;
                    }
                    in_transport_inner.store(false, Ordering::Release);
                    match response {
                        Ok(response) => {
                            let status = response.status;
                            let mut headers = response.headers;
                            sanitize_subresponse_headers(&mut headers);
                            response_header.status = http::StatusCode::from_u16(status)
                                .map_err(|error| -> FilterError { format!("invalid upstream status: {error}").into() })?;
                            response_header.headers.clone_from(&headers);
                            filter_ctx.response_header = Some(&mut response_header);
                            let response_action = pipeline.execute_http_response(&mut filter_ctx).await?;
                            if let FilterAction::Reject(rejection) = response_action {
                                response.body.cancel().await;
                                return Ok(RawResponse::Rejected(rejection));
                            }
                            if self.accounting.exceeds_limit(&filter_ctx.extensions) {
                                response.body.cancel().await;
                                return Ok(RawResponse::Rejected(Rejection::status(413)));
                            }
                            let metadata = filter_ctx.response_header.as_deref().ok_or_else(|| -> FilterError {
                                "filtered_subrequest: response metadata missing after header filters"
                                    .to_owned()
                                    .into()
                            })?;
                            let status = metadata.status;
                            let mut headers = metadata.headers.clone();
                            sanitize_subresponse_headers(&mut headers);
                            Ok(RawResponse::Streaming {
                                body: Box::new(response.body),
                                outcome: SubrequestOutcome {
                                    response: SubResponse { status: status.as_u16(), headers, body: Bytes::new() },
                                    origin: ResponseOrigin::Upstream,
                                    transport_error: None,
                                },
                            })
                        },
                        Err(error) => {
                            let (status, kind) = classify_transport_failure(&error);
                            warn!(step = label, %error, status, "filtered sub-request streaming transport failure");
                            let response = SubResponse { status, headers: HeaderMap::new(), body: Bytes::new() };
                            response_header.status = http::StatusCode::from_u16(status)
                                .map_err(|source| -> FilterError { source.into() })?;
                            filter_ctx.response_header = Some(&mut response_header);
                            let response_action = pipeline.execute_http_response(&mut filter_ctx).await?;
                            if let FilterAction::Reject(rejection) = response_action {
                                return Ok(RawResponse::Rejected(rejection));
                            }
                            if self.accounting.exceeds_limit(&filter_ctx.extensions) {
                                return Ok(RawResponse::Rejected(Rejection::status(413)));
                            }
                            let metadata = filter_ctx.response_header.as_deref().ok_or_else(|| -> FilterError {
                                "filtered_subrequest: response metadata missing after header filters"
                                    .to_owned()
                                    .into()
                            })?;
                            let mut headers = metadata.headers.clone();
                            sanitize_subresponse_headers(&mut headers);
                            Ok(RawResponse::Complete(SubrequestOutcome {
                                response: SubResponse {
                                    status: metadata.status.as_u16(),
                                    headers,
                                    body: response.body,
                                },
                                origin: ResponseOrigin::Transport,
                                transport_error: Some(kind),
                            }))
                        },
                    }
                },
                SubRequestResponseMode::Buffered => {
                    let mut attempt = None;
                    for peer in &peers {
                        // Recompute the remaining budget before every fallback
                        // attempt (see the streaming arm). If the step deadline is
                        // already reached, stop dialing and fall through to the
                        // `attempt.unwrap_or_else(...)` synthesized failure below.
                        let attempt_budget = step_deadline
                            .checked_duration_since(Instant::now())
                            .unwrap_or(Duration::ZERO);
                        if attempt_budget.is_zero() {
                            break;
                        }
                        let outcome = match peer {
                            Ok(peer) => match self
                                .client
                                .execute(peer, &request, self.max_response_bytes, attempt_budget, Some(&framework_headers))
                                .await
                            {
                                Ok(response) => (response, ResponseOrigin::Upstream, None),
                                Err(error) => {
                                    let (status, kind) = classify_transport_failure(&error);
                                    warn!(step = label, %error, status, "filtered sub-request buffered transport failure");
                                    let response =
                                        SubResponse { status, headers: HeaderMap::new(), body: Bytes::new() };
                                    if matches!(error, SubRequestError::Connect(_)) {
                                        // Connection refused/unreachable: remember
                                        // it and try the next validated address.
                                        attempt = Some((response, ResponseOrigin::Transport, Some(kind)));
                                        continue;
                                    }
                                    (response, ResponseOrigin::Transport, Some(kind))
                                },
                            },
                            Err(error) => {
                                warn!(step = label, %error, status = 502_u16, "filtered sub-request buffered transport failure");
                                // Peer construction failed (resolution/SSRF):
                                // remember it and try the next validated address.
                                attempt = Some((
                                    SubResponse { status: 502, headers: HeaderMap::new(), body: Bytes::new() },
                                    ResponseOrigin::Transport,
                                    Some(TransportFailure::Connect),
                                ));
                                continue;
                            },
                        };
                        attempt = Some(outcome);
                        break;
                    }
                    let (mut response, origin, transport_error) = attempt.unwrap_or_else(|| {
                        (
                            SubResponse { status: 502, headers: HeaderMap::new(), body: Bytes::new() },
                            ResponseOrigin::Transport,
                            Some(TransportFailure::Connect),
                        )
                    });
                    in_transport_inner.store(false, Ordering::Release);
                    sanitize_subresponse_headers(&mut response.headers);
                    response_header.status = http::StatusCode::from_u16(response.status)
                        .map_err(|error| -> FilterError { error.into() })?;
                    response_header.headers.clone_from(&response.headers);
                    filter_ctx.response_header = Some(&mut response_header);
                    let response_action = pipeline.execute_http_response(&mut filter_ctx).await?;
                    if let FilterAction::Reject(rejection) = response_action {
                        return Ok(RawResponse::Rejected(rejection));
                    }
                    if self.accounting.exceeds_limit(&filter_ctx.extensions) {
                        return Ok(RawResponse::Rejected(Rejection::status(413)));
                    }
                    let mut body = Some(std::mem::take(&mut response.body));
                    if let Some(limit) = response_body_overflow_limit(
                        pipeline.body_capabilities().response_body_mode,
                        self.max_response_bytes,
                        body.as_ref().map_or(0, Bytes::len),
                    ) {
                        return Ok(RawResponse::ResponseTooLarge {
                            actual: body.as_ref().map_or(0, Bytes::len),
                            limit,
                            message: "filtered_subrequest: step response exceeds configured body limit",
                        });
                    }
                    let body_action = pipeline.execute_http_response_body(&mut filter_ctx, &mut body, true)?;
                    if let FilterAction::Reject(rejection) = body_action {
                        return Ok(RawResponse::Rejected(rejection));
                    }
                    if self.accounting.exceeds_limit(&filter_ctx.extensions) {
                        return Ok(RawResponse::Rejected(Rejection::status(413)));
                    }
                    if let Some(limit) = response_body_overflow_limit(
                        pipeline.body_capabilities().response_body_mode,
                        self.max_response_bytes,
                        body.as_ref().map_or(0, Bytes::len),
                    ) {
                        return Ok(RawResponse::ResponseTooLarge {
                            actual: body.as_ref().map_or(0, Bytes::len),
                            limit,
                            message: "filtered_subrequest: transformed step response exceeds configured body limit",
                        });
                    }
                    response.body = body.unwrap_or_default();
                    if let Some(metadata) = filter_ctx.response_header.as_deref() {
                        response.status = metadata.status.as_u16();
                        response.headers.clone_from(&metadata.headers);
                    }
                    sanitize_subresponse_headers(&mut response.headers);
                    Ok(RawResponse::Complete(SubrequestOutcome { response, origin, transport_error }))
                },
            }
            }
            .instrument(step_span))
            .await;

        let mut raw = match timed {
            Ok(Ok(raw)) => raw,
            Ok(Err(error)) => return Err(FilteredSubrequestError::capture(error, &mut filter_ctx)),
            Err(_) if in_transport.load(Ordering::Acquire) => RawResponse::Complete(SubrequestOutcome {
                response: SubResponse {
                    status: 504,
                    headers: HeaderMap::new(),
                    body: Bytes::new(),
                },
                origin: ResponseOrigin::Transport,
                transport_error: Some(TransportFailure::DeadlineExceeded),
            }),
            Err(_) => RawResponse::Rejected(Rejection::status(504)),
        };

        filter_ctx.response_header = None;
        if filter_ctx.subrequest_response_mode == SubRequestResponseMode::Streaming
            && let RawResponse::Complete(outcome) = &mut raw
            && outcome.origin == ResponseOrigin::Transport
        {
            let cause = outcome
                .transport_error
                .map_or(StreamTerminationCause::Io, stream_termination_cause);
            filter_ctx.extensions.insert(StreamTermination::new(cause));
            let response_snapshot = crate::Response {
                status: http::StatusCode::from_u16(outcome.response.status).unwrap_or(http::StatusCode::BAD_GATEWAY),
                headers: outcome.response.headers.clone(),
            };
            let mut completion_body = None;
            let completion_action = pipeline
                .execute_http_response_body_with_response_header(
                    &mut filter_ctx,
                    &mut completion_body,
                    true,
                    Some(&response_snapshot),
                )
                .map_err(|error| FilteredSubrequestError::capture(error, &mut filter_ctx))?;
            if let FilterAction::Reject(_) = completion_action {
                let error = "filtered_subrequest: step completion filter rejected an abnormal stream"
                    .to_owned()
                    .into();
                return Err(FilteredSubrequestError::capture(error, &mut filter_ctx));
            }
            if self.accounting.exceeds_limit(&filter_ctx.extensions) {
                let error = "filtered_subrequest: retained state limit exceeded during stream completion"
                    .to_owned()
                    .into();
                return Err(FilteredSubrequestError::capture(error, &mut filter_ctx));
            }
            // Mirror the two buffered overflow sites: the effective ceiling is the
            // smaller of `max_response_bytes` and the pipeline's response body mode.
            //
            // Reachability invariant — today this collapses to `max_response_bytes`.
            // A `StreamBuffer` response mode is rejected before streaming is selected
            // (the guard at the top of the streaming arm), and a `SizeLimit` mode only
            // arises when no response-body filter runs — in which case nothing writes
            // `completion_body` and it stays empty. So the only mode under which a
            // completion body can exist is `Stream`, for which the helper returns
            // exactly `max_response_bytes`. Routing through the shared helper keeps all
            // three sites uniform and correct-by-construction should that guard ever
            // be relaxed to admit a tighter response mode here.
            if let Some(limit) = response_body_overflow_limit(
                pipeline.body_capabilities().response_body_mode,
                self.max_response_bytes,
                completion_body.as_ref().map_or(0, Bytes::len),
            ) {
                let error = FilteredSubrequestError::capture(
                    "filtered_subrequest: abnormal completion exceeds response body limit"
                        .to_owned()
                        .into(),
                    &mut filter_ctx,
                )
                .with_too_large(ResponseTooLargeInfo {
                    actual: Some(completion_body.as_ref().map_or(0, Bytes::len)),
                    limit,
                });
                return Err(error);
            }
            outcome.response.body = completion_body.unwrap_or_default();
        }
        let (kind, response_snapshot, completed) = match raw {
            RawResponse::Complete(outcome) => {
                let snapshot = crate::Response {
                    status: http::StatusCode::from_u16(outcome.response.status)
                        .unwrap_or(http::StatusCode::BAD_GATEWAY),
                    headers: outcome.response.headers.clone(),
                };
                (OpenedResponse::Complete(outcome), snapshot, true)
            },
            RawResponse::Rejected(rejection) => {
                let response = subresponse_from_rejection(rejection);
                let snapshot = crate::Response {
                    status: http::StatusCode::from_u16(response.status).unwrap_or(http::StatusCode::BAD_GATEWAY),
                    headers: response.headers.clone(),
                };
                (
                    OpenedResponse::Complete(SubrequestOutcome {
                        response,
                        origin: ResponseOrigin::Local,
                        transport_error: None,
                    }),
                    snapshot,
                    true,
                )
            },
            RawResponse::Streaming { body, outcome } => {
                let snapshot = crate::Response {
                    status: http::StatusCode::from_u16(outcome.response.status)
                        .unwrap_or(http::StatusCode::BAD_GATEWAY),
                    headers: outcome.response.headers.clone(),
                };
                (OpenedResponse::Streaming { body, outcome }, snapshot, false)
            },
            // An executor-side body-limit breach carries the typed overflow
            // detail out as an error, uniformly with the transport path. The
            // response-filter lifecycle already ran before the breach.
            RawResponse::ResponseTooLarge { actual, limit, message } => {
                let error = FilteredSubrequestError::capture(message.to_owned().into(), &mut filter_ctx)
                    .with_too_large(ResponseTooLargeInfo {
                        actual: Some(actual),
                        limit,
                    });
                return Err(error);
            },
        };
        let request_snapshot = crate::Request {
            method: filter_ctx.request.method.clone(),
            uri: filter_ctx.request.uri.clone(),
            headers: filter_ctx.request.headers.clone(),
        };
        let continuation = FilteredSubrequestContinuation::capture(
            Arc::clone(pipeline),
            request_snapshot,
            response_snapshot,
            &mut filter_ctx,
            completed,
            step_deadline,
        );
        Ok(OpenedSubrequest { continuation, kind })
    }
}