dwara_core/dataplane/

ai_proxy.rs

//! The AI proxy action (DW-075): the dataplane half of the
//! provider-adapter pack — extended by DW-076 with routing and
//! failover.
//!
//! Runs when a route's action is `ai`: parse the client's OpenAI
//! chat-completions body, resolve the model alias through the
//! generation's [`AiRuntime`](crate::ai::AiRuntime), translate via the
//! provider's [`ProviderAdapter`],
//! place the call through the provider's UPSTREAM (pooling, TLS,
//! timeouts, breaker, health — the standard machinery), and translate
//! the response back. All failures answer in the OpenAI error shape so
//! standard client SDKs parse them; the gateway's request id rides
//! along inside `error.request_id`.
//!
//! # Routing and failover (DW-076)
//!
//! `AiRuntime::route` turns the alias into an ordered candidate list:
//! `[primary, failover...]` for a chained alias, or exactly one
//! deterministically-picked canary version for a split alias (see
//! `ai::routing`). The action walks the candidates and advances on
//! RETRYABLE outcomes — a 429, a 5xx, a transport error, or a
//! dialect-specific translation rejection — and treats everything
//! else as final. Failover is invisible to the client by
//! construction: a provider response is read and translated WHOLE
//! before any byte reaches the client, so a failing candidate emits
//! nothing client-visible. Same-provider retries are deliberately not
//! the chain's job — the provider's upstream breaker owns those; the
//! chain only ever moves to a DIFFERENT provider/model pair. Usage and
//! request metrics attribute to the provider and canary version that
//! actually served; the access record follows the serving provider.
//!
//! Body bounds: an AI request must be read whole to translate it (the
//! zero-buffering rule yields here by necessity, opt-in by choosing
//! the action), and both directions are capped INCREMENTALLY — the
//! read aborts as soon as the cap is crossed, never after buffering
//! the whole body — at [`MAX_AI_REQUEST_BYTES`] inbound and
//! [`MAX_AI_PROVIDER_RESPONSE_BYTES`] from the provider, so a hostile
//! or runaway peer cannot turn the translation into an unbounded
//! buffer.
//!
//! Streaming (DW-077): `stream: true` requests stream back to the
//! client as `text/event-stream`, translated frame-by-frame with ZERO
//! added buffering ([`AiStreamBody`] + [`StreamTranslator`]): each
//! complete provider frame becomes client frames in the same poll.
//! Token counts are provider-reported only (locked decision) and
//! accumulate mid-stream into one terminal usage chunk; the gateway
//! owns the `data: [DONE]` terminator. The failover chain applies
//! until the streaming response is returned (the commit point); after
//! the first forwarded frame a provider abort ends the stream cleanly
//! with an error chunk — already-forwarded content stands.

use crate::ai::adapter::{adapter_for, ProviderAdapter};
use crate::ai::openai_compat;
use crate::ai::stream::StreamTranslator;
use crate::ai::types::{ChatRequest, Usage};
use crate::ai::CompiledProvider;
use crate::config::ai::UnknownModelPolicy;
use crate::dataplane::proxy::{consumer_type_str, DataPlane, Generation, ProxyBody};
use bytes::Bytes;
use http_body_util::{BodyExt as _, Full};
use hyper::body::Body;
use hyper::{HeaderMap, Method, Request, Response, StatusCode};
use std::pin::pin;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use tracing::Instrument as _;

/// Inbound AI request body cap: 16 MiB, generous against long-context
/// requests while bounding the translation buffer.
pub const MAX_AI_REQUEST_BYTES: usize = 16 * 1024 * 1024;

/// Provider response body cap: 32 MiB (non-streaming completions of
/// extreme length). Provider ERROR bodies are capped tighter at
/// [`MAX_AI_ERROR_BYTES`] — they are never legitimately large.
pub const MAX_AI_PROVIDER_RESPONSE_BYTES: usize = 32 * 1024 * 1024;

/// Provider error body cap: 1 MiB.
pub const MAX_AI_ERROR_BYTES: usize = 1024 * 1024;

/// AI-02: the provider path for each non-chat endpoint. The passthrough
/// forwards the request body as-is to the provider's upstream at this
/// path (no adapter translation). The path is OpenAI-compatible (the
/// most common target); providers with different paths can override
/// the path per-provider via `ai.providers[].path`.
fn passthrough_path(endpoint: crate::config::AiEndpoint) -> &'static str {
    match endpoint {
        crate::config::AiEndpoint::Embeddings => "/v1/embeddings",
        crate::config::AiEndpoint::Images => "/v1/images/generations",
        crate::config::AiEndpoint::Audio => "/v1/audio/speech",
        crate::config::AiEndpoint::Moderation => "/v1/moderations",
        crate::config::AiEndpoint::Chat => "/v1/chat/completions",
        // AI-04 (#194): Batch API uses /v1/batches for create/status.
        // The result file endpoint (/v1/batches/{id}/output) is handled
        // separately in serve_ai_batch_results.
        crate::config::AiEndpoint::Batch => "/v1/batches",
    }
}

/// Resolve the outbound request path for a provider. When the provider
/// has a `path` override configured, it is used verbatim for ALL
/// endpoints (chat, embeddings, etc.). When unset, the adapter's
/// per-kind default path is used.
fn resolve_path<'a>(provider: &'a CompiledProvider, default: &'a str) -> &'a str {
    provider.path.as_deref().unwrap_or(default)
}

/// AI-02: serve a non-chat AI endpoint (embeddings, images, audio,
/// moderation) as a passthrough. The request body is forwarded to the
/// provider's upstream as-is (no adapter translation), and the response
/// is returned as-is. Model governance (allowlist check) still applies
/// using the `model` field from the request body. Budget enforcement
/// and semantic caching do NOT apply (they are chat-specific).
#[allow(clippy::too_many_arguments)]
async fn serve_ai_passthrough(
    json_body: serde_json::Value,
    route_name: &str,
    gen: &Arc<Generation>,
    dp: &Arc<DataPlane>,
    rid: &str,
    rec: &mut crate::observability::AccessRecord,
    identity: Option<&crate::security::authn::Identity>,
    endpoint: crate::config::AiEndpoint,
) -> Response<ProxyBody> {
    // Extract the model alias from the request body for governance and
    // routing. The `model` field is present in all OpenAI endpoint
    // request shapes (embeddings, images, audio, moderation). When
    // absent, the request is forwarded without governance/routing (a
    // passthrough to the first provider's upstream).
    let model_alias = json_body
        .get("model")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");

    // DW-084: model governance pre-route check (same as chat). The
    // requested model alias is checked against the consumer's binding
    // team allowlists BEFORE routing.
    let gateway = gen.snapshot.gateway();
    let governance = dp.ai_governance();
    if !governance.is_empty() {
        let consumer = identity.map(|id| id.consumer_name.as_str());
        let route_cfg = gateway.routes.iter().find(|r| r.name == route_name);
        let route_policies: &[String] = route_cfg.map(|r| r.policies.as_slice()).unwrap_or(&[]);
        let service_policies: &[String] = route_cfg
            .map(|r| crate::ai::budget::service_policies_of(gateway, &r.service))
            .unwrap_or(&[]);
        let listener_policies = crate::ai::budget::listener_policies_of(gateway, &rec.listener);
        let consumer_policies: &[String] = consumer
            .map(|c| crate::ai::budget::consumer_policies_of(gateway, c))
            .unwrap_or(&[]);
        let verdict = governance.check(
            consumer,
            consumer_policies,
            route_policies,
            service_policies,
            listener_policies,
            &[],
            model_alias,
        );
        if let crate::ai::governance::GovernanceVerdict::Deny { reason, .. } = &verdict {
            dp.observability_arc().record_ai_governance_denied(reason);
            return ai_error_response(
                StatusCode::FORBIDDEN,
                &format!("model '{}' is denied by policy: {reason}", model_alias),
                "invalid_request_error",
                Some("model_denied_by_policy"),
                rid,
            );
        }
    }

    // Route the model alias to find the provider (same runtime as
    // chat). When the alias is empty or not found, return 404.
    let Some(runtime) = gen.ai() else {
        return ai_error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "no ai block configured",
            "api_error",
            Some("internal_error"),
            rid,
        );
    };
    let (candidates, _, _) = runtime
        .route_with_policy_and_canary_index(model_alias, rid, "")
        .await;
    if candidates.is_empty() {
        return ai_error_response(
            StatusCode::NOT_FOUND,
            &format!("the model '{}' does not exist", model_alias),
            "invalid_request_error",
            Some("model_not_found"),
            rid,
        );
    }

    // Forward to the first candidate (no failover for passthrough —
    // the response is returned as-is, so retry logic would need to
    // buffer the entire response, which defeats the passthrough
    // design).
    let target = &candidates[0];
    let version = target.version.as_deref().unwrap_or("default");
    let Some(provider) = runtime.provider(&target.provider) else {
        return ai_error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "no provider could serve the model",
            "api_error",
            Some("provider_unreachable"),
            rid,
        );
    };
    let Some(handle) = gen.registry().get(&provider.upstream) else {
        return ai_error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "no provider could serve the model",
            "api_error",
            Some("provider_unreachable"),
            rid,
        );
    };
    rec.upstream = Some(provider.upstream.clone());
    rec.attempts = 1;

    // Build the outbound request: the passthrough path + the original
    // body + the provider's auth headers.
    let path = resolve_path(provider, passthrough_path(endpoint));
    let mut outbound = http::Request::builder()
        .method(http::Method::POST)
        .uri(path);
    outbound = outbound.header("content-type", "application/json");
    let auth_pairs: Vec<(String, String)> = if let Some(pool) = &provider.credential_pool {
        match pool.pick(rid) {
            Some(entry) => vec![(entry.header.clone(), entry.value.clone())],
            None => provider.auth_headers.clone(),
        }
    } else {
        provider.auth_headers.clone()
    };
    for (name, value) in &auth_pairs {
        if let (Ok(n), Ok(v)) = (
            hyper::header::HeaderName::from_bytes(name.as_bytes()),
            hyper::header::HeaderValue::from_str(value),
        ) {
            outbound = outbound.header(n, v);
        }
    }
    let body_bytes = serde_json::to_vec(&json_body).unwrap_or_default();
    let outbound = match outbound.body(Full::new(Bytes::from(body_bytes))) {
        Ok(r) => r,
        Err(_) => {
            return ai_error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                "failed to build provider request",
                "api_error",
                Some("internal_error"),
                rid,
            );
        }
    };

    let obs = dp.observability_arc();
    let started = std::time::Instant::now();
    let upstream_resp = match handle.send(outbound).await {
        Ok(r) => r,
        Err(e) => {
            obs.record_ai_request(&provider.name, route_name, "transport_error", version);
            obs.record_ai_request_duration(
                &provider.name,
                route_name,
                started.elapsed().as_secs_f64(),
            );
            tracing::warn!(
                code = "ai_provider_unreachable",
                request_id = %rid,
                provider = %provider.name,
                upstream = %provider.upstream,
                "ai passthrough call failed: {e}"
            );
            return ai_error_response(
                StatusCode::BAD_GATEWAY,
                "no provider could serve the model",
                "api_error",
                Some("provider_unreachable"),
                rid,
            );
        }
    };

    let status = upstream_resp.status();
    let up_headers = upstream_resp.headers().clone();
    let up_body = upstream_resp.into_body();
    obs.record_ai_request(
        &provider.name,
        route_name,
        if status.is_success() {
            "success"
        } else {
            "error"
        },
        version,
    );
    obs.record_ai_request_duration(&provider.name, route_name, started.elapsed().as_secs_f64());

    if !status.is_success() {
        let err_bytes =
            match bounded_collect(up_body, MAX_AI_ERROR_BYTES, "provider error", rid).await {
                Ok(b) => b,
                Err(resp) => return resp,
            };
        let err_json: serde_json::Value = serde_json::from_slice(&err_bytes).unwrap_or(
            serde_json::json!({"error": {"message": String::from_utf8_lossy(&err_bytes).to_string()}}),
        );
        return response_with_json(status, &err_json, &up_headers);
    }

    // Success: return the response body as-is (no translation).
    let ok_bytes =
        match bounded_collect(up_body, MAX_AI_PROVIDER_RESPONSE_BYTES, "response", rid).await {
            Ok(b) => b,
            Err(resp) => return resp,
        };
    Response::builder()
        .status(status)
        .header("content-type", "application/json")
        .body(ProxyBody::Full(Full::new(ok_bytes)))
        .expect("passthrough response is valid")
}

