pv_harness/

pipeline.rs

//! Ordered hot-path middleware and upstream forwarding.

use crate::config::HarnessConfig;
use crate::reconciler::Finalizer;
use crate::session::{Session, SessionLease, SessionRegistry, Workflow};
use crate::worker::{AtifCapture, ResponsePermit, WorkerHandle, WorkerJob};
use axum::body::Bytes;
use axum::http::{HeaderMap, HeaderName, HeaderValue, StatusCode};
use futures::StreamExt as _;
use pv_bridge::EventBus;
use pv_core::metrics::Registry;
use pv_core::time::elapsed_us;
use pv_events::{AgentIdentity, EventClass, EventMetrics, StatusId, StopReason};
use pv_identity::IdentityValidator;
use pv_loopdetect::{BreakerAction, BreakerState, Embedder, HashEmbedder, NoopVectorSink, VectorSink};
use pv_receipts::Signer;
use pv_sandbox::{Sandbox, ToolVerdict};
use pv_state::{ActionBudget, BudgetDecision, StateStore};
use serde_json::Value;
use std::sync::Arc;
use std::time::Instant;

/// Shared HELP text for the supervised-loop counters that are
/// registered TWICE: eagerly at boot (this module — so `rate() > 0`
/// alerts can see the FIRST fire) and lazily at the observation site
/// (main.rs / reconciler.rs panic arms). The metrics registry is
/// first-wins on HELP and warns on divergence in debug builds; these
/// six pairs had drifted into permanent per-tick warn spam (R285).
/// One const per metric, referenced from both sites.
pub mod supervision_help {
    /// HELP for `pv_reconciler_panics_total` (boot + panic-arm sites).
    pub const RECONCILER_PANICS: &str = "Reconciler tick body panicked and was caught; the reconciler \
         continues on the next tick";
    /// HELP for `pv_bridge_maintenance_panics_total` (boot + panic-arm sites).
    pub const BRIDGE_MAINTENANCE_PANICS: &str = "Bridge maintenance tick panicked and was caught; the loop \
         continues on the next interval";
    /// HELP for `pv_bridge_maintenance_errors_total` (boot + error-arm sites).
    pub const BRIDGE_MAINTENANCE_ERRORS: &str =
        "Bridge maintenance tick returned an error; the loop continues \
         on the next interval";
    /// HELP for `pv_bridge_maintenance_join_errors_total` (boot + join-error sites).
    pub const BRIDGE_MAINTENANCE_JOIN_ERRORS: &str =
        "Bridge maintenance tick's spawned task failed to join (panic \
         or cancellation); the loop continues on the next interval";
    /// HELP for `pv_atif_retention_panics_total` (boot + panic-arm sites).
    pub const ATIF_RETENTION_PANICS: &str = "ATIF retention sweep tick panicked and was caught; the loop \
         continues on the next hour";
    /// HELP for `pv_atif_retention_errors_total` (boot + error-arm sites).
    pub const ATIF_RETENTION_ERRORS: &str =
        "ATIF retention sweep tick returned an error; the loop continues \
         on the next hour";
}

pub(crate) const SESSION_HEADER: &str = "x-pv-session";
const WORKFLOW_HEADER: &str = "x-pv-workflow";
pub(crate) const MIDDLEWARE_US_HEADER: &str = "x-pv-middleware-us";

/// TCP connect timeout for every outbound HTTPS client the harness builds.
pub const HTTP_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);

/// Shared HELP for `pv_shutdown_session_close_timeouts_total`, registered
/// at boot (pre-registration in the pipeline's metric block) and
/// looked up again on the shutdown path (`main.rs` finalize loop). The
/// registry is first-wins on HELP and warns on drift — a single const is
/// the only arrangement that cannot drift.
pub const SESSION_CLOSE_TIMEOUTS_HELP: &str =
    "Per-session close hit the shutdown-time per-session deadline (3 s) and \
     was deferred to restart-time spool recovery. A sustained rate > 0 on \
     every rollout indicates a class of sessions that regularly hang their \
     close (leaked leases, dropped worker permits, unresponsive bridge \
     publish) — the coincident session id in the shutdown warn log is the \
     correlation key. Distinct from `pv_http_shutdown_drain_timeouts_total` \
     which fires on the OUTER phase timeout.";

/// Shared HELP for `pv_shutdown_mcp_drain_timeouts_total` (boot
/// pre-registration and shutdown-path lookup in `main.rs`). Same
/// first-wins/no-drift rationale as [`SESSION_CLOSE_TIMEOUTS_HELP`].
pub const MCP_DRAIN_TIMEOUTS_HELP: &str =
    "Shutdown MCP-inflight drain hit its 5 s deadline before every detached \
     mcp_call_inner spawn completed. Downstream effects: some detached \
     tool bodies were interrupted at close and their sessions may be \
     quarantined at restart-time recovery. Any occurrence during a \
     rollout indicates a class of tool calls that regularly outlive HTTP \
     drain (long-running upstream tool, hung sandbox); check the \
     coincident tracing warn line.";

/// Shared HELP for `pv_idle_close_timeouts_total` (boot pre-registration
/// and reconciler idle-sweep lookup). Same first-wins/no-drift rationale
/// as [`SESSION_CLOSE_TIMEOUTS_HELP`].
pub const IDLE_CLOSE_TIMEOUTS_HELP: &str =
    "Idle-close reached the per-session deadline (90 s) and returned to \
     the next tick — indicates a session with an active lease that never \
     drops (stuck stream, hung worker) or a bridge publish stalled behind \
     an unresponsive broker. A steady rate > 0 requires operator \
     investigation of the coincident session id.";

/// Fallback identity scope when the tool name cannot be parsed from the request.
pub(crate) const TOOL_INVOKE_SCOPE: &str = "tool:invoke";

/// Identity scope required to invoke a specific tool by name.
pub(crate) fn tool_scope(tool: &str) -> String {
    format!("tool:{tool}")
}

/// Shared application state passed through HTTP handlers and background tasks.
#[derive(Clone)]
pub struct AppState {
    /// Versioned harness configuration.
    pub config: Arc<HarnessConfig>,
    /// Atomic quota and action-budget state.
    pub store: Arc<dyn StateStore>,
    /// MCP schema, policy, and action-budget sandbox.
    pub sandbox: Arc<Sandbox>,
    /// Event Bridge backend used by asynchronous workers and by lifecycle
    /// finalization (close/promote publish through it inline).
    pub bridge: Arc<dyn EventBus>,
    /// Live session registry.
    pub sessions: Arc<SessionRegistry>,
    /// Bounded non-blocking worker queue.
    pub worker: WorkerHandle,
    /// Optional NHI validator. Required in production identity mode.
    pub identity: Option<Arc<IdentityValidator>>,
    /// Separate revocation namespace for tokens signed by this gateway.
    pub exchanged_revocations: Arc<dyn pv_identity::RevocationStore>,
    pub(crate) revocation_audit_window: Arc<parking_lot::Mutex<(Instant, u32)>>,
    /// Prometheus-compatible metrics registry.
    pub metrics: Arc<Registry>,
    /// Pre-resolved metric handles for the request hot
    /// path so `observe_stage` doesn't take the registry mutex once
    /// per stage per request. Same series names as the lazy path;
    /// pre-registered at boot in `new_with_backends_and_metrics`.
    pub(crate) hot_metrics: Arc<HotMetrics>,
    /// Reused upstream HTTP client.
    pub client: reqwest::Client,
    /// HTTP client for tool backends (private addresses only when
    /// `require_private_backends` is on; otherwise the shared client).
    pub(crate) backend_client: reqwest::Client,
    /// Static credential injected into every chat-completions forward
    /// (resolved once at startup; value is marked sensitive).
    pub(crate) upstream_auth: Option<(HeaderName, HeaderValue)>,
    /// Bearer credential injected into every tool-upstream forward.
    pub(crate) tool_auth: Option<HeaderValue>,
    /// Serializes tool-audit completion per execution key (see
    /// `routes.rs::complete_tool_audit`): an Unaudited replay racing a
    /// fresh completion for the same key would otherwise submit a second
    /// `tool_completed` worker job — a duplicate execution record on the
    /// audit stream and in the signed chain.
    pub(crate) tool_audit_gates:
        Arc<parking_lot::Mutex<std::collections::HashMap<String, Arc<tokio::sync::Mutex<()>>>>>,
    /// Execution keys whose `tool_completed` audit job has been durably
    /// journaled in this process but whose `.audited` marker write failed;
    /// a retry must re-attempt the marker without re-emitting the event.
    pub(crate) tool_audits_emitted: Arc<parking_lot::Mutex<std::collections::HashSet<String>>>,
    /// The upstream provider's wire dialect, selected by
    /// `config.provider` at boot. `forward_chat` translates requests
    /// and responses for the non-OpenAI dialects so the relay always
    /// sees OpenAI-shaped bytes.
    pub(crate) provider: crate::provider::ProviderKind,
    /// Asynchronous session close and promotion service.
    pub finalizer: Finalizer,
    /// Flips to `true` when shutdown starts. `/readyz` reads this and
    /// reports 503 so a Kubernetes readinessProbe (or any external LB)
    /// stops routing new traffic to a draining pod before axum's
    /// graceful drain begins.
    pub draining: Arc<std::sync::atomic::AtomicBool>,
    /// Sliding one-minute budget for FULL identity-
    /// rejection audit records (window start, records written).
    /// Rejections beyond the budget are counted only — see
    /// `enqueue_transient_failure`.
    pub(crate) identity_rejection_window: Arc<parking_lot::Mutex<(Instant, u32)>>,
    pub(crate) journal_key: [u8; 32],
    /// Tracks in-flight `mcp_call_inner` detached spawns so
    /// `finish_shutdown` can await their completion between HTTP drain
    /// and `worker.wait_idle`. Without this barrier, a client-
    /// disconnected `mcp_call` running past axum's graceful drain
    /// could reach `worker.try_submit` AFTER `wait_idle` returned,
    /// stranding a fresh job across `finalize_sessions` and tripping
    /// `pv_shutdown_session_close_timeouts_total` on an otherwise-
    /// recoverable session.
    pub mcp_inflight: Arc<crate::inflight::InflightTracker>,
    /// Per-principal token-rate limits pushed from the console
    /// (`policy_bundle_file`); empty until a bundle with rate policies is
    /// applied.
    pub pushed_rates: Arc<crate::policy_push::RateLimits>,
    /// Admission semaphore bounding concurrent `mcp_call_inner`
    /// executions to `config.mcp_concurrency`. R70 F1 (landed R71):
    /// without this limit a slow-tool-upstream attack + N
    /// concurrent MCP callers each buffering up to
    /// `MAX_TOOL_RESPONSE_BYTES = 16 MiB` in
    /// `read_limited_tool_response` stacked N × 16 MiB unbounded.
    /// `mcp_inflight` above is a shutdown-drain COUNTER — not a
    /// limiter. Chat path already has admission via
    /// `worker.try_reserve_pair`; MCP was the odd one out.
    ///
    /// Refused admissions increment `pv_mcp_admission_refusals_
    /// total` (see mcp_call in routes.rs) and respond 503 with a
    /// Retry-After header — 503 not 429 because the refusal is a
    /// server-side capacity signal (matches breaker-open 503-not-
    /// 429 discipline documented in finalize_error_response.rs).
    pub mcp_admission: Arc<tokio::sync::Semaphore>,
    /// Pre-resolved `Arc<Counter>` handle for
    /// `pv_mcp_admission_refusals_total`. R72 review F2 (landed
    /// R74): the R72 eager registration removed only
    /// `HashMap::insert` + `String::to_owned` from the refusal-
    /// arm cost; the `.counter(NAME, HELP)` lookup itself still
    /// took `metrics.lock()` (a global parking_lot::Mutex across
    /// the entire registry) on every refusal. Storing the handle
    /// here — matching the HotMetrics discipline — lets the
    /// refusal arm call `.inc()` directly with zero registry
    /// contention.
    pub(crate) mcp_admission_refusals_counter: Arc<pv_core::metrics::Counter>,
    /// Per-backend credential routing (pillar 4). Resolves tool names to
    /// their designated backend and attaches the pre-resolved credential.
    pub backend_router: Arc<crate::backend::BackendRouter>,
    /// Local policy decision point (pillar 2). Maps tool names to business
    /// intents and enforces mission constraints per call.
    pub pdp: Arc<crate::authz::PolicyDecisionPoint>,
    /// Token signer (pillar 3). Signs exchange tokens and intent tokens
    /// with EdDSA. Present only when `token_exchange_seed_file` is configured.
    pub token_signer: Option<Arc<crate::authz::TokenSigner>>,
    /// Redaction engine (pillar 5). Strips sensitive data from event
    /// payloads and ATIF step text before they are journaled or published.
    pub redaction: Option<Arc<pv_redact::RedactionEngine>>,
    /// MCP Streamable HTTP client and per-agent-session backend sessions
    /// for backends with `transport = "mcp"`.
    pub mcp_backends: Arc<crate::mcp_client::McpBackendClient>,
    /// Signs and verifies the `MCP-Session-Id` values this gateway issues.
    pub(crate) mcp_sessions: Arc<crate::mcp::SessionSigner>,
    /// External AuthZEN policy decision point, when `[authzen]` is set.
    pub authzen: Option<Arc<crate::authzen::AuthzenClient>>,
    /// Content sink slot shared with the audit worker (tenant telemetry).
    pub(crate) content_sink: crate::content::SharedContentSink,
    /// Workload assertion verifier, when `[workload_identity]` is set.
    pub(crate) workload: crate::workload::SharedWorkloadIssuer,
}

/// Pre-resolved metric handles for the request hot path so the
/// per-stage histogram observation doesn't take the registry mutex.
/// Same series names as the lazy path — the label suffix matches
/// `pv_stage_duration_seconds{stage="…"}` exactly.
///
/// Ordering of the `[Stage; 5]` arrays mirrors the boot registration
/// in `new_with_backends_and_metrics`: identity, quota, sanitize,
/// compression, dispatch. `Stage::index()` returns the offset so
/// `observe_stage` is O(1) with no string alloc on the hot path.
pub(crate) struct HotMetrics {
    pub(crate) stage_histograms: [Arc<pv_core::metrics::Histogram>; 5],
    pub(crate) stage_strict_budget_counters: [Arc<pv_core::metrics::Counter>; 5],
    /// `config.strict_stage_budget || PV_STRICT_BUDGET`
    /// resolved ONCE at construction. `std::env::var` was previously
    /// called per stage per request because `||` cannot short-circuit
    /// a false config default.
    pub(crate) strict_stage_budget: bool,
    /// Pre-resolved `pv_requests_total{route,status_class}` counters,
    /// indexed by `(Route::index(), StatusClass::index())`. Populated
    /// at boot; every hit is O(1) with zero string alloc and zero
    /// registry-mutex acquisition — R58 audit found the previous
    /// per-request `metrics.counter(&format!("..."))` path did 4
    /// mutex ops on the shared `Registry` per HTTP request plus
    /// ~120 B of allocator churn (at 10k req/s: ~40k mutex ops/s +
    /// 1.2 MB/s allocs on the request middleware alone).
    pub(crate) request_counters: [[Arc<pv_core::metrics::Counter>; 5]; 15],
    /// Pre-resolved `pv_request_duration_seconds{route}` histograms,
    /// indexed by `Route::index()`. Same rationale as
    /// `request_counters` above; the histogram fetch used to hit
    /// TWO Registry mutex ops per request (base_kinds + metrics
    /// map).
    pub(crate) request_duration_histograms: [Arc<pv_core::metrics::Histogram>; 15],
    /// Pre-resolved `pv_upstream_latency_seconds` histogram — hit
    /// on every successful chat forward. R59 continuation of R58's
    /// hot-path pre-resolution: prior code called
    /// `metrics.histogram_with_bounds("pv_upstream_latency_seconds",
    /// ...)` per request, doing 2 Registry mutex ops per chat.
    pub(crate) upstream_latency_histogram: Arc<pv_core::metrics::Histogram>,
    /// Pre-resolved `pv_upstream_errors_total{kind}` counters,
    /// indexed by `UpstreamErrorKind::index()`. Prior code called
    /// `metrics.counter(&format!("pv_upstream_errors_total{...}"))`
    /// per error, doing 2 Registry mutex ops + a String alloc per
    /// upstream failure. Kinds order matches boot pre-registration.
    pub(crate) upstream_errors_counters: [Arc<pv_core::metrics::Counter>; 4],
    /// Pre-resolved `pv_dashboard_requests_total{endpoint,status}`
    /// counters, indexed by `(DashboardEndpoint::index(),
    /// DashboardStatus::index())`. Same rationale as
    /// `request_counters`; the dashboard is lower volume but
    /// hits the same shared Registry mutex.
    pub(crate) dashboard_request_counters: [[Arc<pv_core::metrics::Counter>; 2]; 3],
    /// Pre-resolved `pv_dashboard_request_duration_seconds{endpoint}`
    /// histograms, indexed by `DashboardEndpoint::index()`.
    pub(crate) dashboard_request_duration_histograms: [Arc<pv_core::metrics::Histogram>; 3],
}

#[derive(Copy, Clone, Debug)]
pub(crate) enum Stage {
    Identity,
    Quota,
    Sanitize,
    Compression,
    Dispatch,
}

impl Stage {
    pub(crate) const ORDER: [Stage; 5] = [
        Stage::Identity,
        Stage::Quota,
        Stage::Sanitize,
        Stage::Compression,
        Stage::Dispatch,
    ];

    pub(crate) fn label(self) -> &'static str {
        match self {
            Stage::Identity => "identity",
            Stage::Quota => "quota",
            Stage::Sanitize => "sanitize",
            Stage::Compression => "compression",
            Stage::Dispatch => "dispatch",
        }
    }

    pub(crate) fn index(self) -> usize {
        match self {
            Stage::Identity => 0,
            Stage::Quota => 1,
            Stage::Sanitize => 2,
            Stage::Compression => 3,
            Stage::Dispatch => 4,
        }
    }

    fn from_label(label: &str) -> Option<Self> {
        Stage::ORDER.into_iter().find(|s| s.label() == label)
    }
}

/// HTTP route label enum. Mirrors the closed vocabulary in
/// `route_label(path)` — pre-resolved into a fixed-index array of
/// `Arc<Counter>` / `Arc<Histogram>` so the request-metrics
/// middleware never touches the shared `Registry` mutex on the hot
/// path (see `HotMetrics::request_counters`).
#[derive(Copy, Clone, Debug)]
pub(crate) enum Route {
    Chat,
    Mcp,
    Health,
    Livez,
    Readyz,
    Metrics,
    SessionClose,
    SessionPromote,
    Dashboard,
    TokenExchange,
    Revoke,
    Introspect,
    AdminRevocation,
    Jwks,
    Other,
}

impl Route {
    pub(crate) const ORDER: [Route; 15] = [
        Route::Chat,
        Route::Mcp,
        Route::Health,
        Route::Livez,
        Route::Readyz,
        Route::Metrics,
        Route::SessionClose,
        Route::SessionPromote,
        Route::Dashboard,
        Route::TokenExchange,
        Route::Revoke,
        Route::Introspect,
        Route::AdminRevocation,
        Route::Jwks,
        Route::Other,
    ];

    pub(crate) fn label(self) -> &'static str {
        match self {
            Route::Chat => "chat",
            Route::Mcp => "mcp",
            Route::Health => "health",
            Route::Livez => "livez",
            Route::Readyz => "readyz",
            Route::Metrics => "metrics",
            Route::SessionClose => "session_close",
            Route::SessionPromote => "session_promote",
            Route::Dashboard => "dashboard",
            Route::TokenExchange => "token_exchange",
            Route::Revoke => "revoke",
            Route::Introspect => "introspect",
            Route::AdminRevocation => "admin_revocation",
            Route::Jwks => "jwks",
            Route::Other => "other",
        }
    }

    pub(crate) fn index(self) -> usize {
        match self {
            Route::Chat => 0,
            Route::Mcp => 1,
            Route::Health => 2,
            Route::Livez => 3,
            Route::Readyz => 4,
            Route::Metrics => 5,
            Route::SessionClose => 6,
            Route::SessionPromote => 7,
            Route::Dashboard => 8,
            Route::TokenExchange => 9,
            Route::Revoke => 10,
            Route::Introspect => 11,
            Route::AdminRevocation => 12,
            Route::Jwks => 13,
            Route::Other => 14,
        }
    }
}

/// HTTP status-class label. Mirrors the closed vocabulary in the
/// request-metrics middleware. Kept as a separate enum so future
/// additions (e.g., a hypothetical `2xx-slow`) surface via
/// exhaustive-match compile checks.
#[derive(Copy, Clone, Debug)]
pub(crate) enum StatusClass {
    Informational, // 1xx
    Success,       // 2xx
    Redirection,   // 3xx
    ClientError,   // 4xx
    ServerError,   // 5xx
}

impl StatusClass {
    pub(crate) const ORDER: [StatusClass; 5] = [
        StatusClass::Informational,
        StatusClass::Success,
        StatusClass::Redirection,
        StatusClass::ClientError,
        StatusClass::ServerError,
    ];

    pub(crate) fn label(self) -> &'static str {
        match self {
            StatusClass::Informational => "1xx",
            StatusClass::Success => "2xx",
            StatusClass::Redirection => "3xx",
            StatusClass::ClientError => "4xx",
            StatusClass::ServerError => "5xx",
        }
    }

    pub(crate) fn index(self) -> usize {
        match self {
            StatusClass::Informational => 0,
            StatusClass::Success => 1,
            StatusClass::Redirection => 2,
            StatusClass::ClientError => 3,
            StatusClass::ServerError => 4,
        }
    }

    pub(crate) fn from_u16(status: u16) -> Self {
        match status {
            100..=199 => StatusClass::Informational,
            200..=299 => StatusClass::Success,
            300..=399 => StatusClass::Redirection,
            400..=499 => StatusClass::ClientError,
            _ => StatusClass::ServerError,
        }
    }
}

/// Upstream error classifier used for `pv_upstream_errors_total{kind}`.
/// Mirrors `classify_upstream_error(err)` — same set as the boot
/// pre-registration in `Pipeline::new` at pipeline.rs:511.
#[derive(Copy, Clone, Debug)]
pub(crate) enum UpstreamErrorKind {
    Timeout,
    Connect,
    Send,
    Http5xx,
}

impl UpstreamErrorKind {
    pub(crate) const ORDER: [UpstreamErrorKind; 4] = [
        UpstreamErrorKind::Timeout,
        UpstreamErrorKind::Connect,
        UpstreamErrorKind::Send,
        UpstreamErrorKind::Http5xx,
    ];

    pub(crate) fn label(self) -> &'static str {
        match self {
            UpstreamErrorKind::Timeout => "timeout",
            UpstreamErrorKind::Connect => "connect",
            UpstreamErrorKind::Send => "send",
            UpstreamErrorKind::Http5xx => "http_5xx",
        }
    }

    pub(crate) fn index(self) -> usize {
        match self {
            UpstreamErrorKind::Timeout => 0,
            UpstreamErrorKind::Connect => 1,
            UpstreamErrorKind::Send => 2,
            UpstreamErrorKind::Http5xx => 3,
        }
    }

    /// Convert `classify_upstream_error`'s `&'static str` output into
    /// a typed variant. The classifier at pipeline.rs:2736 already
    /// exhaustively covers this set; this fn is the typed mirror.
    pub(crate) fn from_reqwest(error: &reqwest::Error) -> Self {
        if error.is_timeout() {
            UpstreamErrorKind::Timeout
        } else if error.is_connect() {
            UpstreamErrorKind::Connect
        } else {
            UpstreamErrorKind::Send
        }
    }
}

/// Dashboard endpoint label. Mirrors the `record()` call sites in
/// dashboard.rs (list, detail, stats).
#[derive(Copy, Clone, Debug)]
pub(crate) enum DashboardEndpoint {
    Stats,
    List,
    Detail,
}

impl DashboardEndpoint {
    pub(crate) const ORDER: [DashboardEndpoint; 3] = [
        DashboardEndpoint::Stats,
        DashboardEndpoint::List,
        DashboardEndpoint::Detail,
    ];

    pub(crate) fn label(self) -> &'static str {
        match self {
            DashboardEndpoint::Stats => "stats",
            DashboardEndpoint::List => "list",
            DashboardEndpoint::Detail => "detail",
        }
    }

    pub(crate) fn index(self) -> usize {
        match self {
            DashboardEndpoint::Stats => 0,
            DashboardEndpoint::List => 1,
            DashboardEndpoint::Detail => 2,
        }
    }
}

/// Dashboard endpoint response status label. Mirrors the `record()`
/// call sites (`ok`, `not_found`) in dashboard.rs.
#[derive(Copy, Clone, Debug)]
pub(crate) enum DashboardStatus {
    Ok,
    NotFound,
}

impl DashboardStatus {
    pub(crate) const ORDER: [DashboardStatus; 2] = [DashboardStatus::Ok, DashboardStatus::NotFound];

    pub(crate) fn label(self) -> &'static str {
        match self {
            DashboardStatus::Ok => "ok",
            DashboardStatus::NotFound => "not_found",
        }
    }

    pub(crate) fn index(self) -> usize {
        match self {
            DashboardStatus::Ok => 0,
            DashboardStatus::NotFound => 1,
        }
    }
}

impl HotMetrics {
    fn new(metrics: &Registry, config_strict_stage_budget: bool) -> Self {
        // Registrations here mirror the boot loop below. Using the
        // same `metrics.histogram_with_bounds` / `metrics.counter`
        // getters means these entries are shared with the lazy path
        // — a caller that still uses `metrics.counter(name).inc()`
        // hits the exact same Arc<Counter> we cache here.
        let stage_histograms: [Arc<pv_core::metrics::Histogram>; 5] = Stage::ORDER.map(|stage| {
            metrics.histogram_with_bounds(
                &format!("pv_stage_duration_seconds{{stage=\"{}\"}}", stage.label()),
                "Harness stage latency",
                pv_core::metrics::WIDE_LATENCY_BOUNDS_US,
            )
        });
        let stage_strict_budget_counters: [Arc<pv_core::metrics::Counter>; 5] = Stage::ORDER.map(|stage| {
            metrics.counter(
                &format!("pv_strict_budget_breaches_total{{stage=\"{}\"}}", stage.label()),
                "Middleware stages that exceeded the strict per-stage budget",
            )
        });
        // The post-compression quota debit reports under its own label
        // (`quota_post_compression`, not in `Stage::ORDER`) through
        // observe_stage's lazy fallback. Pre-register it here with the
        // SAME wide bounds as the enum stages: first-registration wins
        // in the Registry, so without this the lazy path created it
        // with the default bounds and the one stage doing real backend
        // work (Redis) had incomparable buckets to its siblings.
        let _ = metrics.histogram_with_bounds(
            "pv_stage_duration_seconds{stage=\"quota_post_compression\"}",
            "Harness stage latency",
            pv_core::metrics::WIDE_LATENCY_BOUNDS_US,
        );
        // Pre-resolve the request-metrics middleware's counters +
        // histograms as a [route][status_class] array so the hot
        // path is O(1) with zero string alloc and zero Registry
        // mutex ops. Registration order matches the enum ORDER
        // constants, matched at read time via `Route::index()` and
        // `StatusClass::index()`.
        let request_counters: [[Arc<pv_core::metrics::Counter>; 5]; 15] = Route::ORDER.map(|route| {
            StatusClass::ORDER.map(|status| {
                metrics.counter(
                    &format!(
                        "pv_requests_total{{route=\"{}\",status_class=\"{}\"}}",
                        route.label(),
                        status.label()
                    ),
                    "HTTP requests by route and status class",
                )
            })
        });
        let request_duration_histograms: [Arc<pv_core::metrics::Histogram>; 15] = Route::ORDER.map(|route| {
            metrics.histogram_with_bounds(
                &format!("pv_request_duration_seconds{{route=\"{}\"}}", route.label()),
                "End-to-end HTTP request latency by route",
                pv_core::metrics::WIDE_LATENCY_BOUNDS_US,
            )
        });
        // Pre-resolve upstream latency + error counters — hit on
        // every chat forward. Same rationale as request_counters.
        let upstream_latency_histogram = metrics.histogram_with_bounds(
            "pv_upstream_latency_seconds",
            "Time to upstream response headers",
            pv_core::metrics::WIDE_LATENCY_BOUNDS_US,
        );
        let upstream_errors_counters: [Arc<pv_core::metrics::Counter>; 4] =
            UpstreamErrorKind::ORDER.map(|kind| {
                metrics.counter(
                    &format!("pv_upstream_errors_total{{kind=\"{}\"}}", kind.label()),
                    "Upstream failures by kind",
                )
            });
        // Pre-resolve dashboard endpoint counters + histograms. Same
        // rationale as request_counters; lower volume but hits the
        // same shared Registry mutex.
        let dashboard_request_counters: [[Arc<pv_core::metrics::Counter>; 2]; 3] = DashboardEndpoint::ORDER
            .map(|endpoint| {
                DashboardStatus::ORDER.map(|status| {
                    metrics.counter(
                        &format!(
                            "pv_dashboard_requests_total{{endpoint=\"{}\",status=\"{}\"}}",
                            endpoint.label(),
                            status.label()
                        ),
                        "Dashboard endpoint requests, labeled by status",
                    )
                })
            });
        let dashboard_request_duration_histograms: [Arc<pv_core::metrics::Histogram>; 3] =
            DashboardEndpoint::ORDER.map(|endpoint| {
                metrics.histogram(
                    &format!(
                        "pv_dashboard_request_duration_seconds{{endpoint=\"{}\"}}",
                        endpoint.label()
                    ),
                    "Dashboard endpoint latency",
                )
            });
        Self {
            stage_histograms,
            stage_strict_budget_counters,
            strict_stage_budget: config_strict_stage_budget || truthy_env("PV_STRICT_BUDGET"),
            request_counters,
            request_duration_histograms,
            upstream_latency_histogram,
            upstream_errors_counters,
            dashboard_request_counters,
            dashboard_request_duration_histograms,
        }
    }
}

/// The header name and scheme the upstream credential is sent with.
///
/// `upstream_auth_header` / `upstream_auth_scheme` default to
/// `authorization` / `Bearer`, which is right for OpenAI and every
/// OpenAI-compatible server. The Anthropic and Gemini native APIs read
/// the key from their own headers (`x-api-key`, `x-goog-api-key`)
/// with no scheme prefix, so when the operator left both settings at
/// their defaults the provider's native header is used instead. An
/// explicit non-default setting is always honoured as written.
pub fn effective_upstream_auth_header(config: &HarnessConfig) -> (String, String) {
    let at_defaults = config.upstream_auth_header.eq_ignore_ascii_case("authorization")
        && config.upstream_auth_scheme.eq_ignore_ascii_case("bearer");
    let native = crate::provider::adapter_for(&config.provider)
        .and_then(crate::provider::ProviderKind::native_auth_header);
    match native {
        Some(header) if at_defaults => (header.to_owned(), String::new()),
        _ => (
            config.upstream_auth_header.to_ascii_lowercase(),
            config.upstream_auth_scheme.clone(),
        ),
    }
}

/// How the harness authenticates to the chat upstream, for startup logs
/// and `pvctl doctor` — never contains key material.
pub fn describe_upstream_auth(config: &HarnessConfig) -> String {
    let (header, _) = effective_upstream_auth_header(config);
    if config.upstream_authorization_passthrough {
        "passthrough(client Authorization)".to_owned()
    } else if let Some(env) = config.upstream_api_key_env.as_deref() {
        format!("api-key from ${env} in header {header:?}")
    } else if let Some(file) = config.upstream_api_key_file.as_deref() {
        format!("api-key from file {file} in header {header:?}")
    } else {
        "none".to_owned()
    }
}

/// Resolve the configured upstream credential into a ready header pair.
///
/// Key material is accepted only from an environment variable or an
/// owner-only file — never from TOML or argv — and the resulting header
/// value is marked sensitive so `Debug` output redacts it.
pub(crate) fn resolve_upstream_auth(
    config: &HarnessConfig,
) -> Result<Option<(HeaderName, HeaderValue)>, PipelineError> {
    let key = match read_secret(
        config.upstream_api_key_env.as_deref(),
        config.upstream_api_key_file.as_deref(),
        "upstream API key",
    )? {
        Some(key) => key,
        None => return Ok(None),
    };
    let (header, scheme) = effective_upstream_auth_header(config);
    let name = HeaderName::try_from(header.as_str()).map_err(|_| {
        PipelineError::upstream(format!(
            "upstream_auth_header {header:?} is not a valid header name"
        ))
    })?;
    let rendered = if scheme.is_empty() {
        key
    } else {
        format!("{scheme} {key}")
    };
    let mut value = HeaderValue::from_str(&rendered).map_err(|_| {
        PipelineError::upstream(
            "upstream API key contains bytes that cannot appear in an HTTP header".to_owned(),
        )
    })?;
    value.set_sensitive(true);
    Ok(Some((name, value)))
}

