praxis-proxy-filter 0.7.2

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

//! Utility functions for HTTP pipeline execution.
//!
//! Provides the per-filter dispatch utilities called by the execution
//! loops in [`http`]: `run_request_filter`, `run_response_filter`,
//! and their body-phase counterparts. Also handles body byte
//! accumulation, response condition evaluation, failure-mode
//! fallback, and optional per-filter duration metrics.
//!
//! [`http`]: super::http

use bytes::Bytes;
use praxis_core::config::FailureMode;
use tracing::{Instrument as _, debug, debug_span, trace, warn};

use super::{check_failure_mode, filter::PipelineFilter};
use crate::{
    FilterError,
    actions::{FilterAction, Rejection, SelectedUpstreamBodyOutcome},
    any_filter::AnyFilter,
    body::BodyAccess,
    condition::{SelectedUpstream, should_execute_from, should_execute_response_ref},
    context::{EffectiveHeaders, HttpFilterContext, Response},
    metrics::{
        PHASE_REQUEST, PHASE_RESPONSE, PHASE_SELECTED_UPSTREAM, STREAM_BODY, STREAM_HEADERS, record_filter_duration,
    },
};
#[cfg(feature = "bound-upstream-request-body")]
use crate::{actions::BoundUpstreamBodyOutcome, metrics::PHASE_BOUND_UPSTREAM};

// -----------------------------------------------------------------------------
// Body Filter Utilities
// -----------------------------------------------------------------------------

/// Add chunk size to accumulator.
pub(super) fn accumulate_body_bytes(counter: &mut u64, body: Option<&Bytes>) {
    if let Some(b) = body {
        *counter += b.len() as u64;
    }
}

/// Return `Release` or `Continue` based on `released` flag.
pub(super) fn released_or_continue(released: bool) -> FilterAction {
    if released {
        FilterAction::Release
    } else {
        FilterAction::Continue
    }
}

/// Borrow the load balancer's published selection as a condition-eval view.
///
/// Both fields are `None` until a load balancer publishes a selection, so a
/// `selected_upstream` predicate evaluated before then fails closed.
pub(super) fn ctx_selected_upstream<'a>(ctx: &'a HttpFilterContext<'_>) -> SelectedUpstream<'a> {
    SelectedUpstream {
        application_protocol: ctx.selected_application_protocol(),
        application_provider: ctx.selected_application_provider(),
    }
}

/// Extract an HTTP filter eligible for request body processing.
///
/// `conditions_resolved` says the request phase already evaluated this
/// filter's conditions (a skipped filter is unmarked in
/// `executed_filter_indices` and filtered out before this call), so the
/// per-chunk re-walk of path/method/header conditions is skipped.
///
/// On the untracked pre-read path conditions are re-evaluated here against
/// the *effective* headers ([`EffectiveHeaders`]): the original request
/// overlaid with headers earlier pre-read filters promoted from the body. An
/// ambiguous effective value fails closed as a [`FilterError`] rather than
/// silently skipping the filter.
///
/// # Errors
///
/// Returns [`FilterError`] when a conditioned header has no unambiguous
/// effective value during pre-read.
pub(super) fn as_request_body_filter<'a>(
    pf: &'a PipelineFilter,
    ctx: &HttpFilterContext<'_>,
    conditions_resolved: bool,
) -> Result<Option<&'a dyn crate::filter::HttpFilter>, FilterError> {
    // Callers reach here only via the precomputed body-filter index
    // lists, which already encode `request_body_access() != None`.
    let AnyFilter::Http(http_filter) = &pf.filter else {
        return Ok(None);
    };
    if !conditions_resolved && !pf.conditions.is_empty() {
        let run = should_execute_from(
            &pf.conditions,
            ctx.request,
            &EffectiveHeaders(ctx),
            ctx.bound_upstream_view(),
            ctx_selected_upstream(ctx),
        )
        .map_err(|e| FilterError::from(format!("{}: {e}", http_filter.name())))?;
        if !run {
            debug!(
                filter = http_filter.name(),
                "body hook skipped by conditions (effective headers)"
            );
            return Ok(None);
        }
    }
    Ok(Some(http_filter.as_ref()))
}

/// Extract an HTTP filter eligible for response body processing.
pub(super) fn as_response_body_filter<'a>(
    filter: &'a AnyFilter,
    resp_conditions: &[praxis_core::config::ResponseCondition],
    response_header: Option<&Response>,
) -> Option<&'a dyn crate::filter::HttpFilter> {
    // Callers reach here only via the precomputed body-filter index
    // lists, which already encode `response_body_access() != None`.
    let http_filter = match filter {
        AnyFilter::Http(f) => f.as_ref(),
        AnyFilter::Tcp(_) => return None,
    };
    if skip_by_response_conditions_with_header(http_filter, resp_conditions, response_header) {
        return None;
    }
    Some(http_filter)
}