/// AI-05 (#195): chat passthrough mode. Forward the chat request body
/// as-is to the provider (no adapter translation), but still apply
/// governance and routing. The model is extracted from the body for
/// routing. The response is returned as-is.
#[allow(clippy::too_many_arguments)]
async fn serve_ai_chat_passthrough(
    json_body: serde_json::Value,
    route_name: &str,
    gen: &Arc<Generation>,
    dp: &Arc<DataPlane>,
    rid: &str,
    rec: &mut crate::observability::AccessRecord,
    identity: Option<&crate::security::authn::Identity>,
) -> Response<ProxyBody> {
    // Extract the model alias from the request body for governance and
    // routing. The `model` field is present in all chat dialects.
    let model_alias = json_body
        .get("model")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");

    // DW-084: model governance pre-route check (same as chat and
    // non-chat passthrough).
    let gateway = gen.snapshot.gateway();
    let governance = dp.ai_governance();
    if !governance.is_empty() {
        let consumer = identity.map(|id| id.consumer_name.as_str());
        let route_cfg = gateway.routes.iter().find(|r| r.name == route_name);
        let route_policies: &[String] = route_cfg.map(|r| r.policies.as_slice()).unwrap_or(&[]);
        let service_policies: &[String] = route_cfg
            .map(|r| crate::ai::budget::service_policies_of(gateway, &r.service))
            .unwrap_or(&[]);
        let listener_policies = crate::ai::budget::listener_policies_of(gateway, &rec.listener);
        let consumer_policies: &[String] = consumer
            .map(|c| crate::ai::budget::consumer_policies_of(gateway, c))
            .unwrap_or(&[]);
        let verdict = governance.check(
            consumer,
            consumer_policies,
            route_policies,
            service_policies,
            listener_policies,
            &[],
            model_alias,
        );
        if let crate::ai::governance::GovernanceVerdict::Deny { reason, .. } = &verdict {
            dp.observability_arc().record_ai_governance_denied(reason);
            return ai_error_response(
                StatusCode::FORBIDDEN,
                &format!("model '{}' is denied by policy: {reason}", model_alias),
                "invalid_request_error",
                Some("model_denied_by_policy"),
                rid,
            );
        }
    }

    // Route the model alias to find the provider.
    let Some(runtime) = gen.ai() else {
        return ai_error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "no ai block configured",
            "api_error",
            Some("internal_error"),
            rid,
        );
    };
    let (candidates, _, _) = runtime
        .route_with_policy_and_canary_index(model_alias, rid, "")
        .await;
    if candidates.is_empty() {
        return ai_error_response(
            StatusCode::NOT_FOUND,
            &format!("the model '{}' does not exist", model_alias),
            "invalid_request_error",
            Some("model_not_found"),
            rid,
        );
    }

    // Forward to the first candidate (no failover for passthrough).
    let target = &candidates[0];
    let version = target.version.as_deref().unwrap_or("default");
    let Some(provider) = runtime.provider(&target.provider) else {
        return ai_error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "no provider could serve the model",
            "api_error",
            Some("provider_unreachable"),
            rid,
        );
    };
    let Some(handle) = gen.registry().get(&provider.upstream) else {
        return ai_error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "no provider could serve the model",
            "api_error",
            Some("provider_unreachable"),
            rid,
        );
    };
    rec.upstream = Some(provider.upstream.clone());
    rec.attempts = 1;

    // Build the outbound request: the chat path + the original body +
    // the provider's auth headers.
    let path = resolve_path(provider, passthrough_path(crate::config::AiEndpoint::Chat));
    let mut outbound = http::Request::builder()
        .method(http::Method::POST)
        .uri(path);
    outbound = outbound.header("content-type", "application/json");
    let auth_pairs: Vec<(String, String)> = if let Some(pool) = &provider.credential_pool {
        match pool.pick(rid) {
            Some(entry) => vec![(entry.header.clone(), entry.value.clone())],
            None => provider.auth_headers.clone(),
        }
    } else {
        provider.auth_headers.clone()
    };
    for (name, value) in &auth_pairs {
        if let (Ok(n), Ok(v)) = (
            hyper::header::HeaderName::from_bytes(name.as_bytes()),
            hyper::header::HeaderValue::from_str(value),
        ) {
            outbound = outbound.header(n, v);
        }
    }
    let body_bytes = serde_json::to_vec(&json_body).unwrap_or_default();
    let outbound = match outbound.body(Full::new(Bytes::from(body_bytes))) {
        Ok(r) => r,
        Err(_) => {
            return ai_error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                "failed to build provider request",
                "api_error",
                Some("internal_error"),
                rid,
            );
        }
    };

    let obs = dp.observability_arc();
    let started = std::time::Instant::now();
    let upstream_resp = match handle.send(outbound).await {
        Ok(r) => r,
        Err(e) => {
            obs.record_ai_request(&provider.name, route_name, "transport_error", version);
            obs.record_ai_request_duration(
                &provider.name,
                route_name,
                started.elapsed().as_secs_f64(),
            );
            tracing::warn!(
                code = "ai_provider_unreachable",
                request_id = %rid,
                provider = %provider.name,
                upstream = %provider.upstream,
                "ai chat passthrough call failed: {e}"
            );
            return ai_error_response(
                StatusCode::BAD_GATEWAY,
                "no provider could serve the model",
                "api_error",
                Some("provider_unreachable"),
                rid,
            );
        }
    };

    let status = upstream_resp.status();
    let up_headers = upstream_resp.headers().clone();
    let up_body = upstream_resp.into_body();
    obs.record_ai_request(
        &provider.name,
        route_name,
        if status.is_success() {
            "success"
        } else {
            "error"
        },
        version,
    );
    obs.record_ai_request_duration(&provider.name, route_name, started.elapsed().as_secs_f64());

    if !status.is_success() {
        let err_bytes =
            match bounded_collect(up_body, MAX_AI_PROVIDER_RESPONSE_BYTES, "error", rid).await {
                Ok(b) => b,
                Err(resp) => return resp,
            };
        let err_json: serde_json::Value = serde_json::from_slice(&err_bytes).unwrap_or(
            serde_json::json!({"error": {"message": String::from_utf8_lossy(&err_bytes).to_string()}}),
        );
        return response_with_json(status, &err_json, &up_headers);
    }

    // Success: return the response body as-is (no translation).
    let ok_bytes =
        match bounded_collect(up_body, MAX_AI_PROVIDER_RESPONSE_BYTES, "response", rid).await {
            Ok(b) => b,
            Err(resp) => return resp,
        };
    Response::builder()
        .status(status)
        .header("content-type", "application/json")
        .body(ProxyBody::Full(Full::new(ok_bytes)))
        .expect("passthrough response is valid")
}