/// Resolve the optional tool-upstream bearer token.
pub(crate) fn resolve_tool_auth(config: &HarnessConfig) -> Result<Option<HeaderValue>, PipelineError> {
    let token = match read_secret(
        config.tool_upstream_bearer_env.as_deref(),
        config.tool_upstream_bearer_file.as_deref(),
        "tool upstream bearer token",
    )? {
        Some(token) => token,
        None => return Ok(None),
    };
    let mut value = HeaderValue::from_str(&format!("Bearer {token}")).map_err(|_| {
        PipelineError::upstream(
            "tool upstream bearer token contains bytes that cannot appear in an HTTP header".to_owned(),
        )
    })?;
    value.set_sensitive(true);
    Ok(Some(value))
}

/// Read a secret from an env var or an owner-only file, trimming
/// surrounding whitespace. A configured-but-missing source is a loud startup error:
/// silently proxying unauthenticated would produce baffling upstream 401s.
pub(crate) fn read_secret(
    env_name: Option<&str>,
    file_path: Option<&str>,
    what: &str,
) -> Result<Option<String>, PipelineError> {
    read_secret_from(|name| std::env::var(name).ok(), env_name, file_path, what)
}

/// Testable core of [`read_secret`] with an injected environment.
fn read_secret_from(
    get_env: impl Fn(&str) -> Option<String>,
    env_name: Option<&str>,
    file_path: Option<&str>,
    what: &str,
) -> Result<Option<String>, PipelineError> {
    if let Some(name) = env_name {
        let value = get_env(name).ok_or_else(|| {
            PipelineError::upstream(format!(
                "{what}: environment variable {name} is not set (export it or update the config)"
            ))
        })?;
        let value = value.trim().to_owned();
        if value.is_empty() {
            return Err(PipelineError::upstream(format!(
                "{what}: environment variable {name} is set but empty"
            )));
        }
        return Ok(Some(value));
    }
    if let Some(path) = file_path {
        require_owner_only_secret(std::path::Path::new(path))
            .map_err(|error| PipelineError::upstream(format!("{what}: {error}")))?;
        let value = std::fs::read_to_string(path).map_err(|error| PipelineError::Upstream {
            context: format!("{what}: read {path}: {error}"),
            source: Some(Box::new(error)),
        })?;
        let value = value.trim().to_owned();
        if value.is_empty() {
            return Err(PipelineError::upstream(format!("{what}: file {path} is empty")));
        }
        return Ok(Some(value));
    }
    Ok(None)
}

/// Same posture as the signing-seed loader: refuse symlinks (a pre-planted
/// link would fool the mode check, CWE-59) and group/other-readable modes
/// on Unix. Windows deployments rely on operator-set ACLs.
pub(crate) fn require_owner_only_secret(path: &std::path::Path) -> Result<(), String> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt as _;
        let metadata = std::fs::symlink_metadata(path)
            .map_err(|error| format!("stat secret file {}: {error}", path.display()))?;
        let file_type = metadata.file_type();
        if file_type.is_symlink() {
            return Err(format!(
                "secret file {} is a symbolic link; refusing to follow",
                path.display()
            ));
        }
        if !file_type.is_file() {
            return Err(format!("secret file {} is not a regular file", path.display()));
        }
        let mode = metadata.mode() & 0o777;
        if mode & 0o077 != 0 {
            return Err(format!(
                "secret file {} has mode 0o{mode:03o}; must be owner-only (chmod 600 {})",
                path.display(),
                path.display()
            ));
        }
    }
    #[cfg(not(unix))]
    {
        let _ = path;
    }
    Ok(())
}

/// A request after all local hot-path gates have passed.
pub struct PreparedRequest {
    /// Session bound to this request.
    pub session: Arc<Session>,
    /// Identity validated for this request.
    pub identity: AgentIdentity,
    /// Payload forwarded to the upstream provider after compression.
    pub payload: Value,
    /// Total local middleware time before upstream I/O.
    pub middleware_us: u64,
    /// Prompt tokens debited from the session and principal budgets at
    /// admission (`ActionBudget::try_tokens`), held as an RAII guard so
    /// BOTH a provably-not-forwarded explicit abort and a cancelled
    /// request future (client disconnect at any pre-dispatch await)
    /// refund exactly what was debited. Disarmed at upstream dispatch.
    admission_debit: AdmissionDebit,
    // Fields drop in declaration order. `capture_guard` MUST drop before
    // `lease` on a cancelled request future — the lease's Drop notifies
    // `wait_for_streams`, unblocking `close_session_locked`, which then
    // checks `wait_for_worker_jobs()`. If the guard submits its terminal
    // job after the lease releases, the close's `pending_jobs == 0` load
    // can win the race, finalize proceeds, and the guard's job lands
    // after the receipt was sealed (chain/receipt divergence for signed
    // sessions; a resurrected step journal after remove_step_journal for
    // unsigned). AbortFinalizingStream preserves the same invariant by
    // convention (its Drop runs before `_lease` drops).
    capture_guard: ResponseCaptureGuard,
    lease: SessionLease,
    /// Client `Authorization` header captured for passthrough mode only.
    client_authorization: Option<HeaderValue>,
    /// Provider-semantic client headers relayed upstream verbatim, from
    /// a strict allowlist only (org/project routing and beta gating —
    /// dropping them broke org-scoped keys and beta features for
    /// otherwise drop-in SDK clients). Never auth-bearing.
    upstream_passthrough_headers: Vec<(axum::http::HeaderName, HeaderValue)>,
}

/// Provider response paired with its active session lease.
///
/// For the OpenAI dialect this is the upstream response as received.
/// For the Anthropic and Gemini dialects the status, headers and body
/// have already been translated into the OpenAI shape (a
/// `chat.completion` object, `chat.completion.chunk` SSE frames, or
/// the OpenAI error envelope), so the relay never sees a native frame.
pub struct ForwardedResponse {
    /// HTTP status to relay.
    pub status: StatusCode,
    /// Response headers to relay (after the forwardable-header filter).
    pub headers: HeaderMap,
    /// Response body bytes, OpenAI-shaped.
    pub body: futures::stream::BoxStream<'static, Result<Bytes, std::io::Error>>,
    /// Prompt tokens debited at admission (post-compression heuristic).
    /// The response relay reconciles this against the
    /// provider's reported `usage.prompt_tokens` — correcting the
    /// session totals and refunding the over-charge.
    pub(crate) billed_prompt_tokens: u64,
    // Same ordering invariant as `PreparedRequest`: guard drops before
    // lease so its terminal job registers `pending_jobs` before the
    // close barrier sees the streams drained. The reserved response
    // slot travels inside the guard until routes.rs takes it.
    pub(crate) capture_guard: ResponseCaptureGuard,
    pub(crate) lease: SessionLease,
}

/// The upstream request(s) planned for one prepared chat request.
struct UpstreamPlan {
    /// Full upstream URL.
    url: String,
    /// Request bodies, sent concurrently: one per choice for the
    /// Anthropic dialect (which has no `n`), otherwise exactly one.
    bodies: Vec<Vec<u8>>,
    /// Extra headers every body is sent with (`anthropic-version`, or
    /// the allowlisted OpenAI passthrough headers).
    headers: Vec<(HeaderName, HeaderValue)>,
    /// The credential header, when any.
    auth: Option<(HeaderName, HeaderValue)>,
    /// How to translate the answer back; `None` relays it unchanged.
    translation: Option<Translation>,
}

/// A natively translated dialect.
#[derive(Clone, Copy)]
enum NativeDialect {
    Anthropic,
    Gemini,
}

/// Everything the response translation needs to know about the
/// request it answers.
struct Translation {
    dialect: NativeDialect,
    /// The client asked for `stream: true`.
    stream: bool,
    /// The client asked for `stream_options.include_usage`.
    include_usage: bool,
    /// The OpenAI completion id shared by every chunk/choice.
    id: String,
    /// The requested model, used until the upstream names its own.
    model: String,
    /// Unix seconds for the `created` field.
    created: u64,
}

/// Longest upstream error body read for the OpenAI error envelope.
const MAX_ERROR_BODY_BYTES: usize = 64 * 1024;

/// Longest buffered native response body (matches the relay's own
/// capture cap).
const MAX_NATIVE_BODY_BYTES: usize = 16 * 1024 * 1024;

/// Read a whole response body, refusing anything over `cap` bytes.
async fn read_bounded(response: reqwest::Response, cap: usize) -> Result<Vec<u8>, std::io::Error> {
    let mut body = Vec::new();
    let mut stream = response.bytes_stream();
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.map_err(std::io::Error::other)?;
        if body.len().saturating_add(chunk.len()) > cap {
            return Err(std::io::Error::other(format!(
                "upstream response exceeds {cap} bytes"
            )));
        }
        body.extend_from_slice(&chunk);
    }
    Ok(body)
}

/// RAII owner of the journalled response attempt and (once created) the
/// durable in-flight response marker, from request admission until a
/// downstream owner takes over (`AbortFinalizingStream`, a terminal
/// failure job, or the worker's capture publish).
///
/// Axum drops the request future wholesale when the client disconnects.
/// Every *explicit* failure path already retires the marker and closes
/// the journalled attempt — but a cancelled future runs none of them:
/// the journal kept a dangling non-terminal attempt (crash recovery then
/// quarantines the whole session, see `track_response_attempt`) and the
/// marker stranded on disk (once the session evicts, every reconciler
/// tick re-quarantines its id forever). Drop submits the same terminal
/// failure job the explicit paths use.
pub(crate) struct ResponseCaptureGuard {
    worker: crate::worker::WorkerHandle,
    session: Arc<Session>,
    identity: AgentIdentity,
    marker: Option<String>,
    attempt_id: String,
    /// The response-capture slot reserved at admission. Owned by the
    /// guard so the cancellation path (Drop) submits its terminal
    /// record through the RESERVED slot, exactly like every explicit
    /// failure path — a saturated worker queue must not be able to
    /// poison an admitted session whose slot was pre-paid
    /// (`mark_capture_failed` quarantines the id permanently).
    /// Explicit paths `take_permit()` when they assume ownership.
    permit: Option<ResponsePermit>,
    armed: bool,
}

impl ResponseCaptureGuard {
    fn new(
        worker: crate::worker::WorkerHandle,
        session: Arc<Session>,
        identity: AgentIdentity,
        attempt_id: String,
        permit: Option<ResponsePermit>,
    ) -> Self {
        Self {
            worker,
            session,
            identity,
            marker: None,
            attempt_id,
            permit,
            armed: true,
        }
    }

    /// Hand the reserved response slot to an explicit owner (the
    /// SSE/buffered relay, or a failure path that submits its own
    /// terminal record). The guard's Drop falls back to the plain
    /// worker queue once the permit is taken.
    pub(crate) fn take_permit(&mut self) -> Option<ResponsePermit> {
        self.permit.take()
    }

    /// Record the durable in-flight marker so a later drop retires it.
    pub(crate) fn set_marker(&mut self, marker: String) {
        self.marker = Some(marker);
    }

    pub(crate) fn attempt_id(&self) -> &str {
        &self.attempt_id
    }

    /// Take ownership of the marker and attempt id, disarming the guard.
    /// The caller now owns terminal-record submission.
    pub(crate) fn disarm(mut self) -> (Option<String>, String) {
        self.armed = false;
        (self.marker.take(), std::mem::take(&mut self.attempt_id))
    }

    /// Disarm in place, returning any marker: for callers that submit the
    /// terminal record themselves while retaining the guard (abandon paths).
    fn defuse(&mut self) -> Option<String> {
        self.armed = false;
        self.marker.take()
    }
}

impl Drop for ResponseCaptureGuard {
    fn drop(&mut self) {
        if !self.armed {
            return;
        }
        let job = refused_response_failure_job(
            Arc::clone(&self.session),
            self.identity.clone(),
            "request_cancelled_before_response_capture".to_owned(),
            self.marker.take(),
            std::mem::take(&mut self.attempt_id),
        );
        // Submit through the reserved response slot when the guard
        // still owns it (cancellation before an explicit owner took
        // over) so worker saturation cannot fail this submission; fall
        // back to the plain queue otherwise. Same fallback as every
        // explicit refusal path: if the submission fails, fail the
        // session closed rather than dropping the capture silently.
        let submitted = match self.permit.take() {
            Some(permit) => permit.submit(&self.worker, job),
            None => self.worker.try_submit(job),
        };
        if submitted.is_err() {
            self.session.mark_capture_failed();
        }
    }
}

/// RAII owner of the admission-time token debit (session AND principal
/// ledgers), from the post-compression charge in `prepare_chat` until
/// the request is handed to the upstream HTTP client.
///
/// Every *explicit* pre-dispatch abort refunds via `abandon_prepared` —
/// but axum drops the request future wholesale on client disconnect,
/// running none of them. Without this guard, a cancellation between
/// admission and `send()` (the `spawn_blocking` join in
/// `prepare_chat_nonblocking`, or the awaited marker write in
/// `forward_chat`) leaked the debit on both ledgers; the principal
/// ledger persists across sessions and never heals, so a client that
/// disconnected mid-admission drained its principal budget with zero
/// LLM work.
///
/// [`Self::take_for_dispatch`] disarms the guard at the last
/// pre-dispatch point: from there the request may reach the provider,
/// and the debit conservatively stands (mirroring `forward_chat`'s
/// `is_connect`-only refund on send errors).
pub(crate) struct AdmissionDebit {
    store: Arc<dyn StateStore>,
    session_id: String,
    budget: pv_state::BudgetSpec,
    principal: Option<(String, pv_state::BudgetSpec)>,
    tokens: u64,
    /// Metrics registry for the Drop-path panic supervisor.
    ///
    /// If the refund closure panics inside the spawn_blocking pool (a
    /// backend allocator failure, a parking_lot mutex panic during
    /// unwind, a HashMap grow OOM), tokio's default
    /// `UnhandledPanic::Ignore` swallows it silently. The debit then
    /// stays permanent on BOTH the session ledger and — the material
    /// consequence — the principal ledger, which persists across
    /// sessions and never heals. Surface the panic via a bumped
    /// counter and a tracing::error so `rate(pv_admission_refund_panics_total)`
    /// catches it.
    metrics: Arc<Registry>,
}

impl AdmissionDebit {
    /// Disarm and hand the debited amount to the dispatch path; the
    /// debit stands from here (reconciled against provider usage, or
    /// refunded by the explicit `is_connect` arm).
    fn take_for_dispatch(&mut self) -> u64 {
        std::mem::take(&mut self.tokens)
    }

    /// Refund both ledgers now, in the order they were spent. Zeroed so
    /// a later drop (or a hypothetical double abandon) cannot
    /// double-refund.
    fn refund_now(&mut self) {
        let tokens = std::mem::take(&mut self.tokens);
        if tokens == 0 {
            return;
        }
        ActionBudget::new(self.store.as_ref(), &self.session_id, &self.budget).refund_tokens(tokens);
        if let Some((principal_id, spec)) = self.principal.as_ref() {
            ActionBudget::for_principal(self.store.as_ref(), principal_id, spec).refund_tokens(tokens);
        }
    }
}

impl Drop for AdmissionDebit {
    fn drop(&mut self) {
        if self.tokens == 0 {
            return;
        }
        // A refund is a synchronous store call (a network round-trip on
        // the redis backend). A cancelled request future drops on an
        // async worker thread, where that round-trip would stall every
        // other stream polled there — move it to the blocking pool when
        // a runtime is available (drops on the `spawn_blocking` path of
        // `prepare_chat_nonblocking` already run on blocking threads,
        // and unit tests may have no runtime at all: refund inline).
        let tokens = std::mem::take(&mut self.tokens);
        let store = Arc::clone(&self.store);
        let session_id = std::mem::take(&mut self.session_id);
        let budget = std::mem::take(&mut self.budget);
        let principal = self.principal.take();
        let metrics = Arc::clone(&self.metrics);
        // `catch_unwind` around the refund body so a panic inside
        // spawn_blocking (backend allocator OOM, parking_lot mutex
        // panic during unwind, a HashMap grow at capacity) doesn't
        // get swallowed by tokio's default `UnhandledPanic::Ignore`.
        // The `store: Arc<dyn StateStore>` and `metrics: Arc<Registry>`
        // captures are already `UnwindSafe` (Arc is UnwindSafe when
        // its inner T is `RefUnwindSafe`, and the store trait's
        // implementors are RefUnwindSafe by convention here).
        // `session_id: String`, `budget: BudgetSpec`, and
        // `principal: Option<(String, BudgetSpec)>` are all
        // Send + UnwindSafe. Wrap the whole thing in
        // `AssertUnwindSafe` to bypass the `RefUnwindSafe` check on
        // the trait object rather than plumb the bound through the
        // whole StateStore hierarchy — the invariant we care about
        // (no shared mutable state leaked from the refund) is
        // upheld by construction.
        let sess_for_log = session_id.clone();
        let refund = move || {
            let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                ActionBudget::new(store.as_ref(), &session_id, &budget).refund_tokens(tokens);
                if let Some((principal_id, spec)) = principal.as_ref() {
                    ActionBudget::for_principal(store.as_ref(), principal_id, spec).refund_tokens(tokens);
                }
            }));
            if let Err(panic) = outcome {
                let msg = panic
                    .downcast_ref::<&'static str>()
                    .copied()
                    .or_else(|| panic.downcast_ref::<String>().map(String::as_str))
                    .unwrap_or("panic payload was not a string");
                metrics
                    .counter(
                        "pv_admission_refund_panics_total",
                        "Admission-debit refund closure panicked in Drop; the session \
                         and principal ledgers retain the debit until operator repair",
                    )
                    .inc();
                tracing::error!(
                    session = %sess_for_log,
                    tokens,
                    panic = %msg,
                    "admission-debit refund PANICKED (caught); \
                     principal ledger drain persists until operator repair"
                );
            }
        };
        match tokio::runtime::Handle::try_current() {
            Ok(handle) => {
                handle.spawn_blocking(refund);
            }
            Err(_) => refund(),
        }
    }
}

/// Terminal failure job used by response-refusal paths (permit missing,
/// unsupported content-encoding, cancelled request future). Carrying
/// `response_marker` lets the worker retire the durable in-flight marker,
/// and `terminal: true` closes the journalled response attempt so crash
/// recovery does not quarantine the session over a request the client
/// already saw fail (or abandoned).
pub(crate) fn refused_response_failure_job(
    session: Arc<Session>,
    identity: AgentIdentity,
    reason: String,
    response_marker: Option<String>,
    response_attempt_id: String,
) -> WorkerJob {
    let atif = (session.workflow == Workflow::Unsigned).then(|| AtifCapture {
        source: pv_atif::Source::Agent,
        message: Value::String(reason.clone()),
        reasoning_content: None,
        model_name: None,
        tool_calls: None,
        observation: None,
        llm_call_count: Some(0),
    });
    WorkerJob {
        session,
        identity,
        class: EventClass::StopReason,
        payload: serde_json::json!({"reason": reason, "direction": "upstream_response"}),
        text: String::new(),
        analyze_loop: false,
        status: StatusId::Failure,
        stop_reason: Some(StopReason::Other),
        native_stop_reason: None,
        metrics: EventMetrics::default(),
        cost_usd_micros: 0,
        prompt_token_correction: 0,
        atif,
        response_marker,
        response_attempt: Some(crate::worker::ResponseAttempt {
            id: response_attempt_id,
            terminal: true,
        }),
    }
}

/// Boxed dynamic error source for harness error variants whose
/// underlying cause can be any of several concrete types (`io::Error`,
/// `serde_json::Error`, `pv_receipts::ReceiptError`, ...). Kept dynamic
/// so one variant covers heterogeneous call sites while still exposing
/// the typed error through [`std::error::Error::source`] — callers can
/// `source().downcast_ref::<std::io::Error>()` to branch on e.g.
/// `io::ErrorKind`, and tracing subscribers receive the full chain.
pub type ErrorSource = Box<dyn std::error::Error + Send + Sync + 'static>;

/// Hot-path failures, each carrying an HTTP status mapping.
///
/// Variants that wrap an underlying failure carry it as a typed
/// `#[source]` instead of a pre-rendered `String`; `context` keeps the
/// exact text the old `String` payloads rendered, so Display output
/// (embedded verbatim in the OpenAI-shaped HTTP error bodies) is
/// unchanged. `Unauthorized` and `Abort` stay plain strings: their
/// messages are deliberately classified policy verdicts (the raw
/// identity error is logged separately and must not travel with the
/// client-facing value).
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum PipelineError {
    /// A request header was malformed or inconsistent.
    #[error("bad request: {context}")]
    BadRequest {
        /// Human-readable description.
        context: String,
        /// Underlying typed error, when one exists.
        #[source]
        source: Option<ErrorSource>,
    },
    /// Identity was missing or invalid.
    #[error("unauthorized: {0}")]
    Unauthorized(String),
    /// A stateful quota or loop breaker blocked execution.
    #[error("request blocked: {context}")]
    Blocked {
        /// Human-readable description.
        context: String,
        /// Underlying typed error, when one exists.
        #[source]
        source: Option<ErrorSource>,
    },
    /// The upstream provider request failed.
    #[error("upstream request failed: {context}")]
    Upstream {
        /// Human-readable description.
        context: String,
        /// Underlying typed error, when one exists.
        #[source]
        source: Option<ErrorSource>,
    },
    /// The upstream provider accepted the connection but did not
    /// finish responding within `upstream_read_timeout_s`. Distinct
    /// from [`Self::Upstream`] so the response layer can map to
    /// `504 Gateway Timeout` (RFC 7231 §6.6.5) rather than
    /// `502 Bad Gateway`, letting operator dashboards distinguish
    /// "provider slow" (504 — check provider health) from
    /// "provider unreachable" (502 — check DNS / network / config).
    /// SDK retry behaviour is unchanged (both retryable).
    #[error("upstream request timed out: {context}")]
    UpstreamTimeout {
        /// Human-readable description.
        context: String,
        /// Underlying typed error, when one exists.
        #[source]
        source: Option<ErrorSource>,
    },
    /// The breaker requested immediate connection closure.
    #[error("connection aborted: {0}")]
    Abort(String),
    /// Required audit capture infrastructure is unavailable.
    #[error("service unavailable: {context}")]
    Unavailable {
        /// Human-readable description.
        context: String,
        /// Underlying typed error, when one exists.
        #[source]
        source: Option<ErrorSource>,
    },
}

impl PipelineError {
    /// Semantic bad-request refusal with no underlying typed error.
    pub fn bad_request(context: impl Into<String>) -> Self {
        Self::BadRequest {
            context: context.into(),
            source: None,
        }
    }

    /// Bad request caused by a typed error. The error text is kept in
    /// `context` so Display matches the old stringified payload, while
    /// the value itself stays reachable via `Error::source()`.
    pub fn bad_request_source<E>(error: E) -> Self
    where
        E: std::error::Error + Send + Sync + 'static,
    {
        Self::BadRequest {
            context: error.to_string(),
            source: Some(Box::new(error)),
        }
    }

    /// Semantic policy block with no underlying typed error.
    pub fn blocked(context: impl Into<String>) -> Self {
        Self::Blocked {
            context: context.into(),
            source: None,
        }
    }

    /// Block caused by a typed error (see [`Self::bad_request_source`]).
    pub fn blocked_source<E>(error: E) -> Self
    where
        E: std::error::Error + Send + Sync + 'static,
    {
        Self::Blocked {
            context: error.to_string(),
            source: Some(Box::new(error)),
        }
    }

    /// Semantic upstream failure with no underlying typed error.
    pub fn upstream(context: impl Into<String>) -> Self {
        Self::Upstream {
            context: context.into(),
            source: None,
        }
    }

    /// Upstream failure caused by a typed error (see
    /// [`Self::bad_request_source`]).
    pub fn upstream_source<E>(error: E) -> Self
    where
        E: std::error::Error + Send + Sync + 'static,
    {
        Self::Upstream {
            context: error.to_string(),
            source: Some(Box::new(error)),
        }
    }

    /// Semantic upstream-timeout failure with no underlying typed
    /// error. Response layer maps this to `504 Gateway Timeout`
    /// (see [`Self::UpstreamTimeout`] docs).
    pub fn upstream_timeout(context: impl Into<String>) -> Self {
        Self::UpstreamTimeout {
            context: context.into(),
            source: None,
        }
    }

    /// Semantic unavailability with no underlying typed error.
    pub fn unavailable(context: impl Into<String>) -> Self {
        Self::Unavailable {
            context: context.into(),
            source: None,
        }
    }

    /// Unavailability caused by a typed error (see
    /// [`Self::bad_request_source`]).
    pub fn unavailable_source<E>(error: E) -> Self
    where
        E: std::error::Error + Send + Sync + 'static,
    {
        Self::Unavailable {
            context: error.to_string(),
            source: Some(Box::new(error)),
        }
    }
    /// HTTP status code corresponding to this failure.
    ///
    /// * `Blocked` / `Abort` are permanent policy verdicts for the current
    ///   session: 403 / 409 stop mainstream SDK auto-retry loops that
    ///   would otherwise interpret 429 as transient rate limiting and
    ///   burn budget re-hitting the same breaker.
    /// * `Unavailable` is transient: 503 is paired with `Retry-After` at
    ///   the response layer so intermediaries and clients back off in
    ///   bounded fashion (RFC 7231 §7.1.3).
    /// * `Unauthorized` is paired with `WWW-Authenticate` at the
    ///   response layer (RFC 7235 §3.1).
    pub fn status(&self) -> axum::http::StatusCode {
        match self {
            Self::BadRequest { .. } => axum::http::StatusCode::BAD_REQUEST,
            Self::Unauthorized(_) => axum::http::StatusCode::UNAUTHORIZED,
            Self::Blocked { .. } => axum::http::StatusCode::FORBIDDEN,
            Self::Upstream { .. } => axum::http::StatusCode::BAD_GATEWAY,
            Self::UpstreamTimeout { .. } => axum::http::StatusCode::GATEWAY_TIMEOUT,
            Self::Abort(_) => axum::http::StatusCode::CONFLICT,
            Self::Unavailable { .. } => axum::http::StatusCode::SERVICE_UNAVAILABLE,
        }
    }
}

impl AppState {
    /// The per-session budget in force: the local `[budget]` tightened by
    /// any console-pushed budget policy (stricter wins on every
    /// dimension, so a pushed bundle can never loosen the host config).
    pub fn effective_budget(&self) -> pv_state::BudgetSpec {
        self.config.budget.tightened_by(&self.sandbox.pushed().budget)
    }

    /// Build fully wired application state with a bounded worker and shared
    /// lifecycle services.
    pub fn new(
        config: HarnessConfig,
        store: Arc<dyn StateStore>,
        sandbox: Arc<Sandbox>,
        bridge: Arc<dyn EventBus>,
        identity: Option<Arc<IdentityValidator>>,
        signer: Arc<dyn Signer>,
    ) -> Result<Self, PipelineError> {
        Self::new_with_embedder(
            config,
            store,
            sandbox,
            bridge,
            identity,
            signer,
            Arc::new(HashEmbedder::default()),
        )
    }

    /// Build application state with an explicit embedding backend.
    #[allow(clippy::too_many_arguments)]
    pub fn new_with_embedder(
        config: HarnessConfig,
        store: Arc<dyn StateStore>,
        sandbox: Arc<Sandbox>,
        bridge: Arc<dyn EventBus>,
        identity: Option<Arc<IdentityValidator>>,
        signer: Arc<dyn Signer>,
        embedder: Arc<dyn Embedder>,
    ) -> Result<Self, PipelineError> {
        Self::new_with_backends(
            config,
            store,
            sandbox,
            bridge,
            identity,
            signer,
            embedder,
            Arc::new(NoopVectorSink),
        )
    }

    /// Build application state with explicit embedding and vector backends.
    #[allow(clippy::too_many_arguments)]
    pub fn new_with_backends(
        config: HarnessConfig,
        store: Arc<dyn StateStore>,
        sandbox: Arc<Sandbox>,
        bridge: Arc<dyn EventBus>,
        identity: Option<Arc<IdentityValidator>>,
        signer: Arc<dyn Signer>,
        embedder: Arc<dyn Embedder>,
        vector_sink: Arc<dyn VectorSink>,
    ) -> Result<Self, PipelineError> {
        Self::new_with_backends_and_metrics(
            config,
            store,
            sandbox,
            bridge,
            identity,
            signer,
            embedder,
            vector_sink,
            Arc::new(Registry::new()),
        )
    }