// -----------------------------------------------------------------------------
// Filter Dispatch Utilities
// -----------------------------------------------------------------------------

/// Outcome of a single body filter invocation.
#[derive(Debug)]
pub(super) enum BodyFilterOutcome {
    /// Filter completed body inspection; skip on remaining chunks.
    BodyDone,

    /// Filter passed; continue to next.
    Continue,

    /// Filter released the body.
    Released,

    /// Filter rejected with the given rejection.
    Rejected(Rejection),
}

/// Classify a body filter result into a [`BodyFilterOutcome`], logging on reject/error.
///
/// When `failure_mode` is [`FailureMode::Open`], errors are logged as
/// warnings and the filter is treated as if it returned `Continue`.
pub(super) fn dispatch_body_result(
    result: Result<FilterAction, FilterError>,
    filter_name: &str,
    phase: &str,
    failure_mode: FailureMode,
) -> Result<BodyFilterOutcome, FilterError> {
    match result {
        Ok(FilterAction::Continue) => Ok(BodyFilterOutcome::Continue),
        Ok(FilterAction::TerminalResponse(_) | FilterAction::StreamingTerminalResponse(_)) => {
            warn!(
                filter = filter_name,
                "{phase}: terminal response ignored; only request-phase filters may synthesize a response"
            );
            Ok(BodyFilterOutcome::Continue)
        },
        Ok(FilterAction::Release) => {
            debug!(filter = filter_name, "filter released body");
            Ok(BodyFilterOutcome::Released)
        },
        Ok(FilterAction::Reject(rejection)) => {
            warn!(
                filter = filter_name,
                status = rejection.status,
                "{phase} rejected by filter"
            );
            Ok(BodyFilterOutcome::Rejected(rejection))
        },
        Ok(FilterAction::BodyDone) => {
            debug!(filter = filter_name, "filter signaled body done");
            Ok(BodyFilterOutcome::BodyDone)
        },
        Err(e) => {
            check_failure_mode(filter_name, e, phase, failure_mode)?;
            Ok(BodyFilterOutcome::Continue)
        },
    }
}

/// Classify a selected-upstream body filter result, logging on reject/error.
///
/// When `failure_mode` is [`FailureMode::Open`], errors are logged as
/// warnings and the filter is treated as if it returned
/// [`SelectedUpstreamBodyOutcome::Continue`].
pub(super) fn dispatch_selected_upstream_body_result(
    result: Result<SelectedUpstreamBodyOutcome, FilterError>,
    filter_name: &str,
    failure_mode: FailureMode,
) -> Result<SelectedUpstreamBodyOutcome, FilterError> {
    match result {
        Ok(SelectedUpstreamBodyOutcome::Continue) => Ok(SelectedUpstreamBodyOutcome::Continue),
        Ok(SelectedUpstreamBodyOutcome::Reject(rejection)) => {
            warn!(
                filter = filter_name,
                status = rejection.status,
                "selected-upstream request body rejected by filter"
            );
            Ok(SelectedUpstreamBodyOutcome::Reject(rejection))
        },
        Err(e) => {
            check_failure_mode(filter_name, e, "selected-upstream request body", failure_mode)?;
            Ok(SelectedUpstreamBodyOutcome::Continue)
        },
    }
}

/// Record a selected-upstream body result on the span's `filter.result` field.
fn record_selected_upstream_result(span: &tracing::Span, result: &Result<SelectedUpstreamBodyOutcome, FilterError>) {
    let label = match result {
        Ok(SelectedUpstreamBodyOutcome::Continue) => "continue",
        Ok(SelectedUpstreamBodyOutcome::Reject(_)) => "reject",
        Err(_) => "error",
    };
    span.record("filter.result", label);
}

