dwara_core/

observability.rs

//! Observability substrate (DW-021, feature analysis sections 4.17 and
//! 4.19): tracing spans per request phase, structured JSON access logs,
//! Prometheus metrics, request IDs, and the unified error envelope.
//!
//! ## Tracing and spans
//!
//! `tracing` (MIT) is the substrate. Every request opens a root span
//! named `request` carrying `request_id`, `method`, `path` (WITHOUT the
//! query string — redaction: query strings carry tokens), `consumer`,
//! `route`, and `listener` fields, plus child spans at each phase:
//! `authn`, `authz`, `ratelimit`, `admission`, `upstream_pick` (emitted
//! inside the upstream handle where the pick actually happens), and one
//! `upstream_attempt` span per send. The binary installs a
//! `tracing-subscriber` JSON formatter over STDOUT filtered by
//! `DWARA_LOG` (RUST_LOG syntax, default `dwara=info`).
//!
//! ## OTLP decision
//!
//! The OTLP exporter (`opentelemetry` + `opentelemetry-otlp`) was
//! evaluated and NOT added in v1: the crates are heavy (their own HTTP
//! transport, tonic/protonic codegen) against a musl <25MB binary-size
//! budget (DW-026) and a compute-conscious CI. The span STRUCTURE ships
//! now and is proven by an in-process span-capture test (one trace shows
//! all phases). Since #126 the exporter exists behind the default-off
//! `otlp` cargo FEATURE on dwara-bin (the env var `DWARA_OTLP_ENDPOINT`
//! is a binary-level knob and the tracing subscriber the exporter must
//! hook lives in the bin; this module stays dependency-light by design).
//! Default build: `DWARA_OTLP_ENDPOINT` remains RESERVED but inert;
//! feature build: the spans defined here export over http/protobuf.
//!
//! ## Access logs
//!
//! One event at target `dwara::access` per completed request (rendered
//! as one JSON line by the binary's subscriber): `ts` (subscriber
//! timestamp), `request_id`, `method`, `path` (sanitized: no query),
//! `status`, `duration_ms`, `route`, `consumer`, `upstream`,
//! `endpoint`, `attempts`, and the `rate_limited` / `broken` / `shed`
//! flags. `bytes_in`/`bytes_out` are deliberately omitted: the proxy
//! path is zero-buffering by design (DW-009) and body sizes are not
//! cheaply available for streamed bodies — recorded here as an honest
//! omission rather than a wrong number. Sampling: `DWARA_ACCESS_LOG_
//! SAMPLE` (0.0-1.0, default 1.0); responses with status >= 500 are
//! ALWAYS logged regardless of sampling (documented rule).
//!
//! ## Metrics
//!
//! Prometheus text format on `/metrics`, reserved on every HTTP(S)
//! listener exactly like `/healthz` (a configured route matching it is
//! shadowed — same accepted v1 behavior). The `prometheus` crate (MIT,
//! default features off: text format only, no protobuf exporter) backs
//! the registry. Metric families:
//!
//! - `requests_total{route,listener,status_class}` counter
//! - `request_duration_seconds{route}` histogram
//! - `upstream_attempts_total{upstream,endpoint,status_class}` counter
//! - `retries_total{upstream}` counter
//! - `dwara_hedge_sent_total{upstream}` counter (DW-063) — speculative
//!   hedge copies sent after the tail-latency timer fired
//! - `dwara_mirror_sent_total{upstream}` counter (DW-062) — shadow
//!   traffic mirror requests sent to a mirror upstream
//! - `rate_limited_total{route}` counter
//! - `dwara_rate_limiter_adaptive_factor{policy}` gauge (DW-089) — the
//!   current adaptive rate-limit factor per policy (1.0 = configured
//!   rate; < 1.0 tightens, > 1.0 relaxes; bounded [min, max])
//! - `dwara_rate_limiter_origin_signal_total{policy,signal}` counter
//!   (DW-089) — upstream origin signals received (`retry_after`), by
//!   policy and signal (both config-bounded)
//! - `dwara_rate_limiter_adaptive_tightened_total{policy}` counter
//!   (DW-089) — adaptive factor decreases, by policy
//! - `dwara_rate_limiter_adaptive_relaxed_total{policy}` counter
//!   (DW-089) — adaptive factor increases, by policy
//! - `dwara_canary_promotions_total{group}` counter (DW-091) — canary
//!   promotions, by group (service name or AI alias)
//! - `dwara_canary_rollbacks_total{group}` counter (DW-091) — canary
//!   rollbacks (including severe-regression immediate rollbacks to 0)
//! - `dwara_canary_weight{group}` gauge (DW-091) — current canary
//!   weight, by group (the total weight stays constant)
//! - `shed_total{priority}` counter
//! - `dwara_policy_dry_run_total{phase,route}` counter (DW-041) —
//!   requests a dry-run (monitor) policy would have rejected, by phase
//!   (`route_limits`, `authz`, `rate_limit`, `load_shed`) and route
//!   (`unrouted` for pre-404 listener/global policies); the request was
//!   ALLOWED. Together with the `dwara::policy` warn events (phase,
//!   would-be status, reason, route, consumer, request id) this family
//!   IS the dry-run report — no separate endpoint.
//! - `breaker_state{upstream}` gauge (0 closed, 1 open, 2 half-open)
//! - `endpoint_health{upstream,endpoint}` gauge (1 available, 0 ejected)
//! - `upstream_fail_open_picks{upstream}` gauge (scrape-time snapshot of
//!   the balancer's monotonic fail-open counter; a gauge rather than a
//!   counter so the hot pick path stays free of registry coupling)
//! - `active_requests` gauge
//! - `config_generation` gauge
//! - `jwks_refresh_total{provider}` counter
//! - `dwara_rate_limiter_evictions_total` gauge (scrape-time snapshot
//!   of the rate-limit engine's monotonic eviction counter, aggregated
//!   over every compiled rule; resets when a reload rebuilds the
//!   engine; a gauge rather than a counter so the hot check path stays
//!   free of registry coupling)
//! - `dwara_rate_limiter_live_keys` gauge (live per-key cells across
//!   every compiled rule, approximate, bounded by the sharded store
//!   cap; aggregate and unlabeled — cardinality is never per key)
//! - `dwara_webhook_events_total{kind,outcome}` counter (DW-044):
//!   webhook deliveries by event kind and outcome — `delivered` (2xx
//!   on some attempt), `failed` (retries exhausted / non-transient
//!   answer / budget spent), `dropped` (never tried: envelope over the
//!   byte cap, or delivery concurrency saturated). Both labels are
//!   closed sets, so cardinality is bounded by construction.
//! - `dwara_events_dropped_total` / `dwara_events_emitted_total`
//!   gauges (DW-044, scrape-time snapshots of the event bus's
//!   monotonic counters — gauges rather than counters so the emit path
//!   stays a plain atomic with zero registry coupling, the same model
//!   as the rate-limiter eviction gauge): events dropped at EMIT time
//!   (full queue / no deliverer) and events handed to the queue.
//! - `dwara_access_records_streamed_total{outcome}` counter (DW-121):
//!   access records counted per flushed-batch delivery outcome —
//!   `delivered` / `failed` (tried and not taken) / `dropped`
//!   (per-record byte cap) — a closed three-value label set, counted
//!   by the stream flusher, never the request path.
//! - `dwara_access_records_offered_total` /
//!   `dwara_access_records_dropped_total` gauges (DW-121, scrape-time
//!   snapshots of the record stream's monotonic counters, the same
//!   gauge-not-counter model as the event bus): records offered at
//!   request completion and records dropped at OFFER time (channel
//!   full / stream disabled).
//! - `dwara_quota_denied_total{consumer,budget}` counter (DW-033) —
//!   requests refused by a consumer request budget (429), by consumer
//!   (the quota-configured set, config-bounded) and binding budget
//!   (`daily`/`monthly`, a closed set). Distinct from
//!   `rate_limited_total`: budgets are not rates.
//! - `dwara_quota_used{consumer,budget}` / `dwara_quota_limit{consumer,
//!   budget}` gauges (DW-033, scrape-time snapshots of the state
//!   store's quota counters and the config caps — the same snapshot
//!   gauge model as the rate-limiter live-keys gauge).
//! - `dwara_admission_queued_total{outcome}` counter (DW-053) —
//!   admission queue outcomes by closed set `admitted` (a permit was
//!   acquired after queueing), `timeout` (the queue timeout expired;
//!   shed with 503), `queue_full` (the queue was at capacity; shed
//!   immediately with 503).
//! - `dwara_admission_queue_depth` gauge (DW-053) — current number of
//!   requests waiting in the admission queue for a concurrency permit.
//! - `dwara_license_status` gauge (DW-032) — 0 = no license (OSS mode),
//!   1 = valid, 2 = expired within grace period, 3 = expired past grace
//!   period (degraded to OSS). Set at startup and on reload; no labels.
//! - `dwara_dns_discovery_endpoints{upstream}` gauge (DW-042) — current
//!   resolved endpoint count for DNS-discovered upstreams. The `upstream`
//!   label is config-bounded; cardinality is never per resolved address.
//! - `dwara_dns_discovery_refresh_total{upstream}` counter (DW-042) — DNS
//!   discovery refresh attempts, by upstream (config-bounded label).
//! - `dwara_dns_discovery_refresh_failures_total{upstream}` counter
//!   (DW-042) — DNS discovery refresh failures, by upstream.
//! - `dwara_mcp_sessions_total{state}` counter (DW-087) — MCP session
//!   lifecycle transitions, by state (`initialized`, `closed`,
//!   `expired`). A closed three-value label set, so cardinality is
//!   three series total.
//! - `dwara_mcp_tool_calls_total{tool,status}` counter (DW-087) — MCP
//!   tool calls, by tool (config-bounded) and status (`success`,
//!   `error`, `denied`). No consumer label (cardinality rule).
//! - `dwara_mcp_tool_duration_seconds{tool}` histogram (DW-087) — MCP
//!   tool call duration (authz check through upstream response), by
//!   tool (config-bounded label).
//! - `dwara_a2a_requests_total{agent,outcome}` counter (DW-114) — A2A
//!   (agent-to-agent) requests, by agent (config-bounded) and outcome
//!   (`success`, `error`, `stubbed`). No consumer label (cardinality
//!   rule).
//! - `dwara_a2a_sessions_total{state}` counter (DW-114) — A2A session
//!   lifecycle transitions, by state (`initialized`, `closed`,
//!   `expired`). A closed three-value label set (mirrors MCP), so
//!   cardinality is three series total.
//! - `dwara_tls_pq_handshakes_total{result}` counter (DW-105) —
//!   post-quantum TLS handshake outcomes, by result (`success`,
//!   `fallback`, `disabled`). A closed three-value label set, so
//!   cardinality is three series total.
//! - `dwara_graphql_requests_total{route,outcome}` counter (DW-099) —
//!   GraphQL check outcomes, by route (config-bounded) and outcome
//!   (`allowed`, `depth_exceeded`, `complexity_exceeded`,
//!   `persisted_query_required` — a closed four-value set).
//! - `dwara_graphql_depth{route}` gauge (DW-099) — last query depth for
//!   a GraphQL-configured route (config-bounded label).
//! - `dwara_nano_service_requests_total{route,outcome}` counter (DW-106)
//!   — nano-service (WASM route handler) request outcomes, by route
//!   (config-bounded) and outcome (`success`, `error`, `timeout` — a
//!   closed three-value set). No consumer label (cardinality rule).
//! - `dwara_nano_service_duration_seconds{route}` histogram (DW-106) —
//!   nano-service `handle` call duration, by route (config-bounded
//!   label).
//! - `dwara_plugin_failures_total{name,reason}` counter (DW-157) —
//!   plugin dispatch failures, by plugin name (config-bounded) and
//!   reason (`crashed`, `disabled`, `not_loaded`, `not_registered`,
//!   `instantiate_failed`, `trap`, `body_too_large`,
//!   `response_stream_ended` — a closed set).
//!   One increment per request the plugin chain answered with an error
//!   (a plugin short-circuit via `send_http_response` is a plugin
//!   DECISION, not a failure, and is not counted; the access-log
//!   `plugin_short_circuit` flag marks those).
//! - `dwara_plugin_total{state}` gauge (DW-158) — plugins loaded per
//!   lifecycle state (`healthy`, `crashed`, `disabled`, `not_loaded`,
//!   `not_registered` — a closed five-value set over the generation's
//!   declared plugins, the same enumeration `GET /plugins` serves).
//!   Refreshed at publish time (every config generation); state changes
//!   that happen between publishes (a crash counted at request time)
//!   land on the next publish or are visible per-request in
//!   `dwara_plugin_failures_total`.
//! - `dwara_graphql_complexity{route}` gauge (DW-099) — last query
//!   complexity for a GraphQL-configured route (config-bounded label).
//! - `dwara_l4_connections_total{listener,protocol}` counter (DW-103) —
//!   L4 connections proxied, by listener (config-bounded) and protocol
//!   (`tcp`, `udp` — a closed two-value set). Counted once per accepted
//!   L4 connection (before the splice is established).
//! - `dwara_l4_connection_duration_seconds{listener}` histogram
//!   (DW-103) — L4 connection duration (splice establishment through
//!   close), by listener (config-bounded label).
//! - `dwara_ai_request_duration_seconds{provider,route}` histogram
//!   (AI-07) — AI request latency (request parse through response
//!   translation), by provider and route (both config-bounded). Observed
//!   once per completed AI request (success and provider/translation
//!   errors alike; client-side rejections never reach a provider).
//! - `dwara_ai_tokens_per_request{provider,model,kind}` histogram
//!   (AI-07) — distribution of provider-reported tokens per request, by
//!   provider (config-bounded), provider_model (config-bounded), and
//!   kind (the closed set `prompt` | `completion`). Provider-reported
//!   only; the gateway never estimates. No consumer label (cardinality).
//!
//! ## OTel GenAI semantic conventions (AI-07)
//!
//! AI request spans carry the OpenTelemetry GenAI semantic-convention
//! `gen_ai.*` attributes (<https://opentelemetry.io/docs/specs/semconv/gen-ai/>):
//! `gen_ai.system` (the provider kind), `gen_ai.request.model`
//! (the client alias), `gen_ai.response.model` (the provider model that
//! served), `gen_ai.request.{max_tokens,temperature,top_p}`, and the
//! provider-reported `gen_ai.usage.{prompt_tokens,completion_tokens,
//! total_tokens}`, `gen_ai.response.finish_reasons`, and
//! `gen_ai.response.id`. The helpers [`record_gen_ai_request_attrs`] and
//! [`record_gen_ai_response_attrs`] record them onto a span created with
//! every `gen_ai.*` field declared `Empty` (the span lives in
//! `dataplane::ai_proxy::serve_ai`).
//!
//! ## Error envelope (section 4.19)
//!
//! Every gateway-generated non-success body — including the reserved
//! `/healthz`/`/readyz` responses, which are aligned to the same shape —
//! is `{"error":{"code","message","request_id"}}`. Messages never leak
//! upstream internals: classification strings only ("upstream
//! unavailable", not hyper error text).
//!
//! ## Request IDs
//!
//! An inbound `X-Request-Id` is respected when it is printable ASCII of
//! at most 128 bytes (anything else is replaced — a hostile ID must not
//! smuggle control bytes into logs); otherwise a process-unique ID is
//! generated (wall-clock nanoseconds + per-process counter — cheaper
//! than a UUID, no extra dependency). The ID is echoed on the response
//! as `X-Request-Id` and available to every span and log line.
//!
//! ## Redaction rules (hard requirements)
//!
//! - Paths are logged WITHOUT query strings.
//! - No Authorization / Proxy-Authorization / Cookie / Set-Cookie /
//!   X-API-Key VALUES are ever emitted by this module or the tracing
//!   calls it fronts; field lists below are exhaustive.
//! - JWKS bodies, credentials, and keys are never logged.

use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use hyper::header::{HeaderMap, HeaderName, HeaderValue};
use prometheus::{
    Encoder, GaugeVec, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge,
    IntGaugeVec, Opts, Registry, TextEncoder,
};

/// Request header carrying the correlation ID (inbound respected,
/// outbound always set by the gateway).
pub const X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id");

/// Request header carrying the business correlation ID (DW-093):
/// inbound respected (falls back to `X-Request-Id`), outbound echoed
/// on the response so a client can tie a reply to its journey across
/// multiple gateway requests.
pub const X_CORRELATION_ID: HeaderName = HeaderName::from_static("x-correlation-id");

/// Extension inserted by the listener frontend (dwara-bin) naming the
/// listener that accepted the request; absent when `proxy::handle` is
/// driven directly (tests), in which case the label is "unknown".
#[derive(Clone, Debug)]
pub struct ListenerLabel(pub std::sync::Arc<str>);

/// Histogram buckets for `request_duration_seconds`: sub-millisecond
/// precision is noise at gateway scale; 10s covers every classified
/// timeout (default connect timeout is 5s).
const DURATION_BUCKETS: &[f64] = &[
    0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
];

/// Histogram buckets for `dwara_ai_first_token_seconds` (DW-077):
/// first-token latency is the metric streaming consumers feel, so the
/// low end is finer than DURATION_BUCKETS; 30s covers a cold provider
/// queue without a tail bucket that swallows real regressions.
const FIRST_TOKEN_BUCKETS: &[f64] = &[
    0.005, 0.01, 0.025, 0.05, 0.1, 0.2, 0.35, 0.5, 0.75, 1.0, 1.5, 2.5, 5.0, 10.0, 30.0,
];

/// Histogram buckets for `dwara_ai_tokens_per_request` (AI-07): token
/// counts span a few (tiny prompts) to hundreds of thousands (long
/// context); an exponential-ish scale keeps both ends observable without
/// a single bucket that swallows the long tail.
const TOKENS_PER_REQUEST_BUCKETS: &[f64] = &[
    1.0, 8.0, 32.0, 128.0, 256.0, 512.0, 1024.0, 2048.0, 4096.0, 8192.0, 16384.0, 32768.0, 65536.0,
    131072.0,
];