    /// Build application state reusing a pre-existing metrics registry.
    ///
    /// `main.rs` uses this so counters registered outside `AppState`
    /// (JWKS refresh errors, HTTP shutdown drain timeouts) live on the
    /// same registry that gets scraped at `/metrics` — otherwise their
    /// samples would be invisible to Prometheus.
    #[allow(clippy::too_many_arguments)]
    pub fn new_with_backends_and_metrics(
        config: HarnessConfig,
        store: Arc<dyn StateStore>,
        sandbox: Arc<Sandbox>,
        bridge: Arc<dyn EventBus>,
        identity: Option<Arc<IdentityValidator>>,
        signer: Arc<dyn Signer>,
        embedder: Arc<dyn Embedder>,
        vector_sink: Arc<dyn VectorSink>,
        metrics: Arc<Registry>,
    ) -> Result<Self, PipelineError> {
        let config = Arc::new(config);
        // Every stage of the `pv_stage_duration_seconds` series MUST
        // share the same bucket bounds, otherwise
        // `histogram_quantile(0.99, sum by(le) (rate(...[5m])))` in
        // Prometheus silently produces nonsense: the `le` values from
        // different label combinations do not align and the summation
        // is undefined. Use WIDE_LATENCY_BOUNDS_US for all stages so
        // a future stage that spans real I/O fits without a bounds
        // migration — the fast stages just get more granular
        // low-end buckets they will never light up, which is fine.
        for stage in ["identity", "quota", "sanitize", "compression", "dispatch"] {
            metrics.histogram_with_bounds(
                &format!("pv_stage_duration_seconds{{stage=\"{stage}\"}}"),
                "Harness stage latency",
                pv_core::metrics::WIDE_LATENCY_BOUNDS_US,
            );
        }
        metrics.counter(
            "pv_events_dropped_total{stage=\"worker_queue\"}",
            "Worker jobs or response-slot reservations dropped, labeled by admission stage",
        );
        // The OPERATIONS.md alert table names the response_slot stage
        // specifically ("Rate > 0 sustained ⇒ response capture
        // backpressure"); a lazily-created series renders that alert
        // blind until the first drop actually happens.
        metrics.counter(
            "pv_events_dropped_total{stage=\"response_slot\"}",
            "Worker jobs or response-slot reservations dropped, labeled by admission stage",
        );
        // Documented as an Escalate alert ("sudden increase"); it must
        // exist from boot for absent()/increase() alerts to arm on
        // healthy nodes that never had a capture failure.
        metrics.counter(
            "pv_incomplete_sessions_total",
            "Sessions refused due to incomplete capture",
        );
        metrics.counter(
            "pv_worker_panics_total",
            "Worker job panics isolated by supervisor",
        );
        metrics.counter("pv_worker_errors_total", "Worker jobs that failed");
        metrics.counter("pv_sessions_finalized_total", "Sessions finalized");
        metrics.counter("pv_sessions_promoted_total", "Unsigned sessions promoted");
        metrics.counter("pv_reconcile_errors_total", "Reconciliation errors");
        // pre-register recovery-skipped counters so
        // Prometheus `absent()` alerts do not fire on healthy nodes
        // that have never hit a per-session recovery error, and so
        // dashboards render flat-zero instead of "No data". Each
        // series must exist on `/metrics` from boot; `Registry::counter`
        // is otherwise lazy and only inserts on first `.inc()`.
        metrics.counter(
            "pv_signed_recovery_skipped_total",
            "Signed sessions skipped during recovery due to per-session errors",
        );
        metrics.counter(
            "pv_signed_recovery_quarantined_total",
            "Signed sessions re-adopted as capture-failed quarantines during recovery",
        );
        metrics.counter(
            "pv_unsigned_recovery_skipped_total",
            "Unsigned step-journal consolidations skipped during recovery due to per-session errors",
        );
        metrics.counter(
            "pv_atif_trajectory_recovery_skipped_total",
            "ATIF trajectories skipped during recovery due to per-session errors",
        );
        metrics.counter(
            "pv_pending_close_completion_failed_total",
            "Pending-close completions that failed to finish their tail",
        );
        // pre-register data-plane series so dashboards
        // render flat-zero from boot instead of "No data", and
        // `rate() > 0` alerts can fire the first time the bad thing
        // happens.
        //
        // A boot-time completeness test in the same crate
        // (`documented_alert_series_are_preregistered_at_boot`) pins
        // this set — new fallible arms added in later work must be
        // added here or fail CI. That test caught three counters
        // (`pv_ephemeral_close_failures_total`,
        // `pv_stream_abort_close_failures_total`,
        // `pv_stream_abort_no_runtime_total`) that leaked past this
        // block as lazy series — the exact defect this pre-registration
        // pass exists to prevent.
        for kind in ["timeout", "connect", "send", "http_5xx"] {
            metrics.counter(
                &format!("pv_upstream_errors_total{{kind=\"{kind}\"}}"),
                "Upstream failures by kind",
            );
        }
        metrics.histogram_with_bounds(
            "pv_upstream_latency_seconds",
            "Time to upstream response headers",
            pv_core::metrics::WIDE_LATENCY_BOUNDS_US,
        );
        for route in ["chat", "mcp", "session_close", "session_promote"] {
            for status_class in ["2xx", "4xx", "5xx"] {
                metrics.counter(
                    &format!("pv_requests_total{{route=\"{route}\",status_class=\"{status_class}\"}}"),
                    "HTTP requests by route and status class",
                );
            }
            metrics.histogram_with_bounds(
                &format!("pv_request_duration_seconds{{route=\"{route}\"}}"),
                "End-to-end HTTP request latency by route",
                pv_core::metrics::WIDE_LATENCY_BOUNDS_US,
            );
        }
        for reason in [
            "too_large",
            "read_error",
            "invalid_json",
            "nonconformant",
            "unauthenticated",
            "provenance",
        ] {
            metrics.counter(
                &format!("pv_atif_recovery_skipped_total{{reason=\"{reason}\"}}"),
                "ATIF spool files skipped during recovery",
            );
        }
        metrics.histogram("pv_receipt_sign_duration_seconds", "Receipt signing latency");
        // pre-register the stream-close failure counters. Each one is
        // written only from an error arm on a background drop path,
        // so left lazy they never appear on `/metrics` until the first
        // failure — the exact "rate() > 0 never fires on a lazily-
        // created series" defect this block exists to defeat. Adding
        // one here is cheaper than an alert that never triggers.
        metrics.counter(
            "pv_ephemeral_close_failures_total",
            "Auto-close of a completed ephemeral one-shot failed; \
             the session is left open until the idle sweeper reaps it",
        );
        metrics.counter(
            "pv_stream_abort_close_failures_total",
            "Background close after a stream abort failed; \
             the session is left open until the idle sweeper reaps it",
        );
        metrics.counter(
            "pv_stream_abort_no_runtime_total",
            "Stream abort observed no tokio runtime; capture marked failed for \
             reconciler retry",
        );
        // Panic-supervision counters — same lazy-series defect the
        // stream-close counters above defeat, applied to every
        // background loop's `catch_unwind` arm. Absent-series alerts
        // (`absent(pv_reconciler_panics_total)`) fire a false-positive
        // on healthy nodes when the counter is registered only inside
        // the panic arm, and `rate() > 0` alerts cannot distinguish
        // "never panicked" from "never registered" — so the FIRST
        // panic that the guardrail exists to catch slips past the
        // alert. Pre-register the five outliers alongside the
        // stream-close set. `pv_jwks_refresh_panics_total` is
        // already eagerly registered at its call site (main.rs
        // beside its interval creation); the five below are the
        // panic-arm-only stragglers. HELP text lives in the
        // `supervision_help` consts (shared with the observation
        // sites in main.rs/reconciler.rs) — the registry is
        // first-wins on HELP and warns on divergence, and these six
        // pairs drifted apart exactly that way (R285).
        metrics.counter("pv_reconciler_panics_total", supervision_help::RECONCILER_PANICS);
        // `pv_worker_shard_panics_total` fires from the OUTER
        // envelope-routing supervisor in worker.rs — NOT from the
        // inner job body (that path uses `pv_worker_panics_total`).
        // The distinction is load-bearing during incident triage:
        // a routing/dispatcher panic points at the shard task
        // structure (backlog, channel, batch-boundary logic), while
        // a job-body panic points at the specific request that
        // panicked. HELP text and comment kept aligned with the
        // write site at worker.rs:864.
        metrics.counter(
            "pv_worker_shard_panics_total",
            "Worker shard driver panicked outside a job; supervised via catch_unwind",
        );
        metrics.counter(
            "pv_bridge_maintenance_panics_total",
            supervision_help::BRIDGE_MAINTENANCE_PANICS,
        );
        metrics.counter(
            "pv_bridge_maintenance_errors_total",
            supervision_help::BRIDGE_MAINTENANCE_ERRORS,
        );
        metrics.counter(
            "pv_bridge_maintenance_join_errors_total",
            supervision_help::BRIDGE_MAINTENANCE_JOIN_ERRORS,
        );
        // ATIF retention loop: same supervision discipline. A silent
        // panic in the hourly sweep would let the spool grow unbounded
        // until it fills the disk (retention is the only in-process
        // reclaim path). Pre-register both counters so `rate() > 0`
        // alerts see the FIRST fire.
        metrics.counter(
            "pv_atif_retention_panics_total",
            supervision_help::ATIF_RETENTION_PANICS,
        );
        metrics.counter(
            "pv_atif_retention_errors_total",
            supervision_help::ATIF_RETENTION_ERRORS,
        );
        // TCP_NODELAY setsockopt failure counter — written from the
        // axum `tap_io` per-accept hook when `set_nodelay(true)` on
        // the freshly-accepted socket fails (typically ENOTCONN /
        // EBADF on a SYN-flood / Slowloris torn-down socket, or
        // ENOPROTOOPT on an embedded target without full socket-
        // option support). The tracing warn is dampened via
        // `std::sync::Once` to avoid a per-accept log storm; this
        // counter keeps operator visibility of the failure rate.
        //
        // Help string MUST match the main.rs registration site
        // byte-exact: `Registry::counter` returns the pre-registered
        // Arc<Counter> and discards subsequent `help` strings, so
        // divergent HELP would silently drop the more informative
        // one from `/metrics`. Same discipline as every sibling
        // `*_panics_total` pre-registration.
        metrics.counter(
            "pv_tcp_nodelay_failures_total",
            "TCP_NODELAY setsockopt failed on an accepted connection; the \
             suboptimal-latency connection still serves. Rate-limited to a \
             single tracing warn per process to avoid a per-accept log storm \
             under half-open flood / SYN flood.",
        );
        // Drop-spawned close-session and admission-refund panics —
        // same rationale. Each is written only from a Drop-spawned
        // background task on the panic arm.
        metrics.counter(
            "pv_stream_abort_panics_total",
            "Background close after a stream abort panicked and was caught; \
             the session is left open until the idle sweeper reaps it",
        );
        metrics.counter(
            "pv_ephemeral_close_panics_total",
            "Auto-close of a completed ephemeral one-shot panicked and was \
             caught; the session is left open until the idle sweeper reaps it",
        );
        metrics.counter(
            "pv_admission_refund_panics_total",
            "Admission-debit refund closure panicked in Drop; the session \
             and principal ledgers retain the debit until operator repair",
        );
        // Idle-close per-session deadline counter. Same pre-
        // registration discipline: this counter is emitted only
        // from the timeout arm of `close_session` in the reconciler
        // idle-close sweep, so a lazy `rate() > 0` alert wouldn't
        // see the FIRST fire — exactly the incident the alert
        // exists to catch. See OPERATIONS.md.
        metrics.counter("pv_idle_close_timeouts_total", IDLE_CLOSE_TIMEOUTS_HELP);
        // Shutdown-time per-session close deadline counter. Fires
        // when the shutdown-path close_session (main.rs
        // finalize_sessions loop) hits its 3 s per-session deadline
        // — distinct from `pv_idle_close_timeouts_total` (steady-
        // state idle sweep) and `pv_http_shutdown_drain_timeouts_
        // total` (outer HTTP-drain phase). Pre-registration matches
        // the discipline for the rest of this block: a lazy
        // `rate() > 0` alert would miss the FIRST fire on the very
        // shutdown that surfaces the class.
        metrics.counter(
            "pv_shutdown_session_close_timeouts_total",
            SESSION_CLOSE_TIMEOUTS_HELP,
        );
        // Shutdown-time MCP-inflight drain-deadline counter. Fires when
        // the shutdown barrier waiting for detached mcp_call_inner
        // spawns to complete hits its 5 s deadline before `wait_idle`.
        // Without this barrier the detached body could reach
        // `worker.try_submit` AFTER `wait_idle` returned, race
        // `finalize_sessions`, and trip
        // `pv_shutdown_session_close_timeouts_total` on an otherwise-
        // recoverable session. Pre-registration matches the rest of
        // the block — any occurrence must be visible to `rate() > 0`.
        metrics.counter("pv_shutdown_mcp_drain_timeouts_total", MCP_DRAIN_TIMEOUTS_HELP);
        // Per-tick recovery-scan cap counter. Every recovery pass that
        // walks `read_dir` yields after `MAX_RECOVERY_ENTRIES_PER_TICK`
        // entries so a poisoned spool (millions of stale files) does
        // not starve legitimate recovery work. Pre-registering each
        // labelled series eagerly closes the same "lazy series that
        // rate()>0 never sees on the FIRST fire" defect the counters
        // above defeat. Keep this labels list in-sync with every call
        // to `bump_recovery_scan_cap`.
        for pass in [
            "adopt_strict_atif",
            "recover_signed_journals",
            "consolidate_step_journals",
            "retry_marked_promotions",
            "remove_acked_outboxes",
            "replay_lifecycle_outboxes",
            "quarantine_orphan_json",
        ] {
            metrics.counter(
                &format!("pv_recovery_scan_capped_total{{pass=\"{pass}\"}}"),
                "Recovery pass hit the per-tick entry-scan cap and returned \
                 early; the next tick will re-open the directory and resume. \
                 A steady rate > 0 indicates a poisoned spool (millions of \
                 stale files) that will starve legitimate recovery work \
                 until an operator quarantines it.",
            );
        }
        // Lifecycle-outbox backlog gauge. Rising baseline is the
        // earliest signal of a broker outage — outboxes persist
        // across restarts, so the disk-bounded queue grows until the
        // bridge is reachable again. Pre-registering the gauge (with
        // value 0) makes `absent(pv_lifecycle_outbox_pending)`
        // alerts distinguish "healthy, no backlog" from
        // "reconciler tick hasn't fired yet" — on a fresh boot the
        // gauge appears at 0 immediately, then updates on the first
        // `remove_acked_lifecycle_outboxes` tick.
        metrics.gauge(
            "pv_lifecycle_outbox_pending",
            crate::reconciler::LIFECYCLE_OUTBOX_PENDING_HELP,
        );
        // Reconciler ticks scan the ATIF spool dir, which can be large;
        // finalisation waits for worker drain + broker publish. Wide
        // bounds keep long-tail p99 useful under load.
        metrics.histogram_with_bounds(
            "pv_reconcile_duration_seconds",
            "Idle reconciliation duration",
            pv_core::metrics::WIDE_LATENCY_BOUNDS_US,
        );
        metrics.histogram_with_bounds(
            "pv_session_finalize_duration_seconds",
            "Session finalization latency",
            pv_core::metrics::WIDE_LATENCY_BOUNDS_US,
        );
        for endpoint in ["stats", "list", "detail"] {
            metrics.histogram(
                &format!("pv_dashboard_request_duration_seconds{{endpoint=\"{endpoint}\"}}"),
                "Dashboard endpoint latency",
            );
            metrics.counter(
                &format!("pv_dashboard_requests_total{{endpoint=\"{endpoint}\",status=\"ok\"}}"),
                "Dashboard endpoint requests, labeled by status",
            );
            metrics.counter(
                &format!("pv_dashboard_requests_total{{endpoint=\"{endpoint}\",status=\"not_found\"}}"),
                "Dashboard endpoint requests, labeled by status",
            );
        }
        let sessions = Arc::new(SessionRegistry::new());
        let journal_key = crate::journal::key_from_signer(signer.as_ref());
        let redaction = if !config.redaction_patterns.is_empty() || !config.redaction_paths.is_empty() {
            let redact_config = pv_redact::RedactionConfig {
                regex_patterns: config.redaction_patterns.clone(),
                pointer_paths: config.redaction_paths.clone(),
                include_builtin_patterns: true,
                ..pv_redact::RedactionConfig::default()
            };
            let engine = pv_redact::RedactionEngine::new(redact_config)
                .map_err(|e| PipelineError::bad_request(format!("redaction engine: {e}")))?;
            Some(Arc::new(engine))
        } else {
            None
        };
        let content_sink = crate::content::SharedContentSink::default();
        let worker = crate::worker::spawn_worker_with_outputs(
            config.worker_channel_capacity,
            Arc::clone(&bridge),
            embedder,
            Arc::clone(&vector_sink),
            Some(std::path::PathBuf::from(&config.atif_spool_dir)),
            journal_key,
            Arc::clone(&metrics),
            crate::content::AuditOutputs {
                redaction: redaction.clone(),
                content: Arc::clone(&content_sink),
            },
        );
        let finalizer = Finalizer::with_bridge(
            signer.clone(),
            std::path::PathBuf::from(&config.atif_spool_dir),
            Arc::clone(&metrics),
            Arc::clone(&bridge),
        )
        .with_state_store(Arc::clone(&store))
        // Closes delete the session's vector-store
        // points (best-effort).
        .with_vector_sink(Arc::clone(&vector_sink));
        // One builder for every outbound client, so the backend client
        // differs from the shared one only by its resolver.
        let build_client = |private_only: bool| -> Result<reqwest::Client, PipelineError> {
            let mut client_builder = reqwest::Client::builder()
                .connect_timeout(HTTP_CONNECT_TIMEOUT)
                .redirect(reqwest::redirect::Policy::none())
                // Providers (OpenAI, Anthropic, Azure, Bedrock fronting)
                // downgrade or block empty / generic UAs; identifying
                // ourselves also gives operators a stable string for
                // provider-side triage. Version is baked in at compile
                // time so a rolling deploy makes the change visible.
                .user_agent(concat!("ProofVisor/", env!("CARGO_PKG_VERSION")))
                // TCP keepalive so pooled connections behind NAT/L4 LBs
                // with short idle windows (AWS NLB 350 s, GCP TCP 600 s,
                // stateful FWs often 60-120 s) do not turn every first
                // request per pool cycle into a full TCP+TLS re-handshake
                // that manifests as a spurious 502/`connection reset`.
                .tcp_keepalive(std::time::Duration::from_secs(30))
                // Cap pool idle at 60 s: shorter than any realistic NAT
                // window, longer than any burst-of-requests batch. Bounds
                // the pool memory footprint and forces frequent-enough
                // TLS refresh under a rolling cert rotation.
                .pool_idle_timeout(std::time::Duration::from_secs(60));
            // Reqwest 0.12 is built here
            // without decompression features, so the proxy cannot decode any
            // content-coding an origin might apply. RFC 9110 §12.5.3 says
            // an ABSENT Accept-Encoding permits any coding, so a CDN in
            // front of the upstream may legally gzip the response — which
            // the chat path then 502's on and the tool path (with the
            // decompression guard applied above) would also refuse. Advertise
            // `Accept-Encoding: identity` so the routine outcome is
            // "identity as requested" and the refusal guards are pure
            // defense-in-depth.
            let mut default_headers = reqwest::header::HeaderMap::new();
            default_headers.insert(
                reqwest::header::ACCEPT_ENCODING,
                reqwest::header::HeaderValue::from_static("identity"),
            );
            client_builder = client_builder.default_headers(default_headers);
            // Apply a read-timeout floor unconditionally so an
            // adversarial or merely broken upstream (chat OR tool) cannot pin
            // a session lease + WorkerPermit + tool-intent claim
            // indefinitely by accepting the request and then never
            // responding. TCP keepalive above only detects a hung
            // *connection*, not a slow/silent HTTP response. Operators can
            // widen or override via `upstream_read_timeout_s`; the shipped
            // default (60 s) is well past any realistic first-token
            // latency (Claude p99 ~15 s, GPT-4 p99 ~30 s) but firm enough
            // that a stalled provider surfaces as a definite 502 rather
            // than a resource-starving hang.
            const DEFAULT_UPSTREAM_READ_TIMEOUT_S: u64 = crate::config::DEFAULT_UPSTREAM_READ_TIMEOUT_S;
            let read_timeout_s = config
                .upstream_read_timeout_s
                .unwrap_or(DEFAULT_UPSTREAM_READ_TIMEOUT_S);
            client_builder = client_builder.read_timeout(std::time::Duration::from_secs(read_timeout_s));
            if config.upstream_http2_prior_knowledge {
                client_builder = client_builder.http2_prior_knowledge();
            }
            if private_only {
                client_builder = client_builder.dns_resolver(Arc::new(crate::backend::PrivateOnlyResolver));
            }
            client_builder.build().map_err(PipelineError::upstream_source)
        };
        let client = build_client(false)?;
        // Tool backends get their own client when `require_private_backends`
        // is on: it refuses to connect to any address that is not private.
        // The LLM upstream and the AuthZEN PDP keep the shared client.
        let backend_client = if config.require_private_backends {
            build_client(true)?
        } else {
            client.clone()
        };
        let upstream_auth = resolve_upstream_auth(&config)?;
        let tool_auth = resolve_tool_auth(&config)?;
        let hot_metrics = Arc::new(HotMetrics::new(&metrics, config.strict_stage_budget));
        // S3: config validation already refused unsupported providers;
        // this defense-in-depth error covers direct constructors that
        // bypassed validate() (test builders).
        let provider = crate::provider::adapter_for(&config.provider).ok_or_else(|| {
            PipelineError::bad_request(format!("unsupported provider {:?}", config.provider))
        })?;
        tracing::info!(provider = provider.name(), "provider adapter selected");
        let mcp_admission = Arc::new(tokio::sync::Semaphore::new(config.mcp_concurrency));
        // R72 review F1 + F2 (landed R74): eager registration
        // stores the returned `Arc<Counter>` handle so the
        // refusal arm calls `.inc()` directly, bypassing the
        // registry mutex on the hot path. Prior R72 shape
        // eagerly registered but discarded the handle,
        // requiring a re-lookup per refusal.
        let mcp_admission_refusals_counter = metrics.counter(
            "pv_mcp_admission_refusals_total",
            "MCP admission was refused because `mcp_concurrency` was already at capacity — \
             each admitted call can buffer up to 16 MiB in `read_limited_tool_response`, \
             so the cap bounds worst-case MCP resident memory. Sustained > 0 indicates \
             tool-upstream slowdown, a burst-traffic event, or an undersized \
             `mcp_concurrency`.",
        );
        for pdp_code in ["UNMAPPED_TOOL", "MISSION_EXPIRED", "MISSION_DENIED", "PDP_DENIED"] {
            metrics.counter(
                &format!("pv_pdp_denials_total{{code=\"{pdp_code}\"}}"),
                "PDP denials by denial code",
            );
        }
        metrics.counter(
            crate::routes::TOKENS_REVOKED_METRIC,
            crate::routes::TOKENS_REVOKED_HELP,
        );
        let backend_router = crate::backend::BackendRouter::new(
            &config.backends,
            config.tool_upstream_url.as_deref(),
            tool_auth.clone(),
        )
        .map_err(|e| PipelineError::bad_request(format!("backend routing: {e}")))?
        .with_fallback_transport(config.tool_upstream_transport);
        let mcp_backends = Arc::new(crate::mcp_client::McpBackendClient::new(backend_client.clone()));
        let mcp_sessions = Arc::new(crate::mcp::SessionSigner::derive(&journal_key));
        let authzen = config
            .authzen
            .as_ref()
            .map(|authzen| crate::authzen::AuthzenClient::from_config(authzen, client.clone()).map(Arc::new))
            .transpose()
            .map_err(|e| PipelineError::bad_request(format!("authzen: {e}")))?;
        let workload = crate::workload::WorkloadIssuer::from_config(&config)
            .map_err(|e| PipelineError::bad_request(format!("workload identity: {e}")))?
            .map(Arc::new);
        let token_signer = if let Some(ref seed_path) = config.token_exchange_seed_file {
            let seed = load_exchange_seed(seed_path)
                .map_err(|e| PipelineError::bad_request(format!("exchange signing key: {e}")))?;
            let ts = Arc::new(crate::authz::TokenSigner::from_seed(&seed));
            if ts.kid() == signer.key_id() {
                return Err(PipelineError::bad_request(
                    "token_exchange_seed_file must use a different key than the receipt signer \
                     (both derive the same kid, which means the same seed)"
                        .to_owned(),
                ));
            }
            Some(ts)
        } else {
            None
        };
        let pdp = crate::authz::PolicyDecisionPoint::from_config(&config, token_signer.clone());
        let exchanged_revocations: Arc<dyn pv_identity::RevocationStore> = if config.state_backend == "redis"
        {
            Arc::new(crate::revocation::GuardedRevocationStore::new(
                Arc::new(crate::revocation::StateRevocationStore::with_namespace(
                    Arc::clone(&store),
                    "exchanged",
                )),
                &metrics,
                "exchanged",
            ))
        } else {
            Arc::new(pv_identity::InMemoryRevocationStore::new())
        };
        Ok(Self {
            config,
            store,
            sandbox,
            bridge,
            sessions,
            worker,
            identity,
            exchanged_revocations,
            revocation_audit_window: Arc::new(parking_lot::Mutex::new((Instant::now(), 0))),
            hot_metrics,
            metrics,
            client,
            backend_client,
            upstream_auth,
            tool_auth,
            tool_audit_gates: Arc::default(),
            tool_audits_emitted: Arc::default(),
            provider,
            finalizer,
            draining: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            identity_rejection_window: Arc::new(parking_lot::Mutex::new((Instant::now(), 0))),
            journal_key,
            mcp_inflight: Arc::new(crate::inflight::InflightTracker::new()),
            pushed_rates: Arc::new(crate::policy_push::RateLimits::default()),
            mcp_admission,
            mcp_admission_refusals_counter,
            backend_router: Arc::new(backend_router),
            pdp: Arc::new(pdp),
            token_signer,
            redaction,
            mcp_backends,
            mcp_sessions,
            authzen,
            content_sink,
            workload,
        })
    }

    /// Install the sink that receives every journaled step in content form
    /// (used by `proofvisord` for full-content tenant telemetry).
    ///
    /// The sink runs on its own thread behind a bounded queue, so it can
    /// neither slow down nor fail the audit worker.
    pub fn set_content_sink(&self, sink: Arc<dyn crate::content::ContentSink>) -> std::io::Result<()> {
        let dispatcher = crate::content::ContentDispatcher::spawn(sink, &self.metrics)?;
        *self.content_sink.write() = Some(Arc::new(dispatcher));
        Ok(())
    }

    /// Run identity, breaker, quota, sanitize, compression, and asynchronous
    /// dispatch in the mandated order without awaiting worker or upstream I/O.
    ///
    /// `identity_hint`, when `Some`, is trusted as the resolved identity for
    /// this request — the internal `resolve_identity` + audit-enqueue step
    /// is skipped. Used by `chat_completions` (R72 F3 pre-admission gate)
    /// which already resolved the identity to fail-fast on unauthenticated
    /// probes before the body parses. Without this hint, the R72 pre-check
    /// path would `enqueue_transient_failure` on a JWKS-rotation race
    /// (pre-check success + re-check failure between the two calls),
    /// double-counting `pv_identity_rejections_total` and burning two
    /// of the sampled full-audit slots per rejected request (R73 review
    /// F3). `None` preserves the historical behaviour for callers that
    /// do NOT pre-resolve.
    pub fn prepare_chat(
        &self,
        headers: &HeaderMap,
        payload: Value,
        identity_hint: Option<AgentIdentity>,
    ) -> Result<PreparedRequest, PipelineError> {
        self.prepare_chat_with_caller(headers, payload, identity_hint, None)
    }