/// Classify a bound-upstream body filter result, logging on reject/error.
///
/// When `failure_mode` is [`FailureMode::Open`], errors are logged as
/// warnings and the filter is treated as if it returned
/// [`BoundUpstreamBodyOutcome::Continue`].
#[cfg(feature = "bound-upstream-request-body")]
pub(super) fn dispatch_bound_upstream_body_result(
    result: Result<BoundUpstreamBodyOutcome, FilterError>,
    filter_name: &str,
    failure_mode: FailureMode,
) -> Result<BoundUpstreamBodyOutcome, FilterError> {
    match result {
        Ok(BoundUpstreamBodyOutcome::Continue) => Ok(BoundUpstreamBodyOutcome::Continue),
        Ok(BoundUpstreamBodyOutcome::Reject(rejection)) => {
            warn!(
                filter = filter_name,
                status = rejection.status,
                "bound-upstream request body rejected by filter"
            );
            Ok(BoundUpstreamBodyOutcome::Reject(rejection))
        },
        Err(e) => {
            check_failure_mode(filter_name, e, "bound-upstream request body", failure_mode)?;
            Ok(BoundUpstreamBodyOutcome::Continue)
        },
    }
}

/// Record a bound-upstream body result on the span's `filter.result` field.
#[cfg(feature = "bound-upstream-request-body")]
fn record_bound_upstream_result(span: &tracing::Span, result: &Result<BoundUpstreamBodyOutcome, FilterError>) {
    let label = match result {
        Ok(BoundUpstreamBodyOutcome::Continue) => "continue",
        Ok(BoundUpstreamBodyOutcome::Reject(_)) => "reject",
        Err(_) => "error",
    };
    span.record("filter.result", label);
}

/// Returns `true` if the filter should be skipped due to
/// response conditions not matching.
pub(super) fn skip_by_response_conditions(
    http_filter: &dyn crate::filter::HttpFilter,
    resp_conditions: &[praxis_core::config::ResponseCondition],
    ctx: &HttpFilterContext<'_>,
) -> bool {
    let response_header = ctx.response_header.as_deref();
    skip_by_response_conditions_with_header(http_filter, resp_conditions, response_header)
}

/// Returns `true` if response conditions fail against the provided header.
pub(super) fn skip_by_response_conditions_with_header(
    http_filter: &dyn crate::filter::HttpFilter,
    resp_conditions: &[praxis_core::config::ResponseCondition],
    response_header: Option<&Response>,
) -> bool {
    let Some(resp) = response_header else {
        return false;
    };
    if !resp_conditions.is_empty() && !should_execute_response_ref(resp_conditions, resp.status, &resp.headers) {
        trace!(filter = http_filter.name(), "skipped by response conditions");
        return true;
    }
    false
}

// -----------------------------------------------------------------------------
// Filter Hook Runners
// -----------------------------------------------------------------------------

/// Outcome of running a single header filter hook (`on_request` or `on_response`).
pub(super) enum HeaderFilterOutcome {
    /// Filter executed successfully; continue pipeline.
    Continue,

    /// Filter rejected the request or response.
    Rejected(Rejection),

    /// Filter produced a complete terminal response (request phase only).
    TerminalResponse(Box<crate::actions::TerminalResponse>),

    /// Filter produced a streaming terminal response (request phase only).
    StreamingTerminalResponse(Box<crate::actions::StreamingTerminalResponse>),
}

/// Run a single request header filter hook with tracing and metrics.
#[expect(clippy::too_many_lines, reason = "metrics instrumentation adds branches per hook")]
pub(super) async fn run_request_filter(
    http_filter: &dyn crate::filter::HttpFilter,
    ctx: &mut HttpFilterContext<'_>,
    failure_mode: FailureMode,
    metrics_enabled: bool,
) -> Result<HeaderFilterOutcome, FilterError> {
    let filter_span = debug_span!(
        "filter",
        "otel.name" = %format_args!("filter:{}:request", http_filter.name()),
        "filter.name" = http_filter.name(),
        "filter.phase" = "request",
        "filter.result" = tracing::field::Empty,
    );
    let request_result = async {
        trace!("on_request");
        let result = if metrics_enabled {
            let start = std::time::Instant::now();
            let result = http_filter.on_request(ctx).await;
            record_filter_duration(
                http_filter.name(),
                PHASE_REQUEST,
                STREAM_HEADERS,
                start.elapsed().as_secs_f64(),
            );
            result
        } else {
            http_filter.on_request(ctx).await
        };
        // Record on the filter span itself: `Span::current()` would resolve
        // to the parent request span whenever the filter span is disabled
        // and pay a subscriber clone per hook for nothing.
        record_filter_result(&filter_span, &result);
        result
    }
    .instrument(filter_span.clone())
    .await;
    match request_result {
        Ok(FilterAction::Continue | FilterAction::Release | FilterAction::BodyDone) => {
            Ok(HeaderFilterOutcome::Continue)
        },
        Ok(FilterAction::Reject(rejection)) => {
            warn!(
                filter = http_filter.name(),
                status = rejection.status,
                "request rejected by filter"
            );
            Ok(HeaderFilterOutcome::Rejected(rejection))
        },
        Ok(FilterAction::TerminalResponse(terminal)) => {
            debug!(
                filter = http_filter.name(),
                status = terminal.status,
                "filter produced terminal response"
            );
            Ok(HeaderFilterOutcome::TerminalResponse(terminal))
        },
        Ok(FilterAction::StreamingTerminalResponse(terminal)) => {
            debug!(
                filter = http_filter.name(),
                status = terminal.status,
                "filter produced streaming terminal response"
            );
            Ok(HeaderFilterOutcome::StreamingTerminalResponse(terminal))
        },
        Err(e) => {
            check_failure_mode(http_filter.name(), e, "request", failure_mode)?;
            Ok(HeaderFilterOutcome::Continue)
        },
    }
}