/// Prometheus status-class label ("2xx", "5xx", ...).
///
/// PERF: a `&'static str` per class, not a `format!` — this fires on
/// the per-request and per-attempt metric recording paths, and the
/// label text is one of ten fixed spellings.
pub fn status_class(status: u16) -> &'static str {
    match status / 100 {
        0 => "0xx",
        1 => "1xx",
        2 => "2xx",
        3 => "3xx",
        4 => "4xx",
        5 => "5xx",
        6 => "6xx",
        7 => "7xx",
        8 => "8xx",
        _ => "9xx",
    }
}

/// The accumulated access-log record for one request. The proxy fills
/// fields as the request progresses; `finish` stamps status/duration.
#[derive(Debug)]
pub struct AccessRecord {
    pub request_id: String,
    /// The correlation ID (DW-093): the inbound `X-Correlation-Id`
    /// when present and valid, else the `request_id` (the fallback
    /// so every request carries a non-empty correlation handle for
    /// the journey/funnel query). Stored in the analytics `raw`
    /// table; echoed on the response as `X-Correlation-Id`.
    pub correlation_id: String,
    pub method: String,
    /// Request path WITHOUT the query string (redaction).
    pub path: String,
    pub listener: String,
    /// Matched route name, or "unrouted" for 404s.
    pub route: String,
    /// Authenticated consumer, or "anonymous".
    pub consumer: String,
    pub upstream: Option<String>,
    pub endpoint: Option<String>,
    pub attempts: u32,
    pub rate_limited: bool,
    pub broken: bool,
    pub shed: bool,
    /// A plugin chain answered this request instead of the route's
    /// action (DW-157): a `send_http_response` short-circuit or a
    /// fail-closed plugin error (500 `plugin_unavailable` /
    /// `plugin_failed` / `plugin_body_too_large`).
    pub plugin_short_circuit: bool,
    pub status: u16,
    pub duration_ms: f64,
    /// Inbound request body bytes (PERF-12, #209): the declared
    /// `Content-Length` when present, 0 for streamed/chunked request
    /// bodies (counting streamed request bytes would require wrapping
    /// the upstream-forwarded body; the response-side count is the
    /// primary signal for capacity planning and top-talker reports).
    pub bytes_in: u64,
    /// Outbound response body bytes (PERF-12, #209): counted frame-by-
    /// frame as the response body streams to the client, WITHOUT
    /// buffering. An `Arc<AtomicU64>` so the `CountingBody` wrapper
    /// can increment it from `poll_frame` while the access record is
    /// held by the caller; the final value is read when the body
    /// completes and the deferred access log / analytics are emitted.
    pub bytes_out: std::sync::Arc<std::sync::atomic::AtomicU64>,
    /// Custom analytics dimensions (DW-043): config-declared
    /// header-sourced tags, captured at record creation. Analytics-
    /// only — deliberately NOT part of `emit_access`'s redacted
    /// field list.
    pub custom: Vec<(String, String)>,
}

impl AccessRecord {
    pub fn new(request_id: String, method: String, path: String, listener: String) -> Self {
        AccessRecord {
            request_id,
            correlation_id: String::new(),
            method,
            path,
            listener,
            route: "unrouted".to_string(),
            consumer: "anonymous".to_string(),
            upstream: None,
            endpoint: None,
            attempts: 0,
            rate_limited: false,
            broken: false,
            shed: false,
            plugin_short_circuit: false,
            status: 0,
            duration_ms: 0.0,
            bytes_in: 0,
            bytes_out: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)),
            custom: Vec::new(),
        }
    }
}

/// Emit one access-log event (one JSON line under the binary's
/// subscriber). The field list is exhaustive and redacted by
/// construction: no headers, no query string, no credentials.
/// `bytes_in`/`bytes_out` (PERF-12, #209): `bytes_in` is the declared
/// `Content-Length` (0 for chunked); `bytes_out` is the frame-counted
/// response body size (read from the atomic the `CountingBody`
/// wrapper updated as frames streamed to the client).
pub fn emit_access(rec: &AccessRecord) {
    let bytes_out = rec.bytes_out.load(std::sync::atomic::Ordering::Relaxed);
    tracing::info!(
        target: "dwara::access",
        request_id = %rec.request_id,
        method = %rec.method,
        path = %rec.path,
        status = rec.status,
        duration_ms = rec.duration_ms,
        route = %rec.route,
        consumer = %rec.consumer,
        upstream = rec.upstream.as_deref().unwrap_or(""),
        endpoint = rec.endpoint.as_deref().unwrap_or(""),
        attempts = rec.attempts,
        rate_limited = rec.rate_limited,
        broken = rec.broken,
        shed = rec.shed,
        plugin_short_circuit = rec.plugin_short_circuit,
        bytes_in = rec.bytes_in,
        bytes_out = bytes_out,
        "access"
    );
}