    fn prepare_chat_with_caller(
        &self,
        headers: &HeaderMap,
        mut payload: Value,
        identity_hint: Option<AgentIdentity>,
        caller: Option<&AuthenticatedCaller>,
    ) -> Result<PreparedRequest, PipelineError> {
        let total_started = Instant::now();
        let session_id = session_id(headers)?;
        let workflow = workflow(headers, &self.config.default_workflow)?;
        // Multi-choice completions (`n > 1`) are accepted up to
        // `MAX_CHOICES`: the capture path accumulates every choice
        // separately (text, tool calls and finish reason keyed by
        // `choices[].index`), so the ATIF step attests each choice as
        // its own message. Refuse anything that is not a plain integer
        // in range up front, before any budget debit: upstream
        // OpenAI-compatible servers lax-coerce `2.0` / `"2"` into real
        // multi-choice requests, so the check fails closed on those
        // encodings instead of forwarding them.
        crate::provider::choice_count(&payload).map_err(PipelineError::bad_request)?;

        // Validate and extract the relayed headers BEFORE any budget
        // debit or admission journaling: `single_header` refuses
        // multi-valued occurrences, and its `?` used to fire from the
        // `PreparedRequest` construction AFTER both ledgers were
        // debited with no refund arm — a client repeating a duplicate
        // `openai-beta` header drained session + principal budgets
        // without a single upstream call.
        let client_authorization = if self.config.upstream_authorization_passthrough {
            single_header(headers, "authorization")?.cloned()
        } else {
            None
        };
        // Strict allowlist of provider-semantic headers. Same
        // single-value discipline as every other relayed header
        // (a multi-valued occurrence is refused at ingress).
        const UPSTREAM_HEADER_ALLOWLIST: [&str; 3] = ["openai-organization", "openai-project", "openai-beta"];
        let mut upstream_passthrough_headers = Vec::new();
        for name in UPSTREAM_HEADER_ALLOWLIST {
            if let Some(value) = single_header(headers, name)? {
                upstream_passthrough_headers.push((axum::http::HeaderName::from_static(name), value.clone()));
            }
        }

        let stage = Instant::now();
        let identity = if let Some(pre_resolved) = identity_hint {
            // R73 review F3: pre-resolved from an earlier caller
            // (chat_completions pre-admission gate). Skip the
            // internal resolve + audit-enqueue so a rotation race
            // between the two calls cannot double-count
            // `pv_identity_rejections_total` or burn a second
            // full-audit sample slot per request.
            pre_resolved
        } else {
            match self.resolve_identity(headers, Some(&self.config.chat_scope)) {
                Ok(identity) => identity,
                Err(error) => {
                    self.audit_authentication_error(headers, &error)?;
                    return Err(error);
                }
            }
        };
        self.observe_stage("identity", stage);

        let session = self
            .sessions
            .get_or_open(&session_id, workflow, &identity, &self.config.breaker);
        let admission = session.admission_guard();
        validate_session_binding(&session, workflow, &identity)?;
        if let Some(caller) = caller {
            caller.bind_session(&session)?;
        }
        session.refresh_identity(&identity);
        if session.is_closed() {
            // Distinguish the PERMANENT quarantine (empty-unsigned close
            // refusal — deliberately never recycled) from the TRANSIENT
            // closing window. `get_or_open` recycles completed-close
            // entries, so a session observed here as closed-but-not-
            // complete is mid-close: an SDK auto-retry (e.g. after a
            // read-timeout disconnect triggered the background close)
            // used to hit a terminal 400 "already closed" — misleading
            // (the client closed nothing) and never retried by SDKs,
            // even though the very next attempt would have succeeded
            // against the recycled id. 503 + Retry-After lets standard
            // retry policies ride out the window.
            // Sealed capture-failed quarantines are ALSO closed-but-
            // never-complete — permanently (never recycled, never
            // evicted). They keep the terminal 400 (permanent
            // refusals route to 400 with no Retry-After);
            // the transient 503 below would otherwise instruct SDK
            // retry policies to poll forever against a session that
            // will refuse forever. Checked BEFORE the empty-quarantine
            // predicate, which capture-failed conversions can also
            // match — the "no captured steps" message was inaccurate
            // for sessions that DID capture steps before failing.
            if session.capture_failed() {
                return Err(PipelineError::bad_request(
                    "session is quarantined (audit capture failed); use a new session id".to_owned(),
                ));
            }
            if session.is_empty_unsigned_quarantine() {
                return Err(PipelineError::bad_request(
                    "session is quarantined (closed with no captured steps); use a new session id".to_owned(),
                ));
            }
            // Enforcement-latched sessions are terminal for the id:
            // `get_or_open` deliberately refuses to recycle them (a
            // fresh incarnation would carry a fresh budget/breaker on
            // the same id). 403, matching the deliberate not-429/not-
            // transient policy for budget and breaker refusals. Keyed
            // on the sticky latch, not the overwritable last stop
            // reason.
            if session.close_complete_flag() && session.enforcement_tripped() {
                return Err(PipelineError::blocked(
                    "session was closed by budget or loop enforcement; use a new session id".to_owned(),
                ));
            }
            return Err(PipelineError::unavailable(
                "session close is completing; retry shortly or use a new session id".to_owned(),
            ));
        }
        if session.capture_failed() {
            return Err(PipelineError::unavailable(
                "session audit capture is incomplete".to_owned(),
            ));
        }
        session.touch();
        if session.loop_state.state() == BreakerState::Open {
            match session.loop_state.action() {
                BreakerAction::Reject => {
                    session.latch_enforcement(StopReason::LoopDetected);
                    return Err(PipelineError::blocked(
                        "semantic loop circuit breaker is open".to_owned(),
                    ));
                }
                BreakerAction::Abort => {
                    session.latch_enforcement(StopReason::LoopDetected);
                    return Err(PipelineError::Abort(
                        "semantic loop circuit breaker requested connection close".to_owned(),
                    ));
                }
                BreakerAction::Inject => {
                    inject_corrective_message(&mut payload)?;
                    session.loop_state.reset();
                }
                _ => {
                    return Err(PipelineError::blocked(
                        "unsupported semantic loop enforcement action".to_owned(),
                    ));
                }
            }
        }

        // Fused acquire: one worker slot + one response slot atomically.
        // If the worker slot succeeds and the response slot fails, the
        // worker permit drops via RAII (its OwnedSemaphorePermit and
        // mpsc reservation release cleanly), so the caller never sees
        // an orphaned half-reservation. Distinct
        // `pv_events_dropped_total{stage=worker_queue|response_slot}`
        // counters let operators tell which one exhausted.
        let permits = self
            .worker
            .try_reserve_pair(&session_id)
            .map_err(PipelineError::unavailable_source)?;
        let worker_permit = permits.worker;
        let response_permit = permits.response;

        let stage = Instant::now();
        let raw_prompt_tokens = pv_core::tokens::approx_tokens_json(&payload);
        // Budget is charged AFTER compression on `compression.tokens_after`
        // (below). Deferring the debit closes the enforcement/attestation
        // gap the review flagged in §6.3: previously the ledger was debited
        // on the uncompressed payload while the receipt attested the
        // post-compression count, over-charging clients by ~40% on any
        // compressible history and making compression's savings invisible
        // to `max_tokens`. The genuine-cap-refusal latch (upstream's
        // enforcement-latch rounds 11-13) is applied at the deferred
        // charge site below.
        self.observe_stage("quota", stage);

        let stage = Instant::now();
        if let Err(reason) = self.sandbox.sanitize("chat/completions", &payload) {
            let error = PipelineError::blocked(reason);
            // No token refund needed: the budget debit happens AFTER
            // compression (below), so a sanitize block exits before
            // anything was charged.
            worker_permit.submit(self.failure_job(
                Arc::clone(&session),
                identity.clone(),
                StopReason::PolicyBlocked,
                error.to_string(),
            ));
            return Err(error);
        }
        self.observe_stage("sanitize", stage);

        let stage = Instant::now();
        let compression = if self.config.compression_enabled {
            pv_compress::compress(
                &payload,
                &pv_compress::CompressionConfig {
                    // Authenticate compression stubs under a key derived
                    // from the journal key, so user text that quotes a
                    // stub cannot impersonate a machine-emitted one and
                    // disable the middle-history pass.
                    marker_key: Some(pv_compress::derive_marker_key(&self.journal_key)),
                    ..pv_compress::CompressionConfig::default()
                },
            )
        } else {
            pv_compress::CompressionOutcome {
                payload,
                tokens_before: raw_prompt_tokens,
                tokens_after: raw_prompt_tokens,
                changed: false,
            }
        };
        self.observe_stage("compression", stage);

        // Console-pushed token-rate policy (per principal, sliding
        // window), checked before either budget ledger is debited so a
        // refusal debits nothing. Admitted tokens count against the
        // window even if a budget below then refuses the request, which
        // errs on the strict side.
        if let Err(refusal) = self
            .pushed_rates
            .admit(&principal_id_for_budget(&identity), compression.tokens_after)
        {
            let error = PipelineError::blocked(refusal.message);
            worker_permit.submit(self.policy_failure_job(
                Arc::clone(&session),
                identity.clone(),
                StopReason::PolicyBlocked,
                error.to_string(),
                Some(refusal.policy),
            ));
            return Err(error);
        }

        // Post-compression budget charge. Both the principal and session
        // ledgers spend `compression.tokens_after` — the same number
        // the signed receipt attests, so budget refusals and the audit
        // artifact always agree. On refusal, if the principal debit
        // landed but the session cap refuses, the principal amount is
        // refunded to keep failed requests non-consumptive.
        let stage = Instant::now();
        let billed_tokens = compression.tokens_after;
        let principal_billing_id = self
            .config
            .principal_budget
            .as_ref()
            .map(|_| principal_id_for_budget(&identity));
        if let (Some(spec), Some(principal_id)) = (
            self.config.principal_budget.as_ref(),
            principal_billing_id.as_deref(),
        ) {
            let quota = match ActionBudget::for_principal(self.store.as_ref(), principal_id, spec)
                .try_tokens(billed_tokens)
            {
                Ok(quota) => quota,
                Err(error) => {
                    let error = PipelineError::Blocked {
                        context: format!("quota backend failed closed (principal): {error}"),
                        source: Some(Box::new(error)),
                    };
                    worker_permit.submit(self.failure_job(
                        Arc::clone(&session),
                        identity.clone(),
                        StopReason::BudgetExceeded,
                        error.to_string(),
                    ));
                    return Err(error);
                }
            };
            if let BudgetDecision::Refused { limit, cap } = quota {
                // Genuine cap exhaustion — latch (upstream rounds 11-13;
                // the backend-ERROR arms deliberately do not: a store
                // outage is not enforcement).
                session.latch_enforcement(StopReason::BudgetExceeded);
                let error = PipelineError::blocked(format!("principal.{limit} exceeded (cap {cap})"));
                worker_permit.submit(self.failure_job(
                    Arc::clone(&session),
                    identity.clone(),
                    StopReason::BudgetExceeded,
                    error.to_string(),
                ));
                return Err(error);
            }
        }
        let quota = match ActionBudget::new(self.store.as_ref(), &session_id, &self.effective_budget())
            .try_tokens(billed_tokens)
        {
            Ok(quota) => quota,
            Err(error) => {
                if let (Some(spec), Some(principal_id)) = (
                    self.config.principal_budget.as_ref(),
                    principal_billing_id.as_deref(),
                ) {
                    ActionBudget::for_principal(self.store.as_ref(), principal_id, spec)
                        .refund_tokens(billed_tokens);
                }
                let error = PipelineError::Blocked {
                    context: format!("quota backend failed closed: {error}"),
                    source: Some(Box::new(error)),
                };
                worker_permit.submit(self.failure_job(
                    Arc::clone(&session),
                    identity.clone(),
                    StopReason::BudgetExceeded,
                    error.to_string(),
                ));
                return Err(error);
            }
        };
        if let BudgetDecision::Refused { limit, cap } = quota {
            if let (Some(spec), Some(principal_id)) = (
                self.config.principal_budget.as_ref(),
                principal_billing_id.as_deref(),
            ) {
                ActionBudget::for_principal(self.store.as_ref(), principal_id, spec)
                    .refund_tokens(billed_tokens);
            }
            // Genuine cap exhaustion — latch (upstream rounds 11-13).
            session.latch_enforcement(StopReason::BudgetExceeded);
            // When the binding cap came from a console-pushed budget
            // policy, the refusal names it, as the tool path does.
            let policy = self
                .sandbox
                .pushed()
                .budget_policy_for(&self.config.budget, &limit, cap);
            let error = PipelineError::blocked(match &policy {
                Some(name) => format!("policy {name:?}: {limit} exceeded (cap {cap})"),
                None => format!("{limit} exceeded (cap {cap})"),
            });
            worker_permit.submit(self.policy_failure_job(
                Arc::clone(&session),
                identity.clone(),
                StopReason::BudgetExceeded,
                error.to_string(),
                policy,
            ));
            return Err(error);
        }
        self.observe_stage("quota_post_compression", stage);

        let stage = Instant::now();
        let text = last_message_text(&compression.payload);
        let atif = match atif_capture_from_request(&compression.payload) {
            Ok(atif) => atif,
            Err(error) => {
                // Never forwarded — refund the post-compression debit on
                // BOTH ledgers (see the refusal arms above).
                ActionBudget::new(self.store.as_ref(), &session_id, &self.effective_budget())
                    .refund_tokens(billed_tokens);
                if let (Some(spec), Some(principal_id)) = (
                    self.config.principal_budget.as_ref(),
                    principal_billing_id.as_deref(),
                ) {
                    ActionBudget::for_principal(self.store.as_ref(), principal_id, spec)
                        .refund_tokens(billed_tokens);
                }
                worker_permit.submit(self.failure_job(
                    Arc::clone(&session),
                    identity.clone(),
                    StopReason::Other,
                    error.to_string(),
                ));
                return Err(error);
            }
        };
        let analyze_loop = atif.source == pv_atif::Source::Agent && !text.is_empty();
        let response_attempt_id = pv_core::new_event_uid();
        let job = WorkerJob {
            session: Arc::clone(&session),
            identity: identity.clone(),
            class: EventClass::Compression,
            payload: serde_json::json!({
                "changed": compression.changed,
                "tokens_before": compression.tokens_before,
                "tokens_after": compression.tokens_after,
            }),
            text,
            analyze_loop,
            status: StatusId::Success,
            stop_reason: None,
            native_stop_reason: None,
            metrics: EventMetrics {
                prompt_tokens: Some(compression.tokens_after),
                completion_tokens: Some(0),
                cached_tokens: Some(0),
                pruned_tokens: Some(compression.pruned_tokens()),
                pruning_ratio_millis: Some(compression.pruning_ratio_millis()),
            },
            cost_usd_micros: 0,
            prompt_token_correction: 0,
            atif: Some(atif),
            response_marker: None,
            response_attempt: Some(crate::worker::ResponseAttempt {
                id: response_attempt_id.clone(),
                terminal: false,
            }),
        };
        worker_permit.submit(job);
        self.observe_stage("dispatch", stage);
        let lease = SessionLease::new(Arc::clone(&session));
        // Armed from here on: any early exit that drops the request —
        // including the `?` on the passthrough header below and, above
        // all, the request future being cancelled by a client disconnect
        // at any later await — submits the terminal failure record that
        // closes the journalled attempt (and retires the marker, once
        // `forward_chat` sets one).
        let capture_guard = ResponseCaptureGuard::new(
            self.worker.clone(),
            Arc::clone(&session),
            identity.clone(),
            response_attempt_id,
            Some(response_permit),
        );
        drop(admission);

        let admission_debit = AdmissionDebit {
            store: Arc::clone(&self.store),
            session_id: session.id.clone(),
            budget: self.effective_budget(),
            principal: self
                .config
                .principal_budget
                .as_ref()
                .map(|spec| (principal_id_for_budget(&identity), spec.clone())),
            tokens: billed_tokens,
            metrics: Arc::clone(&self.metrics),
        };
        Ok(PreparedRequest {
            session,
            identity,
            payload: compression.payload,
            middleware_us: elapsed_us(total_started),
            admission_debit,
            lease,
            capture_guard,
            client_authorization,
            upstream_passthrough_headers,
        })
    }

    /// Run synchronous local gates without waiting for off-path journal,
    /// embedding, or broker work.
    ///
    /// The inline/blocking decision keys off PAYLOAD
    /// SIZE, not off whether a token budget happens to be configured.
    /// The cost of `prepare_chat` (parse-adjacent token scanning,
    /// sanitize, compression) is a function of body size — measured
    /// 1.9 ms p50 of uninterruptible CPU at a 128 KB body, which
    /// stalls every in-flight SSE stream's poll on that tokio worker.
    /// Previously a configured `budget.max_tokens` (the one thing that
    /// makes quotas work) was what moved the work to `spawn_blocking`,
    /// which was backwards. Small bodies stay inline (spawn_blocking
    /// costs ~10-20 µs of handoff — 2× the total gate cost at 75 B).
    ///
    /// The `budget.max_tokens.is_none()` clause below is load-bearing,
    /// not a leftover: with a budget configured, `prepare_chat` calls
    /// `ActionBudget::try_tokens` → `StateStore::try_spend`, which on
    /// the redis backend is a synchronous network round-trip — that
    /// must never run inline on a tokio worker, whatever the body size.
    /// The same applies to `[principal_budget]`: it drives its own
    /// `for_principal(...).try_tokens` store round-trip in
    /// `prepare_chat` even when the session `max_tokens` is unset, so
    /// it must equally force the blocking pool.
    /// Async wrapper over [`Self::prepare_chat`] that spawn-blocks on
    /// bodies large enough to stall the reactor. See `prepare_chat`
    /// for the `identity_hint` contract.
    pub async fn prepare_chat_nonblocking(
        &self,
        headers: &HeaderMap,
        payload: Value,
        body_bytes: usize,
        identity_hint: Option<AgentIdentity>,
    ) -> Result<PreparedRequest, PipelineError> {
        if self.chat_preparation_can_run_inline(body_bytes)
            // Without a trusted hint, a configured validator may perform
            // synchronous shared-revocation I/O even for a tiny request.
            && (identity_hint.is_some() || self.identity.is_none())
        {
            return self.prepare_chat(headers, payload, identity_hint);
        }
        let state = self.clone();
        let headers = headers.clone();
        tokio::task::spawn_blocking(move || state.prepare_chat(&headers, payload, identity_hint))
            .await
            .map_err(PipelineError::unavailable_source)?
    }

    fn chat_preparation_can_run_inline(&self, body_bytes: usize) -> bool {
        // Above 16 KiB, CPU-bound gates can stall the reactor. Any token
        // budget can also call a synchronous state backend, regardless of size.
        body_bytes <= 16 * 1024
            && self.effective_budget().max_tokens.is_none()
            && self.config.principal_budget.is_none()
    }

    pub(crate) async fn authenticate_chat_nonblocking(
        &self,
        headers: &HeaderMap,
    ) -> Result<AuthenticatedCaller, PipelineError> {
        if self.identity.is_none() {
            // Preserve all anonymous/header checks without queueing local work
            // behind durable filesystem writes in the shared blocking pool.
            let caller = self.authenticate(headers)?;
            caller.authorize(self, Some(&self.config.chat_scope))?;
            return Ok(caller);
        }
        let state = self.clone();
        let headers = headers.clone();
        tokio::task::spawn_blocking(move || {
            let caller = state.authenticate(&headers)?;
            caller.authorize(&state, Some(&state.config.chat_scope))?;
            Ok(caller)
        })
        .await
        .map_err(PipelineError::unavailable_source)?
    }

    pub(crate) async fn prepare_authenticated_chat(
        &self,
        headers: &HeaderMap,
        payload: Value,
        body_bytes: usize,
        caller: AuthenticatedCaller,
    ) -> Result<PreparedRequest, PipelineError> {
        if self.chat_preparation_can_run_inline(body_bytes) {
            // Authentication already finished. This rechecks only local
            // expiry/scope and retains principal binding during preparation.
            let identity = caller.authorize(self, Some(&self.config.chat_scope))?;
            return self.prepare_chat_with_caller(headers, payload, Some(identity), Some(&caller));
        }
        let state = self.clone();
        let headers = headers.clone();
        tokio::task::spawn_blocking(move || {
            let identity = caller.authorize(&state, Some(&state.config.chat_scope))?;
            state.prepare_chat_with_caller(&headers, payload, Some(identity), Some(&caller))
        })
        .await
        .map_err(PipelineError::unavailable_source)?
    }

    /// Prepare a request and wait until its audit record is durably captured
    /// before the provider can observe it.
    pub async fn prepare_chat_durable(
        &self,
        headers: &HeaderMap,
        payload: Value,
    ) -> Result<PreparedRequest, PipelineError> {
        if let Ok(id) = session_id(headers) {
            if let Some(session) = self.sessions.get(&id) {
                session.wait_for_worker_jobs().await;
                if session.capture_failed() {
                    return Err(PipelineError::unavailable(
                        "session audit capture is incomplete".to_owned(),
                    ));
                }
            }
        }
        let state = self.clone();
        let headers = headers.clone();
        let mut prepared = tokio::task::spawn_blocking(move || state.prepare_chat(&headers, payload, None))
            .await
            .map_err(PipelineError::unavailable_source)??;
        prepared.session.wait_for_worker_jobs().await;
        if prepared.session.capture_failed() {
            let error = PipelineError::unavailable(
                "request audit capture failed before provider dispatch".to_owned(),
            );
            self.abandon_prepared(&mut prepared, StopReason::Other, &error.to_string());
            return Err(error);
        }
        if prepared.session.loop_state.state() == BreakerState::Open {
            let action = prepared.session.loop_state.action();
            // Latch the terminal enforcement actions only: Inject's
            // refusal here is genuinely transient (the client retry
            // re-enters prepare_chat, which injects the corrective
            // message and proceeds), so it must not make the id
            // terminal. Without this latch, a breaker trip delivered
            // from THIS site (opened during the audit wait) left the
            // id recyclable with a fresh breaker.
            if matches!(action, BreakerAction::Abort | BreakerAction::Reject) {
                prepared.session.latch_enforcement(StopReason::LoopDetected);
            }
            let error = match action {
                BreakerAction::Abort => {
                    PipelineError::Abort("semantic loop circuit breaker opened during audit".to_owned())
                }
                _ => PipelineError::blocked(
                    "semantic loop circuit breaker opened during audit; retry required".to_owned(),
                ),
            };
            self.abandon_prepared(&mut prepared, StopReason::LoopDetected, &error.to_string());
            return Err(error);
        }
        Ok(prepared)
    }

    /// Close out a prepared request that will never reach `forward_chat`.
    ///
    /// `prepare_chat` journals a non-terminal [`crate::worker::ResponseAttempt`]
    /// with its admission record; the matching terminal record normally comes
    /// from `forward_chat`'s failure path or the response relay. A caller that
    /// abandons the request after admission must submit the terminal failure
    /// record itself — otherwise the journal ends with a dangling non-terminal
    /// attempt and a later crash-recovery scan quarantines the whole session
    /// over a request the client already saw fail.
    fn abandon_prepared(&self, prepared: &mut PreparedRequest, stop_reason: StopReason, reason: &str) {
        // Every abandon site runs strictly before upstream dispatch
        // (durable-capture failure, breaker-open, marker-write failure),
        // so the admission token debit provably bought no LLM work —
        // refund it exactly once on BOTH ledgers (the guard zeroes
        // itself, so a hypothetical double abandon — or the guard's own
        // Drop — cannot double-refund).
        prepared.admission_debit.refund_now();
        let Some(permit) = prepared.capture_guard.take_permit() else {
            // Defensive: no permit means the guard's Drop is the only
            // resolver left. Leave it armed so the terminal record lands
            // through the plain worker queue on drop.
            return;
        };
        // Take the marker so the terminal job below retires it too.
        let response_marker = prepared.capture_guard.defuse();
        let mut job = self.failure_job(
            Arc::clone(&prepared.session),
            prepared.identity.clone(),
            stop_reason,
            reason.to_owned(),
        );
        job.response_marker = response_marker;
        job.response_attempt = Some(crate::worker::ResponseAttempt {
            id: prepared.capture_guard.attempt_id().to_owned(),
            terminal: true,
        });
        // Every sibling refusal path (`forward_chat` Err, both
        // `chat_completions` arms, the guard's own Drop) fails the
        // session closed on submit failure so the dangling attempt
        // cannot silently strand. Mirror that here: without it,
        // defusing the guard first would lose the safety net this whole
        // guard machinery exists to provide.
        if permit.submit(&self.worker, job).is_err() {
            prepared.session.mark_capture_failed();
        }
    }

    /// Forward a prepared OpenAI-compatible request to the configured provider.
    ///
    /// For the `openai` dialect the compressed payload is sent as is to
    /// `upstream_url` + `upstream_chat_path` and the response is relayed
    /// unchanged. For `anthropic` and `gemini` the payload is translated
    /// into the native request first (`n` choices become `n` concurrent
    /// Anthropic requests, or Gemini's `candidateCount`), and the native
    /// response is translated back into the OpenAI shape before the
    /// relay sees it, so capture and accounting run one parser.
    pub async fn forward_chat(
        &self,
        mut request: PreparedRequest,
    ) -> Result<ForwardedResponse, PipelineError> {
        // Digest the request payload so the in-flight marker can be
        // matched to the observed response bytes at recovery time.
        //
        // `serde_json::to_vec` on a `Value` is effectively infallible
        // (Value can only carry JSON-serialisable data), but we handle
        // the theoretical error path anyway: falling back to
        // `sha256(b"")` — the well-known empty digest — would make
        // every concurrent failed serialisation collide on the same
        // request_digest, silently violating the marker's
        // one-to-one-with-request invariant. Fall back to a
        // session-id-derived digest so a hypothetical failure at
        // least keeps distinct sessions distinct.
        // Serialize the payload exactly ONCE — the
        // bytes feed both the request digest and (for the OpenAI
        // dialect) the upstream body.
        let payload_bytes = match serde_json::to_vec(&request.payload) {
            Ok(bytes) => Some(bytes),
            Err(error) => {
                tracing::error!(
                    %error,
                    session = %request.session.id,
                    "failed to serialise chat payload for request digest; falling back to session-derived digest"
                );
                None
            }
        };
        let request_digest = match &payload_bytes {
            Some(bytes) => pv_core::digest::sha256_hex(bytes),
            None => pv_core::digest::sha256_hex(request.session.id.as_bytes()),
        };
        // Plan the upstream request(s) before the marker write: a
        // request the native adapter cannot express is a client error
        // that must be refused before any upstream I/O, with the
        // admission debit refunded like every other pre-dispatch abort.
        let plan = match self.plan_upstream(
            &request.payload,
            payload_bytes,
            request.client_authorization.take(),
            std::mem::take(&mut request.upstream_passthrough_headers),
        ) {
            Ok(plan) => plan,
            Err(client_error) => {
                self.abandon_prepared(&mut request, StopReason::Other, &client_error.to_string());
                return Err(client_error);
            }
        };
        // Cancellation fix: arm the guard with the marker id
        // BEFORE the awaited disk write. If the client disconnects
        // between here and the write completing, `spawn_blocking` still
        // runs the write to durability; the guard's terminal-failure
        // job then retires the marker via `clear_response_marker`,
        // which now treats an absent file as success — so a never-written
        // marker is a clean no-op and a written one is deleted. Under
        // the previous ordering, cancellation between the write and
        // `set_marker` left the marker stranded, poisoning quarantine
        // sets and (across restart) bricking receipts.
        let attempt_id = crate::worker::reserve_response_attempt_id();
        request.capture_guard.set_marker(attempt_id.clone());
        if let Err(error) = crate::worker::write_response_marker(
            std::path::Path::new(&self.config.atif_spool_dir),
            &self.journal_key,
            &request.session.id,
            &attempt_id,
            request_digest,
        )
        .await
        {
            // Fail closed: the marker is what lets a restart-time scan
            // (`inflight_response_sessions`) quarantine sessions whose
            // provider response may have been observed but never
            // captured. Dispatching without it would silently drop that
            // crash-durability guarantee, so refuse before upstream I/O
            // — terminating the journaled response attempt like every
            // other post-admission abort (see `abandon_prepared`).
            tracing::warn!(
                %error,
                session = %request.session.id,
                "could not write in-flight response marker; refusing upstream dispatch"
            );
            let client_error =
                PipelineError::unavailable("in-flight response marker could not be persisted".to_owned());
            self.abandon_prepared(&mut request, StopReason::Other, &client_error.to_string());
            return Err(client_error);
        }
        let PreparedRequest {
            session,
            identity,
            lease,
            capture_guard,
            mut admission_debit,
            ..
        } = request;
        let UpstreamPlan {
            url,
            bodies,
            headers,
            auth,
            translation,
        } = plan;
        // Upstream latency (time-to-response-headers)
        // and error counters — the two failure-mode-table rows the
        // review marked "logs only, no metric".
        let upstream_started = Instant::now();
        // Last pre-dispatch point: from here the request may reach the
        // provider, so the admission debit stands (the `is_connect` arm
        // of `upstream_send_failed` refunds the one provably-not-sent
        // case). Disarming here also ends the cancellation-refund
        // window — a disconnect during `send()` keeps the debit,
        // deliberately.
        let debited_tokens = admission_debit.take_for_dispatch();
        let sends = bodies.into_iter().map(|body| {
            let mut upstream_request = self
                .client
                .post(&url)
                .header(reqwest::header::CONTENT_TYPE, "application/json")
                .body(body);
            for (name, value) in &headers {
                upstream_request = upstream_request.header(name.clone(), value.clone());
            }
            if let Some((name, value)) = &auth {
                upstream_request = upstream_request.header(name.clone(), value.clone());
            }
            upstream_request.send()
        });
        // Every choice of a fan-out is in flight at once; the slowest
        // one bounds the time to headers.
        let results = futures::future::join_all(sends).await;
        self.hot_metrics
            .upstream_latency_histogram
            .observe_us(elapsed_us(upstream_started));
        let mut responses = Vec::with_capacity(results.len());
        let mut first_error = None;
        for result in results {
            match result {
                Ok(response) => responses.push(response),
                Err(error) => {
                    if first_error.is_none() {
                        first_error = Some(error);
                    }
                }
            }
        }
        if let Some(error) = first_error {
            // Sibling responses of a fan-out drop here, which cancels
            // them: a request that only partly reached the provider is
            // reported as one upstream failure. A sibling that answered
            // proves the provider saw (and may bill) this request, so a
            // connect failure on another sibling must not refund it.
            let reached_provider = !responses.is_empty();
            drop(responses);
            return Err(self.upstream_send_failed(
                error,
                session,
                identity,
                capture_guard,
                debited_tokens,
                reached_provider,
            ));
        }
        let Some(translation) = translation else {
            let Some(response) = responses.pop() else {
                // Unreachable by construction (the OpenAI plan carries
                // exactly one body); keep the guard armed so its drop
                // retires the marker and closes the attempt.
                drop(capture_guard);
                self.enqueue_failure(
                    session,
                    identity,
                    StopReason::Other,
                    "upstream_no_response".to_owned(),
                )?;
                return Err(PipelineError::upstream("no upstream response".to_owned()));
            };
            if response.status().is_server_error() {
                #[allow(clippy::indexing_slicing)]
                {
                    self.hot_metrics.upstream_errors_counters[UpstreamErrorKind::Http5xx.index()].inc();
                }
            }
            return Ok(ForwardedResponse {
                status: response.status(),
                headers: response.headers().clone(),
                body: response
                    .bytes_stream()
                    .map(|chunk| chunk.map_err(std::io::Error::other))
                    .boxed(),
                billed_prompt_tokens: debited_tokens,
                lease,
                capture_guard,
            });
        };
        Ok(self
            .translate_native_responses(translation, responses, debited_tokens, lease, capture_guard)
            .await)
    }

    /// Build the upstream request(s) for one prepared payload: URL,
    /// bodies (one per Anthropic choice, otherwise exactly one), extra
    /// headers, the credential header, and — for the native dialects —
    /// how to translate the answer back.
    fn plan_upstream(
        &self,
        payload: &Value,
        payload_bytes: Option<Vec<u8>>,
        client_authorization: Option<HeaderValue>,
        upstream_passthrough_headers: Vec<(HeaderName, HeaderValue)>,
    ) -> Result<UpstreamPlan, PipelineError> {
        use crate::provider::{self, ProviderKind};
        let base = self.config.upstream_url.trim_end_matches('/');
        let auth = self.auth_for_dispatch(client_authorization)?;
        match self.provider {
            ProviderKind::OpenAi => {
                let body = match payload_bytes {
                    Some(bytes) => bytes,
                    // Reuse the digest's serialization for the upstream
                    // body; the theoretical to_vec failure above gets a
                    // second attempt here.
                    None => serde_json::to_vec(payload).map_err(|error| {
                        PipelineError::bad_request(format!("chat payload is not serializable: {error}"))
                    })?,
                };
                Ok(UpstreamPlan {
                    url: format!("{base}{}", self.config.upstream_chat_path),
                    bodies: vec![body],
                    // Allowlisted provider-semantic headers (org/project/
                    // beta) relay verbatim so org-scoped keys and
                    // beta-gated features keep working through the proxy.
                    headers: upstream_passthrough_headers,
                    auth,
                    translation: None,
                })
            }
            ProviderKind::Anthropic => {
                let body =
                    provider::anthropic::build_request(payload, self.config.provider_default_max_tokens)
                        .map_err(|reason| {
                            PipelineError::bad_request(format!(
                                "request cannot be expressed in the Anthropic Messages API: {reason}"
                            ))
                        })?;
                let bytes = serde_json::to_vec(&body).map_err(|error| {
                    PipelineError::bad_request(format!("translated request is not serializable: {error}"))
                })?;
                let choices = provider::choice_count(payload).map_err(PipelineError::bad_request)?;
                let count = usize::try_from(choices).unwrap_or(1);
                Ok(UpstreamPlan {
                    url: format!("{base}{}", provider::anthropic::MESSAGES_PATH),
                    bodies: vec![bytes; count],
                    headers: vec![(
                        HeaderName::from_static("anthropic-version"),
                        HeaderValue::from_static(provider::ANTHROPIC_VERSION),
                    )],
                    auth,
                    translation: Some(Translation {
                        dialect: NativeDialect::Anthropic,
                        stream: provider::wants_stream(payload),
                        include_usage: provider::wants_usage_chunk(payload),
                        id: provider::fresh_completion_id(),
                        model: payload
                            .get("model")
                            .and_then(Value::as_str)
                            .unwrap_or_default()
                            .to_owned(),
                        created: provider::created_now(),
                    }),
                })
            }
            ProviderKind::Gemini => {
                let model = payload
                    .get("model")
                    .and_then(Value::as_str)
                    .filter(|model| !model.is_empty())
                    .ok_or_else(|| PipelineError::bad_request("model is required".to_owned()))?;
                let stream = provider::wants_stream(payload);
                let path =
                    provider::gemini::request_path(model, stream).map_err(PipelineError::bad_request)?;
                let body = provider::gemini::build_request(payload).map_err(|reason| {
                    PipelineError::bad_request(format!(
                        "request cannot be expressed in the Gemini generateContent API: {reason}"
                    ))
                })?;
                let bytes = serde_json::to_vec(&body).map_err(|error| {
                    PipelineError::bad_request(format!("translated request is not serializable: {error}"))
                })?;
                Ok(UpstreamPlan {
                    url: format!("{base}{path}"),
                    bodies: vec![bytes],
                    headers: Vec::new(),
                    auth,
                    translation: Some(Translation {
                        dialect: NativeDialect::Gemini,
                        stream,
                        include_usage: provider::wants_usage_chunk(payload),
                        id: provider::fresh_completion_id(),
                        model: model.to_owned(),
                        created: provider::created_now(),
                    }),
                })
            }
        }
    }

    /// The credential header for one dispatch: the static key resolved
    /// at boot, else (passthrough mode) the client's own credential.
    /// For a native dialect the client's bearer token travels under the
    /// provider's key header without the `Bearer` scheme, which is what
    /// those APIs read.
    fn auth_for_dispatch(
        &self,
        client_authorization: Option<HeaderValue>,
    ) -> Result<Option<(HeaderName, HeaderValue)>, PipelineError> {
        if let Some(auth) = &self.upstream_auth {
            return Ok(Some(auth.clone()));
        }
        let Some(client) = client_authorization else {
            return Ok(None);
        };
        // Passthrough mode: the client's own credential travels to the
        // upstream. `validate()` guarantees this is mutually exclusive
        // with static keys and with NHI identity enforcement.
        let Some(header) = self.provider.native_auth_header() else {
            return Ok(Some((axum::http::header::AUTHORIZATION, client)));
        };
        let raw = client
            .to_str()
            .map_err(|_| PipelineError::bad_request("authorization header is not valid text".to_owned()))?;
        let token = raw
            .strip_prefix("Bearer ")
            .or_else(|| raw.strip_prefix("bearer "))
            .unwrap_or(raw)
            .trim();
        let mut value = HeaderValue::from_str(token).map_err(|_| {
            PipelineError::bad_request("authorization token is not a valid header value".to_owned())
        })?;
        value.set_sensitive(true);
        Ok(Some((HeaderName::from_static(header), value)))
    }