/// Run a single request body filter hook with tracing and metrics.
#[expect(clippy::too_many_arguments, reason = "metrics_enabled flag is required per hook")]
pub(super) async fn run_request_body_filter(
    http_filter: &dyn crate::filter::HttpFilter,
    ctx: &mut HttpFilterContext<'_>,
    body: &mut Option<Bytes>,
    end_of_stream: bool,
    failure_mode: FailureMode,
    metrics_enabled: bool,
) -> Result<BodyFilterOutcome, FilterError> {
    let filter_span = debug_span!(
        "filter",
        "otel.name" = %format_args!("filter:{}:request_body", http_filter.name()),
        "filter.name" = http_filter.name(),
        "filter.phase" = "request_body",
        "filter.result" = tracing::field::Empty,
    );
    let body_result = async {
        trace!("on_request_body");
        let result = if metrics_enabled {
            let start = std::time::Instant::now();
            let result = http_filter.on_request_body(ctx, body, end_of_stream).await;
            record_filter_duration(
                http_filter.name(),
                PHASE_REQUEST,
                STREAM_BODY,
                start.elapsed().as_secs_f64(),
            );
            result
        } else {
            http_filter.on_request_body(ctx, body, end_of_stream).await
        };
        record_filter_result(&filter_span, &result);
        result
    }
    .instrument(filter_span.clone())
    .await;
    dispatch_body_result(body_result, http_filter.name(), "request body", failure_mode)
}

/// The tracing span for one body hook invocation.
fn body_hook_span(filter_name: &'static str, phase: &'static str) -> tracing::Span {
    debug_span!(
        "filter",
        "otel.name" = %format_args!("filter:{filter_name}:{phase}"),
        "filter.name" = filter_name,
        "filter.phase" = phase,
        "filter.result" = tracing::field::Empty,
    )
}

/// Await a body hook, recording its duration under `phase` when metrics are
/// on.
async fn timed_body_hook<T>(
    metrics_enabled: bool,
    filter_name: &'static str,
    phase: &'static str,
    hook: impl Future<Output = T>,
) -> T {
    if !metrics_enabled {
        return hook.await;
    }
    let start = std::time::Instant::now();
    let result = hook.await;
    record_filter_duration(filter_name, phase, STREAM_BODY, start.elapsed().as_secs_f64());
    result
}

/// Run a single selected-upstream request body filter hook with tracing
/// and metrics.
///
/// Mirrors [`run_request_body_filter`], but for the post-selection phase:
/// the hook receives the fully buffered body and returns the reduced
/// [`SelectedUpstreamBodyOutcome`] (continue or reject), so there is no
/// `end_of_stream` argument and no release/body-done handling.
pub(super) async fn run_selected_upstream_request_body_filter(
    http_filter: &dyn crate::filter::HttpFilter,
    ctx: &mut HttpFilterContext<'_>,
    body: &mut Option<Bytes>,
    failure_mode: FailureMode,
    metrics_enabled: bool,
) -> Result<SelectedUpstreamBodyOutcome, FilterError> {
    let filter_span = body_hook_span(http_filter.name(), "selected_upstream_request_body");
    // A read-only filter works on a scratch copy so it cannot change the body
    // it was promised not to; a writer that errors gets its partial edit
    // undone before a fail-open swallows the error.
    let before = body.clone();
    let mut scratch = body.clone();
    let target = if http_filter.selected_upstream_request_body_access() == BodyAccess::ReadWrite {
        &mut *body
    } else {
        &mut scratch
    };
    let body_result = async {
        trace!("on_selected_upstream_request_body");
        let result = timed_body_hook(
            metrics_enabled,
            http_filter.name(),
            PHASE_SELECTED_UPSTREAM,
            http_filter.on_selected_upstream_request_body(ctx, target),
        )
        .await;
        record_selected_upstream_result(&filter_span, &result);
        result
    }
    .instrument(filter_span.clone())
    .await;
    if body_result.is_err() {
        *body = before;
    }
    dispatch_selected_upstream_body_result(body_result, http_filter.name(), failure_mode)
}