/// The unified error body (section 4.19): a JSON object an operator can
/// grep and a client can correlate. `code` is a stable machine token,
/// `message` a human string with no upstream internals, `request_id`
/// ties the response to the trace/access log.
pub fn envelope_body(code: &str, message: &str, request_id: &str) -> bytes::Bytes {
    // Manual JSON formatting avoids the serde_json::Value intermediate
    // allocation (the json! macro builds a Value tree, then to_string()
    // serializes it back — two heap allocations). The envelope shape is
    // fixed; we pre-allocate one String and JSON-escape each field.
    fn escape_into(out: &mut String, s: &str) {
        for c in s.chars() {
            match c {
                '"' => out.push_str("\\\""),
                '\\' => out.push_str("\\\\"),
                '\n' => out.push_str("\\n"),
                '\r' => out.push_str("\\r"),
                '\t' => out.push_str("\\t"),
                c if c < ' ' => out.push_str(&format!("\\u{:04x}", c as u32)),
                c => out.push(c),
            }
        }
    }
    let mut buf = String::with_capacity(64 + code.len() + message.len() + request_id.len());
    buf.push_str(r#"{"error":{"code":""#);
    escape_into(&mut buf, code);
    buf.push_str(r#"","message":""#);
    escape_into(&mut buf, message);
    buf.push_str(r#"","request_id":""#);
    escape_into(&mut buf, request_id);
    buf.push_str(r#""}}"#);
    bytes::Bytes::from(buf)
}

/// Validate an inbound `X-Request-Id` value: printable ASCII (0x20-0x7E),
/// at most 128 bytes. Anything else (control bytes, UTF-8 multibyte,
/// overlong) is rejected so hostile IDs cannot smuggle control
/// characters into logs.
pub fn valid_inbound_request_id(v: &[u8]) -> bool {
    !v.is_empty() && v.len() <= 128 && v.iter().all(|b| (0x20..=0x7e).contains(b))
}

/// Resolve the request's correlation ID: a valid inbound `X-Request-Id`
/// verbatim, else a generated one (`req-` + hex wall-clock nanoseconds +
/// hex process counter; unique per process, no dependency pulled in for
/// randomness — request IDs are correlation handles, not secrets).
pub fn resolve_request_id(headers: &HeaderMap) -> String {
    if let Some(v) = headers.get(&X_REQUEST_ID) {
        if valid_inbound_request_id(v.as_bytes()) {
            if let Ok(s) = v.to_str() {
                return s.to_string();
            }
        }
    }
    generate_request_id()
}

/// Per-process request-ID counter (the low bits of the wall clock are
/// already unique per nanosecond; the counter disambiguates coarse
/// clocks and multi-ID calls within one nanosecond).
static REQUEST_COUNTER: AtomicU64 = AtomicU64::new(0);

/// Public for testing the request-id generation contract.
pub fn generate_request_id() -> String {
    let n = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let c = REQUEST_COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("req-{:016x}-{:06x}", n, c & 0xff_ffff)
}

/// Cheap sampling PRNG: a Weyl-sequence over one atomic (add a large
/// odd constant, use the high bits of the product). Not cryptographic —
/// it only decides whether an access-log line is emitted.
struct SampleRng {
    state: AtomicU64,
}

impl SampleRng {
    fn new() -> Self {
        let seed = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos() as u64)
            .unwrap_or(0x9e37_79b9_7f4a_7c15);
        SampleRng {
            state: AtomicU64::new(seed | 1),
        }
    }

    fn next_f64(&self) -> f64 {
        let x = self
            .state
            .fetch_add(0x9e37_79b9_7f4a_7c15, Ordering::Relaxed);
        // Multiply-shift normalization into [0, 1).
        ((x.wrapping_mul(0x9e37_79b9_7f4a_7c15)) >> 11) as f64 / (1u64 << 53) as f64
    }
}

/// Per-dataplane observability state: the metric families plus the
/// access-log sampling knob. One instance per `DataPlane`, so parallel
/// tests never share a registry (duplicate-metric registration errors).
pub struct Observability {
    registry: Registry,
    requests_total: IntCounterVec,
    request_duration: HistogramVec,
    upstream_attempts_total: IntCounterVec,
    retries_total: IntCounterVec,
    /// DW-063: hedge copies sent (label: upstream). Counts every
    /// speculative duplicate the hedge timer fired, regardless of
    /// whether it won the race.
    hedge_sent_total: IntCounterVec,
    /// DW-062: mirror (shadow traffic) requests sent, by upstream.
    mirror_sent_total: IntCounterVec,
    rate_limited_total: IntCounterVec,
    shed_total: IntCounterVec,
    policy_dry_run_total: IntCounterVec,
    breaker_state: IntGaugeVec,
    endpoint_health: IntGaugeVec,
    fail_open_picks: IntGaugeVec,
    jwks_refresh_total: IntCounterVec,
    active_requests: IntGauge,
    config_generation: IntGauge,
    /// #132: rate-limiter eviction/live-key snapshot gauges — aggregate,
    /// unlabeled (config-bounded cardinality: never per key).
    rate_limiter_evictions: IntGauge,
    rate_limiter_live_keys: IntGauge,
    /// DW-044: webhook delivery outcomes by (kind, outcome) — both
    /// labels closed sets (the event kinds and the three outcomes), so
    /// cardinality is bounded by construction.
    webhook_events_total: IntCounterVec,
    /// DW-044: event-bus emit/drop snapshot gauges (see the module
    /// docs for the gauge-not-counter rationale).
    events_dropped: IntGauge,
    events_emitted: IntGauge,
    /// DW-121: access records counted per flushed-batch delivery
    /// outcome — a CLOSED label set (`delivered`, `failed`, `dropped`),
    /// so the family's cardinality is three series total. Counted by
    /// the flusher (never the request path).
    access_records_streamed_total: IntCounterVec,
    /// DW-121: access records offered to the stream channel at request
    /// completion; scrape-time snapshot of the stream's monotonic
    /// counter (the emit path bumps a plain atomic — the same
    /// gauge-not-counter model as the event bus).
    access_records_offered: IntGauge,
    /// DW-121: access records dropped at OFFER time (stream channel
    /// full or the stream disabled); scrape-time snapshot, same model
    /// as `dwara_events_dropped_total`.
    access_records_dropped: IntGauge,
    /// REL-10 (#221): access records dropped by sampled degradation
    /// (probabilistic pre-full drop). Scrape-time snapshot of the
    /// analytics store's counter.
    access_records_sampled_dropped: IntGauge,
    /// DW-039: WebSocket policy decisions — a CLOSED label set
    /// (`origin_denied`, `rate_closed`) over the route label space
    /// (config-bounded, the same cardinality class as
    /// `requests_total`).
    websocket_policy_total: IntCounterVec,
    /// DW-040: split-service dispatch decisions — the `service` and
    /// `upstream` labels are both config-declared names (bounded by
    /// the split configuration), so canary share is directly
    /// readable as `upstream` share within a `service`.
    split_picks_total: IntCounterVec,
    /// DW-040: sticky sessions minted (affinity cookies set on first
    /// response). Plain counter — no label space.
    sticky_sessions_total: IntCounter,
    /// DW-037: response-cache lookup outcomes — a CLOSED label set
    /// (`hit`, `stale`, `miss`, `bypass`), one increment per request on
    /// a cache-configured route, decided at lookup.
    cache_lookups_total: IntCounterVec,
    /// DW-037: response-cache store outcomes — CLOSED set (`stored`,
    /// `vetoed`, `over_cap`, `error`), one increment per cacheable
    /// fetch that reached the store stage.
    cache_stores_total: IntCounterVec,
    /// DW-037: responses served from a stored body after the upstream
    /// answered a conditional revalidation with 304 (the "unchanged"
    /// path; plain counter — no label space to close).
    cache_revalidated_total: IntCounter,
    /// DW-037: purge operations — CLOSED set (`route`, `all`).
    cache_purges_total: IntCounterVec,
    /// DW-037: approximate live cache entries (scrape-time snapshot
    /// walk of the store's `entry_count`, the same gauge model as the
    /// rate-limiter live-keys gauge).
    cache_entries: IntGauge,
    /// DW-038: cache misses that became the coalescing LEADER (their
    /// own upstream fetch ran while followers could attach). Plain
    /// counter — no label space to close.
    coalescing_leaders_total: IntCounter,
    /// DW-038: coalescing follower resolutions — a CLOSED label set
    /// (`served`, `fell_back_timeout`, `fell_back_unshared`,
    /// `fell_back_epoch`), so the family's cardinality is four series
    /// total. `served` = the leader's stored outcome replayed;
    /// `fell_back_timeout` = the wait bound expired; `fell_back_unshared`
    /// = the leader finished without a storable outcome (vetoed,
    /// non-200, error — the follower fetched on its own);
    /// `fell_back_epoch` = a purge/config change invalidated the
    /// generation mid-flight (checked first; the follower fetched on
    /// its own).
    coalescing_followers_total: IntCounterVec,
    /// DW-038: upstream calls avoided by coalescing (one per `served`
    /// follower — the direct answer to "is coalescing paying for
    /// itself").
    coalescing_saved_upstream_calls_total: IntCounter,
    /// DW-038: requests currently parked as coalescing followers.
    coalescing_waiters: IntGauge,
    /// DW-033: quota denials by (consumer, budget) — the consumer label
    /// space is the quota-configured consumer set (config-bounded), and
    /// `budget` is the closed set `daily`/`monthly`.
    quota_denied_total: IntCounterVec,
    /// DW-033: requests counted in each budget's CURRENT window
    /// (scrape-time snapshot of the state store's quota counters, the
    /// same gauge model as the rate-limiter live-keys gauge).
    quota_used: IntGaugeVec,
    /// DW-033: each budget's configured cap (scrape-time, same snapshot
    /// walk as `quota_used`; exported so `used / limit` alerting needs
    /// no config join).
    quota_limit: IntGaugeVec,
    /// DW-052: the SLO burn-rate collector (per-route windowed state
    /// plus the two exported families; values refresh at scrape time).
    slo: std::sync::Arc<SloState>,
    /// DW-053: admission queue outcomes — a CLOSED label set
    /// (`admitted`, `timeout`, `queue_full`), one increment per
    /// request that was queued (or rejected at the queue door).
    admission_queued_total: IntCounterVec,
    /// DW-053: current number of requests waiting in the admission
    /// queue for a concurrency permit. A gauge set from an atomic on
    /// the dataplane (the hot path stays free of registry coupling,
    /// the same model as the coalescing-waiters gauge).
    admission_queue_depth: IntGauge,
    /// DW-051: WAF-lite inspection outcomes — a CLOSED label set
    /// (`blocked`, `logged`, `passed`) over the route and filter label
    /// spaces (both config-bounded). `blocked` = a match was found and
    /// the request was rejected with 403; `logged` = a dry-run match
    /// (the request was allowed); `passed` = no match (counted once per
    /// inspected request, filter label `all`).
    waf_total: IntCounterVec,
    /// DW-090: anomaly-scoring inspection outcomes — a CLOSED label set
    /// (`blocked`, `logged`, `passed`) over the route label space
    /// (config-bounded). `blocked` = the score was at or above the
    /// threshold and the request was rejected with 403; `logged` = a
    /// dry-run match (the request was allowed); `passed` = the score was
    /// below the threshold (counted once per inspected request).
    anomaly_total: IntCounterVec,
    /// DW-099: GraphQL check outcomes — a CLOSED label set (`allowed`,
    /// `depth_exceeded`, `complexity_exceeded`,
    /// `persisted_query_required`) over the route label space
    /// (config-bounded). One increment per GraphQL-configured request.
    graphql_requests_total: IntCounterVec,
    /// DW-099: last query depth for a GraphQL-configured route (gauge;
    /// config-bounded route label).
    graphql_depth: IntGaugeVec,
    /// DW-099: last query complexity for a GraphQL-configured route
    /// (gauge; config-bounded route label).
    graphql_complexity: IntGaugeVec,
    /// DW-032: license status gauge — 0 = no license (OSS), 1 = valid,
    /// 2 = expired within grace, 3 = expired past grace (degraded to
    /// OSS). Set once at startup and on every reload; no labels (closed
    /// four-value set, so cardinality is one series).
    license_status: IntGauge,
    /// DW-042: DNS discovery endpoint count — a gauge of the current
    /// resolved endpoint count per upstream (config-bounded `upstream`
    /// label; never per resolved address). Scrape-time snapshot setter
    /// driven by the discovery task.
    dns_discovery_endpoints: IntGaugeVec,
    /// DW-042: DNS discovery refresh attempts — a counter of resolution
    /// cycles per upstream (config-bounded `upstream` label).
    dns_discovery_refresh_total: IntCounterVec,
    /// DW-042: DNS discovery refresh failures — a counter of failed
    /// resolution cycles per upstream (config-bounded `upstream` label).
    dns_discovery_refresh_failures_total: IntCounterVec,
    /// DW-054: this instance's current convergence generation, by
    /// instance id. The instance label is config-bounded (one series
    /// per process); set by the convergence coordinator after every
    /// local or remote publish.
    config_convergence_generation: IntGaugeVec,
    /// DW-054: total instances the convergence backend currently
    /// reports (the cluster view the drift check reads). Aggregate and
    /// unlabeled -- cardinality is one series.
    config_convergence_instances: IntGauge,
    /// DW-054: drift flag -- 1 when one or more instances serve a
    /// different config hash than the majority, 0 when converged.
    /// Aggregate and unlabeled -- cardinality is one series.
    config_convergence_drift: IntGauge,
    /// DW-054: convergence refresh attempts (polls of the backend).
    /// Aggregate counter -- cardinality is one series.
    config_convergence_refresh_total: IntCounter,
    /// DW-054: convergence refresh failures (backend unreachable or
    /// returned malformed data). Aggregate counter -- cardinality is
    /// one series.
    config_convergence_refresh_failures_total: IntCounter,
    /// DW-075: AI route requests, by provider/route/outcome. The
    /// provider label is the `ai.providers[].name` (config-bounded);
    /// outcomes: success, provider_error, transport_error,
    /// translation_error. DW-076 adds the canary `version` label
    /// (config-bounded; "default" for non-canary aliases).
    /// Client-side rejections (malformed body, unknown model) are NOT
    /// counted here — they never reach a provider and are already
    /// visible in requests_total's status classes.
    ai_requests_total: IntCounterVec,
    /// DW-075: provider-reported token usage, by provider and
    /// direction (prompt|completion), with the canary `version` label
    /// (DW-076). Provider-reported only (the locked M4 decision: the
    /// gateway never estimates).
    ai_tokens_total: IntCounterVec,
    /// DW-079: total priced AI spend in micro-USD, by provider and
    /// provider_model (config-bounded labels — no consumer label,
    /// cardinality rule). Priced through the dataplane's compiled
    /// pricing table; 0 for unknown models.
    ai_cost_micros_total: IntCounterVec,
    /// DW-AI-03: count of requests served by a model with no pricing
    /// entry, by provider_model. Incremented only under the `alert`
    /// unknown-model policy (the `allow` policy is silent; the
    /// `fail_closed` policy rejects before recording). The model label
    /// is the provider model identifier (config-bounded).
    ai_unknown_pricing_total: IntCounterVec,
    /// DW-078: AI budget denials, by kind (tokens | cost). No consumer
    /// label (cardinality); the consumer is in the access log line.
    ai_budget_denied_total: IntCounterVec,
    /// DW-084: AI governance denials, by reason (config-bounded label —
    /// the denial reason string, not the consumer or model). No
    /// consumer/model labels (cardinality rule); the consumer and
    /// model are in the governance audit table and the access log.
    ai_governance_denied_total: IntCounterVec,
    /// DW-082: AI guardrail blocks, by rule name and kind
    /// (config-bounded labels — the rule name and the kind/phase
    /// string). No consumer label (cardinality); the consumer is in
    /// the access log.
    ai_guardrail_blocked_total: IntCounterVec,
    /// DW-083: AI semantic cache hits, by model alias
    /// (config-bounded label). No consumer label (cardinality); the
    /// consumer is in the access log.
    ai_semantic_cache_hits_total: IntCounterVec,
    /// DW-083: AI semantic cache misses, by model alias
    /// (config-bounded label). No consumer label (cardinality); the
    /// consumer is in the access log.
    ai_semantic_cache_misses_total: IntCounterVec,
    /// DW-085: AI routing-policy FallbackChain escalations (complex
    /// prompt -> expensive model), by policy name (config-bounded
    /// label). No consumer label (cardinality); the consumer is in
    /// the access log.
    ai_routing_policy_escalations_total: IntCounterVec,
    /// DW-085: AI routing-policy FallbackChain cheap-model
    /// selections (simple prompt -> cheap model), by policy name
    /// (config-bounded label). No consumer label (cardinality); the
    /// consumer is in the access log.
    ai_routing_policy_cheap_total: IntCounterVec,
    /// DW-085: AI routing-policy LatencyCost selections, by policy
    /// name (config-bounded label). No consumer label (cardinality);
    /// the consumer is in the access log.
    ai_routing_policy_latency_cost_selections_total: IntCounterVec,
    /// DW-086: AI experiment (A/B test) variant selections, by
    /// experiment and variant (config-bounded labels). No consumer
    /// label (cardinality); the consumer is in the access log and
    /// the analytics assignment table.
    ai_experiment_variant_selections_total: IntCounterVec,
    /// DW-077: forwarded streaming chunks (SSE frames), by provider.
    /// One per client-visible delta chunk — the per-chunk
    /// observability of the streaming path.
    ai_stream_chunks_total: IntCounterVec,
    /// DW-077: time to the FIRST forwarded chunk of a stream, by
    /// provider (request send to first delta), in seconds.
    ai_first_token_seconds: HistogramVec,
    /// DW-077: total streaming duration (first request byte to stream
    /// end), by provider, in seconds.
    ai_stream_duration_seconds: HistogramVec,
    /// AI-07: AI request latency (request parse through response
    /// translation), by provider (config-bounded) and route
    /// (config-bounded), in seconds. Observed once per completed AI
    /// request (success and provider/translation errors alike; client-
    /// side rejections never reach a provider and are not observed).
    ai_request_duration_seconds: HistogramVec,
    /// AI-07: distribution of provider-reported tokens per request, by
    /// provider (config-bounded), provider_model (config-bounded), and
    /// kind (the CLOSED set `prompt` | `completion`). Observed once per
    /// request that reported usage (provider-reported only; the gateway
    /// never estimates). No consumer label (cardinality rule).
    ai_tokens_per_request: HistogramVec,
    /// DW-087: MCP session lifecycle transitions, by state
    /// (`initialized`, `closed`, `expired`). A CLOSED three-value label
    /// set, so the family's cardinality is three series total.
    mcp_sessions_total: IntCounterVec,
    /// DW-087: MCP tool calls, by tool and status. `tool` is the tool
    /// name (config-bounded); `status` is the CLOSED set `success`,
    /// `error`, `denied`. No consumer label (cardinality rule); the
    /// consumer is in the access log and the analytics table.
    mcp_tool_calls_total: IntCounterVec,
    /// DW-087: MCP tool call duration (authz check through upstream
    /// response), by tool (config-bounded label), in seconds.
    mcp_tool_duration_seconds: HistogramVec,
    /// DW-114: A2A (agent-to-agent) requests, by agent and outcome.
    /// `agent` is the config-declared agent name (config-bounded);
    /// `outcome` is the CLOSED set `success`, `error`, `stubbed`. No
    /// consumer label (cardinality rule); the consumer is in the
    /// access log.
    a2a_requests_total: IntCounterVec,
    /// DW-114: A2A session lifecycle transitions, by state
    /// (`initialized`, `closed`, `expired`). A CLOSED three-value
    /// label set (mirrors MCP), so the family's cardinality is three
    /// series total.
    a2a_sessions_total: IntCounterVec,
    /// DW-106: nano-service (WASM route handler) request outcomes, by
    /// route (config-bounded) and outcome (the CLOSED set `success`,
    /// `error`, `timeout`). No consumer label (cardinality rule); the
    /// consumer is in the access log.
    nano_service_requests_total: IntCounterVec,
    /// DW-106: nano-service `handle` call duration, by route
    /// (config-bounded label). Observed once per request (success,
    /// error, and timeout alike).
    nano_service_duration_seconds: HistogramVec,
    /// DW-157: plugin dispatch failures, by plugin `name`
    /// (config-bounded) and `reason` (the CLOSED set `crashed`,
    /// `disabled`, `not_loaded`, `not_registered`,
    /// `instantiate_failed`, `trap`, `body_too_large`,
    /// `response_stream_ended`). One increment
    /// per request the plugin chain answered with an error; a plugin's
    /// deliberate `send_http_response` short-circuit is a decision, not
    /// a failure, and is not counted.
    plugin_failures_total: IntCounterVec,
    plugin_callouts_total: IntCounterVec,
    /// DW-158: plugins per lifecycle state — the CLOSED label set
    /// `healthy`, `crashed`, `disabled`, `not_loaded`,
    /// `not_registered` (an unregistered native filter or an
    /// unresolved registry source is unusable exactly like a crashed
    /// plugin — routes referencing it fail closed). One series per
    /// state over the generation's declared plugins (the same
    /// enumeration `GET /plugins` serves), refreshed at publish time.
    plugin_total: IntGaugeVec,
    /// DW-105: post-quantum TLS handshake outcomes — a CLOSED label set
    /// (`success`, `fallback`, `disabled`), one increment per PQ-
    /// configured handshake. `success` = the X25519+ML-KEM hybrid kx
    /// group was negotiated; `fallback` = the client did not support
    /// the hybrid group and a classical group was negotiated;
    /// `disabled` = PQ was off or the experimental API was inert. No
    /// labels beyond `result` (a closed three-value set, so the
    /// family's cardinality is three series total).
    tls_pq_handshakes_total: IntCounterVec,
    /// DW-089: current adaptive rate-limit factor, by policy
    /// (config-bounded label). A factor of 1.0 = the configured rate;
    /// < 1.0 tightens, > 1.0 relaxes. Set on every `record_outcome`.
    rate_limiter_adaptive_factor: GaugeVec,
    /// DW-089: upstream origin signals received, by policy and signal
    /// (both config-bounded labels; `signal` is the closed set
    /// `retry_after`). Counted when the controller honors the signal.
    rate_limiter_origin_signal_total: IntCounterVec,
    /// DW-089: adaptive factor decreased (tightened) for a policy
    /// (config-bounded label). Counted once per stressed update that
    /// moved the factor down.
    rate_limiter_adaptive_tightened_total: IntCounterVec,
    /// DW-089: adaptive factor increased (relaxed) for a policy
    /// (config-bounded label). Counted once per healthy update that
    /// moved the factor up.
    rate_limiter_adaptive_relaxed_total: IntCounterVec,
    /// DW-091: canary promotions, by group (config-bounded label:
    /// service name or AI alias). Counted once per promote action the
    /// auto-canary controller applied.
    canary_promotions_total: IntCounterVec,
    /// DW-091: canary rollbacks, by group (config-bounded label).
    /// Counted once per rollback action (including severe-regression
    /// immediate rollbacks to 0).
    canary_rollbacks_total: IntCounterVec,
    /// DW-091: current canary weight, by group (config-bounded
    /// label). Set on every applied adjustment.
    canary_weight: GaugeVec,
    /// DW-107: sidecar connections proxied, by mode and direction.
    /// `mode` is the CLOSED set `sidecar`; `direction` is the CLOSED
    /// set `inbound`, `outbound`. Counted once per connection the
    /// sidecar proxies (inbound: terminate mTLS + forward to local app;
    /// outbound: apply policies + wrap mTLS + dial remote sidecar).
    mesh_connections_total: IntCounterVec,
    /// DW-107: SPIFFE SVID refresh attempts, by result. `result` is
    /// the CLOSED set `success`, `error`. Counted once per Workload
    /// API refresh attempt (the client refreshes before expiry; an
    /// error keeps serving the previous SVID until it expires).
    spiffe_svid_refresh_total: IntCounterVec,
    /// DW-107: seconds until the current SVID expires (gauge). Set on
    /// every successful refresh; counts down to 0 as the SVID nears
    /// expiry. The refresh loop uses this to schedule the next fetch
    /// (at half the remaining lifetime by default).
    spiffe_svid_expiry_seconds: IntGauge,
    /// DW-103: L4 connections proxied, by listener (config-bounded)
    /// and protocol (the CLOSED set `tcp`, `udp`). Counted once per
    /// accepted L4 connection (before the splice is established).
    l4_connections_total: IntCounterVec,
    /// DW-103: L4 connection duration (splice establishment through
    /// close), by listener (config-bounded label), in seconds.
    /// Observed once per connection (forwarded and closed alike).
    l4_connection_duration_seconds: HistogramVec,
    /// Access-log sample rate [0.0, 1.0] as raw bits (f64 does not fit
    /// an atomic portably); read via [`Self::access_sample`].
    access_sample_bits: AtomicU64,
    rng: SampleRng,
}

impl Default for Observability {
    fn default() -> Self {
        Self::from_env()
    }
}

impl Observability {
    /// Build with `DWARA_ACCESS_LOG_SAMPLE` applied (default 1.0; values
    /// outside [0.0, 1.0] fall back to 1.0 — a malformed knob must never
    /// silence logs).
    pub fn from_env() -> Self {
        let sample = std::env::var("DWARA_ACCESS_LOG_SAMPLE")
            .ok()
            .and_then(|v| v.parse::<f64>().ok())
            .filter(|f| (0.0..=1.0).contains(f))
            .unwrap_or(1.0);
        let obs = Self::new();
        obs.set_access_sample(sample);
        obs
    }

    pub fn new() -> Self {
        let registry = Registry::new();
        let requests_total = IntCounterVec::new(
            Opts::new(
                "requests_total",
                "Requests handled by the gateway, by route/listener/status class.",
            ),
            &["route", "listener", "status_class"],
        )
        .expect("valid metric definition");
        let request_duration = HistogramVec::new(
            HistogramOpts::new(
                "request_duration_seconds",
                "Gateway request latency (route resolution to response headers).",
            )
            .buckets(DURATION_BUCKETS.to_vec()),
            &["route"],
        )
        .expect("valid metric definition");
        let upstream_attempts_total = IntCounterVec::new(
            Opts::new(
                "upstream_attempts_total",
                "Upstream send attempts, by upstream/endpoint/status class.",
            ),
            &["upstream", "endpoint", "status_class"],
        )
        .expect("valid metric definition");
        let retries_total = IntCounterVec::new(
            Opts::new("retries_total", "Retried upstream attempts, by upstream."),
            &["upstream"],
        )
        .expect("valid metric definition");
        let hedge_sent_total = IntCounterVec::new(
            Opts::new(
                "dwara_hedge_sent_total",
                "Speculative hedge copies sent (DW-063), by upstream. Counts \
                 every duplicate the hedge timer fired, regardless of whether \
                 it won the race.",
            ),
            &["upstream"],
        )
        .expect("valid metric definition");
        let mirror_sent_total = IntCounterVec::new(
            Opts::new(
                "dwara_mirror_sent_total",
                "Shadow traffic mirror requests sent (DW-062), by upstream. \
                 Counts every fire-and-forget mirror copy dispatched to a \
                 mirror upstream.",
            ),
            &["upstream"],
        )
        .expect("valid metric definition");
        let rate_limited_total = IntCounterVec::new(
            Opts::new(
                "rate_limited_total",
                "Requests denied by rate limiting, by route.",
            ),
            &["route"],
        )
        .expect("valid metric definition");
        let shed_total = IntCounterVec::new(
            Opts::new(
                "shed_total",
                "Requests shed by the gateway concurrency cap, by priority class.",
            ),
            &["priority"],
        )
        .expect("valid metric definition");
        let policy_dry_run_total = IntCounterVec::new(
            Opts::new(
                "dwara_policy_dry_run_total",
                "Requests a dry-run (monitor) policy would have rejected, by phase and \
                 route; the request was allowed (DW-041). Phases: route_limits, authz, \
                 rate_limit, load_shed.",
            ),
            &["phase", "route"],
        )
        .expect("valid metric definition");
        let breaker_state = IntGaugeVec::new(
            Opts::new(
                "breaker_state",
                "Upstream circuit breaker state: 0 closed, 1 open, 2 half-open.",
            ),
            &["upstream"],
        )
        .expect("valid metric definition");
        let endpoint_health = IntGaugeVec::new(
            Opts::new(
                "endpoint_health",
                "Endpoint availability under passive/active health: 1 available, 0 ejected.",
            ),
            &["upstream", "endpoint"],
        )
        .expect("valid metric definition");
        let fail_open_picks = IntGaugeVec::new(
            Opts::new(
                "upstream_fail_open_picks",
                "Balancer picks that fail-opened over an all-ejected pool (scrape-time snapshot).",
            ),
            &["upstream"],
        )
        .expect("valid metric definition");
        let jwks_refresh_total = IntCounterVec::new(
            Opts::new(
                "jwks_refresh_total",
                "JWKS fetch attempts (stale and rotation-triggered), by provider.",
            ),
            &["provider"],
        )
        .expect("valid metric definition");
        let active_requests = IntGauge::new(
            "active_requests",
            "Requests currently between handle() entry and response-header completion.",
        )
        .expect("valid metric definition");
        let config_generation = IntGauge::new(
            "config_generation",
            "Currently published configuration generation number.",
        )
        .expect("valid metric definition");
        let rate_limiter_evictions = IntGauge::new(
            "dwara_rate_limiter_evictions_total",
            "Rate-limiter per-key cells dropped by eviction sweeps, aggregated over \
             every compiled rule (scrape-time snapshot of the engine's monotonic \
             counter; resets when a reload rebuilds the engine).",
        )
        .expect("valid metric definition");
        let rate_limiter_live_keys = IntGauge::new(
            "dwara_rate_limiter_live_keys",
            "Live per-key rate-limiter cells across every compiled rule \
             (approximate under concurrent checks; bounded by the sharded \
             store cap per window).",
        )
        .expect("valid metric definition");
        let webhook_events_total = IntCounterVec::new(
            Opts::new(
                "dwara_webhook_events_total",
                "Webhook deliveries by event kind and outcome (DW-044). \
                 outcome: delivered (2xx on some attempt), failed (retries \
                 exhausted, non-transient answer, or budget spent), dropped \
                 (never tried: envelope over the byte cap or delivery \
                 concurrency saturated).",
            ),
            &["kind", "outcome"],
        )
        .expect("valid metric definition");
        let events_dropped = IntGauge::new(
            "dwara_events_dropped_total",
            "Events dropped at EMIT time (event queue full or no deliverer \
             draining); scrape-time snapshot of the event bus's monotonic \
             counter (DW-044).",
        )
        .expect("valid metric definition");
        let events_emitted = IntGauge::new(
            "dwara_events_emitted_total",
            "Events handed to the event queue since process start; \
             scrape-time snapshot of the event bus's monotonic counter \
             (DW-044).",
        )
        .expect("valid metric definition");
        let access_records_streamed_total = IntCounterVec::new(
            Opts::new(
                "dwara_access_records_streamed_total",
                "Access records counted per flushed-batch delivery outcome \
                 (DW-121). outcome: delivered (the batch's sink accepted it, \
                 2xx on some attempt), failed (the batch was tried and not \
                 taken: retries exhausted, non-transient answer, or budget \
                 spent), dropped (never delivered: over the per-record byte \
                 cap, or flushed after the stream was disabled with a \
                 queued tail). Emit-time channel drops are the separate \
                 dwara_access_records_dropped_total gauge.",
            ),
            &["outcome"],
        )
        .expect("valid metric definition");
        let access_records_offered = IntGauge::new(
            "dwara_access_records_offered_total",
            "Access records offered to the stream channel at request \
             completion since process start; scrape-time snapshot of the \
             stream's monotonic counter (DW-121).",
        )
        .expect("valid metric definition");
        let access_records_dropped = IntGauge::new(
            "dwara_access_records_dropped_total",
            "Access records dropped at OFFER time (stream channel full, or \
             the stream disabled by the current generation); scrape-time \
             snapshot of the stream's monotonic counter (DW-121) — the \
             never-block posture's honest loss counter.",
        )
        .expect("valid metric definition");
        let access_records_sampled_dropped = IntGauge::new(
            "dwara_access_records_sampled_dropped_total",
            "Access records dropped by sampled degradation (REL-10, #221): \
             records probabilistically dropped when the analytics channel \
             fill ratio exceeded the high watermark, before the channel \
             was full. Scrape-time snapshot of the analytics store's \
             monotonic counter — distinct from \
             dwara_access_records_dropped_total (channel was full).",
        )
        .expect("valid metric definition");
        let split_picks_total = IntCounterVec::new(
            Opts::new(
                "dwara_split_picks_total",
                "Service-split dispatch decisions (DW-040): one per request \
                 dispatched through a weighted split, by service and the \
                 target upstream chosen. The canary share is the upstream's \
                 share of the service's total.",
            ),
            &["service", "upstream"],
        )
        .expect("valid metric definition");
        let sticky_sessions_total = IntCounter::new(
            "dwara_sticky_sessions_total",
            "Sticky sessions minted (DW-040): affinity cookies set on a first \
             response.",
        )
        .expect("valid metric definition");
        let websocket_policy_total = IntCounterVec::new(
            Opts::new(
                "dwara_websocket_policy_total",
                "WebSocket policy decisions (DW-039), by route. outcome: \
                 origin_denied (an upgrade handshake rejected by the route's \
                 origin allowlist, before any upstream contact), rate_closed \
                 (an established tunnel closed by the frame-rate policer, \
                 close code 1008; counted when the tunnel task completes, so \
                 a process exit mid-tunnel can undercount).",
            ),
            &["route", "outcome"],
        )
        .expect("valid metric definition");
        let cache_lookups_total = IntCounterVec::new(
            Opts::new(
                "dwara_cache_lookups_total",
                "Response-cache lookups by outcome (DW-037), one per request on a \
                 cache-configured route. outcome: hit (fresh entry served), stale \
                 (expired entry served inside the stale-while-revalidate window), \
                 miss (no usable entry; the request went upstream), bypass (the \
                 request shape is not cacheable).",
            ),
            &["outcome"],
        )
        .expect("valid metric definition");
        let cache_stores_total = IntCounterVec::new(
            Opts::new(
                "dwara_cache_stores_total",
                "Response-cache store attempts by outcome (DW-037). outcome: \
                 stored, vetoed (a response-side rule forbade storage), over_cap \
                 (body larger than the route's max_body_bytes; passed through \
                 unbuffered), error (the backing store failed).",
            ),
            &["outcome"],
        )
        .expect("valid metric definition");
        let cache_revalidated_total = IntCounter::new(
            "dwara_cache_revalidated_total",
            "Responses served from the stored body after the upstream confirmed \
             the validator with 304 (DW-037).",
        )
        .expect("valid metric definition");
        let cache_purges_total = IntCounterVec::new(
            Opts::new(
                "dwara_cache_purges_total",
                "Cache purge operations by scope (DW-037, DP-04): route \
                 (one named route), all (every route), tag (Cache-Tags \
                 purge), url (exact URL purge), or url_prefix (prefix \
                 URL purge).",
            ),
            &["scope"],
        )
        .expect("valid metric definition");
        let cache_entries = IntGauge::new(
            "dwara_cache_entries",
            "Approximate live response-cache entries (DW-037); scrape-time \
             snapshot of the backing store's entry count.",
        )
        .expect("valid metric definition");
        let coalescing_leaders_total = IntCounter::new(
            "dwara_coalescing_leaders_total",
            "Cache misses that became the request-coalescing leader (DW-038): \
             their upstream fetch ran while later identical misses waited.",
        )
        .expect("valid metric definition");
        let coalescing_followers_total = IntCounterVec::new(
            Opts::new(
                "dwara_coalescing_followers_total",
                "Request-coalescing follower resolutions by outcome (DW-038). \
                 outcome: served (the leader's stored outcome replayed, one \
                 upstream call saved), fell_back_timeout (the wait bound \
                 expired), fell_back_unshared (the leader finished without a \
                 storable outcome; the follower fetched on its own), \
                 fell_back_epoch (a purge/config change invalidated the \
                 generation mid-flight; the follower fetched on its own).",
            ),
            &["outcome"],
        )
        .expect("valid metric definition");
        let coalescing_saved_upstream_calls_total = IntCounter::new(
            "dwara_coalescing_saved_upstream_calls_total",
            "Upstream calls avoided by request coalescing (DW-038); one per \
             served follower.",
        )
        .expect("valid metric definition");
        let coalescing_waiters = IntGauge::new(
            "dwara_coalescing_waiters",
            "Requests currently parked as request-coalescing followers \
             (DW-038).",
        )
        .expect("valid metric definition");
        let quota_denied_total = IntCounterVec::new(
            Opts::new(
                "dwara_quota_denied_total",
                "Requests refused by a consumer request budget (429), by consumer \
                 and budget (DW-033). budget is the binding wall: daily or monthly.",
            ),
            &["consumer", "budget"],
        )
        .expect("valid metric definition");
        let quota_used = IntGaugeVec::new(
            Opts::new(
                "dwara_quota_used",
                "Requests counted in a consumer budget's current UTC window \
                 (scrape-time snapshot of the state store's quota counters; \
                 DW-033).",
            ),
            &["consumer", "budget"],
        )
        .expect("valid metric definition");
        let quota_limit = IntGaugeVec::new(
            Opts::new(
                "dwara_quota_limit",
                "A consumer budget's configured cap for the current window \
                 (scrape-time; DW-033) — pairs with dwara_quota_used for \
                 used/limit alerting without a config join.",
            ),
            &["consumer", "budget"],
        )
        .expect("valid metric definition");
        // DW-052: created once, shared between the recorder methods and
        // the registered collector handle (the window state must be the
        // SAME object the completion path writes and the scrape reads).
        let slo = std::sync::Arc::new(SloState::new());
        let admission_queued_total = IntCounterVec::new(
            Opts::new(
                "dwara_admission_queued_total",
                "Admission queue outcomes (DW-053). outcome: admitted (a \
                 permit was acquired after queueing), timeout (the queue \
                 timeout expired before a permit was available; the request \
                 was shed with 503), queue_full (the queue was at capacity \
                 and the request was shed immediately with 503).",
            ),
            &["outcome"],
        )
        .expect("valid metric definition");
        let admission_queue_depth = IntGauge::new(
            "dwara_admission_queue_depth",
            "Current number of requests waiting in the admission queue for a \
             concurrency permit (DW-053).",
        )
        .expect("valid metric definition");
        let waf_total = IntCounterVec::new(
            Opts::new(
                "dwara_waf_total",
                "WAF-lite inspection outcomes (DW-051), by route and filter. \
                 outcome: blocked (a match was found and the request was \
                 rejected with 403), logged (a dry-run match; the request \
                 was allowed), passed (no match; counted once per inspected \
                 request with filter label 'all').",
            ),
            &["route", "filter", "outcome"],
        )
        .expect("valid metric definition");
        let anomaly_total = IntCounterVec::new(
            Opts::new(
                "dwara_anomaly_total",
                "Anomaly-scoring inspection outcomes (DW-090), by route. \
                 outcome: blocked (the score was at or above the threshold \
                 and the request was rejected with 403), logged (a dry-run \
                 match; the request was allowed), passed (the score was \
                 below the threshold; counted once per inspected request).",
            ),
            &["route", "outcome"],
        )
        .expect("valid metric definition");
        let graphql_requests_total = IntCounterVec::new(
            Opts::new(
                "dwara_graphql_requests_total",
                "GraphQL check outcomes (DW-099), by route and outcome. \
                 outcome: allowed (the query passed all checks), \
                 depth_exceeded (the query depth exceeded the configured \
                 limit), complexity_exceeded (the query complexity exceeded \
                 the configured limit), persisted_query_required (the query \
                 hash was not in the persisted-query store).",
            ),
            &["route", "outcome"],
        )
        .expect("valid metric definition");
        let graphql_depth = IntGaugeVec::new(
            Opts::new(
                "dwara_graphql_depth",
                "Last GraphQL query depth (DW-099), by route. The maximum \
                 brace-nesting level of the most recently checked query.",
            ),
            &["route"],
        )
        .expect("valid metric definition");
        let graphql_complexity = IntGaugeVec::new(
            Opts::new(
                "dwara_graphql_complexity",
                "Last GraphQL query complexity (DW-099), by route. The sum \
                 of per-field costs of the most recently checked query.",
            ),
            &["route"],
        )
        .expect("valid metric definition");
        let license_status = IntGauge::new(
            "dwara_license_status",
            "License status (DW-032): 0 = no license (OSS mode), 1 = valid, \
             2 = expired within grace period, 3 = expired past grace period \
             (degraded to OSS).",
        )
        .expect("valid metric definition");
        let dns_discovery_endpoints = IntGaugeVec::new(
            Opts::new(
                "dwara_dns_discovery_endpoints",
                "Current resolved endpoint count for DNS-discovered upstreams \
                 (DW-042). The upstream label is config-bounded; cardinality is \
                 never per resolved address.",
            ),
            &["upstream"],
        )
        .expect("valid metric definition");
        let dns_discovery_refresh_total = IntCounterVec::new(
            Opts::new(
                "dwara_dns_discovery_refresh_total",
                "DNS discovery refresh attempts (DW-042), by upstream.",
            ),
            &["upstream"],
        )
        .expect("valid metric definition");
        let dns_discovery_refresh_failures_total = IntCounterVec::new(
            Opts::new(
                "dwara_dns_discovery_refresh_failures_total",
                "DNS discovery refresh failures (DW-042), by upstream.",
            ),
            &["upstream"],
        )
        .expect("valid metric definition");
        let config_convergence_generation = IntGaugeVec::new(
            Opts::new(
                "dwara_config_convergence_generation",
                "This instance's current config convergence generation \
                 (DW-054), by instance id. The instance label is \
                 config-bounded (one series per process).",
            ),
            &["instance"],
        )
        .expect("valid metric definition");
        let config_convergence_instances = IntGauge::new(
            "dwara_config_convergence_instances",
            "Total gateway instances the convergence backend currently \
             reports (DW-054); the cluster view the drift check reads.",
        )
        .expect("valid metric definition");
        let config_convergence_drift = IntGauge::new(
            "dwara_config_convergence_drift",
            "Config convergence drift flag (DW-054): 1 when one or more \
             instances serve a different config hash than the majority, 0 \
             when the cluster is converged.",
        )
        .expect("valid metric definition");
        let config_convergence_refresh_total = IntCounter::new(
            "dwara_config_convergence_refresh_total",
            "Config convergence refresh attempts (DW-054): polls of the \
             convergence backend for remote generations.",
        )
        .expect("valid metric definition");
        let config_convergence_refresh_failures_total = IntCounter::new(
            "dwara_config_convergence_refresh_failures_total",
            "Config convergence refresh failures (DW-054): backend \
             unreachable or returned malformed data.",
        )
        .expect("valid metric definition");
        let ai_requests_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_requests_total",
                "AI route requests (DW-075), by provider/route/outcome. Outcomes: \
                 success, provider_error, transport_error, translation_error. \
                 The version label is the canary version that served (DW-076) \
                 or default. Client-side rejections never reach a provider and \
                 are not counted here.",
            ),
            &["provider", "route", "outcome", "version"],
        )
        .expect("valid metric definition");
        let ai_budget_denied_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_budget_denied_total",
                "AI token/cost budget denials (DW-078), by kind (tokens | \
                 cost). Consumer identity is in the access log, not a \
                 label.",
            ),
            &["kind"],
        )
        .expect("valid metric definition");
        let ai_governance_denied_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_governance_denied_total",
                "AI model governance denials (DW-084), by reason. The reason \
                 is a config-bounded label (the denial cause string), not the \
                 consumer or model — cardinality rule. Consumer and model are \
                 in the governance audit table and the access log.",
            ),
            &["reason"],
        )
        .expect("valid metric definition");
        let ai_guardrail_blocked_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_guardrail_blocked_total",
                "AI guardrail blocks (DW-082), by rule name and kind. The rule \
                 name is a config-bounded label (the rule's declared name), and \
                 the kind labels the phase/kind (prompt, response, schema, \
                 pii, injection, banned). No consumer label — cardinality rule.",
            ),
            &["rule", "kind"],
        )
        .expect("valid metric definition");
        let ai_semantic_cache_hits_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_semantic_cache_hits_total",
                "AI semantic cache hits (DW-083), by model alias. The model \
                 is the client-facing alias (config-bounded label). No consumer \
                 label — cardinality rule.",
            ),
            &["model"],
        )
        .expect("valid metric definition");
        let ai_semantic_cache_misses_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_semantic_cache_misses_total",
                "AI semantic cache misses (DW-083), by model alias. The model \
                 is the client-facing alias (config-bounded label). No consumer \
                 label — cardinality rule.",
            ),
            &["model"],
        )
        .expect("valid metric definition");
        let ai_routing_policy_escalations_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_routing_policy_escalations_total",
                "AI routing-policy FallbackChain escalations (DW-085): complex \
                 prompts (classifier score >= threshold) routed to the costlier \
                 model, by policy name (config-bounded label). No consumer \
                 label — cardinality rule.",
            ),
            &["policy"],
        )
        .expect("valid metric definition");
        let ai_routing_policy_cheap_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_routing_policy_cheap_total",
                "AI routing-policy FallbackChain cheap-model selections \
                 (DW-085): simple prompts (classifier score < threshold) routed \
                 to the cheap model, by policy name (config-bounded label). No \
                 consumer label — cardinality rule.",
            ),
            &["policy"],
        )
        .expect("valid metric definition");
        let ai_routing_policy_latency_cost_selections_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_routing_policy_latency_cost_selections_total",
                "AI routing-policy LatencyCost selections (DW-085): requests \
                 routed by the static cost/latency policy, by policy name \
                 (config-bounded label). No consumer label — cardinality rule.",
            ),
            &["policy"],
        )
        .expect("valid metric definition");
        let ai_experiment_variant_selections_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_experiment_variant_selections_total",
                "AI experiment (A/B test) variant selections (DW-086), by \
                 experiment and variant (config-bounded labels). No consumer \
                 label — cardinality rule; the consumer is in the access log \
                 and the analytics assignment table.",
            ),
            &["experiment", "variant"],
        )
        .expect("valid metric definition");
        let ai_stream_chunks_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_stream_chunks_total",
                "Forwarded AI streaming chunks (DW-077), by provider. One per \
                 client-visible delta chunk.",
            ),
            &["provider"],
        )
        .expect("valid metric definition");
        let ai_first_token_seconds = HistogramVec::new(
            HistogramOpts::new(
                "dwara_ai_first_token_seconds",
                "Time to the first forwarded chunk of an AI stream (DW-077), \
                 by provider.",
            )
            .buckets(FIRST_TOKEN_BUCKETS.to_vec()),
            &["provider"],
        )
        .expect("valid metric definition");
        let ai_stream_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                "dwara_ai_stream_duration_seconds",
                "Total AI stream duration (DW-077) — request send to stream \
                 end — by provider.",
            )
            .buckets(DURATION_BUCKETS.to_vec()),
            &["provider"],
        )
        .expect("valid metric definition");
        let ai_request_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                "dwara_ai_request_duration_seconds",
                "AI request latency (AI-07) — request parse through response \
                 translation — by provider (config-bounded) and route \
                 (config-bounded), in seconds. Observed once per completed AI \
                 request (success and provider/translation errors alike).",
            )
            .buckets(DURATION_BUCKETS.to_vec()),
            &["provider", "route"],
        )
        .expect("valid metric definition");
        let ai_tokens_per_request = HistogramVec::new(
            HistogramOpts::new(
                "dwara_ai_tokens_per_request",
                "Distribution of provider-reported tokens per AI request \
                 (AI-07), by provider (config-bounded), provider_model \
                 (config-bounded), and kind (prompt | completion). \
                 Provider-reported only; the gateway never estimates. No \
                 consumer label (cardinality rule).",
            )
            .buckets(TOKENS_PER_REQUEST_BUCKETS.to_vec()),
            &["provider", "model", "kind"],
        )
        .expect("valid metric definition");
        let mcp_sessions_total = IntCounterVec::new(
            Opts::new(
                "dwara_mcp_sessions_total",
                "MCP session lifecycle transitions (DW-087), by state. \
                 state: initialized (a new session created by initialize), \
                 closed (a session deleted by shutdown), expired (a session \
                 reaped by the TTL cleanup).",
            ),
            &["state"],
        )
        .expect("valid metric definition");
        let mcp_tool_calls_total = IntCounterVec::new(
            Opts::new(
                "dwara_mcp_tool_calls_total",
                "MCP tool calls (DW-087), by tool and status. tool is the \
                 config-declared tool name (config-bounded); status is the \
                 closed set success, error, denied. No consumer label \
                 (cardinality rule); the consumer is in the access log and \
                 the analytics mcp_tool_calls table.",
            ),
            &["tool", "status"],
        )
        .expect("valid metric definition");
        let mcp_tool_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                "dwara_mcp_tool_duration_seconds",
                "MCP tool call duration (DW-087) -- authz check through \
                 upstream response -- by tool (config-bounded label).",
            )
            .buckets(DURATION_BUCKETS.to_vec()),
            &["tool"],
        )
        .expect("valid metric definition");
        let a2a_requests_total = IntCounterVec::new(
            Opts::new(
                "dwara_a2a_requests_total",
                "A2A (agent-to-agent) requests (DW-114), by agent and \
                 outcome. agent is the config-declared agent name \
                 (config-bounded); outcome is the closed set success, \
                 error, stubbed (stubbed = the task lifecycle is not \
                 yet implemented pending spec freeze). No consumer label \
                 (cardinality rule); the consumer is in the access log.",
            ),
            &["agent", "outcome"],
        )
        .expect("valid metric definition");
        let a2a_sessions_total = IntCounterVec::new(
            Opts::new(
                "dwara_a2a_sessions_total",
                "A2A session lifecycle transitions (DW-114), by state. \
                 state: initialized (a new task session created), closed \
                 (a session deleted), expired (a session reaped by the \
                 TTL cleanup). A closed three-value label set, so \
                 cardinality is three series total (mirrors MCP).",
            ),
            &["state"],
        )
        .expect("valid metric definition");
        let nano_service_requests_total = IntCounterVec::new(
            Opts::new(
                "dwara_nano_service_requests_total",
                "Nano-service (WASM route handler) request outcomes (DW-106), \
                 by route and outcome. route is the config-declared route \
                 name (config-bounded); outcome is the closed set success, \
                 error, timeout. No consumer label (cardinality rule); the \
                 consumer is in the access log.",
            ),
            &["route", "outcome"],
        )
        .expect("valid metric definition");
        let nano_service_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                "dwara_nano_service_duration_seconds",
                "Nano-service handle call duration (DW-106) -- the WASM \
                 module's request->response execution -- by route \
                 (config-bounded label).",
            )
            .buckets(DURATION_BUCKETS.to_vec()),
            &["route"],
        )
        .expect("valid metric definition");
        let plugin_failures_total = IntCounterVec::new(
            Opts::new(
                "dwara_plugin_failures_total",
                "Plugin dispatch failures (DW-157), by plugin name and \
                 reason. name is the config-declared plugin name \
                 (config-bounded); reason is the closed set crashed, \
                 disabled, not_loaded, not_registered, instantiate_failed, \
                 trap, body_too_large, response_stream_ended, \
                 callout_failed, callout_loop. Counted once \
                 per request the plugin \
                 chain answered with an error (a send_http_response \
                 short-circuit is a plugin decision, not a failure).",
            ),
            &["name", "reason"],
        )
        .expect("valid metric definition");
        let plugin_callouts_total = IntCounterVec::new(
            Opts::new(
                "dwara_plugin_callouts_total",
                "Plugin HTTP callout outcomes (DW-167), by plugin name \
                 and outcome. name is the config-declared plugin name \
                 (config-bounded); outcome is the closed set ok (a \
                 completed response was delivered, any status), timeout \
                 (the clamped whole-exchange deadline expired), error \
                 (connect/DNS/TLS/SSRF/framing/cap failures), loop_guard \
                 (the per-phase round cap stopped a runaway dispatch \
                 loop). Counted once per performed callout (loop_guard \
                 once per guard trip).",
            ),
            &["name", "outcome"],
        )
        .expect("valid metric definition");
        let plugin_total = IntGaugeVec::new(
            Opts::new(
                "dwara_plugin_total",
                "Plugins per lifecycle state (DW-158), refreshed at \
                 publish time. state: healthy, crashed (load or compile \
                 failed; routes referencing it answer 500), disabled, \
                 not_loaded (declared but absent from the runtime -- an \
                 unresolved registry source or a failed engine rebuild), \
                 not_registered (native filter name with no registered \
                 factory). A closed five-value label set over the \
                 generation's declared plugins -- the same enumeration \
                 GET /plugins serves.",
            ),
            &["state"],
        )
        .expect("valid metric definition");
        let tls_pq_handshakes_total = IntCounterVec::new(
            Opts::new(
                "dwara_tls_pq_handshakes_total",
                "Post-quantum TLS handshake outcomes (DW-105), by result. \
                 result: success (the X25519+ML-KEM hybrid kx group was \
                 negotiated), fallback (the client did not support the \
                 hybrid group and a classical group was negotiated), \
                 disabled (PQ was off or the experimental API was inert). \
                 A closed three-value label set, so cardinality is three \
                 series total.",
            ),
            &["result"],
        )
        .expect("valid metric definition");
        let rate_limiter_adaptive_factor = GaugeVec::new(
            Opts::new(
                "dwara_rate_limiter_adaptive_factor",
                "Current adaptive rate-limit factor (DW-089), by policy \
                 (config-bounded). 1.0 = configured rate; < 1.0 tightens, \
                 > 1.0 relaxes. Bounded [min_factor, max_factor].",
            ),
            &["policy"],
        )
        .expect("valid metric definition");
        let rate_limiter_origin_signal_total = IntCounterVec::new(
            Opts::new(
                "dwara_rate_limiter_origin_signal_total",
                "Upstream origin signals received by the adaptive controller \
                 (DW-089), by policy and signal (both config-bounded; signal \
                 is the closed set retry_after).",
            ),
            &["policy", "signal"],
        )
        .expect("valid metric definition");
        let rate_limiter_adaptive_tightened_total = IntCounterVec::new(
            Opts::new(
                "dwara_rate_limiter_adaptive_tightened_total",
                "Adaptive rate-limit factor decreased / tightened (DW-089), by \
                 policy (config-bounded). Counted once per stressed update that \
                 moved the factor down.",
            ),
            &["policy"],
        )
        .expect("valid metric definition");
        let rate_limiter_adaptive_relaxed_total = IntCounterVec::new(
            Opts::new(
                "dwara_rate_limiter_adaptive_relaxed_total",
                "Adaptive rate-limit factor increased / relaxed (DW-089), by \
                 policy (config-bounded). Counted once per healthy update that \
                 moved the factor up.",
            ),
            &["policy"],
        )
        .expect("valid metric definition");
        let canary_promotions_total = IntCounterVec::new(
            Opts::new(
                "dwara_canary_promotions_total",
                "Canary promotions (DW-091), by group (config-bounded: service \
                 name or AI alias). Counted once per promote action the \
                 auto-canary controller applied.",
            ),
            &["group"],
        )
        .expect("valid metric definition");
        let canary_rollbacks_total = IntCounterVec::new(
            Opts::new(
                "dwara_canary_rollbacks_total",
                "Canary rollbacks (DW-091), by group (config-bounded: service \
                 name or AI alias). Counted once per rollback action (including \
                 severe-regression immediate rollbacks to 0).",
            ),
            &["group"],
        )
        .expect("valid metric definition");
        let canary_weight = GaugeVec::new(
            Opts::new(
                "dwara_canary_weight",
                "Current canary weight (DW-091), by group (config-bounded: \
                 service name or AI alias). Set on every applied adjustment. \
                 The total weight stays constant; the baseline absorbs the \
                 difference.",
            ),
            &["group"],
        )
        .expect("valid metric definition");
        let mesh_connections_total = IntCounterVec::new(
            Opts::new(
                "dwara_mesh_connections_total",
                "Sidecar connections proxied (DW-107), by mode and direction. \
                 `mode` is the closed set `sidecar`; `direction` is the closed \
                 set `inbound`, `outbound`. Inbound: terminate mTLS, apply \
                 policies, forward to the local app. Outbound: apply policies, \
                 wrap in mTLS, dial the remote sidecar.",
            ),
            &["mode", "direction"],
        )
        .expect("valid metric definition");
        let spiffe_svid_refresh_total = IntCounterVec::new(
            Opts::new(
                "dwara_spiffe_svid_refresh_total",
                "SPIFFE SVID refresh attempts (DW-107), by result. `result` is \
                 the closed set `success`, `error`. Counted once per Workload \
                 API refresh attempt; an error keeps serving the previous SVID \
                 until it expires.",
            ),
            &["result"],
        )
        .expect("valid metric definition");
        let spiffe_svid_expiry_seconds = IntGauge::new(
            "dwara_spiffe_svid_expiry_seconds",
            "Seconds until the current SPIFFE SVID expires (DW-107). Set on \
             every successful refresh; counts down to 0 as the SVID nears \
             expiry. The refresh loop schedules the next fetch at half the \
             remaining lifetime by default.",
        )
        .expect("valid metric definition");
        let l4_connections_total = IntCounterVec::new(
            Opts::new(
                "dwara_l4_connections_total",
                "L4 connections proxied (DW-103), by listener and protocol. \
                 listener is the config-declared listener name (config-bounded); \
                 protocol is the closed set tcp, udp. Counted once per accepted \
                 L4 connection (before the splice is established).",
            ),
            &["listener", "protocol"],
        )
        .expect("valid metric definition");
        let l4_connection_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                "dwara_l4_connection_duration_seconds",
                "L4 connection duration (DW-103) -- splice establishment through \
                 close -- by listener (config-bounded label), in seconds. Observed \
                 once per connection (forwarded and closed alike).",
            )
            .buckets(DURATION_BUCKETS.to_vec()),
            &["listener"],
        )
        .expect("valid metric definition");
        let ai_tokens_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_tokens_total",
                "Provider-reported AI token usage (DW-075), by provider and \
                 direction (prompt|completion), with the canary version label \
                 (DW-076) or default. Provider-reported only; the gateway never \
                 estimates.",
            ),
            &["provider", "kind", "version"],
        )
        .expect("valid metric definition");
        let ai_cost_micros_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_cost_micros_total",
                "Total priced AI spend in micro-USD (DW-079), by provider and \
                 provider_model. Priced through the dataplane's compiled pricing \
                 table; 0 for unknown models. No consumer label (cardinality).",
            ),
            &["provider", "model"],
        )
        .expect("valid metric definition");
        let ai_unknown_pricing_total = IntCounterVec::new(
            Opts::new(
                "dwara_ai_unknown_pricing_total",
                "Requests served by a model with no pricing entry (DW-AI-03), by \
                 provider_model. Incremented only under the alert unknown-model \
                 policy (allow is silent; fail_closed rejects before recording). \
                 The model label is the provider model identifier (config-bounded).",
            ),
            &["model"],
        )
        .expect("valid metric definition");
        // Clones share state (every prometheus family is a shared handle),
        // so registering clones keeps the originals usable for recording.
        for m in [
            Box::new(requests_total.clone()) as Box<dyn prometheus::core::Collector>,
            Box::new(request_duration.clone()),
            Box::new(upstream_attempts_total.clone()),
            Box::new(retries_total.clone()),
            Box::new(hedge_sent_total.clone()),
            Box::new(mirror_sent_total.clone()),
            Box::new(rate_limited_total.clone()),
            Box::new(shed_total.clone()),
            Box::new(policy_dry_run_total.clone()),
            Box::new(breaker_state.clone()),
            Box::new(endpoint_health.clone()),
            Box::new(fail_open_picks.clone()),
            Box::new(jwks_refresh_total.clone()),
            Box::new(active_requests.clone()),
            Box::new(config_generation.clone()),
            Box::new(rate_limiter_evictions.clone()),
            Box::new(rate_limiter_live_keys.clone()),
            Box::new(webhook_events_total.clone()),
            Box::new(events_dropped.clone()),
            Box::new(events_emitted.clone()),
            Box::new(access_records_streamed_total.clone()),
            Box::new(access_records_offered.clone()),
            Box::new(access_records_dropped.clone()),
            Box::new(access_records_sampled_dropped.clone()),
            Box::new(websocket_policy_total.clone()),
            Box::new(split_picks_total.clone()),
            Box::new(sticky_sessions_total.clone()),
            Box::new(cache_lookups_total.clone()),
            Box::new(cache_stores_total.clone()),
            Box::new(cache_revalidated_total.clone()),
            Box::new(cache_purges_total.clone()),
            Box::new(cache_entries.clone()),
            Box::new(coalescing_leaders_total.clone()),
            Box::new(coalescing_followers_total.clone()),
            Box::new(coalescing_saved_upstream_calls_total.clone()),
            Box::new(coalescing_waiters.clone()),
            Box::new(quota_denied_total.clone()),
            Box::new(quota_used.clone()),
            Box::new(quota_limit.clone()),
            Box::new(SloCollectorHandle(std::sync::Arc::clone(&slo))),
            Box::new(admission_queued_total.clone()),
            Box::new(admission_queue_depth.clone()),
            Box::new(waf_total.clone()),
            Box::new(anomaly_total.clone()),
            Box::new(graphql_requests_total.clone()),
            Box::new(graphql_depth.clone()),
            Box::new(graphql_complexity.clone()),
            Box::new(license_status.clone()),
            Box::new(dns_discovery_endpoints.clone()),
            Box::new(dns_discovery_refresh_total.clone()),
            Box::new(dns_discovery_refresh_failures_total.clone()),
            Box::new(config_convergence_generation.clone()),
            Box::new(config_convergence_instances.clone()),
            Box::new(config_convergence_drift.clone()),
            Box::new(config_convergence_refresh_total.clone()),
            Box::new(config_convergence_refresh_failures_total.clone()),
            Box::new(ai_requests_total.clone()),
            Box::new(ai_tokens_total.clone()),
            Box::new(ai_cost_micros_total.clone()),
            Box::new(ai_unknown_pricing_total.clone()),
            Box::new(ai_stream_chunks_total.clone()),
            Box::new(ai_budget_denied_total.clone()),
            Box::new(ai_governance_denied_total.clone()),
            Box::new(ai_guardrail_blocked_total.clone()),
            Box::new(ai_semantic_cache_hits_total.clone()),
            Box::new(ai_semantic_cache_misses_total.clone()),
            Box::new(ai_routing_policy_escalations_total.clone()),
            Box::new(ai_routing_policy_cheap_total.clone()),
            Box::new(ai_routing_policy_latency_cost_selections_total.clone()),
            Box::new(ai_experiment_variant_selections_total.clone()),
            Box::new(ai_first_token_seconds.clone()),
            Box::new(ai_stream_duration_seconds.clone()),
            Box::new(ai_request_duration_seconds.clone()),
            Box::new(ai_tokens_per_request.clone()),
            Box::new(mcp_sessions_total.clone()),
            Box::new(mcp_tool_calls_total.clone()),
            Box::new(mcp_tool_duration_seconds.clone()),
            Box::new(a2a_requests_total.clone()),
            Box::new(a2a_sessions_total.clone()),
            Box::new(nano_service_requests_total.clone()),
            Box::new(nano_service_duration_seconds.clone()),
            Box::new(plugin_failures_total.clone()),
            Box::new(plugin_callouts_total.clone()),
            Box::new(plugin_total.clone()),
            Box::new(tls_pq_handshakes_total.clone()),
            Box::new(rate_limiter_adaptive_factor.clone()),
            Box::new(rate_limiter_origin_signal_total.clone()),
            Box::new(rate_limiter_adaptive_tightened_total.clone()),
            Box::new(rate_limiter_adaptive_relaxed_total.clone()),
            Box::new(canary_promotions_total.clone()),
            Box::new(canary_rollbacks_total.clone()),
            Box::new(canary_weight.clone()),
            Box::new(mesh_connections_total.clone()),
            Box::new(spiffe_svid_refresh_total.clone()),
            Box::new(spiffe_svid_expiry_seconds.clone()),
            Box::new(l4_connections_total.clone()),
            Box::new(l4_connection_duration_seconds.clone()),
        ] {
            registry
                .register(m)
                .expect("fresh registry accepts every family exactly once");
        }
        Observability {
            registry,
            requests_total,
            request_duration,
            upstream_attempts_total,
            retries_total,
            hedge_sent_total,
            mirror_sent_total,
            rate_limited_total,
            shed_total,
            policy_dry_run_total,
            breaker_state,
            endpoint_health,
            fail_open_picks,
            jwks_refresh_total,
            active_requests,
            config_generation,
            rate_limiter_evictions,
            rate_limiter_live_keys,
            webhook_events_total,
            events_dropped,
            events_emitted,
            access_records_streamed_total,
            access_records_offered,
            access_records_dropped,
            access_records_sampled_dropped,
            websocket_policy_total,
            split_picks_total,
            sticky_sessions_total,
            cache_lookups_total,
            cache_stores_total,
            cache_revalidated_total,
            cache_purges_total,
            cache_entries,
            coalescing_leaders_total,
            coalescing_followers_total,
            coalescing_saved_upstream_calls_total,
            coalescing_waiters,
            quota_denied_total,
            quota_used,
            quota_limit,
            slo,
            admission_queued_total,
            admission_queue_depth,
            waf_total,
            anomaly_total,
            graphql_requests_total,
            graphql_depth,
            graphql_complexity,
            license_status,
            dns_discovery_endpoints,
            dns_discovery_refresh_total,
            dns_discovery_refresh_failures_total,
            config_convergence_generation,
            config_convergence_instances,
            config_convergence_drift,
            config_convergence_refresh_total,
            config_convergence_refresh_failures_total,
            ai_requests_total,
            ai_tokens_total,
            ai_cost_micros_total,
            ai_unknown_pricing_total,
            ai_stream_chunks_total,
            ai_budget_denied_total,
            ai_governance_denied_total,
            ai_guardrail_blocked_total,
            ai_semantic_cache_hits_total,
            ai_semantic_cache_misses_total,
            ai_routing_policy_escalations_total,
            ai_routing_policy_cheap_total,
            ai_routing_policy_latency_cost_selections_total,
            ai_experiment_variant_selections_total,
            ai_first_token_seconds,
            ai_stream_duration_seconds,
            ai_request_duration_seconds,
            ai_tokens_per_request,
            mcp_sessions_total,
            mcp_tool_calls_total,
            mcp_tool_duration_seconds,
            a2a_requests_total,
            a2a_sessions_total,
            nano_service_requests_total,
            nano_service_duration_seconds,
            plugin_failures_total,
            plugin_callouts_total,
            plugin_total,
            tls_pq_handshakes_total,
            rate_limiter_adaptive_factor,
            rate_limiter_origin_signal_total,
            rate_limiter_adaptive_tightened_total,
            rate_limiter_adaptive_relaxed_total,
            canary_promotions_total,
            canary_rollbacks_total,
            canary_weight,
            mesh_connections_total,
            spiffe_svid_refresh_total,
            spiffe_svid_expiry_seconds,
            l4_connections_total,
            l4_connection_duration_seconds,
            access_sample_bits: AtomicU64::new(1.0f64.to_bits()),
            rng: SampleRng::new(),
        }
    }

    /// Access-log sample rate in [0.0, 1.0]. Env-driven at construction
    /// (`DWARA_ACCESS_LOG_SAMPLE`); settable directly for tests.
    pub fn set_access_sample(&self, rate: f64) {
        let rate = rate.clamp(0.0, 1.0);
        self.access_sample_bits
            .store(rate.to_bits(), Ordering::Relaxed);
    }

    pub fn access_sample(&self) -> f64 {
        f64::from_bits(self.access_sample_bits.load(Ordering::Relaxed))
    }

    /// Whether this request's access log line is emitted. Errors
    /// (status >= 500) are ALWAYS logged; everything else follows the
    /// configured sample rate.
    pub fn should_log_access(&self, status: u16) -> bool {
        if status >= 500 {
            return true;
        }
        let rate = self.access_sample();
        rate >= 1.0 || self.rng.next_f64() < rate
    }

    /// Count and observe one completed request.
    pub fn record_request(&self, route: &str, listener: &str, status: u16, elapsed: Duration) {
        self.requests_total
            .with_label_values(&[route, listener, status_class(status)])
            .inc();
        self.request_duration
            .with_label_values(&[route])
            .observe(elapsed.as_secs_f64());
    }

    /// Count one upstream send attempt (endpoint = "unpicked" when the
    /// dispatch never resolved an endpoint).
    pub fn record_upstream_attempt(&self, upstream: &str, endpoint: &str, status: u16) {
        self.upstream_attempts_total
            .with_label_values(&[upstream, endpoint, status_class(status)])
            .inc();
    }

    /// Count one retry (a send attempt after the first).
    pub fn record_retry(&self, upstream: &str) {
        self.retries_total.with_label_values(&[upstream]).inc();
    }

    /// Count one hedge copy sent (DW-063): a speculative duplicate the
    /// hedge timer fired. Called once per copy, not once per request.
    pub fn record_hedge_sent(&self, upstream: &str) {
        self.hedge_sent_total.with_label_values(&[upstream]).inc();
    }

    /// Count one mirror (shadow traffic) request sent (DW-062).
    pub fn record_mirror_sent(&self, upstream: &str) {
        self.mirror_sent_total.with_label_values(&[upstream]).inc();
    }

    /// Count one rate-limit denial (429).
    pub fn record_rate_limited(&self, route: &str) {
        self.rate_limited_total.with_label_values(&[route]).inc();
    }

    /// Set the current adaptive rate-limit factor for a policy (DW-089).
    /// `factor` is the bounded `[min_factor, max_factor]` value the
    /// controller computed on its last `record_outcome`.
    pub fn set_adaptive_factor(&self, policy: &str, factor: f64) {
        self.rate_limiter_adaptive_factor
            .with_label_values(&[policy])
            .set(factor);
    }

    /// Count one upstream origin signal received by the adaptive
    /// controller (DW-089). `signal` is the closed set `retry_after`.
    pub fn record_adaptive_origin_signal(&self, policy: &str, signal: &str) {
        self.rate_limiter_origin_signal_total
            .with_label_values(&[policy, signal])
            .inc();
    }

    /// Count one adaptive factor decrease / tighten (DW-089).
    pub fn record_adaptive_tightened(&self, policy: &str) {
        self.rate_limiter_adaptive_tightened_total
            .with_label_values(&[policy])
            .inc();
    }

    /// Count one adaptive factor increase / relax (DW-089).
    pub fn record_adaptive_relaxed(&self, policy: &str) {
        self.rate_limiter_adaptive_relaxed_total
            .with_label_values(&[policy])
            .inc();
    }

    /// Count one canary promotion (DW-091), by group (config-bounded:
    /// service name or AI alias).
    pub fn record_canary_promotion(&self, group: &str) {
        self.canary_promotions_total
            .with_label_values(&[group])
            .inc();
    }

    /// Count one canary rollback (DW-091), by group (config-bounded:
    /// service name or AI alias). Includes severe-regression
    /// immediate rollbacks to 0.
    pub fn record_canary_rollback(&self, group: &str) {
        self.canary_rollbacks_total
            .with_label_values(&[group])
            .inc();
    }

    /// Set the current canary weight for a group (DW-091). Set on
    /// every applied adjustment; the total weight stays constant.
    pub fn set_canary_weight(&self, group: &str, weight: f64) {
        self.canary_weight.with_label_values(&[group]).set(weight);
    }

    /// Count one sidecar connection proxied (DW-107), by mode and
    /// direction. `mode` is the closed set `sidecar`; `direction` is
    /// the closed set `inbound`, `outbound`. Inbound: terminate mTLS,
    /// apply policies, forward to the local app. Outbound: apply
    /// policies, wrap in mTLS, dial the remote sidecar.
    pub fn record_mesh_connection(&self, mode: &str, direction: &str) {
        self.mesh_connections_total
            .with_label_values(&[mode, direction])
            .inc();
    }

    /// Count one SPIFFE SVID refresh attempt (DW-107), by result.
    /// `result` is the closed set `success`, `error`. An error keeps
    /// serving the previous SVID until it expires.
    pub fn record_spiffe_svid_refresh(&self, result: &str) {
        self.spiffe_svid_refresh_total
            .with_label_values(&[result])
            .inc();
    }

    /// Set the seconds-until-expiry gauge for the current SPIFFE SVID
    /// (DW-107). Set on every successful refresh; counts down to 0 as
    /// the SVID nears expiry.
    pub fn set_spiffe_svid_expiry_seconds(&self, seconds: i64) {
        self.spiffe_svid_expiry_seconds.set(seconds);
    }

    /// Count one quota denial (429, DW-033) by consumer and binding
    /// budget. Deliberately NOT the `rate_limited_total` family: budgets
    /// and rates are separate mechanisms (the issue's headline
    /// distinction); a quota wall is not a rate spike.
    pub fn record_quota_denied(&self, consumer: &str, budget: &str) {
        self.quota_denied_total
            .with_label_values(&[consumer, budget])
            .inc();
    }

    /// Set a budget's current-window usage gauge (DW-033, scrape-time
    /// snapshot; `budget` is `daily` or `monthly`).
    pub fn set_quota_used(&self, consumer: &str, budget: &str, used: i64) {
        self.quota_used
            .with_label_values(&[consumer, budget])
            .set(used);
    }

    /// Set a budget's configured cap gauge (DW-033, scrape-time).
    pub fn set_quota_limit(&self, consumer: &str, budget: &str, limit: i64) {
        self.quota_limit
            .with_label_values(&[consumer, budget])
            .set(limit);
    }

    /// Count one gateway-cap shed (503), by priority class.
    pub fn record_shed(&self, priority: u8) {
        self.shed_total
            .with_label_values(&[&priority.to_string()])
            .inc();
    }

    /// Count one admission queue outcome (DW-053). `outcome` is one of
    /// the closed set `admitted` (a permit was acquired after queueing),
    /// `timeout` (the queue timeout expired; the request was shed with
    /// 503), `queue_full` (the queue was at capacity; the request was
    /// shed immediately with 503).
    pub fn record_admission_queued(&self, outcome: &str) {
        self.admission_queued_total
            .with_label_values(&[outcome])
            .inc();
    }

    /// Set the current admission queue depth gauge (DW-053). Called from
    /// the dataplane's atomic-backed depth counter (the hot path stays
    /// free of registry coupling — the same gauge model as the
    /// coalescing-waiters gauge).
    pub fn set_admission_queue_depth(&self, depth: i64) {
        self.admission_queue_depth.set(depth);
    }

    /// Count one WAF-lite inspection outcome (DW-051). `route` is the
    /// matched route name; `filter` is the filter category that matched
    /// (`sqli`, `xss`, `path_traversal`) or `all` for the `passed`
    /// outcome (no match); `outcome` is `blocked` (403 returned),
    /// `logged` (dry-run match, request allowed), or `passed` (no
    /// match). All three labels are config-bounded (route names and the
    /// closed filter/outcome sets).
    pub fn record_waf(&self, route: &str, filter: &str, outcome: &str) {
        self.waf_total
            .with_label_values(&[route, filter, outcome])
            .inc();
    }

    /// Count one anomaly-scoring inspection outcome (DW-090). `route` is
    /// the matched route name; `outcome` is `blocked` (403 returned),
    /// `logged` (dry-run match, request allowed), or `passed` (score
    /// below the threshold). Both labels are config-bounded (route names
    /// and the closed outcome set).
    pub fn record_anomaly(&self, route: &str, outcome: &str) {
        self.anomaly_total
            .with_label_values(&[route, outcome])
            .inc();
    }

    /// Count one GraphQL check outcome (DW-099). `route` is the matched
    /// route name; `outcome` is one of the closed set `allowed`,
    /// `depth_exceeded`, `complexity_exceeded`,
    /// `persisted_query_required`. Both labels are config-bounded.
    pub fn record_graphql(&self, route: &str, outcome: &str) {
        self.graphql_requests_total
            .with_label_values(&[route, outcome])
            .inc();
    }

    /// Set the last GraphQL query depth gauge (DW-099), by route.
    pub fn set_graphql_depth(&self, route: &str, depth: i64) {
        self.graphql_depth.with_label_values(&[route]).set(depth);
    }

    /// Set the last GraphQL query complexity gauge (DW-099), by route.
    pub fn set_graphql_complexity(&self, route: &str, complexity: i64) {
        self.graphql_complexity
            .with_label_values(&[route])
            .set(complexity);
    }

    /// Count one dry-run (monitor) would-have-rejected observation
    /// (DW-041): a policy phase evaluated a DENY (or shed, or limit
    /// violation) and the request was allowed anyway. `phase` is one of
    /// `route_limits`, `authz`, `rate_limit`, `load_shed`; `route` is
    /// the matched route name (`unrouted` for the pre-404 listener/
    /// global policy pass) — both config-bounded label spaces.
    pub fn record_policy_dry_run(&self, phase: &str, route: &str) {
        self.policy_dry_run_total
            .with_label_values(&[phase, route])
            .inc();
    }

    /// The dry-run report's log side (DW-041): one structured warn
    /// event at target `dwara::policy` per would-have-rejected request,
    /// carrying the phase, the status the phase WOULD have answered, the
    /// route and consumer (when known), and the phase's own reason —
    /// the request-id correlation key ties it to the trace and access
    /// line of the request that was allowed. Field list is exhaustive
    /// and redacted by construction (no headers, no query string).
    pub fn emit_policy_dry_run(
        &self,
        phase: &str,
        would_be_status: u16,
        route: &str,
        consumer: Option<&str>,
        request_id: &str,
        detail: &str,
    ) {
        tracing::warn!(
            target: "dwara::policy",
            code = "policy_dry_run",
            phase = phase,
            would_be_status = would_be_status,
            route = route,
            consumer = consumer.unwrap_or("anonymous"),
            request_id = request_id,
            detail = detail,
            "policy would have rejected this request (dry-run); \
             enforcement of this phase skipped"
        );
    }

    /// The JWKS refresh counter child for one provider (handed to the
    /// authenticator; incremented on every JWKS fetch attempt).
    pub fn jwks_refresh_counter(&self, provider: &str) -> IntCounter {
        self.jwks_refresh_total.with_label_values(&[provider])
    }

    /// Count one response-cache lookup outcome (DW-037). `outcome` is
    /// one of the closed set `hit` / `stale` / `miss` / `bypass` (see
    /// `dataplane::response_cache` for the decision table).
    pub fn record_cache_lookup(&self, outcome: &str) {
        self.cache_lookups_total.with_label_values(&[outcome]).inc();
    }

    /// Count one response-cache store outcome (DW-037). `outcome` is
    /// one of the closed set `stored` / `vetoed` / `over_cap` /
    /// `error`.
    pub fn record_cache_store(&self, outcome: &str) {
        self.cache_stores_total.with_label_values(&[outcome]).inc();
    }

    /// Count one 304-confirmed revalidation served from the stored body
    /// (DW-037).
    pub fn record_cache_revalidated(&self) {
        self.cache_revalidated_total.inc();
    }

    /// Count one cache purge (DW-037). `scope` is `route`, `all`,
    /// `tag`, `url`, or `url_prefix` (the DP-04 tag/url purge axes).
    pub fn record_cache_purge(&self, scope: &str) {
        self.cache_purges_total.with_label_values(&[scope]).inc();
    }

    /// Set the approximate live-entries gauge (DW-037) — a scrape-time
    /// snapshot walk of the backing store (the same gauge model as the
    /// rate-limiter live-keys gauge: state lives outside observability,
    /// the dataplane walks it at scrape time).
    pub fn set_cache_entries(&self, entries: i64) {
        self.cache_entries.set(entries);
    }

    /// DW-038: this cache miss became the coalescing leader.
    pub fn record_coalescing_leader(&self) {
        self.coalescing_leaders_total.inc();
    }

    /// DW-038: a coalescing follower resolved — `outcome` is one of the
    /// closed set `served` / `fell_back_timeout` / `fell_back_unshared`
    /// / `fell_back_epoch` (see the field docs; callers pass literals).
    pub fn record_coalescing_follower(&self, outcome: &str) {
        self.coalescing_followers_total
            .with_label_values(&[outcome])
            .inc();
    }

    /// DW-038: one upstream call avoided (a follower was served).
    pub fn record_coalescing_saved(&self) {
        self.coalescing_saved_upstream_calls_total.inc();
    }

    /// DW-038: follower waiters gauge, +/- on park/resolve.
    pub fn coalescing_waiter(&self, parked: bool) {
        if parked {
            self.coalescing_waiters.inc();
        } else {
            self.coalescing_waiters.dec();
        }
    }

    pub fn active_requests(&self) -> &IntGauge {
        &self.active_requests
    }

    /// Publish the current config generation number.
    pub fn set_config_generation(&self, generation: u64) {
        self.config_generation.set(generation as i64);
    }

    /// Set the `dwara_license_status` gauge (DW-032): 0 = no license
    /// (OSS), 1 = valid, 2 = expired within grace, 3 = expired past
    /// grace (degraded to OSS). Set once at startup and on every reload.
    pub fn set_license_status(&self, status: i64) {
        self.license_status.set(status);
    }

    /// Set the `dwara_dns_discovery_endpoints` gauge for one upstream
    /// (DW-042): the current resolved endpoint count. The `upstream`
    /// label is config-bounded; cardinality is never per resolved
    /// address. Set by the discovery task after each refresh.
    pub fn set_dns_discovery_endpoints(&self, upstream: &str, count: i64) {
        self.dns_discovery_endpoints
            .with_label_values(&[upstream])
            .set(count);
    }

    /// Count one DNS discovery refresh attempt (DW-042) in
    /// `dwara_dns_discovery_refresh_total{upstream}`. The `upstream`
    /// label is config-bounded.
    pub fn record_dns_discovery_refresh(&self, upstream: &str) {
        self.dns_discovery_refresh_total
            .with_label_values(&[upstream])
            .inc();
    }

    /// Count one DNS discovery refresh failure (DW-042) in
    /// `dwara_dns_discovery_refresh_failures_total{upstream}`. The
    /// `upstream` label is config-bounded.
    pub fn record_dns_discovery_refresh_failure(&self, upstream: &str) {
        self.dns_discovery_refresh_failures_total
            .with_label_values(&[upstream])
            .inc();
    }

    /// Set this instance's current convergence generation (DW-054) in
    /// `dwara_config_convergence_generation{instance}`. The instance
    /// label is config-bounded (one series per process). Set by the
    /// convergence coordinator after every local or remote publish.
    pub fn set_config_convergence_generation(&self, instance: &str, generation: i64) {
        self.config_convergence_generation
            .with_label_values(&[instance])
            .set(generation);
    }

    /// Set the total instances the convergence backend reports
    /// (DW-054) in `dwara_config_convergence_instances`. Set by the
    /// coordinator's drift check from the cluster view.
    pub fn set_config_convergence_instances(&self, count: i64) {
        self.config_convergence_instances.set(count);
    }

    /// Set the convergence drift flag (DW-054) in
    /// `dwara_config_convergence_drift`: 1 when one or more instances
    /// serve a different config hash than the majority, 0 when
    /// converged. Set by the coordinator's drift check.
    pub fn set_config_convergence_drift(&self, drift: bool) {
        self.config_convergence_drift.set(i64::from(drift));
    }

    /// Count one convergence refresh attempt (DW-054) in
    /// `dwara_config_convergence_refresh_total`.
    pub fn record_config_convergence_refresh(&self) {
        self.config_convergence_refresh_total.inc();
    }

    /// Count one AI budget denial (DW-078) in
    /// `dwara_ai_budget_denied_total{kind}` (kind: tokens | cost).
    pub fn record_ai_budget_denied(&self, kind: &str) {
        self.ai_budget_denied_total.with_label_values(&[kind]).inc();
    }

    /// Count one AI governance denial (DW-084) in
    /// `dwara_ai_governance_denied_total{reason}` (reason: the denial
    /// cause string, config-bounded — no consumer/model labels).
    pub fn record_ai_governance_denied(&self, reason: &str) {
        self.ai_governance_denied_total
            .with_label_values(&[reason])
            .inc();
    }

    /// Count one AI guardrail block (DW-082) in
    /// `dwara_ai_guardrail_blocked_total{rule,kind}` (rule: the rule
    /// name, config-bounded; kind: the phase/kind label — prompt,
    /// response, schema, pii, injection, banned). No consumer label
    /// (cardinality rule).
    pub fn record_ai_guardrail_blocked(&self, rule: &str, kind: &str) {
        self.ai_guardrail_blocked_total
            .with_label_values(&[rule, kind])
            .inc();
    }

    /// Count one AI semantic cache hit (DW-083) in
    /// `dwara_ai_semantic_cache_hits_total{model}` (model: the
    /// client-facing alias, config-bounded). No consumer label
    /// (cardinality rule).
    pub fn record_ai_semantic_cache_hit(&self, model: &str) {
        self.ai_semantic_cache_hits_total
            .with_label_values(&[model])
            .inc();
    }

    /// Count one AI semantic cache miss (DW-083) in
    /// `dwara_ai_semantic_cache_misses_total{model}` (model: the
    /// client-facing alias, config-bounded). No consumer label
    /// (cardinality rule).
    pub fn record_ai_semantic_cache_miss(&self, model: &str) {
        self.ai_semantic_cache_misses_total
            .with_label_values(&[model])
            .inc();
    }

    /// Count one AI routing-policy FallbackChain escalation (DW-085)
    /// in `dwara_ai_routing_policy_escalations_total{policy}` (policy:
    /// the routing-policy name, config-bounded). No consumer label
    /// (cardinality rule).
    pub fn record_ai_routing_policy_escalation(&self, policy: &str) {
        self.ai_routing_policy_escalations_total
            .with_label_values(&[policy])
            .inc();
    }

    /// Count one AI routing-policy FallbackChain cheap-model
    /// selection (DW-085) in
    /// `dwara_ai_routing_policy_cheap_total{policy}` (policy: the
    /// routing-policy name, config-bounded). No consumer label
    /// (cardinality rule).
    pub fn record_ai_routing_policy_cheap(&self, policy: &str) {
        self.ai_routing_policy_cheap_total
            .with_label_values(&[policy])
            .inc();
    }

    /// Count one AI routing-policy LatencyCost selection (DW-085) in
    /// `dwara_ai_routing_policy_latency_cost_selections_total{policy}`
    /// (policy: the routing-policy name, config-bounded). No consumer
    /// label (cardinality rule).
    pub fn record_ai_routing_policy_latency_cost_selection(&self, policy: &str) {
        self.ai_routing_policy_latency_cost_selections_total
            .with_label_values(&[policy])
            .inc();
    }

    /// Count one AI experiment (A/B test) variant selection (DW-086)
    /// in `dwara_ai_experiment_variant_selections_total{experiment,
    /// variant}` (experiment: the A/B test name, variant: the
    /// selected variant name, both config-bounded labels). No
    /// consumer label (cardinality rule).
    pub fn record_ai_experiment_variant_selection(&self, experiment: &str, variant: &str) {
        self.ai_experiment_variant_selections_total
            .with_label_values(&[experiment, variant])
            .inc();
    }

    /// Count forwarded streaming chunks (DW-077) in
    /// `dwara_ai_stream_chunks_total{provider}`.
    pub fn record_ai_stream_chunk(&self, provider: &str) {
        self.ai_stream_chunks_total
            .with_label_values(&[provider])
            .inc();
    }

    /// Record the time to a stream's first forwarded chunk (DW-077) in
    /// `dwara_ai_first_token_seconds{provider}`. Call once per stream.
    pub fn record_ai_first_token(&self, provider: &str, seconds: f64) {
        self.ai_first_token_seconds
            .with_label_values(&[provider])
            .observe(seconds);
    }

    /// Record a completed stream's total duration (DW-077) in
    /// `dwara_ai_stream_duration_seconds{provider}`.
    pub fn record_ai_stream_end(&self, provider: &str, seconds: f64) {
        self.ai_stream_duration_seconds
            .with_label_values(&[provider])
            .observe(seconds);
    }

    /// Record one AI request's end-to-end latency (AI-07) in
    /// `dwara_ai_request_duration_seconds{provider,route}`. `provider`
    /// and `route` are config-bounded labels. Call once per completed
    /// AI request (success and provider/translation errors alike);
    /// client-side rejections never reach a provider and are not
    /// observed here.
    pub fn record_ai_request_duration(&self, provider: &str, route: &str, seconds: f64) {
        self.ai_request_duration_seconds
            .with_label_values(&[provider, route])
            .observe(seconds);
    }

    /// Observe the per-request token distribution (AI-07) in
    /// `dwara_ai_tokens_per_request{provider,model,kind}`. `kind` is
    /// the closed set `prompt` | `completion`. Call once per request
    /// that reported usage, for each direction that has a value
    /// (provider-reported only; the gateway never estimates). `model`
    /// is the provider's own model identifier (config-bounded).
    pub fn record_ai_tokens_per_request(
        &self,
        provider: &str,
        model: &str,
        prompt: u64,
        completion: u64,
    ) {
        if prompt > 0 {
            self.ai_tokens_per_request
                .with_label_values(&[provider, model, "prompt"])
                .observe(prompt as f64);
        }
        if completion > 0 {
            self.ai_tokens_per_request
                .with_label_values(&[provider, model, "completion"])
                .observe(completion as f64);
        }
    }

    /// Count one MCP session lifecycle transition (DW-087) in
    /// `dwara_mcp_sessions_total{state}`. `state` is one of the closed
    /// set `initialized` (a new session created by `initialize`),
    /// `closed` (a session deleted by `shutdown`), or `expired` (a
    /// session reaped by the TTL cleanup).
    pub fn record_mcp_session(&self, state: &str) {
        self.mcp_sessions_total.with_label_values(&[state]).inc();
    }

    /// Count one MCP tool call (DW-087) in
    /// `dwara_mcp_tool_calls_total{tool,status}`. `tool` is the
    /// config-declared tool name (config-bounded); `status` is one of
    /// the closed set `success`, `error`, `denied`. No consumer label
    /// (cardinality rule); the consumer is in the access log and the
    /// analytics `mcp_tool_calls` table.
    pub fn record_mcp_tool_call(&self, tool: &str, status: &str) {
        self.mcp_tool_calls_total
            .with_label_values(&[tool, status])
            .inc();
    }

    /// Record an MCP tool call duration (DW-087) in
    /// `dwara_mcp_tool_duration_seconds{tool}`. `tool` is the
    /// config-declared tool name (config-bounded label). Call once per
    /// tool call (success, error, and denied alike).
    pub fn record_mcp_tool_duration(&self, tool: &str, seconds: f64) {
        self.mcp_tool_duration_seconds
            .with_label_values(&[tool])
            .observe(seconds);
    }

    /// Count one A2A (agent-to-agent) request (DW-114) in
    /// `dwara_a2a_requests_total{agent,outcome}`. `agent` is the
    /// config-declared agent name (config-bounded); `outcome` is one
    /// of the closed set `success`, `error`, `stubbed` (`stubbed` =
    /// the task lifecycle is not yet implemented pending spec freeze).
    /// No consumer label (cardinality rule); the consumer is in the
    /// access log.
    pub fn record_a2a_request(&self, agent: &str, outcome: &str) {
        self.a2a_requests_total
            .with_label_values(&[agent, outcome])
            .inc();
    }

    /// Count one A2A session lifecycle transition (DW-114) in
    /// `dwara_a2a_sessions_total{state}`. `state` is one of the closed
    /// set `initialized` (a new task session created), `closed` (a
    /// session deleted), or `expired` (a session reaped by the TTL
    /// cleanup). Mirrors MCP's session counter.
    pub fn record_a2a_session(&self, state: &str) {
        self.a2a_sessions_total.with_label_values(&[state]).inc();
    }

    /// Count one nano-service (WASM route handler) request outcome
    /// (DW-106) in `dwara_nano_service_requests_total{route,outcome}`.
    /// `route` is the matched route name (config-bounded); `outcome` is
    /// one of the closed set `success`, `error`, `timeout`. No consumer
    /// label (cardinality rule); the consumer is in the access log.
    pub fn record_nano_service_request(&self, route: &str, outcome: &str) {
        self.nano_service_requests_total
            .with_label_values(&[route, outcome])
            .inc();
    }

    /// Record a nano-service `handle` call duration (DW-106) in
    /// `dwara_nano_service_duration_seconds{route}`. `route` is the
    /// matched route name (config-bounded label). Call once per request
    /// (success, error, and timeout alike).
    pub fn record_nano_service_duration(&self, route: &str, seconds: f64) {
        self.nano_service_duration_seconds
            .with_label_values(&[route])
            .observe(seconds);
    }

    /// Count one plugin dispatch failure (DW-157) in
    /// `dwara_plugin_failures_total{name,reason}`. `name` is the
    /// config-declared plugin name (config-bounded); `reason` is one of
    /// the closed set `crashed`, `disabled`, `not_loaded`,
    /// `not_registered`, `instantiate_failed`, `trap`,
    /// `body_too_large`, `response_stream_ended`. Counted once per
    /// request the plugin chain
    /// answered with an error — a plugin's `send_http_response`
    /// short-circuit is a decision, not a failure, and is not counted
    /// (the access-log `plugin_short_circuit` flag marks those).
    pub fn record_plugin_failure(&self, name: &str, reason: &str) {
        self.plugin_failures_total
            .with_label_values(&[name, reason])
            .inc();
    }

    /// Count one plugin HTTP callout outcome (DW-167) in
    /// `dwara_plugin_callouts_total{name,outcome}`. `name` is the
    /// config-declared plugin name (config-bounded); `outcome` is one
    /// of the closed set `ok` (a completed response was delivered to
    /// the plugin, any status), `timeout`, `error`, `loop_guard`.
    /// Counted once per performed callout (`loop_guard` once per guard
    /// trip); a callout ERROR additionally increments
    /// `dwara_plugin_failures_total{name,reason=callout_failed}`
    /// because the route fails closed.
    pub fn record_plugin_callout(&self, name: &str, outcome: &str) {
        self.plugin_callouts_total
            .with_label_values(&[name, outcome])
            .inc();
    }

    /// Set one state's plugin count (DW-158) in
    /// `dwara_plugin_total{state}`. `state` is one of the closed set
    /// `healthy`, `crashed`, `disabled`, `not_loaded`,
    /// `not_registered`. The dataplane sets every state's series on
    /// each publish (`reload_plugins`), so a reload that removes the
    /// last plugin of a state drives its series back to zero instead of
    /// stranding the previous count.
    pub fn set_plugin_state_total(&self, state: &str, total: i64) {
        self.plugin_total.with_label_values(&[state]).set(total);
    }

    /// Count one post-quantum TLS handshake outcome (DW-105) in
    /// `dwara_tls_pq_handshakes_total{result}`. `result` is one of the
    /// closed set `success` (the X25519+ML-KEM hybrid kx group was
    /// negotiated), `fallback` (the client did not support the hybrid
    /// group and a classical group was negotiated), or `disabled` (PQ
    /// was off or the experimental API was inert). Use
    /// `security::pq::pq_handshake_metric` to derive the label
    /// from a `security::pq::PqHandshakeResult`.
    pub fn record_tls_pq_handshake(&self, result: &str) {
        self.tls_pq_handshakes_total
            .with_label_values(&[result])
            .inc();
    }

    /// Count one convergence refresh failure (DW-054) in
    /// `dwara_config_convergence_refresh_failures_total`.
    pub fn record_config_convergence_refresh_failure(&self) {
        self.config_convergence_refresh_failures_total.inc();
    }

    /// Count one AI route request outcome (DW-075) in
    /// `dwara_ai_requests_total{provider,route,outcome,version}`. The
    /// provider, route, and version labels are config-bounded
    /// (`version` is the canary version that served, DW-076, or
    /// "default" for non-canary aliases).
    pub fn record_ai_request(&self, provider: &str, route: &str, outcome: &str, version: &str) {
        self.ai_requests_total
            .with_label_values(&[provider, route, outcome, version])
            .inc();
    }

    /// Add provider-reported token usage (DW-075) to
    /// `dwara_ai_tokens_total{provider,kind,version}` (kind: prompt |
    /// completion; version: the canary version that served, DW-076, or
    /// "default"). Provider-reported values only.
    pub fn record_ai_tokens(&self, provider: &str, prompt: u64, completion: u64, version: &str) {
        let counters = self
            .ai_tokens_total
            .with_label_values(&[provider, "prompt", version]);
        counters.inc_by(prompt);
        let counters = self
            .ai_tokens_total
            .with_label_values(&[provider, "completion", version]);
        counters.inc_by(completion);
    }

    /// Add priced AI spend (DW-079) to
    /// `dwara_ai_cost_micros_total{provider,model}` in micro-USD. The
    /// provider and model labels are config-bounded (no consumer
    /// label — cardinality rule).
    pub fn record_ai_cost(&self, provider: &str, model: &str, cost_micros: u64) {
        self.ai_cost_micros_total
            .with_label_values(&[provider, model])
            .inc_by(cost_micros);
    }

    /// Count one request served by a model with no pricing entry
    /// (DW-AI-03) under the `alert` unknown-model policy. The model
    /// label is the provider model identifier (config-bounded).
    pub fn record_ai_unknown_pricing(&self, model: &str) {
        self.ai_unknown_pricing_total
            .with_label_values(&[model])
            .inc();
    }

    /// The current `dwara_config_convergence_drift` gauge value (DW-054):
    /// 1 when drift was last reported, 0 when converged. Test seam +
    /// operator introspection (the gauge is also scraped via /metrics).
    pub fn config_convergence_drift(&self) -> i64 {
        self.config_convergence_drift.get()
    }

    /// The current `dwara_config_convergence_instances` gauge value
    /// (DW-054): the last cluster-view instance count. Test seam +
    /// operator introspection.
    pub fn config_convergence_instances(&self) -> i64 {
        self.config_convergence_instances.get()
    }

    /// Set the `breaker_state` gauge for one upstream (0 closed, 1 open,
    /// 2 half-open). Scrape-time snapshot setter: the walk that computes
    /// the values from live upstream state lives on the dataplane (see
    /// `refresh_observation_gauges` in `dataplane::upstream`), keeping
    /// this module free of runtime-state dependencies — it only records
    /// what it is handed.
    pub fn set_breaker_state(&self, upstream: &str, state: i64) {
        self.breaker_state.with_label_values(&[upstream]).set(state);
    }

    /// Set the `upstream_fail_open_picks` gauge for one upstream
    /// (scrape-time snapshot of the balancer's monotonic fail-open
    /// counter; a gauge rather than a counter so the hot pick path stays
    /// free of metrics coupling).
    pub fn set_fail_open_picks(&self, upstream: &str, picks: i64) {
        self.fail_open_picks
            .with_label_values(&[upstream])
            .set(picks);
    }

    /// Set the `dwara_rate_limiter_evictions_total` gauge: aggregate
    /// cells dropped by eviction sweeps across every compiled rule
    /// (#132). Scrape-time snapshot of the engine's monotonic counter —
    /// a gauge rather than a counter so the hot check path stays free
    /// of metrics coupling; the value resets when a reload rebuilds the
    /// engine. See [`Self::set_breaker_state`] for the refresh model.
    pub fn set_rate_limiter_evictions(&self, evictions: i64) {
        self.rate_limiter_evictions.set(evictions);
    }

    /// Set the `dwara_rate_limiter_live_keys` gauge: live per-key cells
    /// across every compiled rule, approximate under concurrent checks
    /// and bounded by the sharded store cap (#132). Aggregate and
    /// unlabeled by design — cardinality is never per key. Scrape-time
    /// snapshot setter; see [`Self::set_breaker_state`].
    pub fn set_rate_limiter_live_keys(&self, keys: i64) {
        self.rate_limiter_live_keys.set(keys);
    }

    /// Count one webhook delivery outcome (DW-044) in
    /// `dwara_webhook_events_total{kind,outcome}`. `kind` is the
    /// event-kind spelling, `outcome` one of `delivered`, `failed`,
    /// `dropped` (the deliverer is the only caller, so both label
    /// spaces stay the closed sets documented on the family).
    pub fn record_webhook_event(&self, kind: &str, outcome: &str) {
        self.webhook_events_total
            .with_label_values(&[kind, outcome])
            .inc();
    }

    /// Set the `dwara_events_dropped_total` gauge (DW-044): events
    /// dropped at EMIT time. Scrape-time snapshot of the event bus's
    /// monotonic counter; see [`Self::set_rate_limiter_evictions`] for
    /// the gauge-not-counter model.
    pub fn set_events_dropped(&self, dropped: i64) {
        self.events_dropped.set(dropped);
    }

    /// Set the `dwara_events_emitted_total` gauge (DW-044): events
    /// handed to the event queue since process start. Scrape-time
    /// snapshot setter; see [`Self::set_events_dropped`].
    pub fn set_events_emitted(&self, emitted: i64) {
        self.events_emitted.set(emitted);
    }

    /// Count `records` access records into
    /// `dwara_access_records_streamed_total{outcome}` (DW-121). The
    /// stream flusher is the only caller; `outcome` is one of
    /// `delivered`, `failed`, `dropped` (the closed set documented on
    /// the family), counted at BATCH granularity by the records each
    /// flushed batch carried.
    pub fn record_access_stream(&self, outcome: &str, records: u64) {
        self.access_records_streamed_total
            .with_label_values(&[outcome])
            .inc_by(records);
    }

    /// Set the `dwara_access_records_offered_total` gauge (DW-121):
    /// records offered to the stream channel since process start.
    /// Scrape-time snapshot of the stream's monotonic counter; see
    /// [`Self::set_events_dropped`] for the gauge-not-counter model.
    pub fn set_access_records_offered(&self, offered: i64) {
        self.access_records_offered.set(offered);
    }

    /// Set the `dwara_access_records_dropped_total` gauge (DW-121):
    /// records dropped at OFFER time. Scrape-time snapshot setter; see
    /// [`Self::set_access_records_offered`].
    pub fn set_access_records_dropped(&self, dropped: i64) {
        self.access_records_dropped.set(dropped);
    }

    /// Set the `dwara_access_records_sampled_dropped_total` gauge
    /// (REL-10, #221): records dropped by sampled degradation.
    /// Scrape-time snapshot setter; see
    /// [`Self::set_access_records_dropped`].
    pub fn set_access_records_sampled_dropped(&self, dropped: i64) {
        self.access_records_sampled_dropped.set(dropped);
    }

    /// Count one split dispatch decision (DW-040) in
    /// `dwara_split_picks_total{service,upstream}`. Both labels are
    /// config-declared names; the caller is the dispatch path.
    pub fn record_split_pick(&self, service: &str, upstream: &str) {
        self.split_picks_total
            .with_label_values(&[service, upstream])
            .inc();
    }

    /// Count one minted sticky session (DW-040).
    pub fn record_sticky_session(&self) {
        self.sticky_sessions_total.inc();
    }

    /// Count one WebSocket policy decision (DW-039) in
    /// `dwara_websocket_policy_total{route,outcome}`. `outcome` is one
    /// of `origin_denied`, `rate_closed` (the closed set documented on
    /// the family); `route` is the config-declared route name.
    pub fn record_websocket_policy(&self, route: &str, outcome: &str) {
        self.websocket_policy_total
            .with_label_values(&[route, outcome])
            .inc();
    }

    /// Set the `endpoint_health` gauge for one endpoint (1 available,
    /// 0 ejected). Scrape-time snapshot setter; see
    /// [`Self::set_breaker_state`] for the refresh model.
    pub fn set_endpoint_health(&self, upstream: &str, endpoint: &str, up: bool) {
        self.endpoint_health
            .with_label_values(&[upstream, endpoint])
            .set(up as i64);
    }

    /// Render every family in Prometheus text format.
    pub fn render(&self) -> String {
        let mut buf = Vec::new();
        let encoder = TextEncoder::new();
        let families = self.registry.gather();
        encoder
            .encode(&families, &mut buf)
            .expect("text encoding of gathered families cannot fail");
        String::from_utf8(buf).expect("prometheus text format is ASCII")
    }

    /// Gather the raw metric families (DW-073: OTLP metrics export).
    /// Used by the OTLP metrics exporter to convert prometheus metric
    /// families into OTLP protobuf for export to a collector.
    pub fn gather(&self) -> Vec<prometheus::proto::MetricFamily> {
        self.registry.gather()
    }
}