    /// Turn the native responses of one request into an OpenAI-shaped
    /// [`ForwardedResponse`]: a non-success status fails the whole
    /// request with that status and the OpenAI error envelope; a
    /// streaming request becomes one merged `chat.completion.chunk`
    /// stream; a buffered request becomes one `chat.completion` object
    /// (choices merged across the fan-out, usage summed).
    async fn translate_native_responses(
        &self,
        translation: Translation,
        mut responses: Vec<reqwest::Response>,
        debited_tokens: u64,
        lease: SessionLease,
        capture_guard: ResponseCaptureGuard,
    ) -> ForwardedResponse {
        use crate::provider::{self, StreamTranslator};
        let Translation {
            dialect,
            stream,
            include_usage,
            id,
            model,
            created,
        } = translation;
        let provider_kind = match dialect {
            NativeDialect::Anthropic => provider::ProviderKind::Anthropic,
            NativeDialect::Gemini => provider::ProviderKind::Gemini,
        };
        if let Some(position) = responses
            .iter()
            .position(|response| !response.status().is_success())
        {
            let failed = responses.swap_remove(position);
            // Dropping the siblings cancels them.
            drop(responses);
            let status = failed.status();
            if status.is_server_error() {
                #[allow(clippy::indexing_slicing)]
                {
                    self.hot_metrics.upstream_errors_counters[UpstreamErrorKind::Http5xx.index()].inc();
                }
            }
            let mut headers = HeaderMap::new();
            headers.insert(
                axum::http::header::CONTENT_TYPE,
                HeaderValue::from_static("application/json"),
            );
            if let Some(retry_after) = failed.headers().get(axum::http::header::RETRY_AFTER) {
                headers.insert(axum::http::header::RETRY_AFTER, retry_after.clone());
            }
            // A failed error-body read still answers with the status;
            // the envelope then carries a generic message.
            let body = read_bounded(failed, MAX_ERROR_BODY_BYTES)
                .await
                .unwrap_or_default();
            let envelope = provider::openai_error_envelope(provider_kind, status.as_u16(), &body);
            return ForwardedResponse {
                status,
                headers,
                body: futures::stream::once(async move { Ok(Bytes::from(envelope.to_string())) }).boxed(),
                billed_prompt_tokens: debited_tokens,
                lease,
                capture_guard,
            };
        }
        let mut headers = HeaderMap::new();
        let body: futures::stream::BoxStream<'static, Result<Bytes, std::io::Error>> = if stream {
            headers.insert(
                axum::http::header::CONTENT_TYPE,
                HeaderValue::from_static("text/event-stream"),
            );
            let streams = responses
                .into_iter()
                .enumerate()
                .map(|(position, response)| {
                    let index = u64::try_from(position).unwrap_or(u64::MAX);
                    let translator: Box<dyn StreamTranslator> = match dialect {
                        NativeDialect::Anthropic => {
                            Box::new(provider::anthropic::AnthropicStreamTranslator::new(
                                index,
                                id.clone(),
                                model.clone(),
                                created,
                            ))
                        }
                        NativeDialect::Gemini => Box::new(provider::gemini::GeminiStreamTranslator::new(
                            id.clone(),
                            model.clone(),
                            created,
                        )),
                    };
                    provider::TranslatingStream::new(
                        response
                            .bytes_stream()
                            .map(|chunk| chunk.map_err(std::io::Error::other))
                            .boxed(),
                        translator,
                    )
                })
                .collect();
            provider::MergedOpenAiStream::new(streams, include_usage, id, model, created).boxed()
        } else {
            headers.insert(
                axum::http::header::CONTENT_TYPE,
                HeaderValue::from_static("application/json"),
            );
            futures::stream::once(async move {
                let mut completions = Vec::with_capacity(responses.len());
                for (position, response) in responses.into_iter().enumerate() {
                    let bytes = read_bounded(response, MAX_NATIVE_BODY_BYTES).await?;
                    let value: Value = serde_json::from_slice(&bytes).map_err(|error| {
                        std::io::Error::other(format!("upstream body is not JSON: {error}"))
                    })?;
                    let index = u64::try_from(position).unwrap_or(u64::MAX);
                    let completion = match dialect {
                        NativeDialect::Anthropic => {
                            provider::anthropic::translate_message(&value, &id, created, index)
                        }
                        NativeDialect::Gemini => {
                            provider::gemini::translate_response(&value, &id, created, &model)
                        }
                    }
                    .map_err(std::io::Error::other)?;
                    completions.push(completion);
                }
                let merged = if completions.len() == 1 {
                    completions
                        .pop()
                        .ok_or_else(|| std::io::Error::other("no upstream completion"))?
                } else {
                    provider::merge_completions(completions).map_err(std::io::Error::other)?
                };
                Ok(Bytes::from(merged.to_string()))
            })
            .boxed()
        };
        ForwardedResponse {
            status: StatusCode::OK,
            headers,
            body,
            billed_prompt_tokens: debited_tokens,
            lease,
            capture_guard,
        }
    }

    /// Account for and report one failed upstream `send()`: metrics,
    /// the connect-failure refund, redacted logging, and the terminal
    /// failure capture that retires the in-flight marker.
    fn upstream_send_failed(
        &self,
        error: reqwest::Error,
        session: Arc<Session>,
        identity: AgentIdentity,
        mut capture_guard: ResponseCaptureGuard,
        debited_tokens: u64,
        reached_provider: bool,
    ) -> PipelineError {
        let kind = UpstreamErrorKind::from_reqwest(&error);
        #[allow(clippy::indexing_slicing)]
        {
            self.hot_metrics.upstream_errors_counters[kind.index()].inc();
        }
        // `reqwest::Error::Display` embeds the
        // request URL (see the redaction rule at
        // routes.rs::read_limited_tool_response — the same
        // rule applies here). `%error` on `reqwest::Error`
        // would render `error sending request for url
        // (https://api.openai.com/v1/chat/completions): dns
        // error: ...` into every OTLP sink. In multi-tenant
        // / air-gapped deployments the upstream URL is not
        // something the operator wants leaked to the
        // customer's SIEM (it may be an internal LiteLLM
        // router, an on-prem Azure resource name, a
        // per-tenant model deployment path). Log the
        // stable classifier + reqwest's structured
        // predicates only.
        //
        // Also propagate the classifier — NOT
        // `error.to_string()` — into the persisted failure
        // event so the URL doesn't land in the on-disk
        // journal or in the Bridge-published failure
        // record either.
        let client_reason = classify_upstream_error(&error);
        // `is_connect()` proves the request never left this
        // process (TCP/TLS establishment failed) — the admission
        // token debit bought no LLM work; refund it on BOTH
        // ledgers, in the same order they were spent. Any other
        // send error (timeout, body, reset mid-response) may have
        // reached the provider, so the debit conservatively
        // stands. Best-effort; a store blip must never turn a
        // 502 into a 5xx cascade.
        // A sibling of an Anthropic fan-out that got an answer counts
        // as "may have reached the provider" for the whole request.
        if error.is_connect() && !reached_provider {
            ActionBudget::new(self.store.as_ref(), &session.id, &self.effective_budget())
                .refund_tokens(debited_tokens);
            if let Some(spec) = self.config.principal_budget.as_ref() {
                let principal_id = principal_id_for_budget(&identity);
                ActionBudget::for_principal(self.store.as_ref(), &principal_id, spec)
                    .refund_tokens(debited_tokens);
            }
        }
        tracing::warn!(
            session = %session.id,
            category = client_reason,
            error.status = ?error.status(),
            error.is_timeout = error.is_timeout(),
            error.is_connect = error.is_connect(),
            error.is_body = error.is_body(),
            "upstream forwarding failed"
        );
        // Route timeouts (upstream accepted the connection but
        // never finished responding) to 504 Gateway Timeout
        // instead of the generic 502 Bad Gateway. Operator
        // dashboards keying off HTTP status can now distinguish
        // "provider slow" (504) from "provider unreachable"
        // (502) without joining against the metric label.
        let client_error = if error.is_timeout() {
            PipelineError::upstream_timeout(client_reason.to_owned())
        } else {
            PipelineError::upstream(client_reason.to_owned())
        };
        let persisted_reason = format!("upstream_{client_reason}");
        if let Some(permit) = capture_guard.take_permit() {
            let (response_marker, response_attempt_id) = capture_guard.disarm();
            let capture_session = Arc::clone(&session);
            let mut job = self.failure_job(session, identity, StopReason::Other, persisted_reason);
            job.response_marker = response_marker;
            job.response_attempt = Some(crate::worker::ResponseAttempt {
                id: response_attempt_id,
                terminal: true,
            });
            // This terminal failure capture is the only thing that
            // retires the durable in-flight marker and the
            // journalled response attempt. If the shard is full,
            // fail the session closed instead of dropping the
            // capture silently: otherwise the marker strands, no
            // terminal response event exists, and the session can
            // still close "cleanly" with a receipt over an
            // unresolved response attempt.
            if permit.submit(&self.worker, job).is_err() {
                capture_session.mark_capture_failed();
            }
        } else {
            // Defensive (a permit always exists here): keep the
            // guard armed so its drop retires the marker and
            // closes the attempt through the plain worker queue.
            drop(capture_guard);
            if let Err(error) = self.enqueue_failure(session, identity, StopReason::Other, persisted_reason) {
                return error;
            }
        }
        client_error
    }

    /// Intercept one MCP JSON-RPC tool call, emit its OCSF verdict
    /// asynchronously, and return the immediate authorization decision.
    pub fn intercept_tool(&self, headers: &HeaderMap, raw: &[u8]) -> Result<ToolVerdict, PipelineError> {
        self.intercept_tool_with_session(headers, raw)
            .map(|(verdict, _session)| verdict)
    }

    /// Cheap pre-parse identity gate for the tool paths. When identity
    /// is REQUIRED, a request with no Authorization header can never be
    /// admitted — refuse before `parse_tool_call` /
    /// `ToolExecution::from_request` run their duplicate-key scan, JSON
    /// parse, canonicalization, and hashing. The chat path resolves
    /// identity before any body processing; the tool paths parsed FIRST
    /// (the required scope depends on the parsed tool name), handing an
    /// unauthenticated flood the expensive pre-auth work — bounded by
    /// the MCP admission cap, but blocking-pool cycles anonymous
    /// garbage shouldn't get. A present-but-garbage bearer still costs
    /// one signature verification before the parse (the full
    /// scope-aware `resolve_identity` still runs afterwards) — the same
    /// cost order as the chat path.
    pub(crate) fn refuse_unauthenticated_tool_call(&self, headers: &HeaderMap) -> Result<(), PipelineError> {
        if !self.config.require_identity {
            return Ok(());
        }
        if single_header(headers, "authorization")?.is_none() {
            // Mirrors resolve_identity's (None, _) require_identity arm,
            // including the identity-rejection audit event.
            let session_id = session_id(headers)?;
            let error = PipelineError::Unauthorized("missing bearer token".to_owned());
            self.enqueue_transient_failure(&session_id, StopReason::IdentityRejected, error.to_string())?;
            return Err(error);
        }
        Ok(())
    }

    /// Core of [`Self::intercept_tool`] that also returns the bound session.
    ///
    /// Callers that must await audit durability need the same session the
    /// verdict was recorded under: re-deriving it from headers would mint a
    /// fresh random id for a header-less request (see [`session_id`]) and
    /// look up a session that does not exist.
    fn intercept_tool_with_session(
        &self,
        headers: &HeaderMap,
        raw: &[u8],
    ) -> Result<(ToolVerdict, Arc<Session>), PipelineError> {
        let caller = match self.authenticate(headers) {
            Ok(caller) => caller,
            Err(error) => {
                self.audit_authentication_error(headers, &error)?;
                return Err(error);
            }
        };
        self.intercept_tool_authenticated(headers, raw, &caller, None)
    }

    fn intercept_tool_authenticated(
        &self,
        headers: &HeaderMap,
        raw: &[u8],
        caller: &AuthenticatedCaller,
        preflight_denial: Option<ToolVerdict>,
    ) -> Result<(ToolVerdict, Arc<Session>), PipelineError> {
        let session_id = session_id(headers)?;
        let workflow = workflow(headers, &self.config.default_workflow)?;
        let parsed_call = pv_sandbox::parse_tool_call(raw).ok();
        let required_scope = parsed_call
            .as_ref()
            .map(|request| tool_scope(&request.tool))
            .unwrap_or_else(|| TOOL_INVOKE_SCOPE.to_owned());
        let identity = match caller.authorize(self, Some(&required_scope)) {
            Ok(identity) => identity,
            Err(error) => {
                self.enqueue_transient_failure(&session_id, StopReason::IdentityRejected, error.to_string())?;
                return Err(error);
            }
        };
        // Tool interception must NOT resurrect a closed session: a
        // tool call extends an in-progress conversation, so silently
        // opening a fresh session under the same id would let the
        // client accumulate tool calls past a signed receipt boundary
        // (chain-of-custody split). Chat requests use `get_or_open`
        // (recycles closed ids); the strict `_no_reopen` variant here
        // makes the `is_closed()` check below fire.
        let session =
            self.sessions
                .get_or_open_no_reopen(&session_id, workflow, &identity, &self.config.breaker);
        let admission = session.admission_guard();
        validate_session_binding(&session, workflow, &identity)?;
        caller.bind_session(&session)?;
        session.refresh_identity(&identity);
        if session.is_closed() {
            return Err(PipelineError::bad_request("session is already closed".to_owned()));
        }
        if session.capture_failed() {
            return Err(PipelineError::unavailable(
                "session audit capture is incomplete".to_owned(),
            ));
        }
        session.touch();
        // The semantic-loop breaker gates CHAT admission (both the
        // prepare arm above and the streaming re-check), but tool
        // execution ignored it entirely: a runaway agent whose chat is
        // already refused/aborted could keep firing otherwise-permitted
        // MCP tool calls on the same session until idle close — the
        // "stop the runaway agent" decision only stopped half the
        // agent. Gate the hard-stop actions here. `Inject` deliberately
        // passes: its remedy is a corrective CHAT message (the next
        // chat turn injects and resets the breaker); blocking tools
        // under Inject would strand tool-driven agents the operator
        // chose to steer rather than stop.
        if session.loop_state.state() == BreakerState::Open {
            match session.loop_state.action() {
                BreakerAction::Reject | BreakerAction::Abort => {
                    session.latch_enforcement(StopReason::LoopDetected);
                    return Err(PipelineError::blocked(
                        "semantic loop circuit breaker is open".to_owned(),
                    ));
                }
                _ => {}
            }
        }
        let worker_permit = self
            .worker
            .try_reserve(&session_id)
            .map_err(PipelineError::unavailable_source)?;
        let budgeted_check = || {
            let principal_binding = self
                .config
                .principal_budget
                .as_ref()
                .map(|spec| (principal_id_for_budget(&identity), spec));
            let principal_ref = principal_binding.as_ref().map(|(id, spec)| (id.as_str(), *spec));
            self.sandbox
                .check_with_principal(self.store.as_ref(), &session_id, principal_ref, raw)
        };
        let verdict = if let Some(denial) = preflight_denial {
            denial
        } else {
            match parsed_call.as_ref() {
                // `sandbox.check`'s budget gate spends before we return, so we must
                // veto workflow-mismatched consequential tools before it runs.
                Some(request)
                    if workflow == Workflow::Unsigned
                        && self
                            .config
                            .consequential_tools
                            .iter()
                            .any(|required| required == &request.tool) =>
                {
                    let reason = format!("tool {:?} requires a signed workflow", request.tool);
                    ToolVerdict::Blocked {
                        tool: request.tool.clone(),
                        stage: "policy",
                        // Built-in workflow gate, named so console-side
                        // attribution has a stable handle an operator can
                        // create a matching policy row for.
                        policy: Some("workflow.signed_required".to_owned()),
                        denial_code: pv_sandbox::DenialCode::PolicyDenied,
                        reason: reason.clone(),
                        response: pv_sandbox::rpc::denial_error(
                            request.id.as_ref(),
                            pv_sandbox::DenialCode::PolicyDenied,
                            &reason,
                        ),
                        elapsed_us: 0,
                    }
                }
                // Same pre-spend veto for the intent map and the active
                // mission: a denied call must neither consume budget nor be
                // audited as allowed. This also covers verdict-only mode, where
                // no forwarding branch ever runs.
                Some(request) => {
                    match self
                        .pdp
                        .decide_for(&request.tool, pv_core::time::now_ms() / 1000, &caller.facts())
                    {
                        crate::authz::AuthzDecision::Deny { code, reason } => {
                            self.metrics
                                .counter(
                                    &format!("pv_pdp_denials_total{{code=\"{}\"}}", code.as_str()),
                                    "PDP denials by denial code",
                                )
                                .inc();
                            ToolVerdict::Blocked {
                                tool: request.tool.clone(),
                                stage: "policy",
                                policy: Some(pdp_policy_name(code).to_owned()),
                                denial_code: code,
                                response: pv_sandbox::rpc::denial_error(request.id.as_ref(), code, &reason),
                                reason,
                                elapsed_us: 0,
                            }
                        }
                        crate::authz::AuthzDecision::Permit { .. } => budgeted_check(),
                    }
                }
                None => budgeted_check(),
            }
        };
        let (status, payload) = match &verdict {
            ToolVerdict::Allowed {
                tool,
                budget_remaining,
                elapsed_us,
                ..
            } => (
                StatusId::Success,
                serde_json::json!({
                    "tool": tool,
                    "allowed": true,
                    "budget_remaining": (*budget_remaining != u64::MAX)
                        .then_some(*budget_remaining),
                    "budget_unlimited": *budget_remaining == u64::MAX,
                    "decision_us": elapsed_us,
                }),
            ),
            ToolVerdict::Blocked {
                tool,
                stage,
                policy,
                denial_code,
                reason,
                elapsed_us,
                ..
            } => (
                StatusId::Failure,
                serde_json::json!({
                    "tool": tool,
                    "allowed": false,
                    "stage": stage,
                    // Present only for named policy-chain denials —
                    // consumed by console-sync as the per-policy
                    // attribution (`payload.policy` → ingest policyName).
                    "policy": policy,
                    "denial_code": denial_code.as_str(),
                    "reason": reason,
                    "decision_us": elapsed_us,
                }),
            ),
            // `ToolVerdict` is `#[non_exhaustive]` (R56 SemVer
            // hardening) — a future variant (e.g. `Warn` for shadow-
            // mode, `Deferred` for async policy evaluation) must NOT
            // be silently forwarded to the tool server. Fail-closed:
            // emit a Blocked-shaped audit payload and log at ERROR
            // level so operators discover the unhandled variant
            // during rollout. Same defensive posture the sandbox's
            // own `is_allowed` uses (whitelist Allowed, treat
            // everything else as blocked).
            _ => {
                tracing::error!(
                    "unhandled ToolVerdict variant reached the harness matcher; \
                     failing closed. This is a bug — extend the match to handle the new variant."
                );
                (
                    StatusId::Failure,
                    serde_json::json!({
                        "tool": "<unknown>",
                        "allowed": false,
                        "stage": "unhandled_verdict",
                        "reason": "unhandled ToolVerdict variant; fail-closed",
                    }),
                )
            }
        };
        let tool_call_id = parsed_call
            .as_ref()
            .and_then(|request| request.id.as_ref())
            .map(|value| value.as_str().map_or_else(|| value.to_string(), str::to_owned))
            .unwrap_or_else(pv_core::new_event_uid);
        let tool_calls = parsed_call.as_ref().map(|request| {
            vec![pv_atif::ToolCall {
                tool_call_id: tool_call_id.clone(),
                function_name: request.tool.clone(),
                arguments: request.arguments.clone(),
                extra: None,
            }]
        });
        let observation = parsed_call.as_ref().map(|_| pv_atif::Observation {
            results: vec![pv_atif::ObservationResult {
                source_call_id: Some(tool_call_id),
                content: Some(Value::String(payload.to_string())),
                subagent_trajectory_ref: None,
                extra: None,
            }],
        });
        let stop_reason = match &verdict {
            ToolVerdict::Allowed { .. } => None,
            ToolVerdict::Blocked { stage: "budget", .. } => Some(StopReason::BudgetExceeded),
            ToolVerdict::Blocked { .. } => Some(StopReason::PolicyBlocked),
            // See the sibling `match &verdict` above for the rationale.
            // R58 correction: use `StopReason::Other` (the ships-with-
            // `#[serde(other)]` forward-compat sink documented for
            // exactly this "unrecognised upstream variant during
            // heterogeneous rolling upgrade" case) rather than
            // `PolicyBlocked`. Returning `PolicyBlocked` here would
            // contaminate policy-block counters / alerts / dashboards
            // with a HARNESS CODE BUG signal, and disagree with the
            // sibling `stage: "unhandled_verdict"` payload label that
            // is honestly synthetic. `Other` preserves fail-closed
            // semantics (non-None, terminates the turn, seals the
            // receipt) while telling the truth on the wire.
            _ => Some(StopReason::Other),
        };
        worker_permit.submit(WorkerJob {
            session: Arc::clone(&session),
            identity,
            class: EventClass::ToolCall,
            payload,
            text: String::from_utf8_lossy(raw).into_owned(),
            analyze_loop: false,
            status,
            stop_reason,
            native_stop_reason: None,
            metrics: EventMetrics::default(),
            cost_usd_micros: 0,
            prompt_token_correction: 0,
            atif: Some(AtifCapture {
                source: pv_atif::Source::Agent,
                message: Value::String("MCP tool authorization decision".to_owned()),
                reasoning_content: None,
                model_name: None,
                tool_calls,
                observation,
                llm_call_count: Some(0),
            }),
            response_marker: None,
            response_attempt: None,
        });
        drop(admission);
        Ok((verdict, session))
    }

    /// Authorize a tool call and wait for its verdict event to become durable.
    pub async fn intercept_tool_durable(
        &self,
        headers: &HeaderMap,
        raw: &[u8],
    ) -> Result<ToolVerdict, PipelineError> {
        let state = self.clone();
        let owned_headers = headers.clone();
        let raw = raw.to_vec();
        let (verdict, session) =
            tokio::task::spawn_blocking(move || state.intercept_tool_with_session(&owned_headers, &raw))
                .await
                .map_err(PipelineError::unavailable_source)??;
        session.wait_for_worker_jobs().await;
        if session.capture_failed() {
            return Err(PipelineError::unavailable(
                "tool authorization audit capture failed".to_owned(),
            ));
        }
        Ok(verdict)
    }

    /// Authorize a tool call on the blocking pool without waiting for the
    /// off-path event journal or broker publication.
    pub async fn intercept_tool_nonblocking(
        &self,
        headers: &HeaderMap,
        raw: &[u8],
    ) -> Result<ToolVerdict, PipelineError> {
        let state = self.clone();
        let owned_headers = headers.clone();
        let raw = raw.to_vec();
        tokio::task::spawn_blocking(move || state.intercept_tool(&owned_headers, &raw))
            .await
            .map_err(PipelineError::unavailable_source)?
    }

    pub(crate) async fn intercept_authenticated_nonblocking(
        &self,
        headers: &HeaderMap,
        raw: &[u8],
        caller: AuthenticatedCaller,
        preflight_denial: Option<ToolVerdict>,
    ) -> Result<ToolVerdict, PipelineError> {
        let state = self.clone();
        let headers = headers.clone();
        let raw = raw.to_vec();
        tokio::task::spawn_blocking(move || {
            state
                .intercept_tool_authenticated(&headers, &raw, &caller, preflight_denial)
                .map(|(verdict, _)| verdict)
        })
        .await
        .map_err(PipelineError::unavailable_source)?
    }

    pub(crate) fn lease_session(&self, headers: &HeaderMap) -> Result<SessionLease, PipelineError> {
        let id = session_id(headers)?;
        let session = self
            .sessions
            .get(&id)
            .ok_or_else(|| PipelineError::bad_request("unknown session".to_owned()))?;
        session
            .try_lease()
            .ok_or_else(|| PipelineError::bad_request("session is already closed".to_owned()))
    }

    pub(crate) fn enqueue_transient_failure(
        &self,
        session_id: &str,
        stop_reason: StopReason,
        reason: String,
    ) -> Result<(), PipelineError> {
        // Every identity rejection
        // used to mint a DISTINCT signed audit session — two fsynced
        // spool files, a bridge publish, a reconciler adoption and an
        // Ed25519 signature per credential-free 401 probe. A curl loop
        // against a hardened deployment turned each rejection into
        // durable server-side work that nothing ever coalesced. Keep
        // the full per-rejection audit record for a bounded number of
        // rejections per minute (enough for forensic sampling), and
        // beyond that budget count-only: the metric and a rate-limited
        // WARN preserve the signal while the signed-receipt lifecycle
        // is no longer attacker-drivable.
        self.metrics
            .counter(
                "pv_identity_rejections_total",
                "Requests refused by identity validation",
            )
            .inc();
        const MAX_AUDITED_REJECTIONS_PER_MINUTE: u32 = 60;
        {
            let now = Instant::now();
            let mut window = self.identity_rejection_window.lock();
            if now.duration_since(window.0) >= std::time::Duration::from_secs(60) {
                *window = (now, 0);
            }
            if window.1 >= MAX_AUDITED_REJECTIONS_PER_MINUTE {
                self.metrics
                    .counter(
                        "pv_identity_rejections_unaudited_total",
                        "Identity rejections beyond the per-minute audit-record budget (counted only)",
                    )
                    .inc();
                return Ok(());
            }
            window.1 += 1;
        }
        let audit_session_id = format!("identity-rejected-{}", pv_core::new_event_uid());
        let identity = AgentIdentity {
            version: "unknown".to_owned(),
            charter: "identity-rejected".into(),
            instance_uid: audit_session_id.clone(),
            ttl_remaining_s: None,
        };
        let session = Arc::new(Session::new(
            audit_session_id,
            Workflow::Signed,
            identity.clone(),
            self.config.breaker.clone(),
        ));
        // Cap the attacker-controlled `session_id` echo so a
        // maliciously long header cannot bloat every audit record.
        // 64 bytes is enough to keep well-formed UUIDs and legitimate
        // client-chosen ids intact; anything larger is truncated with
        // a marker so operators can still tell what the caller sent.
        const MAX_ECHO: usize = 64;
        let mut boundary = MAX_ECHO.min(session_id.len());
        while boundary < session_id.len() && !session_id.is_char_boundary(boundary) {
            boundary -= 1;
        }
        let truncated = boundary < session_id.len();
        let echo = &session_id[..boundary];
        let echo_marker = if truncated { "…(truncated)" } else { "" };
        self.enqueue_failure(
            session,
            identity,
            stop_reason,
            format!("requested session {echo:?}{echo_marker}: {reason}"),
        )
    }

    fn enqueue_failure(
        &self,
        session: Arc<Session>,
        identity: AgentIdentity,
        stop_reason: StopReason,
        reason: String,
    ) -> Result<(), PipelineError> {
        self.worker
            .try_submit(self.failure_job(session, identity, stop_reason, reason))
            .map_err(PipelineError::unavailable_source)
    }

    /// A failure job for a request a console-pushed policy refused: the
    /// stop-reason payload also names the policy, so the console can
    /// attribute the block to it (the tool path does the same through
    /// `ToolVerdict::Blocked::policy`).
    fn policy_failure_job(
        &self,
        session: Arc<Session>,
        identity: AgentIdentity,
        stop_reason: StopReason,
        reason: String,
        policy: Option<String>,
    ) -> WorkerJob {
        let mut job = self.failure_job(session, identity, stop_reason, reason);
        if let (Some(policy), Value::Object(payload)) = (policy, &mut job.payload) {
            payload.insert("policy".to_owned(), Value::String(policy));
        }
        job
    }

    fn failure_job(
        &self,
        session: Arc<Session>,
        identity: AgentIdentity,
        stop_reason: StopReason,
        reason: String,
    ) -> WorkerJob {
        let atif = (session.workflow == Workflow::Unsigned).then(|| AtifCapture {
            source: pv_atif::Source::Agent,
            message: Value::String(reason.clone()),
            reasoning_content: None,
            model_name: None,
            tool_calls: None,
            observation: None,
            llm_call_count: Some(0),
        });
        WorkerJob {
            session,
            identity,
            class: EventClass::StopReason,
            payload: serde_json::json!({"reason": reason}),
            text: String::new(),
            analyze_loop: false,
            status: StatusId::Failure,
            stop_reason: Some(stop_reason),
            native_stop_reason: None,
            metrics: EventMetrics::default(),
            cost_usd_micros: 0,
            prompt_token_correction: 0,
            atif,
            response_marker: None,
            response_attempt: None,
        }
    }

    #[cfg(test)]
    pub(crate) fn authorize_session(
        &self,
        headers: &HeaderMap,
        session: &Session,
        required_scope: &str,
    ) -> Result<(), PipelineError> {
        self.authenticate(headers)?
            .authorize_session(self, session, required_scope)
    }

    pub(crate) fn resolve_identity(
        &self,
        headers: &HeaderMap,
        required_scope: Option<&str>,
    ) -> Result<AgentIdentity, PipelineError> {
        self.authenticate(headers)?.authorize(self, required_scope)
    }

    pub(crate) fn authenticate(&self, headers: &HeaderMap) -> Result<AuthenticatedCaller, PipelineError> {
        let bearer = single_header(headers, "authorization")?
            .and_then(|value| value.to_str().ok())
            .and_then(strip_bearer_scheme);
        if let (Some(token), Some(validator)) = (bearer, &self.identity) {
            self.metrics
                .counter(
                    "pv_identity_validations_total",
                    "Full inbound identity validations",
                )
                .inc();
            let validated = validator.validate(token).map_err(identity_error_to_pipeline)?;
            return Ok(AuthenticatedCaller {
                identity: validated.agent_identity(),
                validated: Some(validated),
            });
        }
        Ok(AuthenticatedCaller {
            identity: self.resolve_anonymous_identity(headers, None)?,
            validated: None,
        })
    }

    pub(crate) fn audit_authentication_error(
        &self,
        headers: &HeaderMap,
        error: &PipelineError,
    ) -> Result<(), PipelineError> {
        if matches!(
            error,
            PipelineError::Unauthorized(_) | PipelineError::Blocked { .. }
        ) {
            self.enqueue_transient_failure(
                &session_id(headers).unwrap_or_default(),
                StopReason::IdentityRejected,
                error.to_string(),
            )?;
        } else if matches!(error, PipelineError::Unavailable { .. }) {
            self.metrics
                .counter(
                    "pv_identity_unavailable_total",
                    "Identity validation dependency failures",
                )
                .inc();
        }
        Ok(())
    }

    fn resolve_anonymous_identity(
        &self,
        headers: &HeaderMap,
        required_scope: Option<&str>,
    ) -> Result<AgentIdentity, PipelineError> {
        // Reuse the duplicate-header refusal
        // pattern for `Authorization` on the identity hot path.
        // Previously `HeaderMap::get(AUTHORIZATION)` returned the
        // first value while the `PreparedRequest.client_authorization`
        // capture and header-smuggling proxies could observe a
        // merged `A, B` form — auth split-brain: the harness
        // authenticates as A while log aggregators / WAFs / OTLP
        // exporters attribute the request to B. Refuse the multi-
        // value case at ingress, symmetric with the X-PV-Session /
        // X-PV-Workflow guards.
        // RFC 7235 §2.1 declares the auth-scheme token
        // case-insensitive. Historically `strip_prefix("Bearer ")`
        // silently dropped `bearer eyJ...` / `BEARER eyJ...` /
        // `Bearer\teyJ...` to `None`, then the outer match arm
        // returned anonymous (when require_identity=false, the
        // shipped default) — the caller believed they had
        // authenticated while the audit trail attributed the
        // request to `anonymous`. That's the exact repudiation
        // vector the surrounding refuse-anonymous-with-token
        // guard was written to close. Now strip the scheme
        // case-insensitively.
        let bearer = single_header(headers, "authorization")?
            .and_then(|value| value.to_str().ok())
            .and_then(strip_bearer_scheme);
        match (bearer, &self.identity) {
            (Some(token), Some(validator)) => {
                let validated = validator.validate(token).map_err(identity_error_to_pipeline)?;
                if self.config.enforce_identity_scopes {
                    if let Some(required) = required_scope {
                        if !scope_allows(&validated.claims.scopes, required) {
                            // RFC 6750 §3.1: a syntactically valid, correctly
                            // signed token that lacks the required scope is an
                            // AUTHORIZATION failure — 403 insufficient_scope —
                            // not a 401. Returning 401 (the old behavior) told
                            // SDK token-refreshers to re-authenticate, which
                            // cannot help: the grant itself lacks the scope.
                            return Err(PipelineError::blocked(format!(
                                "identity scope {required:?} is required"
                            )));
                        }
                    }
                }
                Ok(validated.agent_identity())
            }
            // Client presented a bearer but the validator is not
            // configured. Two legitimate cases and one attack surface:
            //  (a) `upstream_authorization_passthrough = true`: the
            //      client's bearer is the CREDENTIAL FOR THE UPSTREAM
            //      PROVIDER, not an identity claim for us. Accept
            //      anonymous locally so the request proceeds and the
            //      relay layer forwards the header. This is the
            //      documented dev-mode BYO-key posture.
            //  (b) Neither passthrough nor validator: a bearer sent to
            //      a no-op harness is a configuration mismatch —
            //      REJECTED with a diagnostic 401 (see the closing
            //      paragraph below and the arm's else branch) rather
            //      than silently recorded as anonymous.
            //  (c) A validator EXISTS but this arm ran because
            //      match fell through: unreachable because arm 1
            //      catches (Some, Some). No action.
            //
            // The attack case — a caller sends a scoped identity token
            // and we silently attribute the request as `anonymous`
            // producing a repudiation vector — was closed at commit
            // e56... by rejecting here. But that broke passthrough
            // mode. Restore acceptance when passthrough is on so the
            // BYO-key path works, and continue rejecting only when
            // NO passthrough is configured and identity was not
            // required (i.e. the client presenting a bearer to a
            // no-op harness).
            (Some(_), None) => {
                let has_static_key =
                    self.config.upstream_api_key_env.is_some() || self.config.upstream_api_key_file.is_some();
                if self.config.upstream_authorization_passthrough {
                    Ok(AgentIdentity {
                        version: "dev".to_owned(),
                        charter: "anonymous".into(),
                        instance_uid: "anonymous".to_owned(),
                        ttl_remaining_s: None,
                    })
                } else if has_static_key || self.config.ignore_client_authorization {
                    // Static-key posture (the wizard's default): the
                    // operator has declared that THIS PROXY owns the
                    // upstream credential, and OpenAI-compatible SDKs
                    // unconditionally send a placeholder bearer (the
                    // client library refuses to start without one).
                    // Rejecting here made the flagship onboarding fail
                    // on request 1 with the official SDK. The
                    // repudiation rationale for the rejection below
                    // does not apply: exactly like the passthrough arm
                    // above, the config itself assigns the client
                    // bearer a non-identity meaning (a placeholder to
                    // be REPLACED by the static key — it is never
                    // forwarded). Accept anonymous; the placeholder
                    // never reaches the provider.
                    //
                    // `ignore_client_authorization` is the
                    // explicit opt-in covering the same posture for
                    // KEYLESS upstreams (Ollama, LM Studio, vLLM — no
                    // static key configured, so the implicit rule above
                    // cannot fire). validate() refuses the flag in
                    // combination with require_identity (the validator
                    // must see the header) and with passthrough (cannot
                    // both discard and forward).
                    tracing::debug!(
                        "client bearer ignored: static upstream key posture (placeholder \
                         SDK credential, not an identity claim)"
                    );
                    Ok(AgentIdentity {
                        version: "dev".to_owned(),
                        charter: "anonymous".into(),
                        instance_uid: "anonymous".to_owned(),
                        ttl_remaining_s: None,
                    })
                } else {
                    Err(PipelineError::Unauthorized(
                        "bearer token presented but identity validator is not configured — \
                         refusing to silently record the request as anonymous (either \
                         configure identity_jwks_url / identity_hmac_secret_file, or set \
                         upstream_authorization_passthrough=true if the bearer is meant \
                         for the upstream provider)"
                            .to_owned(),
                    ))
                }
            }
            (None, _) if self.config.require_identity => {
                Err(PipelineError::Unauthorized("missing bearer token".to_owned()))
            }
            _ => Ok(AgentIdentity {
                version: "dev".to_owned(),
                charter: "anonymous".into(),
                instance_uid: "anonymous".to_owned(),
                ttl_remaining_s: None,
            }),
        }
    }

    fn observe_stage(&self, stage: &str, started: Instant) {
        let elapsed = elapsed_us(started);
        // Prefer the pre-resolved handle when the caller's stage name
        // maps to a known Stage variant — that path is O(1) with no
        // mutex or allocation. Callers still passing legacy or
        // unrecognized labels fall through to the lazy registry path
        // so the metric remains discoverable (the lazy Registry call
        // uses the same shared Arc<Histogram> the hot cache points
        // to for known stages).
        if let Some(known) = Stage::from_label(stage) {
            // `Stage::index()` is a closed 0..5 map over the same
            // `Stage::ORDER` array both HotMetrics arrays are built
            // from — `.get()` keeps clippy's indexing lint satisfied
            // without a reachable None arm.
            if let Some(histogram) = self.hot_metrics.stage_histograms.get(known.index()) {
                histogram.observe_us(elapsed);
            }
            if self.hot_metrics.strict_stage_budget && elapsed > 2_000 {
                if let Some(counter) = self.hot_metrics.stage_strict_budget_counters.get(known.index()) {
                    counter.inc();
                }
                tracing::warn!(stage, elapsed_us = elapsed, "strict stage budget exceeded");
            }
            return;
        }
        self.metrics
            .histogram(
                &format!("pv_stage_duration_seconds{{stage=\"{stage}\"}}"),
                "Harness stage latency",
            )
            .observe_us(elapsed);
        if self.hot_metrics.strict_stage_budget && elapsed > 2_000 {
            self.metrics
                .counter(
                    &format!("pv_strict_budget_breaches_total{{stage=\"{stage}\"}}"),
                    "Middleware stages that exceeded the strict per-stage budget",
                )
                .inc();
            tracing::warn!(stage, elapsed_us = elapsed, "strict stage budget exceeded");
        }
    }
}

