praxis-proxy-filter 0.7.2

// 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 also scopes the parent's upstream state: on entry it clears the
//! exchange-local upstream selection and, with the `upstream-binding` feature,
//! sets the parent's logical binding aside (an IRR step keeps seeing it, a
//! callout starts unbound), and it restores that checkpoint on every exit.
//! 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;
mod types;

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

use bytes::Bytes;
use http::HeaderMap;
use praxis_core::{
    connectivity::Upstream,
    subrequest::{FrameworkHeaders, StreamLimits, SubRequestError},
};
use tracing::{Instrument as _, warn};

pub use self::types::{CalloutOutcome, CalloutResponse, StagedUpstream, StagedUpstreamFallback, SubrequestRuntime};
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,
    streaming::{CalloutStreamingBody, FilteredStreamingBody},
    types::{
        FilteredSubrequestError, FilteredSubrequestInput, OpenedResponse, OpenedSubrequest, RawResponse,
        ResponseOrigin, ResponseTooLargeInfo, SubrequestOutcome, TransportFailure,
    },
};
#[cfg(feature = "iterative-request-router")]
pub(crate) use self::{continuation::SubrequestCompletion, sanitize::normalize_response_status};
#[cfg(feature = "chain-binding")]
use crate::credentials::{PendingCredentials, ResolvedDestination};
#[cfg(feature = "bound-upstream-request-body")]
use crate::extensions::BoundRequestBodyRewrite;
#[cfg(feature = "upstream-binding")]
use crate::extensions::{BoundUpstream, BoundUpstreamFrozen};
use crate::{
    FilterAction, FilterError, FilterPipeline, StreamTermination, StreamTerminationCause, SubRequest,
    SubRequestResponseMode, SubResponse,
    actions::Rejection,
    context::PendingStreamChunks,
    extensions::{RequestExtensions, SelectedClusterApplication},
    results::RetainedFilterResults,
};

/// Parent routing state shadowed while a nested pipeline executes.
#[cfg(feature = "upstream-binding")]
struct ParentUpstreamState {
    /// Parent request's logical cluster binding.
    binding: Option<BoundUpstream>,
    /// Whether the parent binding had already reached its freeze point.
    frozen: bool,
    /// Body the parent's bound-upstream phase rewrote, kept away from the
    /// nested pipeline so its executor cannot take it as its own.
    #[cfg(feature = "bound-upstream-request-body")]
    body_rewrite: Option<BoundRequestBodyRewrite>,
}

/// Stack form supports filtered sub-requests nested inside another filtered
/// sub-request without overwriting the outer restoration checkpoint.
#[cfg(feature = "upstream-binding")]
#[derive(Default)]
struct ParentUpstreamStates(Vec<ParentUpstreamState>);

/// Save parent routing state before a nested pipeline runs.
///
/// A nested pipeline that `inherits_binding` (an IRR step) keeps seeing the
/// parent's binding; any other (an outbound callout) starts unbound, so its own
/// binding router can publish without colliding with the parent's frozen one.
/// The parent's bound-body rewrite is always set aside so the nested executor
/// cannot take it as its own.
#[cfg(feature = "upstream-binding")]
fn enter_nested_upstream_scope(extensions: &mut RequestExtensions, inherits_binding: bool) {
    let (binding, frozen) = if inherits_binding {
        (
            extensions.get::<BoundUpstream>().cloned(),
            extensions.get::<BoundUpstreamFrozen>().is_some(),
        )
    } else {
        (
            extensions.remove::<BoundUpstream>(),
            extensions.remove::<BoundUpstreamFrozen>().is_some(),
        )
    };
    let state = ParentUpstreamState {
        binding,
        frozen,
        #[cfg(feature = "bound-upstream-request-body")]
        body_rewrite: extensions.remove::<BoundRequestBodyRewrite>(),
    };
    let mut states = extensions.remove::<ParentUpstreamStates>().unwrap_or_default();
    states.0.push(state);
    extensions.insert(states);
    extensions.remove::<SelectedClusterApplication>();
}

/// Without the `upstream-binding` feature there is no binding to shadow; a
/// nested pipeline only starts without the parent's upstream selection.
#[cfg(not(feature = "upstream-binding"))]
fn enter_nested_upstream_scope(extensions: &mut RequestExtensions, _inherits_binding: bool) {
    extensions.remove::<SelectedClusterApplication>();
}

/// Remove nested routing state and restore the most recent parent checkpoint.
#[cfg(feature = "upstream-binding")]
fn restore_parent_upstream_scope(extensions: &mut RequestExtensions) {
    extensions.remove::<SelectedClusterApplication>();
    let Some(mut states) = extensions.remove::<ParentUpstreamStates>() else {
        return;
    };
    let Some(state) = states.0.pop() else {
        return;
    };
    extensions.remove::<BoundUpstream>();
    extensions.remove::<BoundUpstreamFrozen>();
    if let Some(binding) = state.binding {
        extensions.insert(binding);
    }
    if state.frozen {
        extensions.insert(BoundUpstreamFrozen);
    }
    #[cfg(feature = "bound-upstream-request-body")]
    {
        extensions.remove::<BoundRequestBodyRewrite>();
        if let Some(body_rewrite) = state.body_rewrite {
            extensions.insert(body_rewrite);
        }
    }
    if !states.0.is_empty() {
        extensions.insert(states);
    }
}