// --- OTel GenAI semantic conventions (AI-07) -------------------------------

/// Record the request-side OTel GenAI semantic-convention attributes
/// (AI-07) onto an AI span. The span MUST have been created with every
/// `gen_ai.*` field declared as `tracing::field::Empty` (see
/// `gen_ai_span_fields`); `Span::record` only fills fields the span's
/// metadata already declared. Request-side fields are always available
/// once the canonical [`crate::ai::types::ChatRequest`] is parsed; the
/// `system` is the serving provider's kind
/// ([`crate::config::ai::AiProviderKind::gen_ai_system`]).
pub fn record_gen_ai_request_attrs(
    span: &tracing::Span,
    system: &str,
    request_model: &str,
    max_tokens: Option<u64>,
    temperature: Option<f64>,
    top_p: Option<f64>,
) {
    span.record("gen_ai.system", system);
    span.record("gen_ai.request.model", request_model);
    if let Some(m) = max_tokens {
        span.record("gen_ai.request.max_tokens", m);
    }
    if let Some(t) = temperature {
        span.record("gen_ai.request.temperature", t);
    }
    if let Some(p) = top_p {
        span.record("gen_ai.request.top_p", p);
    }
}

/// Record the response-side OTel GenAI semantic-convention attributes
/// (AI-07) onto an AI span. Each argument is recorded only when present
/// (provider-reported only; the gateway never estimates). `finish_reasons`
/// is the per-choice finish-reason token list (one entry per choice).
pub fn record_gen_ai_response_attrs(
    span: &tracing::Span,
    response_model: Option<&str>,
    prompt_tokens: Option<u64>,
    completion_tokens: Option<u64>,
    total_tokens: Option<u64>,
    finish_reasons: &[&str],
    response_id: Option<&str>,
) {
    if let Some(m) = response_model {
        span.record("gen_ai.response.model", m);
    }
    if let Some(p) = prompt_tokens {
        span.record("gen_ai.usage.prompt_tokens", p);
    }
    if let Some(c) = completion_tokens {
        span.record("gen_ai.usage.completion_tokens", c);
    }
    if let Some(t) = total_tokens {
        span.record("gen_ai.usage.total_tokens", t);
    }
    if !finish_reasons.is_empty() {
        span.record(
            "gen_ai.response.finish_reasons",
            tracing::field::display(finish_reasons.join(",")),
        );
    }
    if let Some(id) = response_id {
        span.record("gen_ai.response.id", id);
    }
}