/// Authentication result confined to one request. It cannot be constructed by callers.
#[derive(Clone)]
pub(crate) struct AuthenticatedCaller {
    pub(crate) identity: AgentIdentity,
    pub(crate) validated: Option<pv_identity::ValidatedIdentity>,
}

impl AuthenticatedCaller {
    pub(crate) fn authorize(
        &self,
        state: &AppState,
        scope: Option<&str>,
    ) -> Result<AgentIdentity, PipelineError> {
        if let Some(validated) = &self.validated {
            let now = pv_core::time::now_ms() / 1000;
            let leeway = state.identity.as_ref().map_or(0, |v| v.leeway_secs());
            if now > validated.claims.exp.saturating_add(leeway) {
                return Err(PipelineError::Unauthorized("identity validation failed".into()));
            }
            if state.config.enforce_identity_scopes {
                if let Some(required) = scope {
                    if !scope_allows(&validated.claims.scopes, required) {
                        return Err(PipelineError::blocked(format!(
                            "identity scope {required:?} is required"
                        )));
                    }
                }
            }
            let mut identity = self.identity.clone();
            identity.ttl_remaining_s = Some(validated.claims.exp.saturating_sub(now));
            return Ok(identity);
        }
        Ok(self.identity.clone())
    }

    pub(crate) fn authorize_session(
        &self,
        state: &AppState,
        session: &Session,
        scope: &str,
    ) -> Result<(), PipelineError> {
        let identity = self.authorize(state, Some(scope))?;
        validate_session_binding(session, session.workflow, &identity)?;
        self.bind_session(session)?;
        session.refresh_identity(&identity);
        Ok(())
    }

    /// What the policy decision point may know about this caller.
    pub(crate) fn facts(&self) -> crate::authz::CallerFacts<'_> {
        match &self.validated {
            Some(validated) => crate::authz::CallerFacts {
                instance_uid: &validated.claims.instance_uid,
                charter: &validated.claims.charter,
                subject: Some(&validated.claims.sub),
            },
            None => crate::authz::CallerFacts {
                instance_uid: &self.identity.instance_uid,
                charter: "anonymous",
                subject: None,
            },
        }
    }

    pub(crate) fn principal_digest(&self) -> Option<String> {
        self.validated.as_ref().map(|validated| {
            let pair = serde_json::json!([validated.claims.iss, validated.claims.sub]);
            pv_core::digest::sha256_hex(pair.to_string().as_bytes())
        })
    }

    fn bind_session(&self, session: &Session) -> Result<(), PipelineError> {
        if !session.bind_principal(self.principal_digest().as_deref()) {
            return Err(PipelineError::Unauthorized(
                "session is bound to a different principal".into(),
            ));
        }
        Ok(())
    }
}

fn truthy_env(name: &str) -> bool {
    match std::env::var(name) {
        Ok(value) => matches!(
            value.trim().to_ascii_lowercase().as_str(),
            "1" | "true" | "yes" | "on"
        ),
        Err(_) => false,
    }
}

/// Map a `reqwest::Error` to a client-safe reason (CWE-209): the raw
/// `Display` includes the request URL, which leaks the operator-configured
/// upstream URL — potentially an internal hostname — to the caller. Return a
/// stable, non-identifying category instead; the detailed message is logged
/// server-side by the caller.
pub(crate) fn classify_upstream_error(error: &reqwest::Error) -> &'static str {
    if error.is_timeout() {
        "upstream timed out"
    } else if error.is_connect() {
        "upstream unreachable"
    } else if error.is_request() {
        "upstream request rejected"
    } else if error.is_body() || error.is_decode() {
        "upstream response malformed"
    } else if error.is_status() {
        "upstream returned error status"
    } else {
        "upstream forwarding failed"
    }
}

/// Map an [`IdentityError`] to a client-safe reason (CWE-209).
///
/// The `Display` impl on `IdentityError` embeds attacker-influenced strings
/// (the presented `kid`, `iss`, and structural detail from the underlying
/// JWT parser) and distinguishes between at least sixteen distinct failure
/// modes. Returning that verbatim to the client turns each 401 response
/// into an *enumeration oracle* for the validator's configured JWKS and
/// issuer allowlist: an attacker can iterate candidate `kid`s and issuers
/// and read the response body to discover which values are registered
/// (`UnknownKid` vs `AlgorithmRejected` vs `Verification` vs `BadIssuer`
/// all produce distinguishable text).
///
/// We collapse every attacker-reachable variant to a single stable string.
/// The detailed cause is preserved server-side by the caller via
/// `tracing::warn!` and the failure-job audit chain, so operators keep
/// full diagnostic detail without exposing it to the network.
/// An unreachable revocation list is an availability fault: answer 503 so
/// clients back off and retry, instead of a 401 that tells them to throw
/// away a credential that may be perfectly valid.
pub(crate) fn identity_error_to_pipeline(error: pv_identity::IdentityError) -> PipelineError {
    if matches!(error, pv_identity::IdentityError::RevocationUnavailable(_)) {
        tracing::error!(error = %error, "identity revocation list unavailable; refusing the request");
        return PipelineError::unavailable(
            "identity revocation list is unavailable; retry shortly".to_owned(),
        );
    }
    tracing::warn!(error = %error, "identity validation failed");
    PipelineError::Unauthorized(classify_identity_error(&error).to_owned())
}

pub(crate) fn classify_identity_error(error: &pv_identity::IdentityError) -> &'static str {
    match error {
        // Server-side misconfiguration is not attacker-reachable in the
        // normal request path (validator construction rejects a bad JWKS
        // at startup); report it distinctly so operator dashboards can
        // surface it if it ever appears.
        pv_identity::IdentityError::Jwks(_) => "identity validator misconfigured",
        // Every other variant is at least partially attacker-influenced —
        // collapse to one opaque message. Any finer distinction leaks
        // configured `kid`s, issuers, or acceptable algorithms.
        _ => "identity validation failed",
    }
}

/// Extract the single value for a control header, refusing duplicates.
///
/// `HeaderMap::get(name)` returns only the first entry when a header
/// appears multiple times. A client sending
/// `X-PV-Session: sessA` followed by `X-PV-Session: sessB` would then
/// have `sessA` used for state while an intermediary log-aggregator
/// might observe `sessA, sessB` (some proxies merge on the wire). The
/// resulting session-desync is hard to diagnose and is a header
/// smuggling primitive at boundaries where our proxy is behind
/// another one. Refuse the multi-value case at ingress rather than
/// silently accept the first.
pub(crate) fn single_header<'a>(
    headers: &'a HeaderMap,
    name: &'static str,
) -> Result<Option<&'a axum::http::HeaderValue>, PipelineError> {
    let mut iter = headers.get_all(name).into_iter();
    let first = iter.next();
    if iter.next().is_some() {
        return Err(PipelineError::bad_request(format!(
            "request carries more than one {name} header; provide exactly one"
        )));
    }
    Ok(first)
}

/// Strip the RFC 7235 §2.1 `Bearer` scheme from an Authorization header
/// value, case-insensitively (per the RFC), and allow one or more
/// whitespace characters (`SP` / `HTAB`) between the scheme token and
/// the credential (per `token1` production in §2.1).
///
/// `str::strip_prefix("Bearer ")` used to be the sole
/// parser. It missed `bearer`, `BEARER`, and (per the RFC's own
/// grammar) `Bearer\t...`, silently dropping to `None` and — when
/// `require_identity = false` — letting the request execute as
/// anonymous while the caller believed they had authenticated.
pub(crate) fn strip_bearer_scheme(value: &str) -> Option<&str> {
    let (scheme, rest) = value.split_at_checked(6)?;
    if !scheme.eq_ignore_ascii_case("Bearer") {
        return None;
    }
    let rest = rest.trim_start_matches([' ', '\t']);
    if rest.len() == value.len() - scheme.len() {
        // No whitespace at all after `Bearer` — the credential MUST
        // be separated per the ABNF (`token1 = 1*( SP / HTAB )
        // token`). Refuse `BearerXYZ` — that's a scheme other than
        // Bearer's credential, not a Bearer credential.
        return None;
    }
    if rest.is_empty() {
        // Empty credential.
        return None;
    }
    Some(rest)
}

pub(crate) fn session_id(headers: &HeaderMap) -> Result<String, PipelineError> {
    match single_header(headers, SESSION_HEADER)? {
        Some(value) => {
            let value = value
                .to_str()
                .map_err(|_| PipelineError::bad_request("X-PV-Session is not valid text".to_owned()))?;
            pv_core::SessionId::parse(value)
                .map(|id| id.to_string())
                .map_err(PipelineError::bad_request_source)
        }
        None => Ok(pv_core::new_session_id().to_string()),
    }
}

/// Stable id used as the principal-scoped budget key.
///
/// For authenticated requests this is the JWT's `instance_uid` (the
/// per-instance principal name — a single agent workload can hold several
/// concurrent tokens under one uid). For requests without a validator
/// (dev mode / passthrough) the identity's `instance_uid` is already the
/// literal `"anonymous"` sentinel, so this function is a straight
/// pass-through; the shipping default is a single shared bucket and the
/// config validator refuses `principal_budget` while `require_identity =
/// false` unless the operator explicitly acknowledges that shape.
/// Stable policy handle for a PDP denial, following the
/// `workflow.signed_required` naming: the console attributes denials by
/// this name, and it points the operator at the config block responsible.
fn pdp_policy_name(code: pv_sandbox::DenialCode) -> &'static str {
    match code {
        pv_sandbox::DenialCode::UnmappedTool => "pdp.intent_map",
        pv_sandbox::DenialCode::MissionExpired | pv_sandbox::DenialCode::MissionDenied => "pdp.mission",
        pv_sandbox::DenialCode::PdpDenied => "pdp.authzen",
        _ => "pdp",
    }
}

pub(crate) fn principal_id_for_budget(identity: &AgentIdentity) -> String {
    if identity.instance_uid.is_empty() {
        "anonymous".to_owned()
    } else {
        identity.instance_uid.clone()
    }
}

fn workflow(headers: &HeaderMap, default: &str) -> Result<Workflow, PipelineError> {
    let value = single_header(headers, WORKFLOW_HEADER)?
        .map(|header| {
            header
                .to_str()
                .map_err(|_| PipelineError::bad_request("X-PV-Workflow is not valid text".to_owned()))
        })
        .transpose()?
        .unwrap_or(default);
    Workflow::parse(value).ok_or_else(|| {
        PipelineError::bad_request(format!("X-PV-Workflow must be signed or unsigned, got {value:?}"))
    })
}

fn last_message_text(payload: &Value) -> String {
    let Some(message) = payload
        .get("messages")
        .and_then(Value::as_array)
        .and_then(|messages| messages.last())
    else {
        return String::new();
    };
    // Primary source: string `content`.
    if let Some(text) = message.get("content").and_then(Value::as_str) {
        if !text.is_empty() {
            return text.to_owned();
        }
    }
    // Mirror of the response-side fix: a
    // tool-call-only assistant message carries `content: null` and a
    // `tool_calls` array. `last_message_text` returning "" here would
    // feed a zero-vector embedding into the breaker in worker.rs and
    // false-trip legitimate tool-driven agents on the request side.
    // Synthesize `tool_name(arguments)` for each tool call, in
    // wire-order — same shape as the response-side synthesis in
    // routes.rs::AbortFinalizingStream::submit_response_capture.
    //
    // Beware: `function.arguments` reaches the
    // wire in TWO shapes: a JSON-encoded string (OpenAI reference)
    // OR a bare JSON object (some clients / most Anthropic-shaped
    // gateways). `.as_str()` on the object variant returned None and
    // the synthesis collapsed to `tool_name()` — worse than the
    // pre-fix behavior, because now two different-argument calls to
    // the same tool produced IDENTICAL synthesized text, causing
    // false loop trips. Match the tolerant pattern from
    // `atif_capture_from_request` in the same file: render the raw
    // Value directly when it isn't a stringified JSON. Empty and
    // missing collapse to "" and get gated off analyze_loop
    // downstream.
    if let Some(tool_calls) = message.get("tool_calls").and_then(Value::as_array) {
        let synthesized: Vec<String> = tool_calls
            .iter()
            .filter_map(|call| {
                let name = call
                    .get("function")
                    .and_then(|f| f.get("name"))
                    .and_then(Value::as_str)?;
                let arguments = call.get("function").and_then(|f| f.get("arguments"));
                let args_rendered = match arguments {
                    Some(Value::String(s)) => s.clone(),
                    Some(other) => serde_json::to_string(other).unwrap_or_default(),
                    None => String::new(),
                };
                Some(format!("{name}({args_rendered})"))
            })
            .collect();
        if !synthesized.is_empty() {
            return synthesized.join("\n");
        }
    }
    // Content-parts array (multimodal). Concatenate every text part
    // so an assistant reply built from mixed text+image parts still
    // feeds a non-zero vector to the breaker rather than "".
    if let Some(parts) = message.get("content").and_then(Value::as_array) {
        let text: String = parts
            .iter()
            .filter_map(|part| part.get("text").and_then(Value::as_str))
            .collect::<Vec<_>>()
            .join("\n");
        if !text.is_empty() {
            return text;
        }
    }
    String::new()
}

/// Human-readable JSON type name for diagnostics — mirrors `typeof` in
/// JS. Used by [`atif_capture_from_request`] to say
/// `'messages' must be a JSON array, got object` rather than the
/// historical opaque `chat payload has no messages`.
fn json_type_name(value: &Value) -> &'static str {
    match value {
        Value::Null => "null",
        Value::Bool(_) => "boolean",
        Value::Number(_) => "number",
        Value::String(_) => "string",
        Value::Array(_) => "array",
        Value::Object(_) => "object",
    }
}

fn atif_capture_from_request(payload: &Value) -> Result<AtifCapture, PipelineError> {
    // Split the two failure classes so support engineers can tell
    // "field missing" from "field is the wrong shape" — chasing a
    // phantom "no messages" ticket for a caller who typed
    // `"messages": {}` used to be the top diagnostic-quality
    // complaint. The OpenAI Responses API also uses `input`
    // instead of `messages`; the missing-field message now points
    // directly at the correct fix.
    let messages_value = payload
        .get("messages")
        .ok_or_else(|| PipelineError::bad_request("chat payload is missing 'messages'".to_owned()))?;
    let messages = messages_value.as_array().ok_or_else(|| {
        PipelineError::bad_request(format!(
            "'messages' must be a JSON array, got {}",
            json_type_name(messages_value)
        ))
    })?;
    let message = messages
        .last()
        .ok_or_else(|| PipelineError::bad_request("chat payload 'messages' is empty".to_owned()))?;
    let role = message.get("role").and_then(Value::as_str);
    let source = match role {
        // "function" is the legacy OpenAI function-calling spelling of
        // "tool" (a tool result fed back to the model) — still emitted
        // by older LangChain, AutoGen and llama-index versions.
        // Refusing it 400'd legitimate agent traffic.
        Some("system" | "developer" | "tool" | "function") => pv_atif::Source::System,
        Some("user") => pv_atif::Source::User,
        Some("assistant") => pv_atif::Source::Agent,
        Some(other) => {
            return Err(PipelineError::bad_request(format!(
                "unsupported chat role {other:?}"
            )))
        }
        None => return Err(PipelineError::bad_request("chat message has no role".to_owned())),
    };
    let content = message
        .get("content")
        .filter(|value| value.is_string() || value.is_array())
        .cloned()
        .unwrap_or_else(|| Value::String(String::new()));
    let tool_calls = message.get("tool_calls").and_then(Value::as_array).map(|calls| {
        calls
            .iter()
            .filter_map(|call| {
                let function = call.get("function")?;
                let arguments = function
                    .get("arguments")
                    .cloned()
                    .unwrap_or(Value::Object(Default::default()));
                let arguments = arguments
                    .as_str()
                    .and_then(|raw| serde_json::from_str(raw).ok())
                    .unwrap_or(arguments);
                Some(pv_atif::ToolCall {
                    tool_call_id: call
                        .get("id")
                        .and_then(Value::as_str)
                        .map_or_else(pv_core::new_event_uid, str::to_owned),
                    function_name: function.get("name")?.as_str()?.to_owned(),
                    arguments,
                    extra: None,
                })
            })
            .collect()
    });
    let observation = (role == Some("tool")).then(|| pv_atif::Observation {
        results: vec![pv_atif::ObservationResult {
            source_call_id: None,
            content: Some(content.clone()),
            subagent_trajectory_ref: None,
            extra: message
                .get("tool_call_id")
                .cloned()
                .map(|tool_call_id| serde_json::json!({"tool_call_id": tool_call_id})),
        }],
    });
    Ok(AtifCapture {
        source,
        message: content,
        reasoning_content: if source == pv_atif::Source::Agent {
            message
                .get("reasoning_content")
                .and_then(Value::as_str)
                .map(str::to_owned)
        } else {
            None
        },
        model_name: if source == pv_atif::Source::Agent {
            payload.get("model").and_then(Value::as_str).map(str::to_owned)
        } else {
            None
        },
        tool_calls: if source == pv_atif::Source::Agent {
            tool_calls
        } else {
            None
        },
        observation,
        llm_call_count: (source == pv_atif::Source::Agent).then_some(1),
    })
}

fn validate_session_binding(
    session: &Session,
    workflow: Workflow,
    identity: &AgentIdentity,
) -> Result<(), PipelineError> {
    if session.workflow != workflow {
        return Err(PipelineError::bad_request(
            "session workflow cannot change after open".to_owned(),
        ));
    }
    if session.identity.instance_uid != identity.instance_uid
        || session.identity.charter != identity.charter
        || session.identity.version != identity.version
    {
        return Err(PipelineError::Unauthorized(
            "session is bound to a different agent identity".to_owned(),
        ));
    }
    Ok(())
}

pub(crate) fn scope_allows(scopes: &[String], required: &str) -> bool {
    scopes.iter().any(|scope| {
        scope == "*"
            || scope == required
            // allocation-free prefix match. The old
            // `required.starts_with(&format!("{prefix}:"))` allocated a
            // String per granted scope per request on the identity hot
            // path. `strip_prefix(prefix)` + a `:` head check is the
            // same predicate with zero allocations.
            || scope.strip_suffix(":*").is_some_and(|prefix| {
                required
                    .strip_prefix(prefix)
                    .is_some_and(|rest| rest.starts_with(':'))
            })
    })
}

fn inject_corrective_message(payload: &mut Value) -> Result<(), PipelineError> {
    let messages = payload
        .get_mut("messages")
        .and_then(Value::as_array_mut)
        .ok_or_else(|| PipelineError::bad_request("chat payload has no messages array".to_owned()))?;
    messages.push(serde_json::json!({
        "role": "system",
        "content": "ProofVisor detected a semantic loop. Stop repeating the previous approach, identify new evidence, and choose a materially different next action."
    }));
    Ok(())
}

pub(crate) fn load_exchange_seed(path: &str) -> Result<zeroize::Zeroizing<[u8; 32]>, String> {
    use zeroize::Zeroizing;
    require_owner_only_secret(std::path::Path::new(path)).map_err(|e| format!("exchange seed file: {e}"))?;
    let encoded = Zeroizing::new(
        pv_core::fsutil::read_capped_string(std::path::Path::new(path), pv_core::fsutil::MAX_CONTROL_BYTES)
            .map_err(|e| format!("read exchange seed {path}: {e}"))?,
    );
    let bytes =
        Zeroizing::new(hex::decode(encoded.trim()).map_err(|e| format!("decode exchange seed as hex: {e}"))?);
    let seed: Zeroizing<[u8; 32]> = Zeroizing::new(
        <[u8; 32]>::try_from(bytes.as_slice())
            .map_err(|_| "exchange seed must contain exactly 32 bytes".to_owned())?,
    );
    if *seed == [0u8; 32] || *seed == [0xFFu8; 32] {
        return Err(format!("exchange seed at {path} is a known-weak value; refusing"));
    }
    Ok(seed)
}