/// Run a single bound-upstream request body filter hook with tracing and
/// metrics.
///
/// Mirrors [`run_selected_upstream_request_body_filter`], but for the
/// once-per-request bound-upstream barrier: the hook receives the fully
/// buffered request body and returns the reduced [`BoundUpstreamBodyOutcome`]
/// (continue or reject). Metrics are recorded under [`PHASE_BOUND_UPSTREAM`]
/// so the barrier pass stays separable from the normal request-body phase.
#[cfg(feature = "bound-upstream-request-body")]
pub(super) async fn run_bound_upstream_request_body_filter(
    http_filter: &dyn crate::filter::HttpFilter,
    ctx: &mut HttpFilterContext<'_>,
    body: &mut Option<Bytes>,
    failure_mode: FailureMode,
    metrics_enabled: bool,
) -> Result<(BoundUpstreamBodyOutcome, bool), FilterError> {
    // A read-only participant works on a scratch copy so it cannot change the
    // body it was promised not to; a writer that errors gets its partial edit
    // undone before a fail-open swallows the error. The flag says whether the
    // body may now differ from `before`.
    let writer = http_filter.bound_upstream_request_body_access() == BodyAccess::ReadWrite;
    let before = body.clone();
    let mut scratch = body.clone();
    let target = if writer { &mut *body } else { &mut scratch };
    let filter_span = body_hook_span(http_filter.name(), "bound_upstream_request_body");
    let body_result = async {
        trace!("on_bound_upstream_request_body");
        let result = timed_body_hook(
            metrics_enabled,
            http_filter.name(),
            PHASE_BOUND_UPSTREAM,
            http_filter.on_bound_upstream_request_body(ctx, target),
        )
        .await;
        record_bound_upstream_result(&filter_span, &result);
        result
    }
    .instrument(filter_span.clone())
    .await;
    let rewrote = writer && body_result.is_ok();
    if body_result.is_err() {
        *body = before;
    }
    dispatch_bound_upstream_body_result(body_result, http_filter.name(), failure_mode).map(|outcome| (outcome, rewrote))
}

/// Run a single response body filter hook with tracing and metrics.
#[expect(clippy::too_many_arguments, reason = "metrics_enabled flag is required per hook")]
pub(super) fn run_response_body_filter(
    http_filter: &dyn crate::filter::HttpFilter,
    ctx: &mut HttpFilterContext<'_>,
    body: &mut Option<Bytes>,
    end_of_stream: bool,
    failure_mode: FailureMode,
    metrics_enabled: bool,
) -> Result<BodyFilterOutcome, FilterError> {
    let filter_span = debug_span!(
        "filter",
        "otel.name" = %format_args!("filter:{}:response_body", http_filter.name()),
        "filter.name" = http_filter.name(),
        "filter.phase" = "response_body",
        "filter.result" = tracing::field::Empty,
    );
    let _entered = filter_span.enter();
    trace!("on_response_body");
    let body_result = if metrics_enabled {
        let start = std::time::Instant::now();
        let result = http_filter.on_response_body(ctx, body, end_of_stream);
        record_filter_duration(
            http_filter.name(),
            PHASE_RESPONSE,
            STREAM_BODY,
            start.elapsed().as_secs_f64(),
        );
        result
    } else {
        http_filter.on_response_body(ctx, body, end_of_stream)
    };
    record_filter_result(&filter_span, &body_result);
    dispatch_body_result(body_result, http_filter.name(), "response body", failure_mode)
}