/// Echo a request ID onto a response (insert, never append: exactly one
/// `X-Request-Id`, the gateway's resolved one).
pub fn stamp_request_id(headers: &mut HeaderMap, request_id: &str) {
    if let Ok(v) = HeaderValue::from_str(request_id) {
        headers.insert(&X_REQUEST_ID, v);
    }
}

/// Resolve the request's business correlation ID (DW-093): a valid
/// inbound `X-Correlation-Id` verbatim, else the already-resolved
/// `request_id` (so every request carries a non-empty correlation
/// handle for the journey/funnel query even when the client did not
/// send one). The validity gate mirrors `valid_inbound_request_id`
/// (printable ASCII, at most 128 bytes) so hostile IDs cannot smuggle
/// control characters into the analytics store.
pub fn resolve_correlation_id(headers: &HeaderMap, request_id: &str) -> String {
    if let Some(v) = headers.get(&X_CORRELATION_ID) {
        if valid_inbound_request_id(v.as_bytes()) {
            if let Ok(s) = v.to_str() {
                return s.to_string();
            }
        }
    }
    request_id.to_string()
}

/// Echo the correlation ID onto a response (insert, never append:
/// exactly one `X-Correlation-Id`, the gateway's resolved one).
pub fn stamp_correlation_id(headers: &mut HeaderMap, correlation_id: &str) {
    if let Ok(v) = HeaderValue::from_str(correlation_id) {
        headers.insert(&X_CORRELATION_ID, v);
    }
}