/// Serve one `ai` route action (called from `dispatch_action` in the
/// proxy module). `rec` is the request's access record: the provider's
/// upstream name is attributed there so analytics and the access log
/// see which provider served the call.
///
/// AI-02: the `endpoint` parameter selects which AI endpoint the route
/// serves. `Chat` uses the full adapter translation pipeline (DW-075).
/// The other endpoints (`Embeddings`, `Images`, `Audio`, `Moderation`)
/// are proxied as a passthrough: the request body is forwarded to the
/// provider's upstream as-is, and the response is returned as-is.
/// Model governance (allowlist check) still applies using the `model`
/// field from the request body.
#[allow(clippy::too_many_arguments)]
pub(super) async fn serve_ai<B>(
    req: Request<B>,
    route_name: &str,
    gen: &Arc<Generation>,
    dp: &Arc<DataPlane>,
    rid: &str,
    rec: &mut crate::observability::AccessRecord,
    identity: Option<&crate::security::authn::Identity>,
    listener_name: &str,
    endpoint: crate::config::AiEndpoint,
    dialect: crate::config::AiIngressDialect,
) -> Response<ProxyBody>
where
    B: Body<Data = Bytes> + Send + 'static,
    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
{
    // AI token budget pre-check (DW-078): BEFORE the request body is
    // even read — a holder whose window is exhausted never reaches a
    // provider. The resolution mirrors the rate-limit chain (consumer
    // > route > service > listener > global, most-specific budget
    // governs); unbudgeted holders resolve no guard and skip.
    let budget = {
        let engine = dp.ai_budgets();
        if engine.is_empty() {
            None
        } else {
            let consumer = identity.map(|id| id.consumer_name.as_str());
            let gateway = gen.snapshot.gateway();
            let route_cfg = gateway.routes.iter().find(|r| r.name == route_name);
            let route_policies: &[String] = route_cfg.map(|r| r.policies.as_slice()).unwrap_or(&[]);
            let service_policies: &[String] = route_cfg
                .map(|r| crate::ai::budget::service_policies_of(gateway, &r.service))
                .unwrap_or(&[]);
            let consumer_policies: &[String] = consumer
                .map(|c| crate::ai::budget::consumer_policies_of(gateway, c))
                .unwrap_or(&[]);
            let listener_policies = crate::ai::budget::listener_policies_of(gateway, listener_name);
            engine.resolve(
                consumer,
                consumer_policies,
                route_policies,
                service_policies,
                listener_policies,
                &gateway.global_policies,
            )
        }
    };
    // DW-089: collect the deduplicated applicable policy names (the
    // same resolution order as the rate-limit check) so the AI proxy
    // path can feed provider outcomes back to the adaptive controller.
    let applicable_policies: Vec<String> = {
        let gateway = gen.snapshot.gateway();
        let consumer = identity.map(|id| id.consumer_name.as_str());
        let route_cfg = gateway.routes.iter().find(|r| r.name == route_name);
        let route_policies: &[String] = route_cfg.map(|r| r.policies.as_slice()).unwrap_or(&[]);
        let service_policies: &[String] = route_cfg
            .map(|r| crate::ai::budget::service_policies_of(gateway, &r.service))
            .unwrap_or(&[]);
        let consumer_policies: &[String] = consumer
            .map(|c| crate::ai::budget::consumer_policies_of(gateway, c))
            .unwrap_or(&[]);
        let listener_policies = crate::ai::budget::listener_policies_of(gateway, listener_name);
        let mut seen = std::collections::HashSet::new();
        let mut out = Vec::new();
        for name in consumer_policies
            .iter()
            .chain(route_policies)
            .chain(service_policies)
            .chain(listener_policies)
            .chain(&gateway.global_policies)
        {
            if seen.insert(name.clone()) {
                out.push(name.clone());
            }
        }
        out
    };
    if let Some(guard) = &budget {
        let now_s = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        if let crate::ai::budget::BudgetVerdict::Denied {
            kind,
            retry_after_s,
        } = guard.check(now_s)
        {
            rec.rate_limited = true;
            dp.observability_arc()
                .record_ai_budget_denied(kind.as_str());
            tracing::info!(
                code = "ai_budget_exceeded",
                request_id = %rid,
                route = %route_name,
                kind = kind.as_str(),
                retry_after_s,
                "AI token budget exhausted; rejecting before provider contact"
            );
            let body = openai_compat::error_body(
                "the token budget for this window is exhausted; retry later",
                "rate_limit_error",
                Some("ai_budget_exceeded"),
                rid,
            );
            return Response::builder()
                .status(StatusCode::TOO_MANY_REQUESTS)
                .header("retry-after", retry_after_s.to_string())
                .header(hyper::header::CONTENT_TYPE, "application/json")
                .body(ProxyBody::Full(Full::new(Bytes::from(body.to_string()))))
                .expect("static 429 response is valid");
        }
    }

    // The generation's compiled AI table; absent only when validation
    // was bypassed (defensive: every publish path validates).
    let Some(runtime) = gen.ai() else {
        return ai_error_response(
            // Validation rejects an ai route action without an ai
            // block, so reaching here is a config-invariant violation,
            // not a client error — 5xx, not 404.
            StatusCode::INTERNAL_SERVER_ERROR,
            "this gateway has no ai block configured",
            "api_error",
            Some("ai_not_configured"),
            rid,
        );
    };

    // 1. Read the request body, bounded incrementally.
    let (_parts, body) = req.into_parts();
    let bytes = match bounded_collect(body, MAX_AI_REQUEST_BYTES, "request", rid).await {
        Ok(b) => b,
        Err(resp) => return resp,
    };

    // 2. Parse the OpenAI-shaped body into the canonical request.
    let json_body: serde_json::Value = match serde_json::from_slice(&bytes) {
        Ok(v) => v,
        Err(e) => {
            return ai_error_response(
                StatusCode::BAD_REQUEST,
                &format!("request body is not valid JSON: {e}"),
                "invalid_request_error",
                Some("invalid_json"),
                rid,
            )
        }
    };

    // AI-02: non-chat endpoints are proxied as a passthrough. The
    // request body is forwarded to the provider's upstream as-is (no
    // adapter translation), and the response is returned as-is. Model
    // governance (allowlist check) still applies using the `model`
    // field from the request body. Budget enforcement and semantic
    // caching do NOT apply (they are chat-specific). This branch runs
    // BEFORE chat-request parsing so non-chat bodies (which lack
    // `messages`) do not trigger a 400.
    if endpoint != crate::config::AiEndpoint::Chat {
        return serve_ai_passthrough(json_body, route_name, gen, dp, rid, rec, identity, endpoint)
            .await;
    }

    // AI-05 (#195): multi-dialect ingress. Parse the request body
    // using the route's configured dialect. The default is OpenAI
    // (the historical behavior). Anthropic and Gemini dialects parse
    // the native request format into the canonical ChatRequest.
    // Passthrough mode forwards the body as-is (no translation).
    let mut chat_req: ChatRequest = match dialect {
        crate::config::AiIngressDialect::Openai => {
            match openai_compat::parse_chat_request(&json_body) {
                Ok(r) => r,
                Err(e) => {
                    return ai_error_response(
                        StatusCode::BAD_REQUEST,
                        &e.to_string(),
                        openai_compat::error_type_of(&e),
                        None,
                        rid,
                    );
                }
            }
        }
        crate::config::AiIngressDialect::Anthropic => {
            match crate::ai::ingress::parse_anthropic_request(&json_body) {
                Ok(r) => r,
                Err(e) => {
                    return ai_error_response(
                        StatusCode::BAD_REQUEST,
                        &e,
                        "invalid_request_error",
                        None,
                        rid,
                    );
                }
            }
        }
        crate::config::AiIngressDialect::Gemini => {
            match crate::ai::ingress::parse_gemini_request(&json_body) {
                Ok(r) => r,
                Err(e) => {
                    return ai_error_response(
                        StatusCode::BAD_REQUEST,
                        &e,
                        "invalid_request_error",
                        None,
                        rid,
                    );
                }
            }
        }
        crate::config::AiIngressDialect::Passthrough => {
            // AI-05 (#195): chat passthrough mode. Forward the request
            // body as-is to the provider (no adapter translation), but
            // still apply governance, metering, and budget enforcement.
            // The model is extracted from the body for routing.
            return serve_ai_chat_passthrough(json_body, route_name, gen, dp, rid, rec, identity)
                .await;
        }
    };

    // AI-07: the OTel GenAI semantic-convention span. Every gen_ai.*
    // field is declared Empty so `Span::record` can fill it as the
    // values resolve (request-side here, response-side after the
    // provider answers). The span is instrumented around the provider
    // call so it appears in the trace tree as a child of the request
    // span; response-side fields are recorded on the handle (recording
    // does not require the span to be current).
    let gen_ai_span = tracing::info_span!(
        "gen_ai.chat",
        "gen_ai.system" = tracing::field::Empty,
        "gen_ai.request.model" = tracing::field::Empty,
        "gen_ai.response.model" = tracing::field::Empty,
        "gen_ai.request.max_tokens" = tracing::field::Empty,
        "gen_ai.request.temperature" = tracing::field::Empty,
        "gen_ai.request.top_p" = tracing::field::Empty,
        "gen_ai.usage.prompt_tokens" = tracing::field::Empty,
        "gen_ai.usage.completion_tokens" = tracing::field::Empty,
        "gen_ai.usage.total_tokens" = tracing::field::Empty,
        "gen_ai.response.finish_reasons" = tracing::field::Empty,
        "gen_ai.response.id" = tracing::field::Empty,
    );
    // Request-side attributes are known once the canonical request is
    // parsed. `gen_ai.system` is recorded when the serving provider is
    // selected (it depends on the candidate kind).
    crate::observability::record_gen_ai_request_attrs(
        &gen_ai_span,
        "",
        &chat_req.model,
        chat_req.max_tokens,
        chat_req.temperature,
        chat_req.top_p,
    );
    // AI-07: end-to-end AI request latency, measured from here (request
    // parsed) through the terminal outcome. Recorded alongside each
    // `record_ai_request` outcome below.
    let ai_started = std::time::Instant::now();

    // PERF-01: local token estimation pre-check. The first budget
    // check (above, before the body was read) only rejects when the
    // window is ALREADY exhausted. This second check estimates the
    // request's total token cost (prompt estimate + requested
    // max_tokens) and rejects when `spent + estimated > limit` BEFORE
    // any provider contact — so a holder near the limit cannot overrun
    // by a large prompt. The estimate is conservative (overcounts);
    // the actual spend is still provider-reported after the call.
    if let Some(guard) = &budget {
        let prompt_est = crate::ai::token_estimator::estimate_prompt_tokens(&chat_req);
        let max_out = chat_req.max_tokens.unwrap_or(0);
        let estimated_total = prompt_est.saturating_add(max_out);
        if estimated_total > 0 {
            let now_s = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0);
            if let crate::ai::budget::BudgetVerdict::Denied {
                kind,
                retry_after_s,
            } = guard.check_with_estimate(now_s, estimated_total)
            {
                rec.rate_limited = true;
                dp.observability_arc()
                    .record_ai_budget_denied(kind.as_str());
                tracing::info!(
                    code = "ai_budget_exceeded",
                    request_id = %rid,
                    route = %route_name,
                    kind = kind.as_str(),
                    retry_after_s,
                    estimated_tokens = estimated_total,
                    "AI token budget would be exceeded by estimated \
                     tokens; rejecting before provider contact (PERF-01)"
                );
                let body = openai_compat::error_body(
                    "the token budget for this window would be exceeded \
                     by the estimated request size; retry later",
                    "rate_limit_error",
                    Some("ai_budget_exceeded"),
                    rid,
                );
                return Response::builder()
                    .status(StatusCode::TOO_MANY_REQUESTS)
                    .header("retry-after", retry_after_s.to_string())
                    .header(hyper::header::CONTENT_TYPE, "application/json")
                    .body(ProxyBody::Full(Full::new(Bytes::from(body.to_string()))))
                    .expect("static 429 response is valid");
            }
        }
    }

    // DW-084: model governance pre-route check. The requested model
    // alias is checked against the consumer's binding team allowlists
    // BEFORE routing — a disallowed (or typo'd) alias is blocked at
    // the edge with 403 `model_denied_by_policy` rather than
    // surfacing as a provider 404. The resolution mirrors the
    // rate-limit/budget chain (consumer > route > service > listener
    // > global, deny-wins intersection); a consumer with no binding
    // allowlist policy is allowed (fail-open).
    let governance = dp.ai_governance();
    let governance_verdict = if governance.is_empty() {
        crate::ai::governance::GovernanceVerdict::Allow
    } else {
        let consumer = identity.map(|id| id.consumer_name.as_str());
        let gateway = gen.snapshot.gateway();
        let route_cfg = gateway.routes.iter().find(|r| r.name == route_name);
        let route_policies: &[String] = route_cfg.map(|r| r.policies.as_slice()).unwrap_or(&[]);
        let service_policies: &[String] = route_cfg
            .map(|r| crate::ai::budget::service_policies_of(gateway, &r.service))
            .unwrap_or(&[]);
        let consumer_policies: &[String] = consumer
            .map(|c| crate::ai::budget::consumer_policies_of(gateway, c))
            .unwrap_or(&[]);
        let listener_policies = crate::ai::budget::listener_policies_of(gateway, listener_name);
        governance.check(
            consumer,
            consumer_policies,
            route_policies,
            service_policies,
            listener_policies,
            &gateway.global_policies,
            &chat_req.model,
        )
    };
    let governance_consumer = identity
        .map(|id| id.consumer_name.clone())
        .unwrap_or_else(|| "anonymous".to_string());
    let governance_team = match &governance_verdict {
        crate::ai::governance::GovernanceVerdict::Deny { policy, .. } => policy.clone(),
        crate::ai::governance::GovernanceVerdict::Allow => String::new(),
    };
    if let crate::ai::governance::GovernanceVerdict::Deny { reason, .. } = &governance_verdict {
        dp.observability_arc().record_ai_governance_denied(reason);
        tracing::info!(
            code = "model_denied_by_policy",
            request_id = %rid,
            route = %route_name,
            consumer = %governance_consumer,
            model = %chat_req.model,
            reason = %reason,
            "AI model governance denied the requested model before routing"
        );
        // DW-084: a blocked attempt is ALWAYS audited (the done-when
        // requirement) when the governance block is present — the
        // `audit` flag only extends recording to ALLOWED calls.
        if let Some(analytics) = dp.analytics() {
            analytics.offer_ai_governance_event(crate::analytics::AiGovernanceEvent {
                ts_ms: now_ms(),
                consumer: governance_consumer.clone(),
                team: governance_team.clone(),
                model: chat_req.model.clone(),
                verdict: "deny".to_string(),
                reason: reason.clone(),
            });
        }
        let body = openai_compat::error_body(
            &format!(
                "the model '{}' is not allowed for this consumer by the \
                 team allowlist policy",
                chat_req.model
            ),
            "invalid_request_error",
            Some("model_denied_by_policy"),
            rid,
        );
        return Response::builder()
            .status(StatusCode::FORBIDDEN)
            .header(hyper::header::CONTENT_TYPE, "application/json")
            .body(ProxyBody::Full(Full::new(Bytes::from(body.to_string()))))
            .expect("static 403 response is valid");
    }
    // DW-084: when the audit switch is on, record ALLOWED calls too
    // (shadow review — which team called which model).
    if governance.audit() {
        if let Some(analytics) = dp.analytics() {
            analytics.offer_ai_governance_event(crate::analytics::AiGovernanceEvent {
                ts_ms: now_ms(),
                consumer: governance_consumer.clone(),
                team: governance_team.clone(),
                model: chat_req.model.clone(),
                verdict: "allow".to_string(),
                reason: String::new(),
            });
        }
    }

    // DW-082: prompt guardrails. AFTER the chat request is parsed and
    // AFTER the governance check, BEFORE model resolution and the
    // provider call. A `block` action returns a 400
    // `guardrail_blocked`; a `redact` action scrubs the prompt and
    // continues; a `log` action records and continues (dry-run). The
    // policy chain is the same one governance uses (consumer > route
    // > service > listener > global).
    let guardrails = dp.ai_guardrails();
    if !guardrails.is_empty() {
        let consumer = identity.map(|id| id.consumer_name.as_str());
        let gateway = gen.snapshot.gateway();
        let route_cfg = gateway.routes.iter().find(|r| r.name == route_name);
        let route_policies: &[String] = route_cfg.map(|r| r.policies.as_slice()).unwrap_or(&[]);
        let service_policies: &[String] = route_cfg
            .map(|r| crate::ai::budget::service_policies_of(gateway, &r.service))
            .unwrap_or(&[]);
        let consumer_policies: &[String] = consumer
            .map(|c| crate::ai::budget::consumer_policies_of(gateway, c))
            .unwrap_or(&[]);
        let listener_policies = crate::ai::budget::listener_policies_of(gateway, listener_name);
        let guardrail_result = guardrails.check_prompt(
            consumer_policies,
            route_policies,
            service_policies,
            listener_policies,
            &gateway.global_policies,
            &chat_req,
        );
        match &guardrail_result {
            crate::ai::guardrails::GuardrailResult::Block { rule_name, reason } => {
                dp.observability_arc()
                    .record_ai_guardrail_blocked(rule_name, "prompt");
                tracing::info!(
                    code = "guardrail_blocked",
                    request_id = %rid,
                    route = %route_name,
                    rule = %rule_name,
                    reason = %reason,
                    "AI guardrail blocked the prompt before the provider call"
                );
                let body = openai_compat::error_body(
                    &format!("the prompt was blocked by guardrail rule '{rule_name}': {reason}"),
                    "invalid_request_error",
                    Some("guardrail_blocked"),
                    rid,
                );
                return Response::builder()
                    .status(StatusCode::BAD_REQUEST)
                    .header(hyper::header::CONTENT_TYPE, "application/json")
                    .body(ProxyBody::Full(Full::new(Bytes::from(body.to_string()))))
                    .expect("static 400 response is valid");
            }
            crate::ai::guardrails::GuardrailResult::Redact {
                rule_name,
                redacted_prompt,
            } => {
                tracing::info!(
                    code = "guardrail_redacted",
                    request_id = %rid,
                    route = %route_name,
                    rule = %rule_name,
                    "AI guardrail redacted the prompt before the provider call"
                );
                chat_req = guardrails.apply_redaction(&chat_req, redacted_prompt);
            }
            crate::ai::guardrails::GuardrailResult::Allow => {}
        }
    }

    // AI-13 (#201): fetch remote image URLs for multimodal requests.
    // When the request contains remote image URLs (OpenAI
    // `image_url.url` convention) and the target adapter requires
    // base64 data (Anthropic, Gemini), the gateway fetches the image,
    // converts to base64, and fills `data_b64` + `media_type`. The
    // OpenAI adapter continues to pass the URL through unchanged.
    // The fetcher is skipped entirely when there are no remote images.
    if crate::ai::image_fetch::has_remote_images(&chat_req) {
        let cfg = crate::ai::image_fetch::ImageFetchConfig::default();
        chat_req = crate::ai::image_fetch::fetch_remote_images(chat_req, &cfg).await;
    }

    // 3. Streaming (DW-077): the request is passed through with
    // usage reporting FORCED on the provider call — the stream
    // metrics and the (upcoming) token budgets need provider-reported
    // counts even when the client did not ask for the usage chunk,
    // and the terminal usage chunk we emit is the documented
    // include_usage shape either way.
    let stream_requested = chat_req.stream;
    if stream_requested {
        chat_req.stream_options_include_usage = true;
    }

    // DW-083: semantic cache lookup. AFTER guardrails (the prompt may
    // have been redacted) and AFTER the stream flag is known, BEFORE
    // model routing + the provider call. A cache hit returns the
    // cached response with no provider call and no token spend.
    // PERF-02: streaming requests are now cached too (the stream is
    // tee'd on a miss and replayed on a hit). The lookup is async (it
    // may call the embedding service) but runs in the already-async
    // serve_ai; a miss or error fails open.
    let sem_cache = dp.ai_semantic_cache();
    let can_cache = sem_cache.is_some();
    if can_cache {
        if let Some(sem_cache) = &sem_cache {
            let prompt_text = chat_req
                .messages
                .iter()
                .map(|m| m.text_content())
                .collect::<Vec<_>>()
                .join("\n");
            if let Some(cached) = sem_cache.lookup(&prompt_text, &chat_req.model).await {
                dp.observability_arc()
                    .record_ai_semantic_cache_hit(&chat_req.model);
                tracing::info!(
                    code = "semantic_cache_hit",
                    request_id = %rid,
                    route = %route_name,
                    model = %chat_req.model,
                    "AI semantic cache hit; returning cached response with no provider call"
                );
                return match cached {
                    crate::ai::semantic_cache::CachedResponse::Json(v) => {
                        response_with_json(StatusCode::OK, &v, &HeaderMap::new())
                    }
                    crate::ai::semantic_cache::CachedResponse::Frames(frames) => {
                        // PERF-02: replay the cached SSE frames as a
                        // single concatenated text/event-stream body.
                        let body = frames.join("");
                        Response::builder()
                            .status(StatusCode::OK)
                            .header(hyper::header::CONTENT_TYPE, "text/event-stream")
                            .header(hyper::header::CACHE_CONTROL, "no-cache")
                            .body(ProxyBody::Full(Full::new(Bytes::from(body))))
                            .expect("cached SSE replay response is valid")
                    }
                };
            }
            dp.observability_arc()
                .record_ai_semantic_cache_miss(&chat_req.model);
        }
    }

    // 4. Route (DW-076 / DW-085): the ordered candidate list for this
    // alias — [primary, failover...] for a chained alias, the
    // deterministic canary pick for a split alias, or the policy
    // evaluation result for a policy alias (DW-085: a FallbackChain
    // policy may call an external classifier service async). Empty
    // means the alias does not exist.
    let prompt_text = chat_req
        .messages
        .iter()
        .map(|m| m.text_content())
        .collect::<Vec<_>>()
        .join("\n");
    let (candidates, policy_decision, canary_version_index) = runtime
        .route_with_policy_and_canary_index(&chat_req.model, rid, &prompt_text)
        .await;
    // DW-085: record the routing-policy decision as a metric. The
    // `ai` domain cannot import `observability` (the dependency
    // direction forbids it), so the decision travels back here and
    // the dataplane records it.
    if let Some(decision) = policy_decision {
        if let Some(crate::ai::CompiledModel::Policy(policy)) = runtime.model(&chat_req.model) {
            let name = policy.name();
            match decision {
                crate::ai::policy::PolicyDecision::Escalate => {
                    dp.observability_arc()
                        .record_ai_routing_policy_escalation(name);
                }
                crate::ai::policy::PolicyDecision::Cheap
                | crate::ai::policy::PolicyDecision::ClassifierError => {
                    dp.observability_arc().record_ai_routing_policy_cheap(name);
                }
                crate::ai::policy::PolicyDecision::LatencyCost => {
                    dp.observability_arc()
                        .record_ai_routing_policy_latency_cost_selection(name);
                }
            }
        }
    }
    // DW-086: experiment alias resolution. When the model is an
    // experiment (A/B test) alias, route_with_policy already picked
    // the variant and returned its target in `candidates`. Here we
    // inject the variant's system message into the request (if any)
    // and record the assignment to analytics for attribution.
    // AI-14 (#202): server-side prompt template resolution. When the
    // client references a prompt template by name, resolve it from
    // the experiments config, substitute variables, and prepend the
    // system message BEFORE any existing system message.
    if let Some(prompt_ref) = &chat_req.prompt {
        let experiments = runtime.experiments_config();
        let overrides = runtime.prompt_overrides();
        let system = if let Some((name, version)) = prompt_ref.split_once('/') {
            // Explicit version reference.
            experiments
                .and_then(|exp| exp.prompts.get(name))
                .and_then(|p| p.versions.get(version))
                .map(|v| v.system.clone())
        } else {
            // Active version (with runtime override).
            crate::ai::experiments::active_prompt_system(experiments, overrides, prompt_ref)
        };
        if let Some(mut system) = system {
            // Substitute {{var}} placeholders with prompt_variables.
            for (key, value) in &chat_req.prompt_variables {
                let placeholder = format!("{{{{{key}}}}}");
                system = system.replace(&placeholder, value);
            }
            chat_req.messages.insert(
                0,
                crate::ai::types::ChatMessage::text(crate::ai::types::ChatRole::System, system),
            );
        }
    }
    if let Some(crate::ai::CompiledModel::Experiment(test)) = runtime.model(&chat_req.model) {
        let variant = test.pick(rid);
        // Record the variant selection as a metric.
        dp.observability_arc()
            .record_ai_experiment_variant_selection(&test.name, &variant.name);
        // Inject the variant's prompt-version system message BEFORE
        // any existing system message (the variant's system message
        // is the experiment's prompt; the client's system message, if
        // any, follows it).
        if let Some(sys) = &variant.system_message {
            chat_req.messages.insert(
                0,
                crate::ai::types::ChatMessage::text(
                    crate::ai::types::ChatRole::System,
                    sys.clone(),
                ),
            );
        }
        // Record the assignment to analytics (fire-and-forget).
        if let Some(analytics) = dp.analytics() {
            analytics.offer_ai_experiment_assignment(crate::analytics::AiExperimentAssignment {
                ts_ms: std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_millis() as i64)
                    .unwrap_or(0),
                request_id: rid.to_string(),
                experiment: test.name.clone(),
                variant: variant.name.clone(),
                model: variant.target.provider_model.clone(),
                consumer: identity
                    .map(|id| id.consumer_name.clone())
                    .unwrap_or_else(|| "anonymous".to_string()),
            });
        }
    }

    if candidates.is_empty() {
        return ai_error_response(
            StatusCode::NOT_FOUND,
            &format!("the model '{}' does not exist", chat_req.model),
            "invalid_request_error",
            Some("model_not_found"),
            rid,
        );
    }

    // 5. Walk the candidates (DW-076 failover). Retryable outcomes —
    // a 429, a 5xx, a transport error, or a per-dialect translation
    // rejection — advance to the NEXT candidate; anything else is
    // final. Nothing reaches the client until a candidate SUCCEEDS
    // (the provider response is read whole before translation), so
    // failover is invisible to the client. Same-provider retries are
    // not the chain's job: the provider's upstream breaker owns those.
    let obs = dp.observability_arc();
    let mut attempts: u32 = 0;
    let mut last_transport_error = true;
    let mut fallback_response: Response<ProxyBody> = ai_error_response(
        StatusCode::BAD_GATEWAY,
        "no provider could serve the model",
        "api_error",
        Some("provider_unreachable"),
        rid,
    );
    for target in candidates {
        attempts += 1;
        let version = target.version.as_deref().unwrap_or("default");
        let Some(provider) = runtime.provider(&target.provider) else {
            tracing::error!(
                code = "ai_provider_missing",
                request_id = %rid,
                provider = %target.provider,
                "routed provider is not in the compiled table (validate-vs-build race)"
            );
            obs.record_ai_request(&target.provider, route_name, "transport_error", version);
            obs.record_ai_request_duration(
                &target.provider,
                route_name,
                ai_started.elapsed().as_secs_f64(),
            );
            continue;
        };
        let adapter = adapter_for(provider.kind);
        // AI-07: record the serving provider's kind as gen_ai.system
        // (re-recorded per candidate so the span reflects the provider
        // that actually served; on failover the last candidate wins).
        gen_ai_span.record("gen_ai.system", provider.kind.gen_ai_system());

        // Translate for THIS provider (the provider model differs per
        // candidate). A per-dialect rejection may not exist in the
        // next dialect, so it advances rather than failing the request.
        let provider_req = match adapter.build_request(&chat_req, &target.provider_model) {
            Ok(r) => r,
            Err(e) => {
                obs.record_ai_request(&provider.name, route_name, "translation_error", version);
                obs.record_ai_request_duration(
                    &provider.name,
                    route_name,
                    ai_started.elapsed().as_secs_f64(),
                );
                tracing::info!(
                    code = "ai_provider_translation_rejected",
                    request_id = %rid,
                    provider = %provider.name,
                    "candidate rejected the conversation: {e}"
                );
                fallback_response = ai_error_response(
                    StatusCode::BAD_REQUEST,
                    &e.to_string(),
                    openai_compat::error_type_of(&e),
                    None,
                    rid,
                );
                last_transport_error = false;
                continue;
            }
        };

        let Some(handle) = gen.registry().get(&provider.upstream) else {
            tracing::error!(
                code = "ai_provider_upstream_missing",
                request_id = %rid,
                provider = %provider.name,
                upstream = %provider.upstream,
                "ai provider's upstream is not in the registry (validate-vs-build race)"
            );
            obs.record_ai_request(&provider.name, route_name, "transport_error", version);
            obs.record_ai_request_duration(
                &provider.name,
                route_name,
                ai_started.elapsed().as_secs_f64(),
            );
            continue;
        };
        let outbound_path = resolve_path(provider, &provider_req.path);
        let mut outbound = Request::builder()
            .method(Method::POST)
            .uri(outbound_path)
            .header(hyper::header::CONTENT_TYPE, "application/json")
            .header(hyper::header::ACCEPT, "application/json");
        for (name, value) in &provider_req.headers {
            if let Ok(v) = hyper::header::HeaderValue::from_str(value) {
                outbound = outbound.header(name.clone(), v);
            }
        }
        // Provider auth from the compiled table (resolved at compile
        // time; adapters never see credentials). Applied last so
        // nothing overrides it; unrepresentable values skip loudly.
        // DW-080: when the provider has a credential pool, pick one
        // credential per request and track its index for quarantine
        // on 429. When the pool is exhausted (all keys quarantined),
        // fail this attempt with a 429 + Retry-After.
        let mut pool_entry_index: Option<usize> = None;
        let auth_pairs: Vec<(String, String)> = if let Some(pool) = &provider.credential_pool {
            match pool.pick(rid) {
                Some(entry) => {
                    pool_entry_index = Some(entry.index);
                    vec![(entry.header.clone(), entry.value.clone())]
                }
                None => {
                    // Pool exhaustion: all keys quarantined.
                    let retry_after = pool
                        .earliest_quarantine_expiry()
                        .map(|t| {
                            t.duration_since(std::time::Instant::now())
                                .as_secs()
                                .clamp(1, 600)
                        })
                        .unwrap_or(60);
                    tracing::warn!(
                        code = "ai_credential_pool_exhausted",
                        request_id = %rid,
                        provider = %provider.name,
                        retry_after_secs = retry_after,
                        "ai credential pool exhausted (all keys quarantined)"
                    );
                    obs.record_ai_request(&provider.name, route_name, "pool_exhausted", version);
                    obs.record_ai_request_duration(
                        &provider.name,
                        route_name,
                        ai_started.elapsed().as_secs_f64(),
                    );
                    let body = openai_compat::error_body(
                        "all provider API keys are rate-limited; retry later",
                        "pool_exhausted",
                        Some("429"),
                        rid,
                    );
                    let mut headers = HeaderMap::new();
                    if let Ok(v) = hyper::header::HeaderValue::from_str(&format!("{retry_after}")) {
                        headers.insert("retry-after", v);
                    }
                    fallback_response =
                        response_with_json(hyper::StatusCode::TOO_MANY_REQUESTS, &body, &headers);
                    continue;
                }
            }
        } else {
            provider.auth_headers.clone()
        };
        for (name, value) in &auth_pairs {
            match (
                hyper::header::HeaderName::from_bytes(name.as_bytes()),
                hyper::header::HeaderValue::from_str(value),
            ) {
                (Ok(n), Ok(v)) => {
                    outbound = outbound.header(n, v);
                }
                _ => {
                    tracing::error!(
                        code = "ai_provider_auth_header_invalid",
                        provider = %provider.name,
                        "provider auth header is not representable; call is sent unauthenticated"
                    );
                }
            }
        }
        let body_bytes = serde_json::to_vec(&provider_req.body).unwrap_or_default();
        let outbound = match outbound.body(Full::new(Bytes::from(body_bytes))) {
            Ok(r) => r,
            Err(_) => {
                obs.record_ai_request(&provider.name, route_name, "transport_error", version);
                obs.record_ai_request_duration(
                    &provider.name,
                    route_name,
                    ai_started.elapsed().as_secs_f64(),
                );
                continue;
            }
        };
        // Attribution follows the SERVING (or last attempted)
        // provider across the failover walk (DW-076).
        rec.upstream = Some(provider.upstream.clone());
        rec.attempts = attempts;
        let attempt_started = std::time::Instant::now();
        // AI-07: instrument the upstream send with the gen_ai span so
        // the provider call appears under the GenAI span in the trace
        // tree (the span's duration covers the provider round-trip).
        let upstream_resp = match handle.send(outbound).instrument(gen_ai_span.clone()).await {
            Ok(r) => r,
            Err(e) => {
                obs.record_ai_request(&provider.name, route_name, "transport_error", version);
                obs.record_ai_request_duration(
                    &provider.name,
                    route_name,
                    ai_started.elapsed().as_secs_f64(),
                );
                tracing::warn!(
                    code = "ai_provider_unreachable",
                    request_id = %rid,
                    provider = %provider.name,
                    upstream = %provider.upstream,
                    "ai provider call failed: {e}"
                );
                last_transport_error = true;
                continue;
            }
        };

        let status = upstream_resp.status();
        let (up_parts, up_body) = upstream_resp.into_parts();
        let retryable = status.as_u16() == 429 || status.is_server_error();
        // DW-089: feed the provider outcome to the adaptive controller
        // for every applicable policy. The latency is measured from the
        // attempt start to header resolution; the Retry-After header is
        // parsed from the provider response for the origin-signal
        // backoff. The controller is a no-op for policies without an
        // adaptive block.
        if !applicable_policies.is_empty() {
            let latency = attempt_started.elapsed();
            let retry_after = parse_ai_retry_after(&up_parts.headers);
            let adaptive = Arc::clone(gen.adaptive());
            let status_u16 = status.as_u16();
            for name in &applicable_policies {
                adaptive.record_outcome(name, status_u16, latency, retry_after);
            }
        }
        // DW-091: feed the provider outcome to the auto-canary
        // controller when this request was served by a canary-tracked
        // AI model alias. The canary side is index 1; the baseline is
        // index 0. The latency is measured from the attempt start to
        // header resolution (the same point the adaptive controller
        // observes). The controller is a no-op when the alias has no
        // canary_analysis (canary_version_index is None).
        if let Some(idx) = canary_version_index {
            if let Some(canary) = gen.canary() {
                let latency_ms = attempt_started.elapsed().as_secs_f64() * 1000.0;
                canary.record_outcome(&chat_req.model, idx == 1, status.as_u16(), latency_ms);
            }
        }
        if !status.is_success() {
            let err_bytes =
                match bounded_collect(up_body, MAX_AI_ERROR_BYTES, "provider error", rid).await {
                    Ok(b) => b,
                    Err(resp) => return resp,
                };
            let value: serde_json::Value = serde_json::from_slice(&err_bytes).unwrap_or_default();
            let parsed = adapter.parse_error(&value);
            obs.record_ai_request(&provider.name, route_name, "provider_error", version);
            obs.record_ai_request_duration(
                &provider.name,
                route_name,
                ai_started.elapsed().as_secs_f64(),
            );
            let body = openai_compat::error_body(
                &parsed.message,
                parsed.error_type.as_deref().unwrap_or("api_error"),
                parsed.code.as_deref(),
                rid,
            );
            let resp = response_with_json(status, &body, &up_parts.headers);
            if retryable {
                // DW-080: quarantine the credential pool entry on 429.
                // The pool skips this key for the quarantine window;
                // subsequent requests rotate to the next key. The
                // provider's Retry-After header (if present) overrides
                // the configured quarantine window, capped at 600s.
                if status.as_u16() == 429 {
                    if let (Some(pool), Some(idx)) = (&provider.credential_pool, pool_entry_index) {
                        let retry_after = parse_ai_retry_after(&up_parts.headers);
                        pool.quarantine(idx, retry_after.map(|d| d.as_secs()));
                        tracing::info!(
                            code = "ai_credential_pool_quarantined",
                            request_id = %rid,
                            provider = %provider.name,
                            key_index = idx,
                            "credential pool key quarantined after 429"
                        );
                    }
                }
                // AI-09 (#198): record per-key health from rate-limit
                // headers on every provider response (error or success).
                // When remaining is 0, the pool proactively cools the
                // key until the reset window, preventing the next 429.
                if let (Some(pool), Some(idx)) = (&provider.credential_pool, pool_entry_index) {
                    let health = parse_rate_limit_headers(&up_parts.headers, status.as_u16());
                    pool.record_health(idx, health);
                }
                // 429/5xx is transient — try the next candidate; if
                // none succeed, the LAST provider's answer is the one
                // the client sees (closest to the truth of the outage).
                tracing::info!(
                    code = "ai_provider_failover",
                    request_id = %rid,
                    provider = %provider.name,
                    status = status.as_u16(),
                    "transient provider error; failing over to the next candidate"
                );
                fallback_response = resp;
                last_transport_error = false;
                continue;
            }
            // Other 4xx (bad request, bad key) is deterministic —
            // retrying another provider would only re-diagnose it.
            return resp;
        }
        // AI-09 (#198): record per-key health from rate-limit headers
        // on successful (non-retryable) provider responses too. This
        // captures the remaining quota before a 429 occurs, enabling
        // proactive cooldown.
        if let (Some(pool), Some(idx)) = (&provider.credential_pool, pool_entry_index) {
            let health = parse_rate_limit_headers(&up_parts.headers, status.as_u16());
            pool.record_health(idx, health);
        }
        // Streaming pass-through (DW-077): a 200 SSE response streams
        // to the client frame-by-frame with zero added buffering. This
        // is the failover COMMIT point — returning the streaming
        // response forwards headers (and then chunks) to the client,
        // so no later candidate can replace anything from here on.
        if stream_requested && is_event_stream(&up_parts.headers) {
            // DW-AI-03: unknown-model pricing policy for streaming.
            // The check runs BEFORE the stream commits (the failover
            // commit point is the streaming response return below) so
            // `fail_closed` can reject with 403 without first
            // forwarding frames to the client. `alert` logs + meters
            // here and proceeds; `allow` is silent. The provider call
            // already happened (the 200 SSE body is open), but
            // dropping it cancels the upstream request.
            let pricing = dp.ai_pricing();
            if !pricing.has_pricing(&target.provider_model) {
                let policy = gen
                    .snapshot
                    .gateway()
                    .ai
                    .as_ref()
                    .and_then(|c| c.unknown_model_policy)
                    .unwrap_or_default();
                match policy {
                    UnknownModelPolicy::FailClosed => {
                        obs.record_ai_request(
                            &provider.name,
                            route_name,
                            "unknown_model_pricing",
                            version,
                        );
                        obs.record_ai_request_duration(
                            &provider.name,
                            route_name,
                            ai_started.elapsed().as_secs_f64(),
                        );
                        tracing::warn!(
                            code = "ai_unknown_model_pricing",
                            request_id = %rid,
                            route = %route_name,
                            provider = %provider.name,
                            model = %target.provider_model,
                            "rejecting stream: model has no pricing entry (fail_closed policy)"
                        );
                        return ai_error_response(
                            StatusCode::FORBIDDEN,
                            "the model has no pricing entry and the unknown-model policy is fail_closed",
                            "invalid_request_error",
                            Some("ai_unknown_model_pricing"),
                            rid,
                        );
                    }
                    UnknownModelPolicy::Alert => {
                        obs.record_ai_unknown_pricing(&target.provider_model);
                        tracing::warn!(
                            code = "ai_unknown_model_pricing",
                            request_id = %rid,
                            route = %route_name,
                            provider = %provider.name,
                            model = %target.provider_model,
                            "model has no pricing entry; streaming with cost 0 (alert policy)"
                        );
                    }
                    UnknownModelPolicy::Allow => {}
                }
            }
            obs.record_ai_request(&provider.name, route_name, "success", version);
            // AI-07: end-to-end AI request latency (streaming — measured
            // to stream establishment, the failover commit point; the
            // stream's own duration is dwara_ai_stream_duration_seconds).
            obs.record_ai_request_duration(
                &provider.name,
                route_name,
                ai_started.elapsed().as_secs_f64(),
            );
            tracing::info!(
                code = "ai_stream_started",
                request_id = %rid,
                route = %route_name,
                provider = %provider.name,
                model = %chat_req.model,
                version = %version,
                attempts = attempts,
                upstream = %provider.upstream,
                "ai stream established; forwarding begins"
            );
            // DW-079: compute the consumer and team strings BEFORE
            // moving `budget` into the stream body.
            let consumer_name = identity
                .map(|id| id.consumer_name.clone())
                .unwrap_or_else(|| "anonymous".to_string());
            let consumer_type = identity
                .map(|id| consumer_type_str(id.consumer_type).to_string())
                .unwrap_or_else(|| "user".to_string());
            let team = budget
                .as_ref()
                .map(|g| g.team_key().to_string())
                .unwrap_or_default();
            // DW-081: build the streaming capture context when logging
            // is enabled for this consumer. The prompt is stored
            // redacted at stream close; the response is marked as
            // streamed (the full content is not reassembled).
            let prompt_log_ctx = build_stream_prompt_log_ctx(
                dp,
                gen,
                rid,
                route_name,
                &provider.name,
                &target.provider_model,
                version,
                identity,
                &json_body,
            );
            let body = super::ai_proxy::AiStreamBody::new(
                up_body,
                adapter,
                format!("chatcmpl-{rid}"),
                chat_req.model.clone(),
                provider.name.clone(),
                target.provider_model.clone(),
                version.to_string(),
                attempt_started,
                Arc::clone(&obs),
                rid.to_string(),
                budget,
                dp.ai_pricing(),
                dp.analytics(),
                consumer_name,
                consumer_type,
                team,
                prompt_log_ctx,
                Arc::clone(&guardrails),
                gen_ai_span.clone(),
                // PERF-02: streaming semantic cache tee. Pass the
                // cache and prompt text so the stream body can collect
                // frames and store them at close.
                if can_cache {
                    sem_cache.as_ref().map(Arc::clone)
                } else {
                    None
                },
                chat_req
                    .messages
                    .iter()
                    .map(|m| m.text_content())
                    .collect::<Vec<_>>()
                    .join("\n"),
                chat_req.model.clone(),
            );
            return Response::builder()
                .status(StatusCode::OK)
                .header(hyper::header::CONTENT_TYPE, "text/event-stream")
                .header(hyper::header::CACHE_CONTROL, "no-cache")
                .body(ProxyBody::Ai(Box::new(body)))
                .expect("streaming response is valid");
        }
        let ok_bytes =
            match bounded_collect(up_body, MAX_AI_PROVIDER_RESPONSE_BYTES, "response", rid).await {
                Ok(b) => b,
                Err(resp) => {
                    // An over-cap or unreadable SUCCESS body is
                    // provider-specific (a runaway completion), exactly
                    // like a malformed one below: record it, stash the
                    // 502, and let the next candidate answer. (The
                    // inbound "request" cap and the cheap "provider
                    // error" cap stay FINAL — the first is a client
                    // problem, the second is already on an error path.)
                    obs.record_ai_request(&provider.name, route_name, "translation_error", version);
                    obs.record_ai_request_duration(
                        &provider.name,
                        route_name,
                        ai_started.elapsed().as_secs_f64(),
                    );
                    tracing::warn!(
                        code = "ai_provider_body_over_cap",
                        request_id = %rid,
                        provider = %provider.name,
                        "provider 200 response exceeded the body cap or failed mid-read"
                    );
                    fallback_response = resp;
                    last_transport_error = false;
                    continue;
                }
            };
        let value: serde_json::Value = match serde_json::from_slice(&ok_bytes) {
            Ok(v) => v,
            Err(e) => {
                obs.record_ai_request(&provider.name, route_name, "translation_error", version);
                obs.record_ai_request_duration(
                    &provider.name,
                    route_name,
                    ai_started.elapsed().as_secs_f64(),
                );
                tracing::warn!(
                    code = "ai_provider_body_invalid",
                    request_id = %rid,
                    provider = %provider.name,
                    "provider 200 response was not valid JSON: {e}"
                );
                // A malformed 200 body is provider-specific; the next
                // candidate may be healthy.
                fallback_response = ai_error_response(
                    StatusCode::BAD_GATEWAY,
                    "the model provider returned a malformed response",
                    "api_error",
                    Some("provider_malformed_response"),
                    rid,
                );
                last_transport_error = false;
                continue;
            }
        };
        let chat_resp = match adapter.parse_response(&value) {
            Ok(r) => r,
            Err(e) => {
                obs.record_ai_request(&provider.name, route_name, "translation_error", version);
                obs.record_ai_request_duration(
                    &provider.name,
                    route_name,
                    ai_started.elapsed().as_secs_f64(),
                );
                tracing::warn!(
                    code = "ai_provider_body_untranslatable",
                    request_id = %rid,
                    provider = %provider.name,
                    "provider 200 response could not be translated: {e}"
                );
                fallback_response = ai_error_response(
                    StatusCode::BAD_GATEWAY,
                    "the model provider returned a response the gateway could not translate",
                    "api_error",
                    Some("provider_untranslatable_response"),
                    rid,
                );
                last_transport_error = false;
                continue;
            }
        };
        // AI-07: record the response-side gen_ai.* attributes on the
        // span (provider-reported only; None leaves the field empty).
        let finish_reasons: Vec<&str> = chat_resp
            .choices
            .iter()
            .map(|c| c.finish_reason.as_gen_ai())
            .collect();
        crate::observability::record_gen_ai_response_attrs(
            &gen_ai_span,
            chat_resp.model.as_deref(),
            chat_resp.usage.and_then(|u| u.prompt_tokens),
            chat_resp.usage.and_then(|u| u.completion_tokens),
            chat_resp.usage.and_then(|u| u.total_tokens),
            &finish_reasons,
            chat_resp.id.as_deref(),
        );
        if let Some(usage) = chat_resp.usage {
            // Usage attributes to the SERVING provider and canary
            // version (DW-076; the input DW-079 cost metering reads).
            obs.record_ai_tokens(
                &provider.name,
                usage.prompt_tokens.unwrap_or(0),
                usage.completion_tokens.unwrap_or(0),
                version,
            );
            // AI-07: per-request token distribution histogram.
            obs.record_ai_tokens_per_request(
                &provider.name,
                &target.provider_model,
                usage.prompt_tokens.unwrap_or(0),
                usage.completion_tokens.unwrap_or(0),
            );
            // DW-079: price the call through the dataplane's compiled
            // pricing table (swapped on reload, so a pricing change
            // applies to the next request with no restart). Unknown
            // model -> None (the unknown-model policy is applied
            // below, DW-AI-03).
            let pricing = dp.ai_pricing();
            let cost_opt = pricing.cost_micros(&target.provider_model, usage, false);
            let cost = cost_opt.unwrap_or(0);
            if cost > 0 {
                obs.record_ai_cost(&provider.name, &target.provider_model, cost);
            }
            // Budget spend (DW-078): the provider-reported usage is
            // recorded against the requesting holder's windows AFTER
            // the call — check-then-spend; the crossing (if any) is
            // already priced into the next pre-check.
            if let Some(guard) = &budget {
                guard.spend(now_epoch_s(), usage, cost);
            }
            // DW-079: record the spend dimensions into the analytics
            // store for billing reconciliation. The team field is the
            // policy-scoped budget's key (the policy name) when a
            // team budget binds, else empty.
            if let Some(analytics) = dp.analytics() {
                let consumer_name = identity
                    .map(|id| id.consumer_name.as_str())
                    .unwrap_or("anonymous");
                let team = budget
                    .as_ref()
                    .map(|g| g.team_key().to_string())
                    .unwrap_or_default();
                let prompt = usage.prompt_tokens.unwrap_or(0);
                let completion = usage.completion_tokens.unwrap_or(0);
                let total = usage
                    .total_tokens
                    .unwrap_or_else(|| prompt.saturating_add(completion));
                analytics.offer_ai_spend(crate::analytics::AiSpendRecord {
                    ts_ms: now_ms(),
                    consumer: consumer_name.to_string(),
                    consumer_type: identity
                        .map(|i| consumer_type_str(i.consumer_type))
                        .unwrap_or("user")
                        .to_string(),
                    team,
                    provider: provider.name.clone(),
                    model: target.provider_model.clone(),
                    version: version.to_string(),
                    prompt_tokens: prompt,
                    completion_tokens: completion,
                    total_tokens: total,
                    cost_micros: cost,
                });
            }
        }
        // DW-AI-03: unknown-model pricing policy. After cost
        // computation, if the serving provider_model has no pricing
        // entry, apply the configured policy. `fail_closed` rejects
        // the response with 403 `ai_unknown_model_pricing` (the
        // provider call already happened, but the unpriced model
        // cannot serve traffic silently); `alert` logs a WARN and
        // increments the `dwara_ai_unknown_pricing_total{model}`
        // metric; `allow` (the default) proceeds with cost 0 as
        // before.
        let pricing = dp.ai_pricing();
        if !pricing.has_pricing(&target.provider_model) {
            let policy = gen
                .snapshot
                .gateway()
                .ai
                .as_ref()
                .and_then(|c| c.unknown_model_policy)
                .unwrap_or_default();
            match policy {
                UnknownModelPolicy::FailClosed => {
                    obs.record_ai_request(
                        &provider.name,
                        route_name,
                        "unknown_model_pricing",
                        version,
                    );
                    tracing::warn!(
                        code = "ai_unknown_model_pricing",
                        request_id = %rid,
                        route = %route_name,
                        provider = %provider.name,
                        model = %target.provider_model,
                        "rejecting response: model has no pricing entry (fail_closed policy)"
                    );
                    return ai_error_response(
                        StatusCode::FORBIDDEN,
                        "the model has no pricing entry and the unknown-model policy is fail_closed",
                        "invalid_request_error",
                        Some("ai_unknown_model_pricing"),
                        rid,
                    );
                }
                UnknownModelPolicy::Alert => {
                    obs.record_ai_unknown_pricing(&target.provider_model);
                    tracing::warn!(
                        code = "ai_unknown_model_pricing",
                        request_id = %rid,
                        route = %route_name,
                        provider = %provider.name,
                        model = %target.provider_model,
                        "model has no pricing entry; proceeding with cost 0 (alert policy)"
                    );
                }
                UnknownModelPolicy::Allow => {}
            }
        }
        obs.record_ai_request(&provider.name, route_name, "success", version);
        // AI-07: end-to-end AI request latency (non-streaming success).
        obs.record_ai_request_duration(
            &provider.name,
            route_name,
            ai_started.elapsed().as_secs_f64(),
        );
        // DW-082: response guardrails. AFTER the response is parsed
        // and BEFORE it is returned to the client. A `block` action
        // returns a 400 (`response_schema_violation` for schema kind,
        // `guardrail_blocked` otherwise); a `log` action records and
        // continues. Non-streaming only (streaming responses cannot
        // be schema-validated on partial content; banned-content
        // checks run per-chunk in the stream body).
        if !guardrails.is_empty() {
            let consumer = identity.map(|id| id.consumer_name.as_str());
            let gateway = gen.snapshot.gateway();
            let route_cfg = gateway.routes.iter().find(|r| r.name == route_name);
            let route_policies: &[String] = route_cfg.map(|r| r.policies.as_slice()).unwrap_or(&[]);
            let service_policies: &[String] = route_cfg
                .map(|r| crate::ai::budget::service_policies_of(gateway, &r.service))
                .unwrap_or(&[]);
            let consumer_policies: &[String] = consumer
                .map(|c| crate::ai::budget::consumer_policies_of(gateway, c))
                .unwrap_or(&[]);
            let listener_policies = crate::ai::budget::listener_policies_of(gateway, listener_name);
            let guardrail_result = guardrails.check_response(
                consumer_policies,
                route_policies,
                service_policies,
                listener_policies,
                &gateway.global_policies,
                &chat_resp,
            );
            if let crate::ai::guardrails::GuardrailResult::Block { rule_name, reason } =
                &guardrail_result
            {
                let kind_label = if reason == "response_schema_violation" {
                    "schema"
                } else {
                    "response"
                };
                dp.observability_arc()
                    .record_ai_guardrail_blocked(rule_name, kind_label);
                tracing::info!(
                    code = "guardrail_blocked",
                    request_id = %rid,
                    route = %route_name,
                    rule = %rule_name,
                    reason = %reason,
                    "AI guardrail blocked the response before returning to the client"
                );
                let error_code = if reason == "response_schema_violation" {
                    "response_schema_violation"
                } else {
                    "guardrail_blocked"
                };
                let body = openai_compat::error_body(
                    &format!("the response was blocked by guardrail rule '{rule_name}': {reason}"),
                    "invalid_request_error",
                    Some(error_code),
                    rid,
                );
                return Response::builder()
                    .status(StatusCode::BAD_REQUEST)
                    .header(hyper::header::CONTENT_TYPE, "application/json")
                    .body(ProxyBody::Full(Full::new(Bytes::from(body.to_string()))))
                    .expect("static 400 response is valid");
            }
        }
        let body = openai_compat::response_to_openai(&chat_resp, &chat_req.model, rid);
        // DW-081: prompt/response logging capture hook (non-streaming).
        // After the response is parsed and BEFORE it is sent, check
        // whether logging is enabled (global + per-consumer override),
        // apply sampling, redact, and offer to analytics. The capture
        // is fire-and-forget (never blocks the response path).
        capture_prompt_log(
            dp,
            gen,
            rid,
            route_name,
            &provider.name,
            &target.provider_model,
            version,
            identity,
            &json_body,
            &body,
            false,
        );
        tracing::info!(
            code = "ai_request_served",
            request_id = %rid,
            route = %route_name,
            provider = %provider.name,
            model = %chat_req.model,
            version = %version,
            attempts = attempts,
            upstream = %provider.upstream,
            "ai chat completion translated and served"
        );
        // DW-083: store the response in the semantic cache
        // (non-streaming only). Fire-and-forget — the embedding call
        // and HNSW insert happen in a spawned task so they never
        // block the response path. The prompt text is the joined
        // message text (the same shape the lookup used); the body is
        // the OpenAI-shaped response JSON.
        if can_cache {
            if let Some(sem_cache) = &sem_cache {
                let sem_cache = Arc::clone(sem_cache);
                let prompt_text = chat_req
                    .messages
                    .iter()
                    .map(|m| m.text_content())
                    .collect::<Vec<_>>()
                    .join("\n");
                let body_clone = body.clone();
                let model = chat_req.model.clone();
                tokio::spawn(async move {
                    sem_cache.store(&prompt_text, &body_clone, &model).await;
                });
            }
        }
        return response_with_json(StatusCode::OK, &body, &up_parts.headers);
    }

    // 6. Every candidate failed. The last stashed provider answer is
    // the most truthful response; a pure-transport exhaustion keeps
    // the generic 502.
    rec.attempts = attempts.max(1);
    if attempts > 1 {
        tracing::warn!(
            code = "ai_failover_exhausted",
            request_id = %rid,
            model = %chat_req.model,
            attempts = attempts,
            transport_only = last_transport_error,
            "all providers failed for the model"
        );
    }
    fallback_response
}