/// Without the `upstream-binding` feature the only nested state to clear is
/// the nested pipeline's upstream selection.
#[cfg(not(feature = "upstream-binding"))]
fn restore_parent_upstream_scope(extensions: &mut RequestExtensions) {
    extensions.remove::<SelectedClusterApplication>();
}

// -----------------------------------------------------------------------------
// Constants
// -----------------------------------------------------------------------------

/// 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);

// -----------------------------------------------------------------------------
// Retained-State Accounting
// -----------------------------------------------------------------------------

/// 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;
}

// -----------------------------------------------------------------------------
// Sub-Request Error
// -----------------------------------------------------------------------------

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 parent's upstream scope (its selection and, with binding, its
    /// logical binding) has already been restored into these extensions.
    ///
    /// 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(mut self) -> (FilterError, RequestExtensions) {
        restore_parent_upstream_scope(&mut self.extensions);
        (self.error, self.extensions)
    }
}

// -----------------------------------------------------------------------------
// Executor
// -----------------------------------------------------------------------------

/// 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
    }
}

/// Restore a caller-staged destination and discard metadata published for any
/// load-balanced selection it replaces.
fn repin_staged_upstream(ctx: &mut crate::HttpFilterContext<'_>, pinned_upstream: Option<&Upstream>) {
    if let Some(upstream) = pinned_upstream {
        ctx.upstream = Some(upstream.clone());
        ctx.extensions.remove::<SelectedClusterApplication>();
    }
}

/// 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`.
    ///     praxis_tls::provider::install(); // required before any connector is built
    ///     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::callout(pipeline, request, 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());
    ///
    ///     praxis_tls::provider::install(); // required before any connector is built
    ///     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::callout(pipeline, request, 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,
            inherits_binding,
        } = input;
        // Every error below is handed back through `into_parts`, which restores
        // one parent checkpoint, so enter the scope before the first of them.
        enter_nested_upstream_scope(&mut extensions, inherits_binding);

        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;
                filter_ctx.buffered_request_body.clone_from(&request_body);
            }

            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 let Some(rewritten) = filter_ctx.take_bound_request_body_rewrite() {
                request_body = Some(rewritten);
            }
            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.
            // #1138 Correction 2: a staged upstream carries no application
            // metadata. Any value a chain filter published for the discarded
            // load-balancer selection would misdescribe the pinned destination,
            // so the body phase must not read it.
            repin_staged_upstream(&mut filter_ctx, pinned_upstream.as_ref());

            // #1138: selected-upstream request-body phase. Mirrors the Pingora
            // path (crates/protocol .../request_filter/mod.rs run_pipeline): after
            // upstream selection and the staged re-pin, run any selected-upstream
            // body filters over the buffered body before dialing. `request_body`
            // is a step-local clone of `current_request.body`, so the adapted
            // bytes flow into the SubRequest (built below) without disturbing the
            // canonical iteration input (D5). The transport reframes from the
            // adapted bytes, so no explicit Content-Length stamping is needed.
            // Gated on participation and a selected upstream.
            if pipeline.body_capabilities().needs_selected_upstream_request_body
                && filter_ctx.upstream.is_some()
            {
                let action = pipeline
                    .execute_http_selected_upstream_request_body(&mut filter_ctx, &mut request_body)
                    .await?;
                if let FilterAction::Reject(rejection) = action {
                    return Ok(RawResponse::Rejected(rejection));
                }
                // A 413 is only meaningful when a writer may have grown the body;
                // check against the same listener-clamped limit as every path.
                if pipeline.body_capabilities().any_selected_upstream_request_body_writer
                    && request_body.as_ref().map_or(0, Bytes::len)
                        > pipeline.selected_upstream_request_body_limit()
                {
                    return Ok(RawResponse::Rejected(Rejection::status(413)));
                }
                // Enforce the retained-state ceiling before dialing, like every
                // preceding filter boundary. A phase filter may have grown
                // cross-sub-request state (for the IRR, the retained
                // `IterationState`); an oversized retention must be rejected here
                // rather than carried across a dial and held until the response or
                // timeout.
                if self.accounting.exceeds_limit(&filter_ctx.extensions) {
                    return Ok(RawResponse::Rejected(Rejection::status(413)));
                }
                // Re-pin the staged upstream once more. This phase received a
                // mutable `ctx` and may have rewritten `ctx.upstream` after the
                // pre-phase re-pin, so reassert the prepared destination before the
                // dial. Otherwise a body adapted for the pinned upstream — for a
                // body-authenticated provider, the request body itself carries the
                // credential — could be redirected to an authority the callout never
                // prepared, the exact exfiltration the staged-upstream invariant
                // blocks. Mirror the pre-phase re-pin: clear any discarded-selection
                // metadata a rewrite may have republished.
                repin_staged_upstream(&mut filter_ctx, pinned_upstream.as_ref());
            }

            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.
            #[cfg(feature = "chain-binding")]
            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.
            #[cfg(feature = "chain-binding")]
            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);
            filter_ctx.apply_trace_propagation(&mut framework_headers);
            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 utility returns
            // exactly `max_response_bytes`. Routing through the shared utility 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 })
    }
}