// --- SLO & error budgets (DW-052) -----------------------------------------
//
// Per-route objectives (config: routes[].slo) exported as burn-rate
// metrics for multiwindow alerting. Availability counts a request BAD
// when the gateway answered 5xx (client errors are the caller's
// policy, not availability); latency counts a request BAD when the
// end-to-end duration exceeds the configured threshold. Burn rate is
// the standard error-budget consumption signal:
//
//   burn = (bad_fraction over window) / (1 - target)
//
// 1.0 = consuming budget at exactly the allowed rate (the budget lasts
// its whole period); 14.4 over 1h = the 28-day budget exhausted in
// ~49 hours (the classic fast-burn page); 6 over 1h the slow-burn
// signal. The exported windows are the alerting pair 5m and 1h.
//
// Values are computed from in-process sliding-window counters (15 s
// buckets) and published at SCRAPE time through a custom Collector —
// always fresh, never a stale sampled gauge, and no background task.
// The windows are process-local: they start empty at boot (documented
// — burn rates become meaningful as traffic accumulates, and a restart
// is itself an availability event the SLOs of OTHER processes catch).

/// Plain per-route SLO targets. Primitives on purpose: observability
/// depends on nothing, so the config types are converted at the call
/// site (dataplane refresh).
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct SloTargets {
    /// Availability target as a fraction in (0, 1]: the share of
    /// requests that must NOT be a gateway-answered 5xx.
    pub availability: f64,
    /// Latency objective threshold in milliseconds. `None` = no
    /// latency objective for this route.
    pub latency_threshold_ms: Option<f64>,
    /// Latency target as a fraction in (0, 1): the share of requests
    /// that must complete within `latency_threshold_ms`.
    pub latency_target: f64,
}