/// Run a single response header filter.
///
/// When `failure_mode` is [`FailureMode::Open`], errors are logged as
/// warnings and the filter is treated as if it returned `Continue`.
///
/// This hook does **not** infer header modification. Comparing the header
/// count before and after the filter misses same-count edits (a removal
/// paired with an addition), and the protocol layer relies on that signal
/// to pick a write-back strategy. Detection therefore lives at the
/// write-back site, which compares the actual header name sequence.
/// [`response_headers_modified`] remains available for filters to set as
/// an explicit hint.
///
/// [`response_headers_modified`]: HttpFilterContext::response_headers_modified
#[expect(clippy::too_many_lines, reason = "metrics instrumentation adds branches per hook")]
pub(super) async fn run_response_filter(
    http_filter: &dyn crate::filter::HttpFilter,
    ctx: &mut HttpFilterContext<'_>,
    failure_mode: FailureMode,
    metrics_enabled: bool,
) -> Result<HeaderFilterOutcome, FilterError> {
    let filter_span = debug_span!(
        "filter",
        "otel.name" = %format_args!("filter:{}:response", http_filter.name()),
        "filter.name" = http_filter.name(),
        "filter.phase" = "response",
        "filter.result" = tracing::field::Empty,
    );
    let response_result = async {
        trace!("on_response");
        let result = if metrics_enabled {
            let start = std::time::Instant::now();
            let result = http_filter.on_response(ctx).await;
            record_filter_duration(
                http_filter.name(),
                PHASE_RESPONSE,
                STREAM_HEADERS,
                start.elapsed().as_secs_f64(),
            );
            result
        } else {
            http_filter.on_response(ctx).await
        };
        record_filter_result(&filter_span, &result);
        result
    }
    .instrument(filter_span.clone())
    .await;
    match response_result {
        Ok(
            FilterAction::Continue
            | FilterAction::Release
            | FilterAction::BodyDone
            | FilterAction::TerminalResponse(_)
            | FilterAction::StreamingTerminalResponse(_),
        ) => Ok(HeaderFilterOutcome::Continue),
        Ok(FilterAction::Reject(rejection)) => {
            warn!(
                filter = http_filter.name(),
                status = rejection.status,
                "response rejected by filter"
            );
            Ok(HeaderFilterOutcome::Rejected(rejection))
        },
        Err(e) => {
            check_failure_mode(http_filter.name(), e, "response", failure_mode)?;
            Ok(HeaderFilterOutcome::Continue)
        },
    }
}

/// Record the filter result on the span's `filter.result` field.
fn record_filter_result(span: &tracing::Span, result: &Result<FilterAction, FilterError>) {
    let label = match result {
        Ok(FilterAction::Continue) => "continue",
        Ok(FilterAction::Release) => "release",
        Ok(FilterAction::BodyDone) => "body_done",
        Ok(FilterAction::Reject(_)) => "reject",
        Ok(FilterAction::TerminalResponse(_)) => "terminal",
        Ok(FilterAction::StreamingTerminalResponse(_)) => "streaming_terminal",
        Err(_) => "error",
    };
    span.record("filter.result", label);
}

// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------

#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic,
    reason = "tests"
)]
mod tests {
    use super::*;
    use crate::HttpFilter;

    #[test]
    fn accumulate_body_bytes_some_adds_to_counter() {
        let mut counter = 0_u64;
        let body = Some(Bytes::from_static(b"hello"));
        accumulate_body_bytes(&mut counter, body.as_ref());
        assert_eq!(counter, 5, "counter should equal byte length of body");
    }

    #[test]
    fn accumulate_body_bytes_none_does_not_change_counter() {
        let mut counter = 42_u64;
        accumulate_body_bytes(&mut counter, None);
        assert_eq!(counter, 42, "counter should remain unchanged for None body");
    }

    #[test]
    fn accumulate_body_bytes_multiple_sums_correctly() {
        let mut counter = 0_u64;
        accumulate_body_bytes(&mut counter, Some(&Bytes::from_static(b"abc")));
        accumulate_body_bytes(&mut counter, Some(&Bytes::from_static(b"defgh")));
        accumulate_body_bytes(&mut counter, None);
        accumulate_body_bytes(&mut counter, Some(&Bytes::from_static(b"ij")));
        assert_eq!(counter, 10, "counter should be sum of all Some chunk lengths");
    }

    #[test]
    fn released_or_continue_true_returns_release() {
        assert!(
            matches!(released_or_continue(true), FilterAction::Release),
            "true should produce FilterAction::Release"
        );
    }

    #[test]
    fn released_or_continue_false_returns_continue() {
        assert!(
            matches!(released_or_continue(false), FilterAction::Continue),
            "false should produce FilterAction::Continue"
        );
    }

    #[test]
    fn dispatch_body_result_ok_continue() {
        let outcome = dispatch_body_result(Ok(FilterAction::Continue), "test", "request", FailureMode::Closed).unwrap();
        assert!(
            matches!(outcome, BodyFilterOutcome::Continue),
            "Ok(Continue) should produce BodyFilterOutcome::Continue"
        );
    }