/// DW-081: prompt/response logging capture hook. Checks whether
/// logging is enabled (global + per-consumer override), applies
/// sampling, redacts the prompt and response JSON, and offers the
/// record to the analytics store. Fire-and-forget — never blocks.
#[allow(clippy::too_many_arguments)]
fn capture_prompt_log(
    dp: &Arc<DataPlane>,
    gen: &Arc<Generation>,
    rid: &str,
    route_name: &str,
    provider: &str,
    model: &str,
    version: &str,
    identity: Option<&crate::security::authn::Identity>,
    prompt_json: &serde_json::Value,
    response_json: &serde_json::Value,
    stream: bool,
) {
    let Some(logging) = dp.ai_logging() else {
        return;
    };
    // Resolve the per-consumer override from the config.
    let consumer_name = identity
        .map(|id| id.consumer_name.as_str())
        .unwrap_or("anonymous");
    let consumer_override = gen
        .snapshot
        .gateway()
        .consumers
        .iter()
        .find(|c| c.name == consumer_name)
        .and_then(|c| c.ai_logging);
    if !logging.capture_for(consumer_override) {
        return;
    }
    // Sampling: deterministic per request id.
    if !logging.should_sample(rid) {
        return;
    }
    // Redact both the prompt and the response.
    let redactor = logging.redactor();
    let redacted_prompt = redactor.redact_json(prompt_json);
    let redacted_response = redactor.redact_json(response_json);
    // Offer to analytics (fire-and-forget).
    if let Some(analytics) = dp.analytics() {
        analytics.offer_ai_prompt_log(crate::analytics::AiPromptLogRecord {
            ts_ms: now_ms(),
            request_id: rid.to_string(),
            consumer: consumer_name.to_string(),
            route: route_name.to_string(),
            provider: provider.to_string(),
            model: model.to_string(),
            version: version.to_string(),
            prompt_json: redacted_prompt.to_string(),
            response_json: redacted_response.to_string(),
            stream,
        });
    }
}