/// Sliding-window bucket width (15 s): the 5m window is 20 buckets,
/// the 1h window 240 — per route, trivially small, and granular
/// enough that scrape-time computation is never more than one bucket
/// stale.
const SLO_BUCKET_MS: i64 = 15_000;

/// The exported windows: (milliseconds, label). The standard
/// multiwindow-burn-rate alerting pair.
const SLO_WINDOWS: [(i64, &str); 2] = [(300_000, "5m"), (3_600_000, "1h")];

#[derive(Debug, Default, Clone, Copy)]
struct SloBucket {
    start_ms: i64,
    total: u64,
    errors: u64,
    over_latency: u64,
}

#[derive(Debug)]
struct SloRouteState {
    targets: SloTargets,
    /// One ring per [`SLO_WINDOWS`] entry; the ring at index i has
    /// `SLO_WINDOWS[i].0 / SLO_BUCKET_MS` buckets.
    rings: Vec<Vec<SloBucket>>,
}

impl SloRouteState {
    fn new(targets: SloTargets) -> Self {
        let rings = SLO_WINDOWS
            .iter()
            .map(|(ms, _)| {
                let n = (*ms / SLO_BUCKET_MS) as usize;
                vec![SloBucket::default(); n.max(1)]
            })
            .collect();
        SloRouteState { targets, rings }
    }