    #[test]
    fn dispatch_body_result_terminal_response_is_ignored_not_honored() {
        let terminal = FilterAction::TerminalResponse(Box::new(crate::TerminalResponse::new(200)));
        let outcome = dispatch_body_result(Ok(terminal), "test", "request_body", FailureMode::Closed).unwrap();
        assert!(
            matches!(outcome, BodyFilterOutcome::Continue),
            "a terminal response from a body phase is dropped and the body continues"
        );
    }

    #[test]
    fn dispatch_body_result_ok_release() {
        let outcome = dispatch_body_result(Ok(FilterAction::Release), "test", "request", FailureMode::Closed).unwrap();
        assert!(
            matches!(outcome, BodyFilterOutcome::Released),
            "Ok(Release) should produce BodyFilterOutcome::Released"
        );
    }

    #[test]
    fn dispatch_body_result_ok_reject() {
        let rejection = Rejection::status(429);
        let outcome = dispatch_body_result(
            Ok(FilterAction::Reject(rejection)),
            "test",
            "request",
            FailureMode::Closed,
        )
        .unwrap();
        assert!(
            matches!(&outcome, BodyFilterOutcome::Rejected(r) if r.status == 429),
            "Ok(Reject(429)) should produce BodyFilterOutcome::Rejected with status 429"
        );
    }

    #[test]
    fn dispatch_body_result_ok_body_done() {
        let outcome = dispatch_body_result(Ok(FilterAction::BodyDone), "test", "request", FailureMode::Closed).unwrap();
        assert!(
            matches!(outcome, BodyFilterOutcome::BodyDone),
            "Ok(BodyDone) should produce BodyFilterOutcome::BodyDone"
        );
    }

    #[test]
    fn dispatch_body_result_err_failure_mode_open_swallows_error() {
        let err: FilterError = "test error".into();
        let outcome = dispatch_body_result(Err(err), "test", "request", FailureMode::Open).unwrap();
        assert!(
            matches!(outcome, BodyFilterOutcome::Continue),
            "error with FailureMode::Open should produce BodyFilterOutcome::Continue"
        );
    }

    #[test]
    fn dispatch_body_result_err_failure_mode_closed_propagates() {
        let err: FilterError = "test error".into();
        let result = dispatch_body_result(Err(err), "test", "request", FailureMode::Closed);
        assert!(result.is_err(), "error with FailureMode::Closed should propagate");
    }

    #[test]
    fn skip_by_response_conditions_empty_conditions() {
        let filter = crate::builtins::StaticResponseFilter::from_config(
            &serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
        )
        .unwrap();
        let req = crate::test_utils::make_request(http::Method::GET, "/");
        let mut resp = crate::test_utils::make_response();
        let mut ctx = crate::test_utils::make_filter_context(&req);
        ctx.response_header = Some(&mut resp);
        assert!(
            !skip_by_response_conditions(filter.as_ref(), &[], &ctx),
            "empty conditions should not skip"
        );
    }

    #[test]
    fn skip_by_response_conditions_matching_when_does_not_skip() {
        use praxis_core::config::{ResponseCondition, ResponseConditionMatch};

        let filter = crate::builtins::StaticResponseFilter::from_config(
            &serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
        )
        .unwrap();
        let conds = vec![ResponseCondition::When(ResponseConditionMatch {
            status: Some(vec![200]),
            headers: None,
        })];
        let req = crate::test_utils::make_request(http::Method::GET, "/");
        let mut resp = crate::test_utils::make_response();
        let mut ctx = crate::test_utils::make_filter_context(&req);
        ctx.response_header = Some(&mut resp);
        assert!(
            !skip_by_response_conditions(filter.as_ref(), &conds, &ctx),
            "matching 'when' condition should not skip"
        );
    }

    #[test]
    fn skip_by_response_conditions_non_matching_when_skips() {
        use praxis_core::config::{ResponseCondition, ResponseConditionMatch};

        let filter = crate::builtins::StaticResponseFilter::from_config(
            &serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
        )
        .unwrap();
        let conds = vec![ResponseCondition::When(ResponseConditionMatch {
            status: Some(vec![404]),
            headers: None,
        })];
        let req = crate::test_utils::make_request(http::Method::GET, "/");
        let mut resp = crate::test_utils::make_response();
        let mut ctx = crate::test_utils::make_filter_context(&req);
        ctx.response_header = Some(&mut resp);
        assert!(
            skip_by_response_conditions(filter.as_ref(), &conds, &ctx),
            "non-matching 'when' condition should skip"
        );
    }