/// DW-081: build the streaming capture context. Returns None when
/// logging is off for this consumer or the sampling decision is
/// negative — the stream body then skips the capture at close. The
/// prompt JSON is carried to stream close where it is redacted and
/// offered (the response is marked as streamed).
#[allow(clippy::too_many_arguments)]
fn build_stream_prompt_log_ctx(
    dp: &Arc<DataPlane>,
    gen: &Arc<Generation>,
    rid: &str,
    route_name: &str,
    provider: &str,
    model: &str,
    version: &str,
    identity: Option<&crate::security::authn::Identity>,
    prompt_json: &serde_json::Value,
) -> Option<PromptLogCtx> {
    let logging = dp.ai_logging()?;
    let consumer_name = identity
        .map(|id| id.consumer_name.as_str())
        .unwrap_or("anonymous");
    let consumer_override = gen
        .snapshot
        .gateway()
        .consumers
        .iter()
        .find(|c| c.name == consumer_name)
        .and_then(|c| c.ai_logging);
    if !logging.capture_for(consumer_override) {
        return None;
    }
    if !logging.should_sample(rid) {
        return None;
    }
    // Redact the prompt at build time (the stream body does not carry
    // the redactor).
    let redactor = logging.redactor();
    let redacted_prompt = redactor.redact_json(prompt_json);
    Some(PromptLogCtx {
        route: route_name.to_string(),
        provider: provider.to_string(),
        model: model.to_string(),
        version: version.to_string(),
        consumer: consumer_name.to_string(),
        prompt_json: redacted_prompt.to_string(),
        recorded: false,
    })
}