    /// Record one outcome into every ring (each ring buckets
    /// independently by its own wraparound index).
    fn record(&mut self, now_ms: i64, error: bool, over_latency: bool) {
        let idx = (now_ms / SLO_BUCKET_MS) as usize;
        let stamp = idx as i64 * SLO_BUCKET_MS;
        for ring in &mut self.rings {
            let slot = idx % ring.len();
            let b = &mut ring[slot];
            if b.start_ms != stamp {
                *b = SloBucket {
                    start_ms: stamp,
                    ..SloBucket::default()
                };
            }
            b.total += 1;
            if error {
                b.errors += 1;
            }
            if over_latency {
                b.over_latency += 1;
            }
        }
    }

    /// Sum a window ring over the buckets whose start is within the
    /// window ending at `now_ms` (whole buckets; a partially-expired
    /// edge bucket is included — the 15 s granularity is the
    /// documented precision).
    fn sums(&self, ring: usize, now_ms: i64) -> (u64, u64, u64) {
        let cutoff = now_ms - SLO_WINDOWS[ring].0;
        let mut total = 0u64;
        let mut errors = 0u64;
        let mut over = 0u64;
        for b in &self.rings[ring] {
            if b.start_ms > cutoff && b.total > 0 {
                total += b.total;
                errors += b.errors;
                over += b.over_latency;
            }
        }
        (total, errors, over)
    }
}

/// The SLO collector: owns the per-route window state and the two
/// output families, refreshing the gauges from the windows when
/// Prometheus gathers (scrape time — see the section docs).
struct SloState {
    routes: std::sync::RwLock<std::collections::HashMap<String, SloRouteState>>,
    /// PERF: `record` runs on EVERY completed request, but most fleets
    /// configure no route SLOs — the empty-map case still paid a
    /// `RwLock::write` (exclusive across every worker thread) per
    /// request. This flag gates that lock away entirely; it is
    /// maintained under the same write guard as the map swap, so
    /// `false` implies the map is empty and `record`'s `get_mut` would
    /// have missed anyway. A record racing a generation swap can slip
    /// one sample into the outgoing map or skip one in the incoming
    /// map — the same tolerance the pre-existing swap race already
    /// had, and far below one window's noise floor.
    configured: std::sync::atomic::AtomicBool,
    burn_rate: prometheus::GaugeVec,
    target: prometheus::GaugeVec,
}

impl SloState {
    fn new() -> Self {
        let burn_rate = prometheus::GaugeVec::new(
            prometheus::Opts::new(
                "dwara_slo_burn_rate",
                "Error-budget consumption rate per route SLO objective \
                 (DW-052): (bad fraction over the window) / (1 - target). \
                 1.0 consumes the budget at exactly the allowed rate; \
                 14.4 over 1h is the classic fast-burn page. window: 5m or \
                 1h (process-local sliding windows — they start empty at \
                 boot). objective: availability (bad = gateway-answered \
                 5xx) or latency (bad = duration over the configured \
                 threshold).",
            ),
            &["route", "objective", "window"],
        )
        .expect("valid metric definition");
        let target = prometheus::GaugeVec::new(
            prometheus::Opts::new(
                "dwara_slo_target",
                "The configured target per route SLO objective (DW-052), as \
                 a fraction: availability or latency-target share.",
            ),
            &["route", "objective"],
        )
        .expect("valid metric definition");
        SloState {
            routes: std::sync::RwLock::new(std::collections::HashMap::new()),
            configured: std::sync::atomic::AtomicBool::new(false),
            burn_rate,
            target,
        }
    }

    /// Swap the configured route set (dataplane refresh). Routes that
    /// vanish from the config lose their gauge children so a removed
    /// SLO stops exporting stale series.
    fn set_routes(&self, slos: Vec<(String, SloTargets)>) {
        let mut routes = self.routes.write().expect("slo state lock");
        let names: std::collections::HashSet<String> =
            slos.iter().map(|(n, _)| n.clone()).collect();
        for gone in routes.keys().filter(|n| !names.contains(*n)) {
            let name = gone.as_str();
            let objectives: &[&str] = if routes[gone].targets.latency_threshold_ms.is_some() {
                &["availability", "latency"]
            } else {
                &["availability"]
            };
            for objective in objectives {
                for window in ["5m", "1h"] {
                    let _ = self
                        .burn_rate
                        .remove_label_values(&[name, objective, window]);
                }
                let _ = self.target.remove_label_values(&[name, objective]);
            }
        }
        routes.clear();
        for (name, targets) in slos {
            routes.insert(name, SloRouteState::new(targets));
        }
        // Maintained under the write guard so a `record` in flight sees
        // either the fully swapped map with the flag set, or the gate
        // closed and no lock at all (see the field docs).
        self.configured
            .store(!routes.is_empty(), std::sync::atomic::Ordering::Release);
    }

    /// Record one completed request's outcome for a configured route
    /// (no-op for routes without an SLO — the common case).
    fn record(&self, route: &str, status: u16, duration_ms: f64, now_ms: i64) {
        if !self.configured.load(std::sync::atomic::Ordering::Acquire) {
            return;
        }
        if let Some(state) = self.routes.write().expect("slo state lock").get_mut(route) {
            let error = status >= 500;
            let over_latency = state
                .targets
                .latency_threshold_ms
                .is_some_and(|t| duration_ms > t);
            state.record(now_ms, error, over_latency);
        }
    }

    /// Refresh the gauge children from the current windows (scrape
    /// time). No traffic in a window => burn 0.0 (no consumption), by
    /// explicit decision: NaN would break alerting comparisons.
    fn refresh(&self, now_ms: i64) {
        let routes = self.routes.read().expect("slo state lock");
        for (route_name, state) in routes.iter() {
            let name = route_name.as_str();
            let allowed_avail = 1.0 - state.targets.availability;
            for (ring, (_, window)) in SLO_WINDOWS.iter().enumerate() {
                let (total, errors, over) = state.sums(ring, now_ms);
                let burn = if total > 0 && allowed_avail > 0.0 {
                    (errors as f64 / total as f64) / allowed_avail
                } else {
                    0.0
                };
                self.burn_rate
                    .with_label_values(&[name, "availability", window])
                    .set(burn);
                if state.targets.latency_threshold_ms.is_some() {
                    let allowed_lat = 1.0 - state.targets.latency_target;
                    let burn = if total > 0 && allowed_lat > 0.0 {
                        (over as f64 / total as f64) / allowed_lat
                    } else {
                        0.0
                    };
                    self.burn_rate
                        .with_label_values(&[name, "latency", window])
                        .set(burn);
                }
            }
            self.target
                .with_label_values(&[name, "availability"])
                .set(state.targets.availability);
            if state.targets.latency_threshold_ms.is_some() {
                self.target
                    .with_label_values(&[name, "latency"])
                    .set(state.targets.latency_target);
            }
        }
    }
}

impl prometheus::core::Collector for SloState {
    fn desc(&self) -> Vec<&prometheus::core::Desc> {
        let mut d = self.burn_rate.desc();
        d.extend(self.target.desc());
        d
    }

    fn collect(&self) -> Vec<prometheus::proto::MetricFamily> {
        let now = unix_now_ms();
        self.refresh(now);
        let mut families = self.burn_rate.collect();
        families.extend(self.target.collect());
        // A family with zero children (no route has an SLO — the
        // default) would make the text encoder PANIC on "no metrics";
        // emit nothing instead (an unconfigured feature exports no
        // series, exactly like a family that was never registered).
        families.retain(|f| !f.get_metric().is_empty());
        families
    }
}

/// Registration handle: shares the live [`SloState`] so the scrape
/// reads the same windows the completion path writes (a cloned
/// GaugeVec would share only the metric handles, not the state).
struct SloCollectorHandle(std::sync::Arc<SloState>);

impl prometheus::core::Collector for SloCollectorHandle {
    fn desc(&self) -> Vec<&prometheus::core::Desc> {
        prometheus::core::Collector::desc(&*self.0)
    }
    fn collect(&self) -> Vec<prometheus::proto::MetricFamily> {
        prometheus::core::Collector::collect(&*self.0)
    }
}

/// Wall-clock ms since the Unix epoch (the SLO windows' time domain).
fn unix_now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

impl Observability {
    /// DW-052: swap the configured per-route SLO set. Called by the
    /// dataplane on every generation refresh; routes without an `slo`
    /// block record nothing (the completion-path call is a no-op for
    /// them).
    pub fn set_route_slos(&self, slos: Vec<(String, SloTargets)>) {
        self.slo.set_routes(slos);
    }

    /// DW-052: record one completed request against its route's SLO
    /// windows (request-completion path; no-op when unconfigured).
    pub fn record_slo(&self, route: &str, status: u16, duration_ms: f64) {
        self.slo.record(route, status, duration_ms, unix_now_ms());
    }

    /// Deterministic-clock twin of [`Self::record_slo`] for the unit
    /// tests (window math is time-driven; the real clock cannot make
    /// "the 5m window expired" deterministic).
    #[doc(hidden)]
    pub fn record_slo_at(&self, route: &str, status: u16, duration_ms: f64, now_ms: i64) {
        self.slo.record(route, status, duration_ms, now_ms);
    }

    /// DW-052: scrape-time refresh + gather of the SLO families only
    /// (the full render is [`Self::render`]; this is the tests' window
    /// into the collector's output).
    #[doc(hidden)]
    pub fn render_slo(&self, now_ms: i64) -> String {
        use prometheus::core::Collector as _;
        self.slo.refresh(now_ms);
        let mut families = self.slo.burn_rate.collect();
        families.extend(self.slo.target.collect());
        families.retain(|f| !f.get_metric().is_empty());
        let mut buf = Vec::new();
        TextEncoder::new()
            .encode(&families, &mut buf)
            .expect("text encoding of gathered families cannot fail");
        String::from_utf8(buf).expect("prometheus text format is ASCII")
    }

    /// Count one L4 connection (DW-103) in
    /// `dwara_l4_connections_total{listener,protocol}`. `listener` is
    /// the config-declared listener name (config-bounded); `protocol`
    /// is one of the closed set `tcp`, `udp`. Counted once per accepted
    /// L4 connection (before the splice is established).
    pub fn record_l4_connection(&self, listener: &str, protocol: &str) {
        self.l4_connections_total
            .with_label_values(&[listener, protocol])
            .inc();
    }

    /// Record an L4 connection duration (DW-103) in
    /// `dwara_l4_connection_duration_seconds{listener}`. `listener` is
    /// the config-declared listener name (config-bounded label). Call
    /// once per connection (forwarded and closed alike).
    pub fn record_l4_connection_duration(&self, listener: &str, seconds: f64) {
        self.l4_connection_duration_seconds
            .with_label_values(&[listener])
            .observe(seconds);
    }
}