    #[test]
    fn skip_by_response_conditions_no_response_header_does_not_skip() {
        use praxis_core::config::{ResponseCondition, ResponseConditionMatch};

        let filter = crate::builtins::StaticResponseFilter::from_config(
            &serde_yaml::from_str::<serde_yaml::Value>("status: 200").unwrap(),
        )
        .unwrap();
        let conds = vec![ResponseCondition::When(ResponseConditionMatch {
            status: Some(vec![200]),
            headers: None,
        })];
        let req = crate::test_utils::make_request(http::Method::GET, "/");
        let ctx = crate::test_utils::make_filter_context(&req);
        assert!(
            !skip_by_response_conditions(filter.as_ref(), &conds, &ctx),
            "no response header should not skip"
        );
    }

    #[test]
    fn skip_by_response_conditions_unless_match_skips() {
        use http::StatusCode;
        use praxis_core::config::{ResponseCondition, ResponseConditionMatch};

        let req = crate::test_utils::make_request(http::Method::GET, "/");
        let mut resp = crate::test_utils::make_response();
        resp.status = StatusCode::BAD_REQUEST;

        let mut ctx = crate::test_utils::make_filter_context(&req);
        ctx.response_header = Some(&mut resp);

        let conditions = vec![ResponseCondition::Unless(ResponseConditionMatch {
            status: Some(vec![400]),
            headers: None,
        })];

        let filter = StubFilter;
        assert!(
            skip_by_response_conditions(&filter, &conditions, &ctx),
            "Unless with matching status should cause skip"
        );
    }

    // -------------------------------------------------------------------------
    // Span Event Tests
    // -------------------------------------------------------------------------

    #[test]
    fn dispatch_body_result_rejection_returns_status() {
        let rejection = Rejection::status(413);
        let outcome = dispatch_body_result(
            Ok(FilterAction::Reject(rejection)),
            "size_limit",
            "request body",
            FailureMode::Closed,
        )
        .unwrap();
        assert!(
            matches!(&outcome, BodyFilterOutcome::Rejected(r) if r.status == 413),
            "body rejection should carry status 413 for span event"
        );
    }

    #[test]
    fn dispatch_body_result_response_rejection_returns_status() {
        let rejection = Rejection::status(500);
        let outcome = dispatch_body_result(
            Ok(FilterAction::Reject(rejection)),
            "transform_filter",
            "response body",
            FailureMode::Closed,
        )
        .unwrap();
        assert!(
            matches!(&outcome, BodyFilterOutcome::Rejected(r) if r.status == 500),
            "response body rejection should carry status 500 for span event"
        );
    }

    #[tokio::test]
    async fn run_request_filter_rejection_returns_rejected_outcome() {
        let filter = RejectingFilter(429);
        let req = crate::test_utils::make_request(http::Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = run_request_filter(&filter, &mut ctx, FailureMode::Closed, false)
            .await
            .unwrap();
        assert!(
            matches!(&outcome, HeaderFilterOutcome::Rejected(r) if r.status == 429),
            "rejecting filter should produce Rejected outcome with status for span event"
        );
    }

    #[tokio::test]
    async fn run_response_filter_rejection_returns_rejected_outcome() {
        let filter = ResponseRejectingFilter(503);
        let req = crate::test_utils::make_request(http::Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = run_response_filter(&filter, &mut ctx, FailureMode::Closed, false)
            .await
            .unwrap();
        assert!(
            matches!(&outcome, HeaderFilterOutcome::Rejected(r) if r.status == 503),
            "response rejection should produce Rejected outcome with status for span event"
        );
    }

    // -------------------------------------------------------------------------
    // Test Utilities
    // -------------------------------------------------------------------------

    /// Minimal HTTP filter stub for unit tests.
    struct StubFilter;

    #[async_trait::async_trait]
    impl HttpFilter for StubFilter {
        fn name(&self) -> &'static str {
            "stub"
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Ok(FilterAction::Continue)
        }
    }

    /// HTTP filter that always rejects requests with the given status.
    struct RejectingFilter(u16);

    #[async_trait::async_trait]
    impl HttpFilter for RejectingFilter {
        fn name(&self) -> &'static str {
            "rejecting_filter"
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Ok(FilterAction::Reject(Rejection::status(self.0)))
        }
    }

    /// HTTP filter that always rejects responses with the given status.
    struct ResponseRejectingFilter(u16);

    #[async_trait::async_trait]
    impl HttpFilter for ResponseRejectingFilter {
        fn name(&self) -> &'static str {
            "response_rejecting_filter"
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Ok(FilterAction::Continue)
        }

        async fn on_response(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Ok(FilterAction::Reject(Rejection::status(self.0)))
        }
    }
}