/// Collect a body up to `cap` bytes, aborting the read the moment the
/// cap is crossed (the producer is dropped mid-stream — an over-cap
/// body is never buffered whole). Over-cap answers 413 inbound / 502
/// provider-side, in the AI error shape.
async fn bounded_collect<B>(
    body: B,
    cap: usize,
    what: &str,
    rid: &str,
) -> Result<Bytes, Response<ProxyBody>>
where
    B: Body<Data = Bytes> + Send + 'static,
    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
{
    let mut buf: Vec<u8> = Vec::new();
    let mut body = pin!(body);
    while let Some(frame) = body.as_mut().frame().await {
        let frame = frame.map_err(|e| {
            let e = e.into();
            ai_error_response(
                StatusCode::BAD_REQUEST,
                &format!("the {what} body could not be read: {e}"),
                if what == "request" {
                    "invalid_request_error"
                } else {
                    "api_error"
                },
                None,
                rid,
            )
        })?;
        if let Some(data) = frame.data_ref() {
            if buf.len() + data.len() > cap {
                return Err(ai_error_response(
                    if what == "request" {
                        StatusCode::PAYLOAD_TOO_LARGE
                    } else {
                        StatusCode::BAD_GATEWAY
                    },
                    &format!("the {what} body exceeds the gateway's {cap}-byte limit"),
                    if what == "request" {
                        "invalid_request_error"
                    } else {
                        "api_error"
                    },
                    Some("body_too_large"),
                    rid,
                ));
            }
            buf.extend_from_slice(data);
        }
    }
    Ok(Bytes::from(buf))
}

/// Build an AI-route error response (OpenAI error shape).
fn ai_error_response(
    status: StatusCode,
    message: &str,
    error_type: &str,
    code: Option<&str>,
    rid: &str,
) -> Response<ProxyBody> {
    let body = openai_compat::error_body(message, error_type, code, rid);
    response_with_json(status, &body, &HeaderMap::new())
}

/// Current Unix seconds (budget window indexing).
fn now_epoch_s() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Current Unix milliseconds (DW-079 spend record timestamping).
fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

/// Whether provider response headers declare an SSE body.
fn is_event_stream(headers: &HeaderMap) -> bool {
    headers
        .get(hyper::header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .map(|v| v.starts_with("text/event-stream"))
        .unwrap_or(false)
}

/// Parse the `Retry-After` header from an AI provider response (DW-089)
/// into a duration from now. Supports both the integer-seconds and
/// HTTP-date forms (see `config::versioning::parse_retry_after`).
/// Returns `None` when the header is absent or unparseable.
fn parse_ai_retry_after(headers: &HeaderMap) -> Option<std::time::Duration> {
    let value = headers.get(hyper::header::RETRY_AFTER)?;
    let s = value.to_str().ok()?;
    let now_unix_seconds = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);
    crate::config::versioning::parse_retry_after(s, now_unix_seconds)
}