#[cfg(test)]
mod tests {
    #![allow(
        clippy::expect_used,
        clippy::indexing_slicing,
        clippy::panic,
        clippy::unwrap_used
    )]

    /// R285: the twice-registered supervised-loop metrics (eager boot
    /// registration here + lazy observation-site registration in
    /// main.rs/reconciler.rs) must take their HELP from the shared
    /// `supervision_help` consts / `LIFECYCLE_OUTBOX_PENDING_HELP`.
    /// Six counter pairs and the outbox gauge had drifted into
    /// per-tick "divergent HELP" warn spam; first-wins semantics also
    /// let a terse boot HELP shadow the rich forensic text alerting
    /// playbooks key on. A string literal beside any of these names
    /// reintroduces the drift the consts exist to prevent.
    #[test]
    fn twice_registered_metrics_use_shared_help_consts() {
        let sources = [
            ("pipeline.rs", include_str!("pipeline.rs")),
            ("main.rs", include_str!("main.rs")),
            ("reconciler.rs", include_str!("reconciler.rs")),
        ];
        let shared = [
            ("pv_reconciler_panics_total", "RECONCILER_PANICS"),
            ("pv_bridge_maintenance_panics_total", "BRIDGE_MAINTENANCE_PANICS"),
            ("pv_bridge_maintenance_errors_total", "BRIDGE_MAINTENANCE_ERRORS"),
            (
                "pv_bridge_maintenance_join_errors_total",
                "BRIDGE_MAINTENANCE_JOIN_ERRORS",
            ),
            ("pv_atif_retention_panics_total", "ATIF_RETENTION_PANICS"),
            ("pv_atif_retention_errors_total", "ATIF_RETENTION_ERRORS"),
            ("pv_lifecycle_outbox_pending", "LIFECYCLE_OUTBOX_PENDING_HELP"),
        ];
        for (file, src) in sources {
            for (metric, const_name) in shared {
                let needle = format!("\"{metric}\"");
                let mut from = 0;
                while let Some(pos) = src[from..].find(&needle) {
                    let at = from + pos;
                    // The ~200 chars after the name must reference the
                    // shared const (registration site) — doc comments,
                    // alert examples, and scrape asserts don't register
                    // and carry no adjacent `.counter(`/`.gauge(` call.
                    let window = &src[at..(at + 220).min(src.len())];
                    let registers = src[..at]
                        .rfind(|c: char| !c.is_whitespace())
                        .map(|p| {
                            src[p.saturating_sub(20)..at].contains(".counter(")
                                || src[p.saturating_sub(20)..at].contains(".gauge(")
                        })
                        .unwrap_or(false);
                    if registers {
                        assert!(
                            window.contains(const_name),
                            "{file}: registration of {metric} does not use \
                             supervision_help::{const_name} — string-literal HELP \
                             reintroduces the divergent-HELP drift (R285)"
                        );
                    }
                    from = at + needle.len();
                }
            }
        }
    }

    use super::*;
    use axum::http::HeaderValue;
    use pv_bridge::{BusError, PublishAck, StoredEvent};
    use pv_receipts::Ed25519Signer;
    use pv_sandbox::SandboxConfig;
    use pv_state::InMemoryStore;
    use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};

    /// `last_message_text`
    /// must produce distinct output for two different tool-call
    /// argument shapes so the loop breaker's embedding varies.
    /// Both string-shaped and object-shaped arguments MUST reach the
    /// synthesized text — the first take collapsed
    /// object-shaped arguments to `tool_name()`.
    #[test]
    fn last_message_text_tool_call_synthesis_varies_with_object_arguments() {
        let payload_string_args = serde_json::json!({
            "messages": [{
                "role": "assistant",
                "content": null,
                "tool_calls": [{
                    "id": "c1",
                    "type": "function",
                    "function": {"name": "read", "arguments": "{\"path\":\"a\"}"}
                }]
            }]
        });
        let payload_object_args_a = serde_json::json!({
            "messages": [{
                "role": "assistant",
                "content": null,
                "tool_calls": [{
                    "id": "c1",
                    "type": "function",
                    "function": {"name": "read", "arguments": {"path": "a"}}
                }]
            }]
        });
        let payload_object_args_b = serde_json::json!({
            "messages": [{
                "role": "assistant",
                "content": null,
                "tool_calls": [{
                    "id": "c1",
                    "type": "function",
                    "function": {"name": "read", "arguments": {"path": "b"}}
                }]
            }]
        });
        let t_string = last_message_text(&payload_string_args);
        let t_obj_a = last_message_text(&payload_object_args_a);
        let t_obj_b = last_message_text(&payload_object_args_b);
        assert!(!t_string.is_empty(), "string-arg synthesis must not be empty");
        assert!(!t_obj_a.is_empty(), "object-arg synthesis must not be empty");
        assert_ne!(
            t_obj_a, t_obj_b,
            "object-shape args with different content must produce different text; \
             got identical {t_obj_a:?}"
        );
        assert!(
            t_obj_a.contains("\"path\":\"a\"") || t_obj_a.contains("path"),
            "object-arg synthesis must include the argument content, got {t_obj_a:?}"
        );
    }

    /// A tool-call message with EMPTY arguments still synthesizes to
    /// `tool_name()` and reaches the breaker as a non-empty string
    /// — otherwise every tool call with no arguments would false-trip
    /// the breaker via zero-vector embedding.
    #[test]
    fn last_message_text_tool_call_with_empty_arguments_still_synthesizes() {
        let payload = serde_json::json!({
            "messages": [{
                "role": "assistant",
                "content": null,
                "tool_calls": [{
                    "id": "c1",
                    "type": "function",
                    "function": {"name": "ping"}
                }]
            }]
        });
        let text = last_message_text(&payload);
        assert_eq!(text, "ping()");
    }

    struct NullBus;

    struct BlockingSink {
        entered: AtomicBool,
        release: tokio::sync::Notify,
    }

    impl VectorSink for BlockingSink {
        fn record<'a>(
            &'a self,
            _session_id: &'a str,
            _vector: &'a [f32],
        ) -> pv_loopdetect::VectorSinkFuture<'a> {
            Box::pin(async move {
                self.entered.store(true, AtomicOrdering::Release);
                self.release.notified().await;
                Ok(())
            })
        }
    }

    impl EventBus for NullBus {
        fn publish(&self, topic: &str, _key: &str, _value: &Value) -> Result<PublishAck, BusError> {
            Ok(PublishAck {
                topic: topic.to_owned(),
                partition: 0,
                offset: 0,
            })
        }

        fn fetch(
            &self,
            _topic: &str,
            _partition: u32,
            _offset: u64,
            _max: usize,
        ) -> Result<Vec<StoredEvent>, BusError> {
            Ok(Vec::new())
        }

        fn partitions(&self, _topic: &str) -> Result<u32, BusError> {
            Ok(1)
        }

        fn topics(&self) -> Vec<String> {
            EventClass::all()
                .iter()
                .map(|class| class.topic().to_owned())
                .collect()
        }
    }

    fn state(mut config: HarnessConfig) -> AppState {
        config.atif_spool_dir = tempfile::tempdir().unwrap().keep().to_string_lossy().into_owned();
        AppState::new(
            config,
            Arc::new(InMemoryStore::new()),
            Arc::new(Sandbox::new(SandboxConfig::default(), Vec::new()).unwrap()),
            Arc::new(NullBus),
            None,
            Arc::new(Ed25519Signer::from_seed(&[9; 32])),
        )
        .unwrap()
    }

    fn payload() -> Value {
        serde_json::json!({
            "model": "test",
            "messages": [{"role": "user", "content": "hello"}],
        })
    }

    fn admission_test_runtime() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .max_blocking_threads(1)
            .enable_all()
            .build()
            .unwrap()
    }

    async fn occupy_admission_blocking_pool(
    ) -> (tokio::sync::oneshot::Sender<()>, tokio::task::JoinHandle<()>) {
        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
        let (release_tx, release_rx) = tokio::sync::oneshot::channel();
        let task = tokio::task::spawn_blocking(move || {
            let _ = started_tx.send(());
            // Dropping the sender also releases this thread if a test panics.
            let _ = release_rx.blocking_recv();
        });
        started_rx.await.unwrap();
        (release_tx, task)
    }

    async fn poll_admission_once<F: std::future::Future>(
        mut future: std::pin::Pin<&mut F>,
    ) -> std::task::Poll<F::Output> {
        std::future::poll_fn(|context| std::task::Poll::Ready(future.as_mut().poll(context))).await
    }

    #[test]
    fn small_chat_admission_bypasses_saturated_blocking_pool() {
        admission_test_runtime().block_on(async {
            for authenticated in [false, true] {
                let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
                config.require_identity = authenticated;
                config.enforce_identity_scopes = authenticated;
                let mut state = state(config);
                let mut headers = HeaderMap::new();
                headers.insert(SESSION_HEADER, HeaderValue::from_static("small-local-admission"));
                if authenticated {
                    let validator = IdentityValidator::new("proofvisor");
                    validator
                        .add_key(
                            "scope-key",
                            pv_identity::KeyMaterial::HmacSecret(b"scope-secret".to_vec()),
                        )
                        .unwrap();
                    state.identity = Some(Arc::new(validator));
                    headers.insert(
                        axum::http::header::AUTHORIZATION,
                        format!("Bearer {}", scoped_token(&["chat:write"]))
                            .parse()
                            .unwrap(),
                    );
                }
                let caller = state.authenticate_chat_nonblocking(&headers).await.unwrap();
                let principal = caller.principal_digest();
                let mut other_principal = caller.clone();
                if let Some(validated) = &mut other_principal.validated {
                    validated.claims.sub = "another-principal".into();
                }
                let (release, blocker) = occupy_admission_blocking_pool().await;
                if !authenticated {
                    let mut authenticate = Box::pin(state.authenticate_chat_nonblocking(&headers));
                    assert!(matches!(
                        poll_admission_once(authenticate.as_mut()).await,
                        std::task::Poll::Ready(Ok(_))
                    ));
                    let mut duplicated = headers.clone();
                    duplicated.append(
                        axum::http::header::AUTHORIZATION,
                        HeaderValue::from_static("Bearer a"),
                    );
                    duplicated.append(
                        axum::http::header::AUTHORIZATION,
                        HeaderValue::from_static("Bearer b"),
                    );
                    let mut authenticate = Box::pin(state.authenticate_chat_nonblocking(&duplicated));
                    assert!(matches!(
                        poll_admission_once(authenticate.as_mut()).await,
                        std::task::Poll::Ready(Err(PipelineError::BadRequest { .. }))
                    ));
                }
                let body = payload();
                let body_bytes = body.to_string().len();
                let mut prepare =
                    Box::pin(state.prepare_authenticated_chat(&headers, body, body_bytes, caller));
                let outcome = poll_admission_once(prepare.as_mut()).await;
                let completed_inline = outcome.is_ready();
                release.send(()).unwrap();
                blocker.await.unwrap();
                let prepared = match outcome {
                    std::task::Poll::Ready(result) => result.unwrap(),
                    std::task::Poll::Pending => prepare.await.unwrap(),
                };
                assert!(
                    completed_inline,
                    "small local admission queued behind unrelated blocking work"
                );
                assert_eq!(prepared.session.principal_binding(), principal);
                if authenticated {
                    assert_eq!(
                        state
                            .metrics
                            .counter(
                                "pv_identity_validations_total",
                                "Full inbound identity validations"
                            )
                            .get(),
                        1
                    );
                    // The inline path must still refuse another principal on
                    // this session even when agent instance claims coincide.
                    let refused = state
                        .prepare_authenticated_chat(&headers, payload(), body_bytes, other_principal)
                        .await;
                    assert!(matches!(refused, Err(PipelineError::Unauthorized(_))));
                }
                drop(prepared);
                state.worker.wait_idle().await;
            }
        });
    }

    #[test]
    fn chat_preparation_offloads_large_bodies_and_both_quota_classes() {
        admission_test_runtime().block_on(async {
            for (session_quota, principal_quota, large_body) in [(true, false, false), (false, true, false), (false, false, true)] {
                let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
                if session_quota {
                    config.budget.max_tokens = Some(1_000_000);
                }
                if principal_quota {
                    config.principal_budget = Some(pv_state::BudgetSpec { max_tokens: Some(1_000_000), ..Default::default() });
                }
                let state = state(config);
                let mut headers = HeaderMap::new();
                headers.insert(SESSION_HEADER, HeaderValue::from_static("offloaded-admission"));
                let caller = state.authenticate_chat_nonblocking(&headers).await.unwrap();
                let body = if large_body {
                    serde_json::json!({"model":"test", "messages":[{"role":"user", "content":"x".repeat(32 * 1024)}]})
                } else {
                    payload()
                };
                let body_bytes = body.to_string().len();
                let (release, blocker) = occupy_admission_blocking_pool().await;
                let mut prepare = Box::pin(state.prepare_authenticated_chat(&headers, body, body_bytes, caller));
                let outcome = poll_admission_once(prepare.as_mut()).await;
                let offloaded = outcome.is_pending();
                let sessions_before_release = state.sessions.len();
                release.send(()).unwrap();
                blocker.await.unwrap();
                let prepared = match outcome {
                    std::task::Poll::Ready(result) => result.unwrap(),
                    std::task::Poll::Pending => prepare.await.unwrap(),
                };
                assert!(offloaded, "large-body or quota preparation ran inline");
                assert_eq!(sessions_before_release, 0, "preparation ran before the blocking worker was available");
                drop(prepared);
                state.worker.wait_idle().await;
            }
        });
    }

    #[test]
    fn chat_route_passes_actual_body_size_to_admission() {
        use tower::ServiceExt as _;

        admission_test_runtime().block_on(async {
            let state = state(HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp"));
            let body = serde_json::json!({
                "model": "test",
                "messages": [{"role": "user", "content": "x".repeat(32 * 1024)}],
            })
            .to_string();
            let request = axum::http::Request::builder()
                .method("POST")
                .uri("/v1/chat/completions")
                .header("content-type", "application/json")
                .header(SESSION_HEADER, "large-route-admission")
                .body(axum::body::Body::from(body))
                .unwrap();
            let (release, blocker) = occupy_admission_blocking_pool().await;
            let mut request = Box::pin(crate::build_router(state.clone()).oneshot(request));
            let outcome = poll_admission_once(request.as_mut()).await;
            let offloaded = outcome.is_pending();
            // If the route passed zero instead of body.len(), preparation
            // would already have opened a session before forwarding blocked.
            let sessions_before_release = state.sessions.len();
            release.send(()).unwrap();
            blocker.await.unwrap();
            let response = match outcome {
                std::task::Poll::Ready(result) => result.unwrap(),
                std::task::Poll::Pending => request.await.unwrap(),
            };
            assert!(offloaded);
            assert_eq!(sessions_before_release, 0);
            assert_eq!(response.status(), axum::http::StatusCode::BAD_GATEWAY);
            assert_eq!(state.sessions.len(), 1);
            state.worker.wait_idle().await;
        });
    }

    #[test]
    fn unresolved_identity_never_runs_revocation_on_the_admission_task() {
        struct CountRevocations(std::sync::atomic::AtomicUsize);
        impl pv_identity::RevocationStore for CountRevocations {
            fn revoke(&self, _jti: &str, _expires_at: u64) -> Result<(), String> {
                Ok(())
            }
            fn is_revoked(&self, _jti: &str) -> Result<bool, String> {
                self.0.fetch_add(1, AtomicOrdering::SeqCst);
                Ok(false)
            }
        }
        admission_test_runtime().block_on(async {
            let revocations = Arc::new(CountRevocations(std::sync::atomic::AtomicUsize::new(0)));
            let mut validator = IdentityValidator::new("proofvisor");
            validator
                .add_key(
                    "scope-key",
                    pv_identity::KeyMaterial::HmacSecret(b"scope-secret".to_vec()),
                )
                .unwrap();
            validator.set_revocation_store(revocations.clone());
            let mut state = state(HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp"));
            state.identity = Some(Arc::new(validator));
            let mut headers = HeaderMap::new();
            headers.insert(SESSION_HEADER, HeaderValue::from_static("unresolved-admission"));
            headers.insert(
                axum::http::header::AUTHORIZATION,
                format!("Bearer {}", scoped_token(&["chat:write"]))
                    .parse()
                    .unwrap(),
            );
            let (release, blocker) = occupy_admission_blocking_pool().await;
            let mut authenticate = Box::pin(state.authenticate_chat_nonblocking(&headers));
            let outcome = poll_admission_once(authenticate.as_mut()).await;
            let offloaded = outcome.is_pending();
            let calls_before_release = revocations.0.load(AtomicOrdering::SeqCst);
            release.send(()).unwrap();
            blocker.await.unwrap();
            let _caller = match outcome {
                std::task::Poll::Ready(result) => result.unwrap(),
                std::task::Poll::Pending => authenticate.await.unwrap(),
            };
            assert!(offloaded);
            assert_eq!(calls_before_release, 0);
            assert_eq!(revocations.0.load(AtomicOrdering::SeqCst), 1);

            let (release, blocker) = occupy_admission_blocking_pool().await;
            let body = payload();
            let body_bytes = body.to_string().len();
            let mut prepare = Box::pin(state.prepare_chat_nonblocking(&headers, body, body_bytes, None));
            let outcome = poll_admission_once(prepare.as_mut()).await;
            let offloaded = outcome.is_pending();
            let calls_before_release = revocations.0.load(AtomicOrdering::SeqCst);
            release.send(()).unwrap();
            blocker.await.unwrap();
            let prepared = match outcome {
                std::task::Poll::Ready(result) => result.unwrap(),
                std::task::Poll::Pending => prepare.await.unwrap(),
            };
            assert!(offloaded);
            assert_eq!(calls_before_release, 1);
            assert_eq!(revocations.0.load(AtomicOrdering::SeqCst), 2);
            drop(prepared);
            state.worker.wait_idle().await;
        });
    }

    async fn trip_loop(state: &AppState, headers: &HeaderMap, repeated: &Value) {
        for expected in 1..=4u64 {
            let mut prepared = state.prepare_chat(headers, repeated.clone(), None).unwrap();
            // This helper abandons the prepared request without forwarding.
            // Defuse the capture guard so its terminal failure job does not
            // pad the chain — the chain-count waits below must count only
            // the admission jobs, or the analyze jobs race the assertion.
            prepared.capture_guard.defuse();
            drop(prepared);
            let session = state.sessions.get("loop-session").unwrap();
            // Generous budget: the chain append rides an async worker job and
            // must survive heavily loaded parallel test runs.
            tokio::time::timeout(std::time::Duration::from_secs(10), async {
                while session.chain.lock().count() < expected {
                    tokio::task::yield_now().await;
                }
            })
            .await
            .unwrap();
        }
    }

    #[tokio::test]
    async fn prepare_binds_session_and_stays_below_hot_path_budget() {
        let config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("session-1"));

        let prepared = state.prepare_chat(&headers, payload(), None).unwrap();
        assert_eq!(prepared.session.id, "session-1");
        // Debug-build ceiling: 100ms is 3 orders of magnitude above the release
        // SLA (~33us p95); a regression an order of magnitude worse still trips
        // this. The strict perf gate lives in the SLA bench suite.
        assert!(
            prepared.middleware_us < 100_000,
            "middleware took {}us",
            prepared.middleware_us
        );
        assert!(state.metrics.render().contains("pv_stage_duration_seconds"));
    }

    #[tokio::test]
    async fn missing_identity_fails_closed_when_required() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.require_identity = true;
        let error = match state(config).prepare_chat(&HeaderMap::new(), payload(), None) {
            Ok(_) => panic!("missing identity was accepted"),
            Err(error) => error,
        };
        assert!(matches!(error, PipelineError::Unauthorized(_)));
    }

    #[tokio::test]
    async fn token_quota_blocks_before_dispatch() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.budget.max_tokens = Some(1);
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("quota-failure"));
        let error = match state.prepare_chat(&headers, payload(), None) {
            Ok(_) => panic!("over-budget request was accepted"),
            Err(error) => error,
        };
        assert!(matches!(error, PipelineError::Blocked { .. }));
        assert_eq!(state.sessions.len(), 1);
        let session = state.sessions.get("quota-failure").unwrap();
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            while session.atif_steps_count() == 0 {
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();
    }

    /// The principal-scoped budget (§3.2 fix) binds across rotating
    /// X-PV-Session headers: sessions A and B pointed at the same
    /// principal share one ledger. Without the principal budget, a
    /// caller could reset the session bucket per request and never
    /// hit `max_tokens`.
    #[tokio::test]
    async fn principal_budget_binds_across_rotated_session_headers() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        // Wide session budget so it never binds — we're proving the
        // *principal* ledger is what refuses the second request.
        config.budget.max_tokens = Some(1_000_000);
        // The test payload approximates to ~30 tokens (roughly 2 tokens per
        // ASCII word + one per punctuation glyph, on the ~60-char JSON body).
        // A cap of 40 admits the first request and refuses the second under
        // the same principal — the anti-rotation property in one assertion.
        config.principal_budget = Some(pv_state::BudgetSpec {
            max_tokens: Some(40),
            ..pv_state::BudgetSpec::default()
        });
        let state = state(config);

        // First request under session A: fits the principal budget.
        let mut headers_a = HeaderMap::new();
        headers_a.insert(SESSION_HEADER, HeaderValue::from_static("rotated-a"));
        let first = state.prepare_chat(&headers_a, payload(), None);
        assert!(first.is_ok(), "first principal request must be admitted");
        // Mark the admission dispatched before dropping: a dropped
        // UNdispatched request now (correctly) refunds via
        // `AdmissionDebit`'s cancellation guard, which would empty the
        // very ledger this test needs to stay charged.
        if let Ok(mut prepared) = first {
            let _ = prepared.admission_debit.take_for_dispatch();
        }

        // Second request under a DIFFERENT session id but the same
        // principal (anonymous — both requests unauthenticated). The
        // session ledger for `rotated-b` is virgin; only the principal
        // ledger can refuse. Refusal proves the header-rotation attack
        // is closed.
        let mut headers_b = HeaderMap::new();
        headers_b.insert(SESSION_HEADER, HeaderValue::from_static("rotated-b"));
        match state.prepare_chat(&headers_b, payload(), None) {
            Ok(_) => panic!("header rotation defeated principal budget — the fix from §3.2 has regressed"),
            Err(PipelineError::Blocked { context: msg, .. }) => {
                assert!(
                    msg.contains("principal."),
                    "refusal must name the principal-scoped limit, got {msg:?}"
                );
            }
            Err(other) => panic!("unexpected refusal shape {other:?}"),
        }
    }

    /// `abandon_prepared` runs strictly pre-dispatch, so its refund must
    /// mirror the DUAL-ledger admission debit. Pre-fix it refunded only
    /// the session ledger: every pre-dispatch abort (durable-capture
    /// failure, breaker-open, marker-write failure) permanently drained
    /// principal tokens for zero LLM work — under header rotation the
    /// principal ledger is the only one that binds, so aborts alone
    /// could exhaust an identity's budget.
    #[tokio::test]
    async fn abandon_prepared_refunds_the_principal_ledger_too() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.budget.max_tokens = Some(1_000_000);
        // Fits ONE ~30-token payload; a leaked (unrefunded) debit makes
        // the next same-principal request refuse.
        config.principal_budget = Some(pv_state::BudgetSpec {
            max_tokens: Some(40),
            ..pv_state::BudgetSpec::default()
        });
        let state = state(config);

        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("abandon-a"));
        let mut prepared = state.prepare_chat(&headers, payload(), None).unwrap();
        state.abandon_prepared(&mut prepared, StopReason::Other, "test abandon");
        drop(prepared);

        // Same (anonymous) principal, different session: only the
        // principal ledger can refuse here. It must not — the abandoned
        // request's debit was refunded on both ledgers.
        let mut headers_b = HeaderMap::new();
        headers_b.insert(SESSION_HEADER, HeaderValue::from_static("abandon-b"));
        let second = state.prepare_chat(&headers_b, payload(), None);
        assert!(
            second.is_ok(),
            "an abandoned request must not consume principal budget; got {:?}",
            second.err()
        );
    }

    /// Axum drops the request future wholesale on client disconnect —
    /// no explicit abort path runs. The admission debit must refund via
    /// `AdmissionDebit`'s Drop (both ledgers), or every mid-admission
    /// disconnect permanently drains the never-healing principal
    /// ledger with zero LLM work.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dropping_a_prepared_request_refunds_both_ledgers() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.budget.max_tokens = Some(1_000_000);
        // Fits ONE ~30-token payload; a leaked (unrefunded) debit makes
        // the next same-principal request refuse.
        config.principal_budget = Some(pv_state::BudgetSpec {
            max_tokens: Some(40),
            ..pv_state::BudgetSpec::default()
        });
        let state = state(config);

        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("cancelled-a"));
        let prepared = state.prepare_chat(&headers, payload(), None).unwrap();
        // Simulate the cancelled future: drop with no explicit abort.
        drop(prepared);

        // The Drop refund runs on the blocking pool; poll until it lands.
        let mut headers_b = HeaderMap::new();
        headers_b.insert(SESSION_HEADER, HeaderValue::from_static("cancelled-b"));
        let mut second = state.prepare_chat(&headers_b, payload(), None);
        for _ in 0..200 {
            if second.is_ok() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            second = state.prepare_chat(&headers_b, payload(), None);
        }
        assert!(
            second.is_ok(),
            "a dropped (cancelled) prepared request must refund the principal ledger; got {:?}",
            second.err()
        );
    }

    /// The refund window ends at upstream dispatch: once
    /// `take_for_dispatch` ran (immediately before `send()`), dropping
    /// the request must NOT refund — the request may have reached the
    /// provider, and the debit conservatively stands (mirroring the
    /// `is_connect`-only refund on send errors).
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dispatched_debit_stands_when_the_request_drops() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.budget.max_tokens = Some(1_000_000);
        config.principal_budget = Some(pv_state::BudgetSpec {
            max_tokens: Some(40),
            ..pv_state::BudgetSpec::default()
        });
        let state = state(config);

        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("dispatched-a"));
        let mut prepared = state.prepare_chat(&headers, payload(), None).unwrap();
        let debited = prepared.admission_debit.take_for_dispatch();
        assert!(debited > 0, "admission must have debited tokens");
        drop(prepared);

        // Give a hypothetical (buggy) Drop refund time to land, then
        // verify the principal ledger still holds the debit.
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        let mut headers_b = HeaderMap::new();
        headers_b.insert(SESSION_HEADER, HeaderValue::from_static("dispatched-b"));
        assert!(
            matches!(
                state.prepare_chat(&headers_b, payload(), None),
                Err(PipelineError::Blocked { .. })
            ),
            "a dispatched request's debit must stand after drop"
        );
    }

    /// The relayed-header single-value checks must run BEFORE the
    /// budget debit: pre-fix, `single_header`'s `?` fired from the
    /// `PreparedRequest` construction after BOTH ledgers were debited,
    /// with no refund arm — a client repeating a duplicate
    /// `openai-beta` header drained budgets without one upstream call.
    #[tokio::test]
    async fn duplicate_passthrough_header_is_refused_without_consuming_budget() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        // Fits ONE ~30-token payload: a pre-refusal debit leak makes the
        // follow-up legitimate request refuse.
        config.budget.max_tokens = Some(40);
        let state = state(config);

        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("dup-header"));
        headers.append("openai-beta", HeaderValue::from_static("assistants=v1"));
        headers.append("openai-beta", HeaderValue::from_static("assistants=v2"));
        assert!(
            matches!(
                state.prepare_chat(&headers, payload(), None),
                Err(PipelineError::BadRequest { .. })
            ),
            "duplicate allowlisted headers must be refused"
        );

        // The refusal must not have consumed the session budget.
        let mut clean = HeaderMap::new();
        clean.insert(SESSION_HEADER, HeaderValue::from_static("dup-header"));
        clean.insert("openai-beta", HeaderValue::from_static("assistants=v2"));
        let second = state.prepare_chat(&clean, payload(), None);
        assert!(
            second.is_ok(),
            "a header-refused request must not consume budget; got {:?}",
            second.err()
        );
    }

    #[tokio::test]
    async fn invalid_workflow_is_rejected() {
        let config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("sometimes"));
        assert!(matches!(
            state.prepare_chat(&headers, payload(), None),
            Err(PipelineError::BadRequest { .. })
        ));
    }

    #[tokio::test]
    async fn open_session_cannot_change_workflow() {
        let config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("bound-session"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        state.prepare_chat(&headers, payload(), None).unwrap();
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("unsigned"));
        assert!(matches!(
            state.prepare_chat(&headers, payload(), None),
            Err(PipelineError::BadRequest { .. })
        ));
    }

    #[tokio::test]
    async fn asynchronous_loop_verdict_blocks_the_next_request() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.breaker.min_tokens = 0;
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("loop-session"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        let repeated = serde_json::json!({
            "model": "test",
            "messages": [{
                "role": "assistant",
                "content": "I should check the database again for pending orders"
            }]
        });
        trip_loop(&state, &headers, &repeated).await;
        assert!(matches!(
            state.prepare_chat(&headers, repeated, None),
            Err(PipelineError::Blocked { .. })
        ));
    }

    #[tokio::test]
    async fn loop_inject_action_adds_correction_and_resets_breaker() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.breaker.min_tokens = 0;
        config.breaker.action = BreakerAction::Inject;
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("loop-session"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        let repeated = serde_json::json!({
            "messages": [{"role": "assistant", "content": "repeat the database query"}]
        });
        trip_loop(&state, &headers, &repeated).await;
        let prepared = state.prepare_chat(&headers, repeated, None).unwrap();
        let correction = prepared
            .payload
            .get("messages")
            .and_then(Value::as_array)
            .and_then(|messages| messages.last())
            .and_then(|message| message.get("content"))
            .and_then(Value::as_str)
            .unwrap();
        assert!(correction.contains("semantic loop"));
        assert_eq!(prepared.session.loop_state.state(), BreakerState::Closed);
    }

    #[tokio::test]
    async fn loop_abort_action_returns_abort_error() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.breaker.min_tokens = 0;
        config.breaker.action = BreakerAction::Abort;
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("loop-session"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        let repeated = serde_json::json!({
            "messages": [{"role": "assistant", "content": "repeat the database query"}]
        });
        trip_loop(&state, &headers, &repeated).await;
        assert!(matches!(
            state.prepare_chat(&headers, repeated, None),
            Err(PipelineError::Abort(_))
        ));
    }

    /// A breaker trip used to replace the job's Compression class with
    /// StopReason *before* the accounting decisions ran, so the tripped
    /// admission's prompt tokens vanished from the session totals (and the
    /// journal record) — receipts undercounted exactly the runaway sessions
    /// the breaker exists to attest. Accounting must key on the submitted
    /// class, not the swapped one.
    #[tokio::test]
    async fn breaker_trip_does_not_drop_the_admissions_prompt_token_accounting() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.breaker.min_tokens = 0;
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("loop-session"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        let repeated = serde_json::json!({
            "model": "test",
            "messages": [{
                "role": "assistant",
                "content": "I should check the database again for pending orders"
            }]
        });
        // Four identical admissions, tracking chain growth like `trip_loop`;
        // the breaker trips on one of the later worker jobs. Capture the
        // per-admission token amount after the first job lands.
        let mut per_request = 0u64;
        for expected in 1..=4u64 {
            state.prepare_chat(&headers, repeated.clone(), None).unwrap();
            let session = state.sessions.get("loop-session").unwrap();
            tokio::time::timeout(std::time::Duration::from_secs(10), async {
                while session.chain.lock().count() < expected {
                    tokio::task::yield_now().await;
                }
            })
            .await
            .unwrap();
            if expected == 1 {
                tokio::time::timeout(std::time::Duration::from_secs(10), session.wait_for_worker_jobs())
                    .await
                    .unwrap();
                per_request = session
                    .totals
                    .prompt_tokens
                    .load(std::sync::atomic::Ordering::Acquire);
                assert!(per_request > 0, "admission must account prompt tokens");
            }
        }
        let session = state.sessions.get("loop-session").unwrap();
        tokio::time::timeout(std::time::Duration::from_secs(10), session.wait_for_worker_jobs())
            .await
            .unwrap();
        assert_eq!(
            session.loop_state.state(),
            BreakerState::Open,
            "precondition: the breaker must have tripped during the four admissions",
        );
        assert_eq!(
            session
                .totals
                .prompt_tokens
                .load(std::sync::atomic::Ordering::Acquire),
            per_request * 4,
            "every admission's prompt tokens must be accounted, including the one whose \
             worker job carried the breaker trip",
        );
    }

    /// Budget counters are dead weight once a session is sealed (admission
    /// rejects closed sessions before any quota check); finalization must
    /// clear them or the in-memory state store grows by a few cells per
    /// client-chosen session id forever.
    #[tokio::test]
    async fn finalization_clears_the_sessions_budget_counters() {
        let store: Arc<dyn StateStore> = Arc::new(InMemoryStore::new());
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.budget.max_tokens = Some(1_000_000);
        config.atif_spool_dir = tempfile::tempdir().unwrap().keep().to_string_lossy().into_owned();
        let state = AppState::new(
            config,
            Arc::clone(&store),
            Arc::new(Sandbox::new(SandboxConfig::default(), Vec::new()).unwrap()),
            Arc::new(NullBus),
            None,
            Arc::new(Ed25519Signer::from_seed(&[9; 32])),
        )
        .unwrap();
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("budget-cleanup"));
        let mut prepared = state.prepare_chat(&headers, payload(), None).unwrap();
        // Root cause of the intermittent CI-only failure (3× as of
        // 2026-09-08, never in 40+ local runs): dropping the prepared
        // request arms AdmissionDebit::drop, whose refund runs on the
        // blocking pool and races the precondition read below — on a
        // loaded runner the refund won and `observed` read Ok(0). The
        // race also made the FINAL assert vacuous: a drop-refund zeroes
        // the cells with no help from the finalizer. Disarm the debit
        // exactly like a real dispatch (`take_for_dispatch`, the
        // :4021 idiom) so the cells stay genuinely spent until
        // `close_session` clears them — the behavior this test pins.
        let _ = prepared.admission_debit.take_for_dispatch();
        drop(prepared);
        let tokens_key = format!(
            "{}tokens",
            pv_state::ActionBudget::session_prefix("budget-cleanup")
        );
        let observed = store.get(&tokens_key);
        assert!(
            observed.as_ref().copied().unwrap_or(0) > 0,
            "precondition: admission must have spent from the token budget \
             (key {tokens_key:?}, observed {observed:?})",
        );
        let session = state.sessions.get("budget-cleanup").unwrap();
        session.wait_for_worker_jobs().await;
        state
            .finalizer
            .close_session(session, StopReason::SessionClosed)
            .await
            .unwrap();
        assert_eq!(
            store.get(&tokens_key).unwrap(),
            0,
            "finalization must remove the sealed session's budget counters",
        );
    }

    #[tokio::test]
    async fn consequential_tools_require_signed_workflows() {
        let config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        let state = state(config);
        let raw = serde_json::to_vec(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": "db_write", "arguments": {}}
        }))
        .unwrap();
        let mut unsigned = HeaderMap::new();
        unsigned.insert(SESSION_HEADER, HeaderValue::from_static("unsigned-write"));
        assert!(matches!(
            state.intercept_tool(&unsigned, &raw).unwrap(),
            ToolVerdict::Blocked { stage: "policy", .. }
        ));

        let mut signed = HeaderMap::new();
        signed.insert(SESSION_HEADER, HeaderValue::from_static("signed-write"));
        signed.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        assert!(state.intercept_tool(&signed, &raw).unwrap().is_allowed());
    }

    /// The semantic-loop breaker's hard-stop actions (Reject/Abort) must
    /// gate TOOL admission exactly like they gate chat: pre-fix, a
    /// breaker-open session — whose chat requests were already refused —
    /// kept executing otherwise-permitted MCP tool calls on the same
    /// session until idle close, so the "stop the runaway agent"
    /// decision only stopped half the agent.
    #[tokio::test]
    async fn open_breaker_hard_stop_refuses_tool_calls() {
        let config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        let state = state(config);
        let identity = pv_events::AgentIdentity {
            version: "dev".into(),
            charter: "anonymous".into(),
            instance_uid: "anonymous".into(),
            ttl_remaining_s: None,
        };
        let session = state.sessions.get_or_open(
            "breaker-tools",
            crate::session::Workflow::Unsigned,
            &identity,
            &state.config.breaker,
        );
        // Trip the breaker directly: identical embeddings above the
        // min-token floor are zero-novelty steps; the default window
        // trips after three.
        for _ in 0..4 {
            session.loop_state.observe_embedding(vec![1.0, 2.0, 3.0], 2_000);
        }
        assert_eq!(
            session.loop_state.state(),
            BreakerState::Open,
            "precondition: breaker must be open",
        );

        let raw = serde_json::to_vec(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": "db_read", "arguments": {}}
        }))
        .unwrap();
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("breaker-tools"));
        let refusal = state.intercept_tool(&headers, &raw);
        assert!(
            matches!(&refusal, Err(PipelineError::Blocked { context, .. }) if context.contains("circuit breaker")),
            "an open hard-stop breaker must refuse tool execution, got {refusal:?}",
        );
        assert!(
            session.enforcement_tripped(),
            "the tool-path refusal latches enforcement like the chat path",
        );
    }

    /// Required-identity tool calls with NO Authorization header are
    /// refused by the cheap pre-parse gate — an unauthenticated flood
    /// must not reach the MCP parse/canonicalize/hash work, and the
    /// refusal shape matches resolve_identity's missing-bearer arm.
    #[tokio::test]
    async fn required_identity_tool_call_without_bearer_is_refused_before_parse() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.require_identity = true;
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("anon-tools"));
        // Deliberately unparseable body: the gate fires on the header
        // alone, so the parse must never get a say in the verdict.
        let refusal = state.intercept_tool(&headers, b"{this is not json");
        assert!(
            matches!(&refusal, Err(PipelineError::Unauthorized(reason)) if reason.contains("missing bearer token")),
            "missing bearer under require_identity must refuse pre-parse, got {refusal:?}",
        );
    }

    /// `intercept_tool_durable` used to re-derive the session id from headers
    /// AFTER the verdict. For a header-less request `session_id()` mints a
    /// fresh random id on every call, so the post-verdict lookup targeted a
    /// session that never existed and the call always failed with "tool
    /// session disappeared" — after the verdict had already been computed and
    /// audited under the real (generated) session.
    #[tokio::test]
    async fn intercept_tool_durable_works_without_a_session_header() {
        let config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        let state = state(config);
        let raw = serde_json::to_vec(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": "safe_tool", "arguments": {}}
        }))
        .unwrap();
        let verdict = state
            .intercept_tool_durable(&HeaderMap::new(), &raw)
            .await
            .unwrap();
        assert!(
            verdict.is_allowed(),
            "durable interception must await the verdict's own session, not a re-derived id",
        );
    }

    /// When `prepare_chat_durable` refuses a request AFTER admission (the
    /// request's own loop analysis opened the breaker during the audit wait),
    /// it must submit the terminal failure record for the response attempt
    /// journaled at admission. Otherwise the active journal ends with a
    /// dangling non-terminal attempt and a later crash-recovery scan
    /// quarantines the session over a request the client already saw fail.
    #[tokio::test]
    async fn durable_breaker_refusal_journals_a_terminal_response_attempt() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.breaker.min_tokens = 0;
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("durable-loop"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        let repeated = serde_json::json!({
            "model": "test",
            "messages": [{
                "role": "assistant",
                "content": "I should check the database again for pending orders"
            }],
        });
        // Attempts the test itself abandons after a successful prepare: the
        // capture guard terminates them on drop (same as a client
        // disconnect), so the journal must end with NO dangling attempts.
        let mut abandoned_by_test = std::collections::HashSet::new();
        let mut refusal = None;
        for _ in 0..8 {
            match state.prepare_chat_durable(&headers, repeated.clone()).await {
                Ok(prepared) => {
                    abandoned_by_test.insert(prepared.capture_guard.attempt_id().to_owned());
                }
                Err(error) => {
                    refusal = Some(error);
                    break;
                }
            }
        }
        assert!(
            matches!(refusal, Some(PipelineError::Blocked { .. })),
            "the breaker must refuse during the audit wait, got {refusal:?}",
        );
        assert!(
            !abandoned_by_test.is_empty(),
            "at least one prepare must succeed first"
        );
        let session = state.sessions.get("durable-loop").unwrap();
        session.wait_for_worker_jobs().await;

        let digest = pv_core::digest::sha256_hex("durable-loop".as_bytes());
        let stem = digest.get(..32).unwrap();
        let journal_path =
            std::path::Path::new(&state.config.atif_spool_dir).join(format!("{stem}.events.ndjson"));
        let journal = std::fs::read_to_string(&journal_path).unwrap();
        let mut dangling = std::collections::HashSet::new();
        for (index, line) in journal.lines().enumerate() {
            let record: crate::worker::ActiveJournalRecord = crate::journal::open(
                &state.journal_key,
                "durable-loop:active",
                index as u64,
                line.as_bytes(),
            )
            .unwrap();
            if let Some(attempt) = record.response_attempt {
                if attempt.terminal {
                    assert!(
                        dangling.remove(&attempt.id),
                        "terminal record for attempt {} has no admission record",
                        attempt.id,
                    );
                } else {
                    dangling.insert(attempt.id);
                }
            }
        }
        assert!(
            dangling.is_empty(),
            "every response attempt must be terminated: the refusal terminates its own \
             attempt, and dropping a PreparedRequest (client disconnect) terminates via \
             the capture guard; still dangling: {dangling:?} (test dropped: {abandoned_by_test:?})",
        );
    }

    /// `forward_chat` must fail closed when the in-flight response marker
    /// cannot be persisted: the marker is what lets the restart-time scan
    /// (`inflight_response_sessions`) quarantine sessions whose provider
    /// response may have been observed but never captured. Dispatching
    /// without it would silently drop that crash-durability guarantee.
    /// The refusal must also terminate the journaled response attempt so
    /// no dangling non-terminal record quarantines the session later.
    #[tokio::test]
    async fn forward_chat_fails_closed_when_marker_cannot_persist() {
        let config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("marker-outage"));
        let prepared = state.prepare_chat(&headers, payload(), None).unwrap();
        let attempt_id = prepared.capture_guard.attempt_id().to_owned();
        // Sabotage the marker directory: a regular file at
        // <spool>/inflight-responses makes `create_dir_all` inside
        // `write_atomic` fail for every marker write.
        std::fs::write(
            std::path::Path::new(&state.config.atif_spool_dir).join(crate::spool::INFLIGHT_RESPONSES),
            b"",
        )
        .unwrap();
        let error = state.forward_chat(prepared).await.err().unwrap();
        // Unavailable (not Upstream) proves the refusal happened BEFORE
        // any upstream dispatch was attempted.
        assert!(
            matches!(error, PipelineError::Unavailable { .. }),
            "marker persist failure must map to Unavailable, got {error:?}",
        );

        let session = state.sessions.get("marker-outage").unwrap();
        session.wait_for_worker_jobs().await;
        let digest = pv_core::digest::sha256_hex("marker-outage".as_bytes());
        let stem = digest.get(..32).unwrap();
        let journal_path =
            std::path::Path::new(&state.config.atif_spool_dir).join(format!("{stem}.events.ndjson"));
        let journal = std::fs::read_to_string(&journal_path).unwrap();
        let mut saw_admission = false;
        let mut saw_terminal = false;
        for (index, line) in journal.lines().enumerate() {
            let record: crate::worker::ActiveJournalRecord = crate::journal::open(
                &state.journal_key,
                "marker-outage:active",
                index as u64,
                line.as_bytes(),
            )
            .unwrap();
            if let Some(attempt) = record.response_attempt {
                if attempt.id == attempt_id {
                    if attempt.terminal {
                        saw_terminal = true;
                    } else {
                        saw_admission = true;
                    }
                }
            }
        }
        assert!(saw_admission, "admission must have journaled the attempt");
        assert!(
            saw_terminal,
            "the refused request's response attempt must be terminated in the journal",
        );
    }

    /// A session that only routes MCP tool calls (no chat completions) must
    /// still be considered active by `idle_sessions` — otherwise the
    /// reconciler tick force-closes tool-only sessions after
    /// `session_idle_close_s` and every subsequent tool call fails with
    /// "session is already closed". `prepare_chat` refreshes
    /// `last_activity_ms` via `touch()`; `intercept_tool` did not.
    #[tokio::test]
    async fn intercept_tool_refreshes_session_activity() {
        use std::sync::atomic::Ordering as AtomicOrdering;

        let config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        let state = state(config);
        let raw = serde_json::to_vec(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": "safe_tool", "arguments": {}}
        }))
        .unwrap();
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("tool-only-session"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));

        state.intercept_tool(&headers, &raw).unwrap();
        let session = state.sessions.get("tool-only-session").unwrap();
        // Backdate the session to before the epoch's first second — this stands
        // in for a real session that has been quiet long enough to hit the
        // idle sweeper's cutoff.
        session.last_activity_ms.store(0, AtomicOrdering::Release);

        state.intercept_tool(&headers, &raw).unwrap();

        let refreshed = session.last_activity_ms.load(AtomicOrdering::Acquire);
        assert!(
            refreshed > 0,
            "intercept_tool must refresh last_activity_ms so a tool-only session is not force-closed by the idle sweeper — got {refreshed}",
        );
    }

    /// `intercept_tool` used to call `sandbox.check` unconditionally and then
    /// post-hoc override an `Allowed` verdict to `Blocked` when the workflow
    /// gate vetoes a consequential tool on an unsigned session. But
    /// `sandbox.check`'s budget gate (`try_spend_many`) spends before the
    /// override runs, so a legitimate client asking for a consequential tool
    /// on the wrong workflow got their `max_total_tool_calls` counter
    /// decremented for a call the sandbox never actually authorized. Once the
    /// client upgraded to a signed session to make the call for real, they
    /// hit the cap short of the number they were configured for.
    #[tokio::test]
    async fn unsigned_consequential_veto_does_not_spend_budget() {
        let store: Arc<dyn StateStore> = Arc::new(InMemoryStore::new());
        let sandbox_config = SandboxConfig {
            budget: pv_state::BudgetSpec {
                max_total_tool_calls: Some(5),
                ..pv_state::BudgetSpec::default()
            },
            ..SandboxConfig::default()
        };
        let sandbox = Arc::new(Sandbox::new(sandbox_config, Vec::new()).unwrap());
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.atif_spool_dir = tempfile::tempdir().unwrap().keep().to_string_lossy().into_owned();
        let state = AppState::new(
            config,
            Arc::clone(&store),
            sandbox,
            Arc::new(NullBus),
            None,
            Arc::new(Ed25519Signer::from_seed(&[9; 32])),
        )
        .unwrap();

        let raw = serde_json::to_vec(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": "db_write", "arguments": {}}
        }))
        .unwrap();
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("unsigned-consequential"));
        // Default workflow is unsigned; no explicit x-pv-workflow header.

        let verdict = state.intercept_tool(&headers, &raw).unwrap();
        assert!(
            matches!(verdict, ToolVerdict::Blocked { stage: "policy", .. }),
            "consequential tool on unsigned workflow must block at the policy gate; got {verdict:?}",
        );

        let session_digest = pv_core::digest::sha256_hex("unsigned-consequential".as_bytes());
        let key = format!("budget:{{{}}}:total_calls", session_digest.get(..32).unwrap());
        let count = store.get(&key).unwrap();
        assert_eq!(
            count, 0,
            "sandbox.check must not spend the budget for a call the workflow gate vetoes — otherwise a legitimate client that upgrades to signed and retries hits the cap short of the configured limit",
        );
    }

    #[tokio::test]
    async fn full_capture_queue_fails_closed_before_upstream() {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        // The fused permit split (worker_capacity + response_capacity)
        // gives distinct exhaustion counters but keeps the *effective*
        // total admission the same as before, because response
        // capture doesn't hold an mpsc slot up front — only the
        // response semaphore. So with worker_channel_capacity=2 the
        // first request consumes 1 worker semaphore + 1 response
        // semaphore + 1 mpsc slot; the second request consumes the
        // remaining worker semaphore + response semaphore + mpsc slot;
        // the third would fail. Set to 1 for a clean single-request
        // saturation.
        config.worker_channel_capacity = 1;
        config.breaker.min_tokens = u64::MAX;
        let sink = Arc::new(BlockingSink {
            entered: AtomicBool::new(false),
            release: tokio::sync::Notify::new(),
        });
        let state = AppState::new_with_backends(
            config,
            Arc::new(InMemoryStore::new()),
            Arc::new(Sandbox::new(SandboxConfig::default(), Vec::new()).unwrap()),
            Arc::new(NullBus),
            None,
            Arc::new(Ed25519Signer::from_seed(&[10; 32])),
            Arc::new(HashEmbedder::default()),
            sink.clone(),
        )
        .unwrap();
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("overload"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        let mut agent_payload = payload();
        agent_payload["messages"][0]["role"] = Value::String("assistant".to_owned());
        let mut first = state.prepare_chat(&headers, agent_payload.clone(), None).unwrap();
        // Abandoning without forwarding: defuse the guard, or its terminal
        // job (refused by the size-1 queue) marks the session
        // capture-failed and the worker skips the sink this test blocks on.
        first.capture_guard.defuse();
        drop(first);
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            while !sink.entered.load(AtomicOrdering::Acquire) {
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();
        assert!(matches!(
            state.prepare_chat(&headers, agent_payload, None),
            Err(PipelineError::Unavailable { .. })
        ));
        sink.release.notify_waiters();
    }

    #[test]
    fn identity_scope_matching_supports_exact_and_namespace_wildcards() {
        assert!(scope_allows(&["chat:write".into()], "chat:write"));
        assert!(scope_allows(&["tool:*".into()], "tool:db_write"));
        assert!(scope_allows(&["*".into()], "session:promote"));
        assert!(!scope_allows(&["chat:read".into()], "chat:write"));
        assert!(!scope_allows(&["tooling:*".into()], "tool:db_write"));
    }

    #[test]
    fn developer_and_tool_roles_map_to_valid_system_steps() {
        for (role, expects_observation) in [("developer", false), ("tool", true)] {
            let capture = atif_capture_from_request(&serde_json::json!({
                "messages": [{
                    "role": role,
                    "content": "context",
                    "tool_call_id": "call-1"
                }]
            }))
            .unwrap();
            assert_eq!(capture.source, pv_atif::Source::System);
            assert_eq!(capture.observation.is_some(), expects_observation);
            assert!(capture.model_name.is_none());
        }
    }

    /// `messages` field with the wrong type used to return
    /// the same "chat payload has no messages" as an absent field,
    /// steering support tickets at a phantom bug. Verify each class
    /// now returns a discriminating error message.
    #[test]
    fn atif_capture_rejects_non_array_messages_with_precise_diagnostics() {
        fn bad_request_message(payload: serde_json::Value) -> String {
            match atif_capture_from_request(&payload) {
                Ok(_) => panic!("expected BadRequest, got Ok"),
                Err(PipelineError::BadRequest { context: msg, .. }) => msg,
                Err(other) => panic!("expected BadRequest, got {other:?}"),
            }
        }
        // (1) Field entirely missing (e.g., OpenAI Responses API caller
        //     who used `input` instead of `messages`).
        let msg = bad_request_message(serde_json::json!({
            "input": [{ "role": "user", "content": "hi" }]
        }));
        assert!(msg.contains("missing 'messages'"), "got {msg}");
        // (2) Field present but wrong JSON shape — object.
        let msg = bad_request_message(serde_json::json!({
            "messages": { "0": { "role": "user", "content": "hi" } }
        }));
        assert!(msg.contains("must be a JSON array"), "got {msg}");
        assert!(msg.contains("object"), "got {msg}");
        // (3) Field present, is an array, but empty.
        let msg = bad_request_message(serde_json::json!({ "messages": [] }));
        assert!(msg.contains("empty"), "got {msg}");
    }

    /// Multiple X-PV-Session (or X-PV-Workflow) headers on
    /// a single request must be refused. HeaderMap::get() returns
    /// only the first, so an intermediary that merges duplicates on
    /// the wire and downstream code that reads them separately can
    /// disagree on which session id was in effect — a header
    /// smuggling desync. Refuse loudly at ingress.
    #[test]
    fn duplicate_x_pv_session_header_is_refused() {
        let mut headers = HeaderMap::new();
        headers.append(SESSION_HEADER, HeaderValue::from_static("sessA"));
        headers.append(SESSION_HEADER, HeaderValue::from_static("sessB"));
        match session_id(&headers) {
            Ok(id) => panic!("expected duplicate-header refusal, got session_id {id:?}"),
            Err(PipelineError::BadRequest { context: msg, .. }) => {
                assert!(msg.contains("more than one"), "got {msg}");
                assert!(msg.contains("x-pv-session"), "got {msg}");
            }
            Err(other) => panic!("expected BadRequest, got {other:?}"),
        }
    }

    #[test]
    fn duplicate_x_pv_workflow_header_is_refused() {
        let mut headers = HeaderMap::new();
        headers.append(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        headers.append(WORKFLOW_HEADER, HeaderValue::from_static("unsigned"));
        match workflow(&headers, "signed") {
            Ok(w) => panic!("expected duplicate-header refusal, got workflow {w:?}"),
            Err(PipelineError::BadRequest { context: msg, .. }) => {
                assert!(msg.contains("more than one"), "got {msg}");
                assert!(msg.contains("x-pv-workflow"), "got {msg}");
            }
            Err(other) => panic!("expected BadRequest, got {other:?}"),
        }
    }

    /// Single X-PV-Session must continue to work — this is the
    /// happy-path regression guard for the duplicate-header refusal.
    #[test]
    fn single_x_pv_session_header_still_flows() {
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("only-one"));
        assert_eq!(session_id(&headers).unwrap(), "only-one");
    }

    /// Identity hot path (`resolve_identity`) must refuse
    /// duplicate `Authorization` headers, symmetric with X-PV-Session.
    /// Previously `HeaderMap::get(AUTHORIZATION)` returned the first
    /// value only — the harness would authenticate as `A` while
    /// log aggregators / WAFs seeing a merged `A, B` form would
    /// attribute the request to `B`. This is the same identity split-brain
    /// already closed for session headers.
    #[tokio::test]
    async fn resolve_identity_refuses_duplicate_authorization_header() {
        let state = null_state();
        let mut headers = HeaderMap::new();
        headers.append(
            axum::http::header::AUTHORIZATION,
            HeaderValue::from_static("Bearer aaaa"),
        );
        headers.append(
            axum::http::header::AUTHORIZATION,
            HeaderValue::from_static("Bearer bbbb"),
        );
        match state.resolve_identity(&headers, None) {
            Ok(_) => panic!("expected duplicate-header refusal, got Ok"),
            Err(PipelineError::BadRequest { context: msg, .. }) => {
                assert!(msg.contains("more than one"), "got {msg}");
                assert!(msg.contains("authorization"), "got {msg}");
            }
            Err(other) => panic!("expected BadRequest, got {other:?}"),
        }
    }

    /// Single-value Authorization still flows through (happy-path
    /// regression guard for the dedup refusal).
    #[tokio::test]
    async fn resolve_identity_accepts_single_authorization_header() {
        let state = null_state();
        let mut headers = HeaderMap::new();
        headers.insert(
            axum::http::header::AUTHORIZATION,
            HeaderValue::from_static("Bearer aaaa"),
        );
        // No validator configured → the refuse-401 posture fires.
        // The important assertion is that we did NOT get a
        // BadRequest("more than one") on a single header.
        let outcome = state.resolve_identity(&headers, None);
        assert!(
            !matches!(&outcome, Err(PipelineError::BadRequest { context: msg, .. }) if msg.contains("more than one")),
            "single Authorization header must not be refused as duplicate; got {outcome:?}"
        );
    }

    /// With `ignore_client_authorization`
    /// opted in and no validator configured, a stock OpenAI SDK's
    /// mandatory `Authorization` header must be accepted-and-discarded
    /// — the request proceeds anonymously instead of hard-401ing on
    /// request one. Without the opt-in, the diagnostic 401 stays.
    #[tokio::test]
    async fn ignore_client_authorization_accepts_and_discards_sdk_bearer() {
        let mut headers = HeaderMap::new();
        headers.insert(
            axum::http::header::AUTHORIZATION,
            HeaderValue::from_static("Bearer sk-test-openai-sdk-key"),
        );

        // Default (flag off): diagnostic 401.
        let refused = null_state().resolve_identity(&headers, None);
        assert!(
            matches!(&refused, Err(PipelineError::Unauthorized(_))),
            "bearer with no validator and no opt-in must stay 401; got {refused:?}"
        );

        // Flag on: accepted as anonymous.
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.require_identity = false;
        config.ignore_client_authorization = true;
        let accepted = state(config).resolve_identity(&headers, None);
        match accepted {
            Ok(identity) => {
                assert_eq!(identity.instance_uid, "anonymous");
            }
            Err(error) => {
                panic!("ignore_client_authorization must accept the SDK bearer anonymously; got {error:?}")
            }
        }
    }

    /// The two contradictory combinations must be refused at
    /// validate(): require_identity needs the header, passthrough
    /// forwards it — neither can coexist with discarding it.
    #[test]
    fn ignore_client_authorization_is_refused_with_identity_or_passthrough() {
        let mut with_identity = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        with_identity.ignore_client_authorization = true;
        with_identity.require_identity = true;
        with_identity.identity_hmac_secret_file = Some("/tmp/hmac".into());
        let err = with_identity.validate().unwrap_err();
        assert!(err.contains("ignore_client_authorization"), "{err}");
        assert!(err.contains("require_identity"), "{err}");

        let mut with_passthrough = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        with_passthrough.ignore_client_authorization = true;
        with_passthrough.upstream_authorization_passthrough = true;
        let err = with_passthrough.validate().unwrap_err();
        assert!(err.contains("ignore_client_authorization"), "{err}");
        assert!(err.contains("passthrough"), "{err}");
    }

    /// A credential-free 401 flood must not mint an
    /// unbounded stream of durable signed audit sessions. The first
    /// 60 rejections per minute get full audit records; the rest are
    /// counted only. Proven by driving 100 rejections and asserting
    /// the overflow counter carries the surplus.
    #[tokio::test]
    async fn identity_rejection_audit_records_are_bounded_per_minute() {
        let state = null_state();
        for i in 0..100u32 {
            state
                .enqueue_transient_failure(
                    &format!("probe-{i}"),
                    StopReason::IdentityRejected,
                    "missing bearer token".to_owned(),
                )
                .unwrap();
        }
        let total = state
            .metrics
            .counter(
                "pv_identity_rejections_total",
                "Requests refused by identity validation",
            )
            .get();
        let unaudited = state
            .metrics
            .counter(
                "pv_identity_rejections_unaudited_total",
                "Identity rejections beyond the per-minute audit-record budget (counted only)",
            )
            .get();
        assert_eq!(total, 100, "every rejection must be counted");
        assert_eq!(
            unaudited, 40,
            "rejections beyond the 60/min budget must be count-only, not signed audit sessions"
        );
    }

    /// RFC 7235 §2.1 auth-scheme is case-insensitive.
    /// A caller sending `Authorization: bearer eyJ...` or `BEARER`
    /// used to be silently downgraded to anonymous (when
    /// require_identity=false, the shipped default) — repudiation
    /// class. Verify each case now parses to the same credential.
    #[test]
    fn strip_bearer_scheme_matches_case_insensitively() {
        for scheme in ["Bearer aaaa", "bearer aaaa", "BEARER aaaa", "BeArEr aaaa"] {
            assert_eq!(
                strip_bearer_scheme(scheme),
                Some("aaaa"),
                "scheme {scheme:?} did not parse to `aaaa`"
            );
        }
        // Tab between scheme and credential (per the RFC's `SP / HTAB` grammar).
        assert_eq!(strip_bearer_scheme("Bearer\taaaa"), Some("aaaa"));
        // Multiple spaces are allowed.
        assert_eq!(strip_bearer_scheme("Bearer   aaaa"), Some("aaaa"));
        // Missing whitespace or empty credential — refused.
        assert_eq!(strip_bearer_scheme("BearerXYZ"), None);
        assert_eq!(strip_bearer_scheme("Bearer "), None);
        assert_eq!(strip_bearer_scheme("Bearer"), None);
        // Different scheme — refused.
        assert_eq!(strip_bearer_scheme("Basic ZmY6bGFyZQ=="), None);
        // Too short to hold the scheme name — refused.
        assert_eq!(strip_bearer_scheme(""), None);
    }

    fn null_state() -> AppState {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.require_identity = false;
        let sandbox = Sandbox::new(SandboxConfig::default(), Vec::new()).unwrap();
        let store = Arc::new(InMemoryStore::new());
        let bridge: Arc<dyn EventBus> = Arc::new(NullBus);
        let signer: Arc<dyn Signer> = Arc::new(Ed25519Signer::generate());
        let embedder = Arc::new(HashEmbedder::default());
        let vector_sink: Arc<dyn VectorSink> = Arc::new(NoopVectorSink);
        AppState::new_with_backends(
            config,
            store,
            Arc::new(sandbox),
            bridge,
            None,
            signer,
            embedder,
            vector_sink,
        )
        .unwrap()
    }

    fn scoped_token(scopes: &[&str]) -> String {
        let now = pv_core::time::now_ms() / pv_core::units::MS_PER_SEC;
        let claims = pv_identity::NhiClaims {
            sub: "agent:test".into(),
            iss: "https://idp.example".into(),
            aud: "proofvisor".into(),
            iat: now,
            nbf: None,
            exp: now + 600,
            jti: pv_core::new_event_uid(),
            azp: None,
            act: None,
            instance_uid: "scoped-instance".into(),
            charter: "scoped-charter".into(),
            version: "1".into(),
            scopes: scopes.iter().map(|scope| (*scope).to_owned()).collect(),
            parent_token: None,
        };
        let mut header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::HS256);
        header.kid = Some("scope-key".into());
        jsonwebtoken::encode(
            &header,
            &claims,
            &jsonwebtoken::EncodingKey::from_secret(b"scope-secret"),
        )
        .unwrap()
    }

    fn scoped_state() -> AppState {
        let mut config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        config.require_identity = true;
        config.enforce_identity_scopes = true;
        let validator = pv_identity::IdentityValidator::new("proofvisor");
        validator
            .add_key(
                "scope-key",
                pv_identity::KeyMaterial::HmacSecret(b"scope-secret".to_vec()),
            )
            .unwrap();
        AppState::new(
            config,
            Arc::new(InMemoryStore::new()),
            Arc::new(Sandbox::new(SandboxConfig::default(), Vec::new()).unwrap()),
            Arc::new(NullBus),
            Some(Arc::new(validator)),
            Arc::new(Ed25519Signer::from_seed(&[12; 32])),
        )
        .unwrap()
    }

    #[tokio::test]
    async fn real_jwt_scopes_gate_chat_and_lifecycle_operations() {
        let state = scoped_state();
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("scoped-session"));
        headers.insert(
            axum::http::header::AUTHORIZATION,
            HeaderValue::from_str(&format!("Bearer {}", scoped_token(&["chat:write"]))).unwrap(),
        );
        let prepared = state.prepare_chat(&headers, payload(), None).unwrap();
        // RFC 6750 §3.1: missing scope on a valid token is 403
        // insufficient_scope (Blocked), NOT 401 — a refreshed token would
        // carry the same grant, so telling the SDK to re-authenticate
        // (401's semantic) is wrong and loops.
        assert!(matches!(
            state.authorize_session(&headers, &prepared.session, "session:close"),
            Err(PipelineError::Blocked { .. })
        ));
        headers.insert(
            axum::http::header::AUTHORIZATION,
            HeaderValue::from_str(&format!(
                "Bearer {}",
                scoped_token(&["chat:write", "session:close"])
            ))
            .unwrap(),
        );
        state
            .authorize_session(&headers, &prepared.session, "session:close")
            .unwrap();
    }

    #[tokio::test]
    async fn upstream_failure_is_added_to_signed_chain() {
        let config = HarnessConfig::for_tests("http://127.0.0.1:9", "/tmp", "/tmp");
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("upstream-failure"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        let prepared = state.prepare_chat(&headers, payload(), None).unwrap();
        let session = Arc::clone(&prepared.session);
        assert!(matches!(
            state.forward_chat(prepared).await,
            Err(PipelineError::Upstream { .. })
        ));
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            while session.chain.lock().count() < 2 {
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();
    }

    /// Regression lock for CWE-209 information exposure. Before the fix,
    /// `PipelineError::upstream(reqwest_err.to_string())` embedded the
    /// operator-configured upstream URL in the JSON error body — any client
    /// that triggered an upstream failure could discover the URL, potentially
    /// leaking an internal hostname. The client-facing message must now be a
    /// stable category ("upstream unreachable", "upstream timed out", …) that
    /// does not depend on the URL.
    #[tokio::test]
    async fn upstream_failure_message_does_not_leak_configured_url() {
        // Use a distinctive private-network host so a regression is unmissable
        // in the assertion below.
        let sentinel_url = "http://internal-sentinel-host.corp.example:65001";
        let config = HarnessConfig::for_tests(sentinel_url, "/tmp", "/tmp");
        let state = state(config);
        let mut headers = HeaderMap::new();
        headers.insert(SESSION_HEADER, HeaderValue::from_static("cwe-209-check"));
        headers.insert(WORKFLOW_HEADER, HeaderValue::from_static("signed"));
        let prepared = state.prepare_chat(&headers, payload(), None).unwrap();
        let error = match state.forward_chat(prepared).await {
            Ok(_) => panic!("connect to unroutable host must fail"),
            Err(error) => error,
        };
        let message = error.to_string();
        assert!(
            !message.contains("internal-sentinel-host"),
            "upstream error message {message:?} leaks the configured URL"
        );
        assert!(
            !message.contains("65001"),
            "upstream error message {message:?} leaks the configured port"
        );
        assert!(
            !message.contains("corp.example"),
            "upstream error message {message:?} leaks the configured domain"
        );
    }

    /// Regression lock for CWE-209 information exposure. Before the fix,
    /// `PipelineError::Unauthorized(identity_err.to_string())` embedded the
    /// specific `IdentityError` variant's `Display` text in the JSON error
    /// body. Because that text distinguishes at least sixteen distinct
    /// failure modes — and echoes the attacker-supplied `kid`, `iss`, and
    /// `alg` values — each response became an enumeration oracle: an
    /// attacker could iterate candidate `kid`s and read the response body
    /// to discover which are configured on our validator. The client-facing
    /// message must now be a single stable string, and must never echo
    /// attacker-controlled token fields.
    #[test]
    fn identity_validation_failure_does_not_leak_kid_or_issuer() {
        // Every variant that an attacker can reach via a crafted token must
        // classify to exactly the same client-facing string. Any divergence
        // is an enumeration oracle for validator configuration.
        //
        // Sentinels that would appear in the raw `Display` output are
        // asserted absent — this catches accidental future variants that
        // slip through the `_` catch-all with a different classification.
        let sentinel_kid = "sentinel-attacker-kid";
        let sentinel_iss = "https://sentinel-attacker-iss.example.invalid";
        let sentinel_alg = "HS512";
        let cases: Vec<(&'static str, pv_identity::IdentityError)> = vec![
            (
                "Malformed",
                pv_identity::IdentityError::Malformed("jwt parse".into()),
            ),
            ("MissingKid", pv_identity::IdentityError::MissingKid),
            (
                "UnknownKid",
                pv_identity::IdentityError::UnknownKid(sentinel_kid.into()),
            ),
            (
                "AlgorithmRejected",
                pv_identity::IdentityError::AlgorithmRejected {
                    alg: sentinel_alg.into(),
                    kid: sentinel_kid.into(),
                },
            ),
            (
                "Verification",
                pv_identity::IdentityError::Verification("bad sig".into()),
            ),
            ("TtlTooLong", pv_identity::IdentityError::TtlTooLong(9999)),
            (
                "BadTimestamps",
                pv_identity::IdentityError::BadTimestamps { iat: 10, exp: 5 },
            ),
            (
                "FutureIat",
                pv_identity::IdentityError::FutureIat {
                    iat: 999_999_999,
                    now: 1,
                },
            ),
            ("EmptyField", pv_identity::IdentityError::EmptyField("charter")),
            (
                "SpoofingCharacter",
                pv_identity::IdentityError::SpoofingCharacter("charter"),
            ),
            (
                "BadIssuer",
                pv_identity::IdentityError::BadIssuer(sentinel_iss.into()),
            ),
            (
                "ScopeEscalation",
                pv_identity::IdentityError::ScopeEscalation {
                    scope: "chat:write".into(),
                },
            ),
            (
                "ExpEscalation",
                pv_identity::IdentityError::ExpEscalation {
                    child: 100,
                    parent: 50,
                },
            ),
            ("ChainTooDeep", pv_identity::IdentityError::ChainTooDeep(5)),
        ];

        let mut classifications: std::collections::BTreeSet<&'static str> = std::collections::BTreeSet::new();
        for (name, error) in &cases {
            let client_msg = super::classify_identity_error(error);
            classifications.insert(client_msg);
            assert!(
                !client_msg.contains(sentinel_kid),
                "{name}: client message {client_msg:?} echoes attacker kid"
            );
            assert!(
                !client_msg.contains(sentinel_iss),
                "{name}: client message {client_msg:?} echoes attacker iss"
            );
            assert!(
                !client_msg.contains(sentinel_alg),
                "{name}: client message {client_msg:?} echoes attacker alg"
            );
            // The full PipelineError as rendered to the client must also
            // not leak — this is what actually goes on the wire via
            // `pipeline_error()` -> `Json(json!({"error": err.to_string()}))`.
            let wire = PipelineError::Unauthorized(client_msg.to_owned()).to_string();
            assert!(
                !wire.contains(sentinel_kid),
                "{name}: wire error {wire:?} echoes attacker kid"
            );
            assert!(
                !wire.contains(sentinel_iss),
                "{name}: wire error {wire:?} echoes attacker iss"
            );
            assert!(
                !wire.contains(sentinel_alg),
                "{name}: wire error {wire:?} echoes attacker alg"
            );
        }
        // All attacker-reachable variants collapse to exactly one class —
        // the `Jwks` variant (server-side misconfig only, not on the
        // request path) may distinctly classify.
        assert_eq!(
            classifications.len(),
            1,
            "attacker-reachable IdentityError variants classify to multiple messages {classifications:?}: this is an enumeration oracle"
        );

        // And Jwks classifies distinctly (server-side, not attacker-reachable).
        let jwks_msg = super::classify_identity_error(&pv_identity::IdentityError::Jwks("bad jwks".into()));
        assert_ne!(
            jwks_msg,
            *classifications.iter().next().unwrap(),
            "server-side misconfig should classify distinctly for operator dashboards"
        );
    }

    /// The secret reader must fail loudly on missing/empty sources and
    /// trim whitespace — silent unauthenticated proxying is the worst
    /// failure mode because the operator sees only upstream 401s.
    #[test]
    fn read_secret_source_handling() {
        let env = |name: &str| -> Option<String> {
            match name {
                "GOOD_KEY" => Some("sk-live-123\n".into()),
                "EMPTY_KEY" => Some("   ".into()),
                _ => None,
            }
        };
        assert_eq!(
            super::read_secret_from(env, Some("GOOD_KEY"), None, "upstream API key")
                .unwrap()
                .as_deref(),
            Some("sk-live-123"),
            "value must be trimmed"
        );
        assert!(super::read_secret_from(env, Some("MISSING_KEY"), None, "upstream API key").is_err());
        assert!(super::read_secret_from(env, Some("EMPTY_KEY"), None, "upstream API key").is_err());
        assert!(super::read_secret_from(env, None, None, "upstream API key")
            .unwrap()
            .is_none());

        // File source: owner-only accepted (with trim), world-readable refused.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("key");
        std::fs::write(&path, "sk-file-456\n").unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt as _;
            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
            let error = super::read_secret_from(env, None, path.to_str(), "upstream API key").unwrap_err();
            assert!(error.to_string().contains("must be owner-only"), "{error}");
            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).unwrap();
        }
        assert_eq!(
            super::read_secret_from(env, None, path.to_str(), "upstream API key")
                .unwrap()
                .as_deref(),
            Some("sk-file-456")
        );
        // Missing file is a hard error, not silent None.
        assert!(
            super::read_secret_from(env, None, dir.path().join("absent").to_str(), "upstream API key")
                .is_err()
        );
    }

    /// Auth resolution renders scheme-prefixed and raw header values,
    /// marks them sensitive, and never leaks the key through `describe`.
    #[test]
    fn upstream_auth_resolution_and_description() {
        let dir = tempfile::tempdir().unwrap();
        let key_path = dir.path().join("azure.key");
        std::fs::write(&key_path, "azure-raw-key").unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt as _;
            std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)).unwrap();
        }

        // Bearer scheme.
        let mut config = HarnessConfig::for_tests("http://up", "spool", "bridge");
        config.upstream_api_key_file = Some(key_path.to_string_lossy().into_owned());
        let (name, value) = super::resolve_upstream_auth(&config).unwrap().unwrap();
        assert_eq!(name.as_str(), "authorization");
        assert_eq!(value.to_str().unwrap(), "Bearer azure-raw-key");
        assert!(value.is_sensitive(), "credential must redact in Debug output");

        // Raw scheme (Azure api-key style).
        config.upstream_auth_header = "api-key".into();
        config.upstream_auth_scheme = String::new();
        let (name, value) = super::resolve_upstream_auth(&config).unwrap().unwrap();
        assert_eq!(name.as_str(), "api-key");
        assert_eq!(value.to_str().unwrap(), "azure-raw-key");

        // Description names the source but never the value.
        let described = super::describe_upstream_auth(&config);
        assert!(described.contains("api-key from file"), "{described}");
        assert!(!described.contains("azure-raw-key"), "{described}");

        // No auth configured resolves to None and describes as none.
        let bare = HarnessConfig::for_tests("http://up", "spool", "bridge");
        assert!(super::resolve_upstream_auth(&bare).unwrap().is_none());
        assert_eq!(super::describe_upstream_auth(&bare), "none");

        // Tool bearer renders Bearer form from a file source.
        let mut tooling = HarnessConfig::for_tests("http://up", "spool", "bridge");
        tooling.tool_upstream_url = Some("http://tools/mcp".into());
        tooling.tool_upstream_bearer_file = Some(key_path.to_string_lossy().into_owned());
        let bearer = super::resolve_tool_auth(&tooling).unwrap().unwrap();
        assert_eq!(bearer.to_str().unwrap(), "Bearer azure-raw-key");
        assert!(bearer.is_sensitive());
    }
}