/// AI-09 (#198): Parse provider rate-limit headers into per-key health.
/// Covers the common header names used by OpenAI, Anthropic, and
/// Gemini. Returns a `KeyHealth` with whatever fields were present.
fn parse_rate_limit_headers(headers: &HeaderMap, status: u16) -> crate::ai::credentials::KeyHealth {
    use crate::ai::credentials::KeyHealth;
    use std::time::{Duration, Instant};

    let mut health = KeyHealth {
        last_status: Some(status),
        updated_at: Some(Instant::now()),
        ..Default::default()
    };

    // OpenAI: x-ratelimit-remaining-requests, x-ratelimit-remaining-tokens
    // Anthropic: anthropic-ratelimit-requests-remaining, anthropic-ratelimit-tokens-remaining
    // Gemini: x-ratelimit-remaining (generic)
    // We try all known header names and take the first that parses.
    let remaining_request_headers = [
        "x-ratelimit-remaining-requests",
        "anthropic-ratelimit-requests-remaining",
        "x-ratelimit-remaining",
    ];
    let remaining_token_headers = [
        "x-ratelimit-remaining-tokens",
        "anthropic-ratelimit-tokens-remaining",
    ];
    let reset_headers = [
        "x-ratelimit-reset-requests",
        "anthropic-ratelimit-requests-reset",
        "x-ratelimit-reset",
    ];

    for name in &remaining_request_headers {
        if let Some(v) = headers.get(*name).and_then(|h| h.to_str().ok()) {
            if let Ok(n) = v.trim().parse::<u64>() {
                health.remaining_requests = Some(n);
                break;
            }
            // Some providers send a duration like "1s" or "6m0s"
            if let Some(n) = parse_duration_seconds(v) {
                health.remaining_requests = Some(n);
                break;
            }
        }
    }

    for name in &remaining_token_headers {
        if let Some(v) = headers.get(*name).and_then(|h| h.to_str().ok()) {
            if let Ok(n) = v.trim().parse::<u64>() {
                health.remaining_tokens = Some(n);
                break;
            }
        }
    }

    for name in &reset_headers {
        if let Some(v) = headers.get(*name).and_then(|h| h.to_str().ok()) {
            if let Some(secs) = parse_duration_seconds(v) {
                health.reset_at = Some(Instant::now() + Duration::from_secs(secs));
                break;
            }
            // Try as integer seconds
            if let Ok(secs) = v.trim().parse::<u64>() {
                health.reset_at = Some(Instant::now() + Duration::from_secs(secs));
                break;
            }
        }
    }

    health
}

/// Parse a duration string like "1s", "6m0s", "1h2m3s" into seconds.
fn parse_duration_seconds(s: &str) -> Option<u64> {
    let s = s.trim();
    if s.is_empty() {
        return None;
    }
    // Try plain integer first
    if let Ok(n) = s.parse::<u64>() {
        return Some(n);
    }
    // Parse h/m/s suffixes
    let mut total: u64 = 0;
    let mut num = String::new();
    for ch in s.chars() {
        if ch.is_ascii_digit() {
            num.push(ch);
        } else {
            let n: u64 = num.parse().ok()?;
            num.clear();
            match ch {
                'h' => total += n * 3600,
                'm' => total += n * 60,
                's' => total += n,
                _ => return None,
            }
        }
    }
    // Trailing number without suffix = seconds
    if !num.is_empty() {
        let n: u64 = num.parse().ok()?;
        total += n;
    }
    Some(total)
}

/// Build a JSON response. The provider's `x-request-id` (when sent) is
/// carried through for correlation; every other provider header is
/// dropped — the gateway authored this response.
fn response_with_json(
    status: StatusCode,
    body: &serde_json::Value,
    provider_headers: &HeaderMap,
) -> Response<ProxyBody> {
    let mut builder = Response::builder()
        .status(status)
        .header(hyper::header::CONTENT_TYPE, "application/json");
    if let Some(v) = provider_headers.get("x-request-id") {
        builder = builder.header("x-request-id", v);
    }
    builder
        .body(ProxyBody::Full(Full::new(Bytes::from(body.to_string()))))
        .expect("static JSON response is valid")
}

// ---------------------------------------------------------------------------
// The streaming body (DW-077)
// ---------------------------------------------------------------------------

/// Per-stream accounting shared by the body and its metrics hooks.
struct StreamGauges {
    provider: String,
    version: String,
    chunks: u64,
    first_token_recorded: bool,
    started: std::time::Instant,
    /// Set once terminal metrics have been recorded (clean end OR
    /// drop): the counters must fire exactly once per stream.
    closed: bool,
}

/// The zero-buffer AI streaming body (DW-077): wraps the provider's
/// upstream body, translates each COMPLETE provider SSE frame into
/// OpenAI-shaped client frames as the bytes arrive, and forwards them
/// in the same poll — nothing waits for the stream to finish. Usage
/// events accumulate (provider-reported only) and are emitted as one
/// terminal chunk; the gateway owns the `data: [DONE]` terminator. A
/// provider body error after forwarding becomes a terminal error
/// chunk (already-forwarded content stands), never a reset.
///
/// Infallible by construction: every failure mode is expressed as
/// terminal frames, so the client stream always ends cleanly.
pub struct AiStreamBody {
    /// The provider's body — until the budget cutoff DROPS it (the
    /// upstream cancel then propagates immediately, regardless of how
    /// the client behaves; see `cut_off`).
    inner: Option<crate::dataplane::upstream::UpstreamBody>,
    translator: StreamTranslator,
    adapter: &'static dyn ProviderAdapter,
    /// Synthesized terminal frames waiting to be forwarded.
    tail: std::collections::VecDeque<String>,
    gauges: StreamGauges,
    obs: std::sync::Arc<crate::observability::Observability>,
    /// The requesting holder's budget (DW-078): checked as usage
    /// events accumulate — a crossing cuts the stream off.
    budget: Option<crate::ai::budget::BudgetGuard>,
    /// The serving provider's model id (the DW-079 pricing key).
    provider_model: String,
    /// The request id (the cutoff event's correlation handle).
    rid: String,
    /// Budget cutoff fired: the provider body was dropped AT the
    /// cutoff so the upstream request is cancelled immediately — a
    /// stalled client that never drains the tail must not keep the
    /// provider generating billed tokens. Only the synthesized tail
    /// remains to forward.
    cut_off: bool,
    /// Budget spend watermarks (DW-078): the accumulated totals
    /// already spent. The provider's usage report GROWS as the stream
    /// runs (input tokens at message_start, output at message_delta),
    /// so every spend is the DELTA above the watermark — each
    /// reported token is counted exactly once no matter how many
    /// chunks the report arrived in.
    budget_spent_tokens: u64,
    budget_spent_cost_micros: u64,
    /// DW-079: the dataplane's compiled pricing table (for pricing
    /// the stream's terminal usage).
    pricing: std::sync::Arc<crate::ai::cost::PricingTable>,
    /// DW-079: the analytics store (for recording the spend record at
    /// stream close). None when analytics is off.
    analytics: Option<std::sync::Arc<crate::analytics::EmbeddedAnalytics>>,
    /// DW-079: the consumer name for spend attribution.
    consumer: String,
    /// DW-113: the consumer type for spend attribution (`user` or
    /// `agent`).
    consumer_type: String,
    /// DW-079: the team key for spend attribution (policy name when
    /// a team budget binds, else empty).
    team: String,
    /// DW-079: whether the spend record has been recorded (exactly
    /// once per stream, at close).
    spend_recorded: bool,
    /// DW-081: prompt/response logging context for streaming capture.
    /// The prompt JSON (the original OpenAI-shaped request body) is
    /// stored redacted at stream close; the response is marked as
    /// streamed (the full content is not reassembled — buffering the
    /// entire stream would contradict the zero-buffer design).
    prompt_log_ctx: Option<PromptLogCtx>,
    /// DW-082: the compiled guardrail engine for streaming banned-
    /// content checks. Each forwarded chunk's text is checked against
    /// banned-content rules; a match cuts the stream off (similar to
    /// the budget cutoff). Schema enforcement does not apply to
    /// streaming (partial content cannot be validated).
    guardrails: std::sync::Arc<crate::ai::guardrails::GuardrailEngine>,
    /// DW-082: whether a banned-content cutoff has fired (the tail
    /// is the stream's remainder).
    guardrail_cut_off: bool,
    /// AI-07: the OTel GenAI semantic-convention span. Response-side
    /// `gen_ai.*` attributes (usage, finish reasons, response id/model)
    /// are recorded on this handle at stream close (the values arrive
    /// across the stream and only the terminal usage is complete).
    gen_ai_span: tracing::Span,
    /// PERF-02: the semantic cache for streaming tee. When present,
    /// forwarded SSE frames are collected into `stream_tee_frames`
    /// (bounded by `stream_tee_cap_bytes`); at stream close, the
    /// collected frames are stored in the cache. None when the cache
    /// is disabled or not configured.
    stream_tee_cache: Option<std::sync::Arc<crate::ai::semantic_cache::SemanticCacheEngine>>,
    /// PERF-02: the collected SSE frames for the streaming tee. Each
    /// forwarded `data: ...\n\n` frame is pushed here; at stream close
    /// the collected frames are stored in the cache.
    stream_tee_frames: Vec<String>,
    /// PERF-02: the byte cap for the streaming tee buffer. When the
    /// collected frames exceed this cap, the tee is abandoned (the
    /// stream is still forwarded to the client, but NOT cached — a
    /// single long stream must not consume unbounded memory).
    stream_tee_cap_bytes: usize,
    /// PERF-02: whether the tee has been abandoned (the cap was
    /// exceeded). When true, no store happens at stream close.
    stream_tee_abandoned: bool,
    /// PERF-02: the prompt text for the streaming store (the joined
    /// message text, the same shape the lookup used).
    stream_tee_prompt: String,
    /// PERF-02: the model alias for the streaming store.
    stream_tee_model: String,
}

/// DW-081: the streaming capture context carried by AiStreamBody.
pub(super) struct PromptLogCtx {
    route: String,
    provider: String,
    model: String,
    version: String,
    consumer: String,
    /// The REDACTED prompt as a JSON string (redacted at build time
    /// so the stream body does not need to carry the redactor).
    prompt_json: String,
    /// Whether the prompt log has been recorded (exactly once per
    /// stream, at close).
    recorded: bool,
}

impl AiStreamBody {
    /// Build the body for a provider stream whose headers resolved
    /// successfully. `started` is the request's send instant (first-
    /// token latency measures from there).
    #[allow(clippy::too_many_arguments)]
    pub(super) fn new(
        inner: crate::dataplane::upstream::UpstreamBody,
        adapter: &'static dyn ProviderAdapter,
        response_id: String,
        model_alias: String,
        provider: String,
        provider_model: String,
        version: String,
        started: std::time::Instant,
        obs: std::sync::Arc<crate::observability::Observability>,
        rid: String,
        budget: Option<crate::ai::budget::BudgetGuard>,
        pricing: std::sync::Arc<crate::ai::cost::PricingTable>,
        analytics: Option<std::sync::Arc<crate::analytics::EmbeddedAnalytics>>,
        consumer: String,
        consumer_type: String,
        team: String,
        prompt_log_ctx: Option<PromptLogCtx>,
        guardrails: std::sync::Arc<crate::ai::guardrails::GuardrailEngine>,
        gen_ai_span: tracing::Span,
        stream_tee_cache: Option<std::sync::Arc<crate::ai::semantic_cache::SemanticCacheEngine>>,
        stream_tee_prompt: String,
        stream_tee_model: String,
    ) -> Self {
        AiStreamBody {
            inner: Some(inner),
            translator: StreamTranslator::new(response_id, model_alias, openai_compat::unix_now()),
            adapter,
            tail: std::collections::VecDeque::new(),
            gauges: StreamGauges {
                provider,
                version,
                chunks: 0,
                first_token_recorded: false,
                started,
                closed: false,
            },
            obs,
            budget,
            provider_model,
            rid,
            cut_off: false,
            budget_spent_tokens: 0,
            budget_spent_cost_micros: 0,
            pricing,
            analytics,
            consumer,
            consumer_type,
            team,
            spend_recorded: false,
            prompt_log_ctx,
            guardrails,
            guardrail_cut_off: false,
            gen_ai_span,
            stream_tee_cap_bytes: 1024 * 1024,
            stream_tee_frames: Vec::new(),
            stream_tee_abandoned: false,
            stream_tee_cache,
            stream_tee_prompt,
            stream_tee_model,
        }
    }

    /// Spend the provider-reported usage GROWTH since the last spend
    /// (DW-078). The translator's usage is an ACCUMULATOR that grows
    /// as the provider reports (input tokens early, output late), so
    /// only the delta above the watermark is NEW spend — each
    /// reported token is counted exactly once regardless of how many
    /// chunks its report spanned. Returns whether the token window
    /// crossed its limit on THIS spend (the mid-stream cutoff signal).
    fn budget_spend_delta(&mut self) -> bool {
        let Some(guard) = &self.budget else {
            return false;
        };
        let usage = self.translator.usage();
        let total = usage.total_tokens.unwrap_or_else(|| {
            usage
                .prompt_tokens
                .unwrap_or(0)
                .saturating_add(usage.completion_tokens.unwrap_or(0))
        });
        let tokens = total.saturating_sub(self.budget_spent_tokens);
        let cost_total = self
            .pricing
            .cost_micros(&self.provider_model, usage, false)
            .unwrap_or(0);
        let cost = cost_total.saturating_sub(self.budget_spent_cost_micros);
        if tokens == 0 && cost == 0 {
            return false;
        }
        // Watermarks advance to the accumulated totals (max, so a
        // provider that revises a count downward never re-spends the
        // difference later).
        self.budget_spent_tokens = total.max(self.budget_spent_tokens);
        self.budget_spent_cost_micros = cost_total.max(self.budget_spent_cost_micros);
        guard.spend(
            now_epoch_s(),
            Usage {
                total_tokens: Some(tokens),
                ..Usage::default()
            },
            cost,
        )
    }

    /// Mid-stream cutoff check (DW-078). Called after each translated
    /// batch: the batch's NEW usage is spent against the holder's
    /// window, and a crossing (only detectable when the dialect
    /// reports usage mid-stream — Anthropic's message_start input
    /// tokens) cuts the stream.
    fn budget_tick(&mut self) {
        if self.budget.is_none() {
            return;
        }
        let crossed = self.budget_spend_delta();
        if crossed && !self.cut_off && !self.translator.is_ended() {
            self.cut_off = true;
            // Drop the provider body NOW: an upstream request is
            // cancelled the moment its body is dropped, however slowly
            // (or never) the client drains the cutoff tail below — the
            // cancel must not wait on client behavior. The tail still
            // reaches the client: the cutoff event, then [DONE].
            self.inner = None;
            self.obs.record_ai_budget_denied("tokens");
            tracing::warn!(
                code = "ai_budget_exceeded_midstream",
                request_id = %self.rid,
                "token budget crossed mid-stream; cutting off and cancelling the provider stream"
            );
            self.tail
                .push_back(crate::ai::budget::BudgetGuard::cutoff_frame(&self.rid));
            self.tail.push_back("data: [DONE]\n\n".to_string());
        }
    }

    /// DW-082: mid-stream banned-content check. Called after each
    /// translated batch: the batch's forwarded text is checked against
    /// banned-content rules; a match cuts the stream off (similar to
    /// the budget cutoff). Schema enforcement does not apply to
    /// streaming (partial content cannot be validated).
    fn guardrail_tick(&mut self, batch_text: &str) {
        if self.guardrails.is_empty() || self.guardrail_cut_off {
            return;
        }
        // AI-08 (#197): PII redaction on streaming chunks. The
        // redacted text replaces the batch text in the translated
        // frames. This runs BEFORE the banned-content check so the
        // banned patterns see the redacted text.
        let batch_text = self.guardrails.redact_stream_chunk(batch_text);
        if let Some(rule_name) = self.guardrails.check_stream_chunk(&batch_text) {
            self.guardrail_cut_off = true;
            // Drop the provider body NOW (eager upstream cancel).
            self.inner = None;
            self.obs.record_ai_guardrail_blocked(&rule_name, "banned");
            tracing::warn!(
                code = "guardrail_blocked_midstream",
                request_id = %self.rid,
                rule = %rule_name,
                "banned-content guardrail matched mid-stream; cutting off and cancelling the provider stream"
            );
            // Synthesize an error chunk and the terminator.
            let error_payload = serde_json::json!({
                "error": {
                    "code": "guardrail_blocked",
                    "message": "the response was blocked by a banned-content guardrail",
                    "request_id": self.rid,
                }
            });
            self.tail.push_back(format!("data: {}\n\n", error_payload));
            self.tail.push_back("data: [DONE]\n\n".to_string());
        }
    }

    /// Record per-chunk accounting for a translated batch.
    fn note_chunks(&mut self, count: usize, first: bool) {
        for _ in 0..count {
            self.obs.record_ai_stream_chunk(&self.gauges.provider);
        }
        self.gauges.chunks += count as u64;
        if first && !self.gauges.first_token_recorded {
            self.gauges.first_token_recorded = true;
            self.obs.record_ai_first_token(
                &self.gauges.provider,
                self.gauges.started.elapsed().as_secs_f64(),
            );
        }
    }

    /// PERF-02: collect forwarded SSE frames into the tee buffer for
    /// streaming cache storage at close. Each frame is a `data: ...\n\n`
    /// string. The buffer is bounded by `stream_tee_cap_bytes`: when
    /// the total collected bytes exceed the cap, the tee is abandoned
    /// (no store at close — the stream is still forwarded to the
    /// client unmodified).
    fn tee_collect(&mut self, frames: &[String]) {
        if self.stream_tee_cache.is_none() || self.stream_tee_abandoned {
            return;
        }
        let mut total: usize = self.stream_tee_frames.iter().map(|f| f.len()).sum();
        for f in frames {
            total += f.len();
            if total > self.stream_tee_cap_bytes {
                // Cap exceeded: abandon the tee. The stream is still
                // forwarded to the client; it just is not cached.
                self.stream_tee_abandoned = true;
                self.stream_tee_frames.clear();
                tracing::info!(
                    code = "semantic_cache_stream_tee_abandoned",
                    cap_bytes = self.stream_tee_cap_bytes,
                    "streaming semantic cache tee abandoned (cap exceeded); stream not cached"
                );
                return;
            }
            self.stream_tee_frames.push(f.clone());
        }
    }

    /// PERF-02: store the collected streaming frames in the semantic
    /// cache. Called at stream close when the tee was not abandoned
    /// and the cache is present. Fire-and-forget (spawned): the
    /// embedding call and HNSW insert happen after the response is
    /// fully sent.
    fn tee_store(&self) {
        if self.stream_tee_abandoned || self.stream_tee_frames.is_empty() {
            return;
        }
        let Some(cache) = &self.stream_tee_cache else {
            return;
        };
        let cache = Arc::clone(cache);
        let prompt = self.stream_tee_prompt.clone();
        let model = self.stream_tee_model.clone();
        let frames = self.stream_tee_frames.clone();
        tokio::spawn(async move {
            cache.store_streaming(&prompt, frames, &model).await;
        });
    }

    /// Terminal metrics: exactly once per stream. Records duration and
    /// the accumulated provider-reported usage (the DW-079 input; the
    /// version label carries the canary attribution, DW-076).
    fn close(&mut self) {
        if self.gauges.closed {
            return;
        }
        self.gauges.closed = true;
        self.obs.record_ai_stream_end(
            &self.gauges.provider,
            self.gauges.started.elapsed().as_secs_f64(),
        );
        let usage = self.translator.usage();
        // Terminal budget spend (DW-078): the not-yet-spent delta. A
        // dialect that reports usage only in its final events spends
        // here (its mid-stream ticks saw nothing) — enforced by the
        // next pre-check; mid-stream cutoff was impossible for it.
        // After a cutoff the stream's spend is already recorded.
        if !self.cut_off {
            self.budget_spend_delta();
        }
        if usage.prompt_tokens.is_some() || usage.completion_tokens.is_some() {
            self.obs.record_ai_tokens(
                &self.gauges.provider,
                usage.prompt_tokens.unwrap_or(0),
                usage.completion_tokens.unwrap_or(0),
                &self.gauges.version,
            );
            // AI-07: per-request token distribution histogram.
            self.obs.record_ai_tokens_per_request(
                &self.gauges.provider,
                &self.provider_model,
                usage.prompt_tokens.unwrap_or(0),
                usage.completion_tokens.unwrap_or(0),
            );
        }
        // AI-07: record the response-side gen_ai.* attributes on the
        // span at stream close. The response model is the serving
        // provider model (streaming providers do not echo a model in a
        // single frame); the response id is the gateway-synthesized
        // stream id; finish reasons are not available mid-stream
        // without reassembling choices (left empty — "where available").
        crate::observability::record_gen_ai_response_attrs(
            &self.gen_ai_span,
            Some(&self.provider_model),
            usage.prompt_tokens,
            usage.completion_tokens,
            usage.total_tokens,
            &[],
            Some(self.translator.id()),
        );
        // DW-079: record the spend dimensions into the analytics
        // store (exactly once per stream). The terminal usage is the
        // accumulated provider-reported total; cost is priced through
        // the dataplane's pricing table.
        if !self.spend_recorded {
            self.spend_recorded = true;
            let prompt = usage.prompt_tokens.unwrap_or(0);
            let completion = usage.completion_tokens.unwrap_or(0);
            let total = usage
                .total_tokens
                .unwrap_or_else(|| prompt.saturating_add(completion));
            let cost = self
                .pricing
                .cost_micros(&self.provider_model, usage, false)
                .unwrap_or(0);
            if cost > 0 {
                self.obs
                    .record_ai_cost(&self.gauges.provider, &self.provider_model, cost);
            }
            if let Some(analytics) = &self.analytics {
                analytics.offer_ai_spend(crate::analytics::AiSpendRecord {
                    ts_ms: now_ms(),
                    consumer: self.consumer.clone(),
                    consumer_type: self.consumer_type.clone(),
                    team: self.team.clone(),
                    provider: self.gauges.provider.clone(),
                    model: self.provider_model.clone(),
                    version: self.gauges.version.clone(),
                    prompt_tokens: prompt,
                    completion_tokens: completion,
                    total_tokens: total,
                    cost_micros: cost,
                });
            }
        }
        // DW-081: streaming prompt/response capture. The prompt is
        // stored redacted (redacted at build time); the response is
        // marked as streamed (the full content is not reassembled —
        // buffering the entire stream would contradict the zero-buffer
        // design). The context was built at stream start only when
        // logging is enabled for this consumer and the sampling
        // decision was positive.
        if let Some(ctx) = &mut self.prompt_log_ctx {
            if !ctx.recorded {
                ctx.recorded = true;
                if let Some(analytics) = &self.analytics {
                    analytics.offer_ai_prompt_log(crate::analytics::AiPromptLogRecord {
                        ts_ms: now_ms(),
                        request_id: self.rid.clone(),
                        consumer: ctx.consumer.clone(),
                        route: ctx.route.clone(),
                        provider: ctx.provider.clone(),
                        model: ctx.model.clone(),
                        version: ctx.version.clone(),
                        prompt_json: ctx.prompt_json.clone(),
                        response_json: serde_json::json!({"streamed": true}).to_string(),
                        stream: true,
                    });
                }
            }
        }
        // PERF-02: store the collected streaming frames in the
        // semantic cache (fire-and-forget). No-op when the tee was
        // abandoned or the cache is absent.
        self.tee_store();
    }
}

impl hyper::body::Body for AiStreamBody {
    type Data = Bytes;
    type Error = super::proxy::ProxyBodyError;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<hyper::body::Frame<Bytes>, super::proxy::ProxyBodyError>>> {
        let this = self.get_mut();
        loop {
            // Drain synthesized terminal frames first: one data frame
            // per poll (flushes reach the client per frame).
            if let Some(text) = this.tail.pop_front() {
                if this.tail.is_empty() && this.translator.is_ended() {
                    // The last terminal frame: close after it is
                    // handed over. The NEXT poll returns None.
                    this.tee_collect(std::slice::from_ref(&text));
                    this.close();
                }
                return Poll::Ready(Some(Ok(hyper::body::Frame::data(Bytes::from(text)))));
            }
            if this.translator.is_ended() {
                return Poll::Ready(None);
            }
            // The provider body is gone once the budget cutoff dropped
            // it (eager upstream cancel): the synthesized tail above is
            // the stream's entire remainder.
            let Some(inner) = this.inner.as_mut() else {
                return Poll::Ready(None);
            };
            match Pin::new(inner).poll_frame(cx) {
                Poll::Ready(Some(Ok(frame))) => {
                    let Some(data) = frame.data_ref().cloned() else {
                        continue; // trailers / non-data frames pass by
                    };
                    let (frames, chunks, first) = this.translator.feed(data.as_ref(), this.adapter);
                    this.note_chunks(chunks, first);
                    this.budget_tick();
                    if this.translator.is_ended() {
                        this.tail.extend(this.translator.finish());
                    }
                    if frames.is_empty() {
                        // Bytes only continued a partial frame; keep
                        // polling (or drain the tail next loop).
                        continue;
                    }
                    let mut joined = String::new();
                    for f in &frames {
                        joined.push_str(f);
                    }
                    // DW-082: banned-content guardrail check on the
                    // forwarded batch text. A match cuts the stream
                    // off (the tail is synthesized above); the
                    // already-joined batch is still forwarded (the
                    // match is in THIS batch — the cutoff prevents
                    // FURTHER content, not retroactive suppression).
                    this.guardrail_tick(&joined);
                    if this.guardrail_cut_off {
                        // The guardrail fired: forward this batch
                        // (already-joined) then the tail (error +
                        // [DONE]) drains on subsequent polls. The
                        // provider body was dropped by guardrail_tick.
                        this.tee_collect(&frames);
                        return Poll::Ready(Some(Ok(hyper::body::Frame::data(Bytes::from(
                            joined,
                        )))));
                    }
                    this.tee_collect(&frames);
                    return Poll::Ready(Some(Ok(hyper::body::Frame::data(Bytes::from(joined)))));
                }
                Poll::Ready(Some(Err(e))) => {
                    // Mid-stream provider abort: already-forwarded
                    // content stands; the client stream ends cleanly
                    // with an error chunk and the terminator.
                    tracing::warn!(
                        code = "ai_stream_aborted",
                        provider = %this.gauges.provider,
                        "provider stream aborted mid-flight: {e}"
                    );
                    this.tail.extend(
                        this.translator
                            .abort_tail("the model provider closed the stream"),
                    );
                    continue;
                }
                Poll::Ready(None) => {
                    // Clean end: flush any unterminated frame, then
                    // the terminal usage chunk and the terminator.
                    let (frames, chunks, first) = this.translator.flush_partial(this.adapter);
                    this.note_chunks(chunks, first);
                    let mut joined = String::new();
                    for f in frames {
                        joined.push_str(&f);
                    }
                    this.tail.extend(this.translator.finish());
                    if joined.is_empty() {
                        continue;
                    }
                    return Poll::Ready(Some(Ok(hyper::body::Frame::data(Bytes::from(joined)))));
                }
                Poll::Pending => return Poll::Pending,
            }
        }
    }

    fn is_end_stream(&self) -> bool {
        false
    }

    fn size_hint(&self) -> hyper::body::SizeHint {
        hyper::body::SizeHint::default()
    }
}

impl Drop for AiStreamBody {
    fn drop(&mut self) {
        // A client that hangs up mid-stream still owes its terminal
        // metrics (duration histogram; usage if reported).
        self.close();
    }
}