nautalid 0.1.0

//! Optional **AetherDB over WebTransport** — route LID wire frames (`command`, `dma`, `control`)
//! through HTTP/3 bidi streams when the deployment requires QUIC instead of native ZMQ/TCP.

use std::collections::HashSet;
use std::net::SocketAddr;
use std::sync::{Arc, OnceLock, RwLock, Weak};
use std::time::Duration;

use bytes::{BufMut, Bytes, BytesMut};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::sync::Semaphore;
use tracing::{debug, info, warn};

use crate::kwt_replay::runtime::AetherRuntime;

const MAGIC: &[u8; 4] = b"AWT1";

/// Hard ceiling for `AWT1` payload bytes (request or response body).
pub const MAX_AWT1_FRAME_PAYLOAD: usize = 64 * 1024 * 1024;

/// Default configured payload cap (override with `NAUTALID_WEBTRANSPORT_AETHER_MAX_FRAME_BYTES`).
const DEFAULT_AWT1_FRAME_PAYLOAD: usize = 4 * 1024 * 1024;

/// Maximum UTF-8 address field length inside an `AWT1` request.
pub const MAX_AWT1_ADDRESS_LEN: usize = 4096;

/// Maximum bus-token field length inside an `AWT1` request (`flags` bit `0x02`).
pub const MAX_AWT1_BUS_TOKEN_LEN: usize = 512;

const AWT1_FLAG_ADDRESS: u8 = 0x01;
const AWT1_FLAG_BUS_TOKEN: u8 = 0x02;
const AWT1_FLAG_REPLICATE_KEY: u8 = 0x04;

/// Fixed byte length of the replicate master key in an `AWT1` request (`flags` bit `0x04`).
pub const AWT1_REPLICATE_KEY_LEN: usize = 32;

const DEFAULT_MAX_BIDI_STREAMS: usize = 32;
const DEFAULT_FRAME_TIMEOUT_SECS: u64 = 30;

/// Per-session TLS identity established at QUIC connect time.
#[derive(Debug, Clone)]
pub struct WtAetherSessionAuth {
    pub peer_cert: crate::tls::PeerClientCert,
    pub peer_ip: Option<std::net::IpAddr>,
}

impl WtAetherSessionAuth {
    #[must_use]
    pub fn anonymous() -> Self {
        Self {
            peer_cert: crate::tls::PeerClientCert(Arc::new(Vec::new())),
            peer_ip: None,
        }
    }

    #[must_use]
    pub fn from_peer_cert(peer_cert: crate::tls::PeerClientCert) -> Self {
        Self {
            peer_cert,
            peer_ip: None,
        }
    }

    #[must_use]
    pub fn with_peer(peer_cert: crate::tls::PeerClientCert, peer_ip: std::net::IpAddr) -> Self {
        Self {
            peer_cert,
            peer_ip: Some(peer_ip),
        }
    }

    #[must_use]
    pub fn peer_authenticated(&self) -> bool {
        !self.peer_cert.is_empty()
    }

    /// Non-empty cert chain for unit tests (not a real PKI identity).
    #[doc(hidden)]
    #[must_use]
    pub fn authenticated() -> Self {
        use rustls::pki_types::CertificateDer;
        Self::from_peer_cert(crate::tls::PeerClientCert(Arc::new(vec![
            CertificateDer::from(Vec::from(b"wt-test-leaf-cert")),
        ])))
    }
}

/// Which AetherDB wire routes may traverse WebTransport.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum WtAetherRoute {
    Command = 0,
    Dma = 1,
    Control = 2,
    Replicate = 3,
}

impl WtAetherRoute {
    fn from_tag(tag: u8) -> Option<Self> {
        match tag {
            0 => Some(Self::Command),
            1 => Some(Self::Dma),
            2 => Some(Self::Control),
            3 => Some(Self::Replicate),
            _ => None,
        }
    }

    fn name(self) -> &'static str {
        match self {
            Self::Command => "command",
            Self::Dma => "dma",
            Self::Control => "control",
            Self::Replicate => "replicate",
        }
    }
}

/// Dispatch target for WT frames.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WtAetherBackend {
    /// In-process [`AetherRuntime`] (colocated LID).
    Local,
    /// Forward to a remote AetherDB wire bind (`tcp://…`).
    Wire { upstream: String },
}

/// Live WebTransport ↔ Aether routing policy (env + bus hot-reload).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WtAetherConfig {
    pub enabled: bool,
    /// WT CONNECT paths that terminate in the Aether bridge (e.g. `/wt/aether`, `/lid`).
    pub session_paths: Vec<String>,
    pub backend: WtAetherBackend,
    pub routes: HashSet<WtAetherRoute>,
    /// When true, `control` and `replicate` routes require a verified mTLS client certificate.
    pub require_mtls_for_privileged: bool,
    /// When true, `command` and `dma` routes require a verified mTLS client certificate.
    pub require_mtls_for_data: bool,
    /// When false, `replicate` is rejected even if listed in [`Self::routes`] (default off).
    pub allow_replicate_route: bool,
    /// When true, only leaf certs matching configured SHA-256 fingerprints may use WT routes.
    pub restrict_client_certs: bool,
    /// Global leaf-cert SHA-256 allowlist (lowercase hex); applies when tier list is empty.
    pub cert_sha256_allow: HashSet<String>,
    /// Leaf-cert SHA-256 allowlist for `control` / `replicate`.
    pub cert_sha256_allow_privileged: HashSet<String>,
    /// Leaf-cert SHA-256 allowlist for `command` / `dma`.
    pub cert_sha256_allow_data: HashSet<String>,
    /// Concurrent client bidi streams allowed per WebTransport session.
    pub max_bidi_streams: usize,
    /// Wall-clock limit for reading/dispatching one `AWT1` frame on a bidi stream.
    pub frame_timeout_secs: u64,
    /// Maximum `AWT1` payload bytes accepted on this deployment.
    pub max_frame_payload_bytes: usize,
    /// When true, apply L7 filter policy to WT aether session paths and frame dispatch (project opt-in).
    pub apply_http_kernel: bool,
}

impl Default for WtAetherConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            session_paths: vec!["/wt/aether".into()],
            backend: WtAetherBackend::Local,
            routes: default_wt_routes(),
            require_mtls_for_privileged: privileged_routes_require_mtls_from_env(),
            require_mtls_for_data: data_routes_require_mtls_from_env(),
            allow_replicate_route: allow_replicate_route_from_env(),
            restrict_client_certs: restrict_client_certs_from_env(),
            cert_sha256_allow: cert_sha256_allow_from_env("NAUTALID_WEBTRANSPORT_AETHER_CERT_SHA256"),
            cert_sha256_allow_privileged: cert_sha256_allow_from_env(
                "NAUTALID_WEBTRANSPORT_AETHER_CERT_SHA256_PRIVILEGED",
            ),
            cert_sha256_allow_data: cert_sha256_allow_from_env(
                "NAUTALID_WEBTRANSPORT_AETHER_CERT_SHA256_DATA",
            ),
            max_bidi_streams: max_bidi_streams_from_env(),
            frame_timeout_secs: frame_timeout_secs_from_env(),
            max_frame_payload_bytes: max_frame_payload_from_env(),
            apply_http_kernel: apply_http_kernel_from_env(),
        }
    }
}

impl WtAetherConfig {
    #[must_use]
    pub fn from_env() -> Self {
        let mut cfg = Self::default();
        if crate::surfaces::env_enabled("NAUTALID_WEBTRANSPORT_AETHER") {
            cfg.enabled = true;
        }
        if let Some(raw) = std::env::var("NAUTALID_WEBTRANSPORT_AETHER_PATH")
            .ok()
            .filter(|s| !s.trim().is_empty())
        {
            cfg.session_paths = parse_path_list(&raw);
        }
        if let Some(raw) = std::env::var("NAUTALID_WEBTRANSPORT_AETHER_ROUTES")
            .ok()
            .filter(|s| !s.trim().is_empty())
        {
            cfg.routes = parse_routes(&raw);
        }
        if let Some(upstream) = std::env::var("NAUTALID_WEBTRANSPORT_AETHER_UPSTREAM")
            .ok()
            .filter(|s| !s.trim().is_empty())
        {
            cfg.backend = WtAetherBackend::Wire {
                upstream: upstream.trim().to_string(),
            };
        } else if crate::surfaces::env_enabled("NAUTALID_WEBTRANSPORT_AETHER_WIRE") {
            let upstream = std::env::var("AETHERDB_HOST_BIND")
                .ok()
                .filter(|s| !s.trim().is_empty())
                .unwrap_or_else(|| "tcp://127.0.0.1:5556".into());
            cfg.backend = WtAetherBackend::Wire { upstream };
        }
        cfg.require_mtls_for_privileged = privileged_routes_require_mtls_from_env();
        cfg.require_mtls_for_data = data_routes_require_mtls_from_env();
        cfg.allow_replicate_route = allow_replicate_route_from_env();
        cfg.restrict_client_certs = restrict_client_certs_from_env();
        cfg.cert_sha256_allow =
            cert_sha256_allow_from_env("NAUTALID_WEBTRANSPORT_AETHER_CERT_SHA256");
        cfg.cert_sha256_allow_privileged =
            cert_sha256_allow_from_env("NAUTALID_WEBTRANSPORT_AETHER_CERT_SHA256_PRIVILEGED");
        cfg.cert_sha256_allow_data =
            cert_sha256_allow_from_env("NAUTALID_WEBTRANSPORT_AETHER_CERT_SHA256_DATA");
        cfg.max_bidi_streams = max_bidi_streams_from_env();
        cfg.frame_timeout_secs = frame_timeout_secs_from_env();
        cfg.max_frame_payload_bytes = max_frame_payload_from_env();
        cfg.apply_http_kernel = apply_http_kernel_from_env();
        cfg
    }

    pub fn apply_kdl(&mut self, node: &kdl::KdlNode) -> Result<(), String> {
        if let Some(enabled) = node.get("enable").and_then(|v| v.as_bool()) {
            self.enabled = enabled;
        }
        if let Some(paths) = node.get("path").and_then(|v| v.as_string()) {
            self.session_paths = parse_path_list(paths);
        }
        if let Some(paths) = node.get("paths").and_then(|v| v.as_string()) {
            self.session_paths = parse_path_list(paths);
        }
        if let Some(routes) = node.get("routes").and_then(|v| v.as_string()) {
            self.routes = parse_routes(routes);
        }
        if let Some(mode) = node.get("mode").and_then(|v| v.as_string()) {
            match mode.to_ascii_lowercase().as_str() {
                "local" | "colocated" | "embedded" => self.backend = WtAetherBackend::Local,
                "wire" | "remote" | "zmq" => {
                    let upstream = node
                        .get("upstream")
                        .and_then(|v| v.as_string())
                        .map(str::to_string)
                        .or_else(|| {
                            std::env::var("AETHERDB_HOST_BIND")
                                .ok()
                                .filter(|s| !s.trim().is_empty())
                        })
                        .unwrap_or_else(|| "tcp://127.0.0.1:5556".into());
                    self.backend = WtAetherBackend::Wire { upstream };
                }
                other => return Err(format!("unknown WT aether mode `{other}`")),
            }
        }
        if let Some(upstream) = node.get("upstream").and_then(|v| v.as_string()) {
            if !upstream.trim().is_empty() {
                self.backend = WtAetherBackend::Wire {
                    upstream: upstream.trim().to_string(),
                };
            }
        }
        if let Some(require) = node.get("require_mtls").and_then(|v| v.as_bool()) {
            self.require_mtls_for_privileged = require;
            self.require_mtls_for_data = require;
        }
        if let Some(require) = node.get("require_mtls_data").and_then(|v| v.as_bool()) {
            self.require_mtls_for_data = require;
        }
        if let Some(allow) = node.get("allow_replicate").and_then(|v| v.as_bool()) {
            self.allow_replicate_route = allow;
        }
        if let Some(restrict) = node.get("restrict_certs").and_then(|v| v.as_bool()) {
            self.restrict_client_certs = restrict;
        }
        if let Some(raw) = node.get("cert_sha256").and_then(|v| v.as_string()) {
            self.cert_sha256_allow = parse_cert_sha256_list(raw);
        }
        if let Some(raw) = node.get("cert_sha256_privileged").and_then(|v| v.as_string()) {
            self.cert_sha256_allow_privileged = parse_cert_sha256_list(raw);
        }
        if let Some(raw) = node.get("cert_sha256_data").and_then(|v| v.as_string()) {
            self.cert_sha256_allow_data = parse_cert_sha256_list(raw);
        }
        if let Some(max) = node.get("max_bidi").and_then(|v| v.as_integer()) {
            if max > 0 {
                self.max_bidi_streams = max as usize;
            }
        }
        if let Some(secs) = node.get("frame_timeout_secs").and_then(|v| v.as_integer()) {
            if secs > 0 {
                self.frame_timeout_secs = secs as u64;
            }
        }
        if let Some(max) = node.get("max_frame_bytes").and_then(|v| v.as_integer()) {
            if max > 0 {
                self.max_frame_payload_bytes =
                    (max as usize).clamp(1024, MAX_AWT1_FRAME_PAYLOAD);
            }
        }
        if let Some(apply) = node.get("apply_http_kernel").and_then(|v| v.as_bool()) {
            self.apply_http_kernel = apply;
        }
        Ok(())
    }

    #[must_use]
    pub fn effective_cert_allowlist(&self, route: WtAetherRoute) -> &HashSet<String> {
        let tier = match route {
            WtAetherRoute::Control | WtAetherRoute::Replicate => &self.cert_sha256_allow_privileged,
            WtAetherRoute::Command | WtAetherRoute::Dma => &self.cert_sha256_allow_data,
        };
        if !tier.is_empty() {
            return tier;
        }
        &self.cert_sha256_allow
    }

    #[must_use]
    pub fn status_kdl(&self) -> String {
        let routes: Vec<_> = [
            WtAetherRoute::Command,
            WtAetherRoute::Dma,
            WtAetherRoute::Control,
            WtAetherRoute::Replicate,
        ]
        .into_iter()
        .filter(|route| self.routes.contains(route))
        .map(WtAetherRoute::name)
        .collect();
        let backend = match &self.backend {
            WtAetherBackend::Local => "local".into(),
            WtAetherBackend::Wire { upstream } => format!("wire upstream={upstream}"),
        };
        format!(
            "enabled={} paths={:?} routes={routes:?} backend={backend}",
            self.enabled, self.session_paths
        )
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WtSessionKind {
    Demo,
    Aether,
}

struct WtAetherState {
    config: WtAetherConfig,
    runtime: Weak<AetherRuntime>,
    #[cfg(feature = "filter")]
    filter: Option<crate::filter::FilterStore>,
}

static WT_AETHER: OnceLock<Arc<RwLock<WtAetherState>>> = OnceLock::new();

/// Install or refresh the global WT ↔ Aether registry (called at startup).
pub fn install(runtime: Arc<AetherRuntime>, config: WtAetherConfig) {
    #[cfg(feature = "filter")]
    install_with_filter(runtime, config, None);
    #[cfg(not(feature = "filter"))]
    {
        let state = WtAetherState {
            config,
            runtime: Arc::downgrade(&runtime),
        };
        if let Some(slot) = WT_AETHER.get() {
            let mut guard = slot.write().expect("wt aether lock poisoned");
            guard.config = state.config;
            guard.runtime = state.runtime;
        } else {
            let _ = WT_AETHER.set(Arc::new(RwLock::new(state)));
        }
        if config_snapshot().enabled {
            info!(
                paths = ?config_snapshot().session_paths,
                "WebTransport Aether bridge enabled"
            );
        }
    }
}

/// Install with optional L7 filter store for [`WtAetherConfig::apply_http_kernel`].
#[cfg(feature = "filter")]
pub fn install_with_filter(
    runtime: Arc<AetherRuntime>,
    config: WtAetherConfig,
    filter: Option<crate::filter::FilterStore>,
) {
    let state = WtAetherState {
        config,
        runtime: Arc::downgrade(&runtime),
        filter,
    };
    if let Some(slot) = WT_AETHER.get() {
        let mut guard = slot.write().expect("wt aether lock poisoned");
        guard.config = state.config;
        guard.runtime = state.runtime;
        guard.filter = state.filter;
    } else {
        let _ = WT_AETHER.set(Arc::new(RwLock::new(state)));
    }
    if config_snapshot().enabled {
        info!(
            paths = ?config_snapshot().session_paths,
            "WebTransport Aether bridge enabled"
        );
    }
}

fn apply_http_kernel_from_env() -> bool {
    crate::surfaces::env_enabled("NAUTALID_WEBTRANSPORT_AETHER_APPLY_HTTP_KERNEL")
}

#[cfg(feature = "filter")]
fn filter_snapshot() -> Option<crate::filter::FilterStore> {
    WT_AETHER.get().and_then(|slot| {
        slot.read()
            .ok()
            .and_then(|guard| guard.filter.clone())
    })
}

#[cfg(feature = "filter")]
fn wt_http_kernel_denies(peer_ip: Option<std::net::IpAddr>, path: &str) -> bool {
    if !config_snapshot().apply_http_kernel {
        return false;
    }
    let Some(filter) = filter_snapshot() else {
        return false;
    };
    !filter.read(|policy| crate::filter::policy_allows(policy, peer_ip, path))
}

#[cfg(feature = "filter")]
fn wt_frame_path(route: WtAetherRoute) -> String {
    format!("/wt/aether/{}", route.name())
}

#[must_use]
pub fn config_snapshot() -> WtAetherConfig {
    WT_AETHER
        .get()
        .map(|slot| slot.read().expect("wt aether lock poisoned").config.clone())
        .unwrap_or_else(WtAetherConfig::from_env)
}

pub fn update_config(mutator: impl FnOnce(&mut WtAetherConfig)) {
    let Some(slot) = WT_AETHER.get() else {
        return;
    };
    let mut guard = slot.write().expect("wt aether lock poisoned");
    mutator(&mut guard.config);
}

/// Optional dedicated UDP listener for WT-only HTTP/3 (separate from `PORT` / `NAUTALID_H3_PORT`).
#[must_use]
pub fn dedicated_listen_from_env() -> Option<SocketAddr> {
    std::env::var("NAUTALID_WEBTRANSPORT_LISTEN")
        .ok()
        .filter(|s| !s.trim().is_empty())
        .and_then(|raw| raw.trim().parse().ok())
}

/// Classify a WT CONNECT path.
#[must_use]
pub fn session_kind_for_path(path: &str) -> Option<WtSessionKind> {
    if !crate::webtransport::is_enabled() {
        return None;
    }
    let cfg = config_snapshot();
    if cfg.enabled && cfg.session_paths.iter().any(|p| p == path) {
        return Some(WtSessionKind::Aether);
    }
    if path == crate::webtransport::path_from_env() {
        return Some(WtSessionKind::Demo);
    }
    None
}

/// All WT CONNECT paths currently advertised (demo + aether when enabled).
#[must_use]
pub fn all_session_paths() -> Vec<String> {
    let mut paths = vec![crate::webtransport::path_from_env()];
    let cfg = config_snapshot();
    if cfg.enabled {
        for path in &cfg.session_paths {
            if !paths.iter().any(|p| p == path) {
                paths.push(path.clone());
            }
        }
    }
    paths
}

#[derive(Clone)]
struct WtSessionLimits {
    bidi: Arc<Semaphore>,
    frame_timeout: Duration,
    max_frame_payload: usize,
}

impl WtSessionLimits {
    fn from_config(cfg: &WtAetherConfig) -> Self {
        Self {
            bidi: Arc::new(Semaphore::new(cfg.max_bidi_streams.max(1))),
            frame_timeout: Duration::from_secs(cfg.frame_timeout_secs.max(1)),
            max_frame_payload: cfg.max_frame_payload_bytes,
        }
    }
}

/// Drive an Aether-only WebTransport session (bidi streams carry `AWT1` frames).
pub async fn drive_aether_session(
    session: crate::webtransport::WtSession,
    auth: WtAetherSessionAuth,
) {
    let session_id = session.session_id();
    let limits = WtSessionLimits::from_config(&config_snapshot());
    info!(?session_id, peer_authenticated = auth.peer_authenticated(), "webtransport aether session established");
    loop {
        match session.accept_bi().await {
            Ok(Some(h3_webtransport::server::AcceptedBi::BidiStream(_sid, mut stream))) => {
                let auth = auth.clone();
                let limits = limits.clone();
                tokio::spawn(async move {
                    let Some(_permit) =
                        crate::webtransport::acquire_bidi_permit(limits.bidi.clone()).await
                    else {
                        warn!(?session_id, "wt aether session bidi stream limit reached");
                        let _ = stream.shutdown().await;
                        return;
                    };
                    bridge_bidi_stream(stream, auth, limits.frame_timeout, limits.max_frame_payload)
                        .await;
                });
            }
            Ok(Some(h3_webtransport::server::AcceptedBi::Request(req, mut stream))) => {
                if let Err(err) = serve_aether_nested_request(req, &mut stream, auth.clone()).await {
                    warn!(%err, "webtransport aether nested request failed");
                }
            }
            Ok(None) => {
                debug!(?session_id, "webtransport aether session closed");
                break;
            }
            Err(err) => {
                warn!(?session_id, %err, "webtransport aether session error");
                break;
            }
        }
    }
}

async fn serve_aether_nested_request(
    req: http::Request<()>,
    stream: &mut crate::webtransport::H3RequestStream,
    auth: WtAetherSessionAuth,
) -> Result<(), h3::error::StreamError> {
    let cfg = config_snapshot();
    #[cfg(feature = "filter")]
    if wt_http_kernel_denies(auth.peer_ip, req.uri().path()) {
        let response = http::Response::builder()
            .status(http::StatusCode::FORBIDDEN)
            .header("content-type", "application/json")
            .body(())
            .expect("response");
        stream.send_response(response).await?;
        stream.send_data(Bytes::from(r#"{"error":"denied by filter"}"#))
            .await?;
        return stream.finish().await;
    }
    let (status, body) = aether_nested_http_response(auth, &cfg, req.method().as_str(), req.uri().path());
    let response = http::Response::builder()
        .status(status)
        .header("content-type", "application/json")
        .body(())
        .expect("response");
    stream.send_response(response).await?;
    stream.send_data(Bytes::from(body.to_string())).await?;
    stream.finish().await
}

/// JSON + status for nested HTTP requests inside an Aether WT session (testable policy surface).
#[must_use]
pub fn aether_nested_http_response(
    auth: WtAetherSessionAuth,
    cfg: &WtAetherConfig,
    method: &str,
    path: &str,
) -> (http::StatusCode, serde_json::Value) {
    let status = if cfg.require_mtls_for_privileged && !auth.peer_authenticated() {
        http::StatusCode::FORBIDDEN
    } else {
        http::StatusCode::OK
    };
    let _ = (method, path);
    let body = if status == http::StatusCode::FORBIDDEN {
        serde_json::json!({ "error": "mTLS client certificate required" })
    } else {
        serde_json::json!({ "ok": true })
    };
    (status, body)
}

async fn bridge_bidi_stream(
    mut stream: h3_webtransport::stream::BidiStream<h3_quinn::BidiStream<Bytes>, Bytes>,
    auth: WtAetherSessionAuth,
    frame_timeout: Duration,
    max_frame_payload: usize,
) {
    loop {
        let frame = match tokio::time::timeout(
            frame_timeout,
            read_frame(&mut stream, max_frame_payload),
        )
        .await
        {
            Ok(Ok(frame)) => frame,
            Ok(Err(err)) => {
                debug!(%err, "wt aether frame read ended");
                break;
            }
            Err(_) => {
                debug!("wt aether frame read timed out");
                break;
            }
        };
        let response = match tokio::time::timeout(
            frame_timeout,
            dispatch_frame(frame, &auth),
        )
        .await
        {
            Ok(Ok(bytes)) => bytes,
            Ok(Err(err)) => encode_error_frame(&err),
            Err(_) => encode_error_frame("wt aether frame dispatch timed out"),
        };
        if write_bytes(&mut stream, &response).await.is_err() {
            break;
        }
    }
    let _ = stream.shutdown().await;
}

#[derive(Debug)]
struct WtFrame {
    route: WtAetherRoute,
    bus_token: Option<String>,
    address: Option<String>,
    replicate_key: Option<Vec<u8>>,
    payload: Bytes,
}

/// Credentials carried inside an `AWT1` request frame.
#[derive(Debug, Clone, Copy, Default)]
pub struct WtRouteAuth<'a> {
    pub bus_token: Option<&'a str>,
    pub replicate_key: Option<&'a [u8]>,
}

/// Integration-test entry: parse one AWT1 request blob and return an encoded AWT1 response.
pub async fn dispatch_test_awt1_request(
    request: Bytes,
    auth: WtAetherSessionAuth,
) -> Result<Bytes, String> {
    let frame = decode_request_frame(&request, config_snapshot().max_frame_payload_bytes)?;
    let timeout = Duration::from_secs(config_snapshot().frame_timeout_secs.max(1));
    match tokio::time::timeout(timeout, dispatch_frame(frame, &auth)).await {
        Ok(Ok(bytes)) => Ok(bytes),
        Ok(Err(err)) => Ok(encode_error_frame(&err)),
        Err(_) => Ok(encode_error_frame("wt aether frame dispatch timed out")),
    }
}

fn decode_request_frame(bytes: &[u8], max_payload: usize) -> Result<WtFrame, String> {
    if bytes.len() < 10 {
        return Err("AWT1 request too short".into());
    }
    if &bytes[..4] != MAGIC {
        return Err("expected AWT1 frame magic".into());
    }
    let route = WtAetherRoute::from_tag(bytes[4])
        .ok_or_else(|| format!("unknown route tag {}", bytes[4]))?;
    let flags = bytes[5];
    let mut offset = 6usize;
    let bus_token = if flags & AWT1_FLAG_BUS_TOKEN != 0 {
        if bytes.len() < offset + 2 {
            return Err("truncated AWT1 bus token length".into());
        }
        let len = u16::from_be_bytes(bytes[offset..offset + 2].try_into().expect("len")) as usize;
        offset += 2;
        if len > MAX_AWT1_BUS_TOKEN_LEN {
            return Err(format!("AWT1 bus token longer than {MAX_AWT1_BUS_TOKEN_LEN} bytes"));
        }
        if bytes.len() < offset + len {
            return Err("truncated AWT1 bus token".into());
        }
        let token = String::from_utf8(bytes[offset..offset + len].to_vec())
            .map_err(|_| "bus token not utf-8")?;
        offset += len;
        Some(token)
    } else {
        None
    };
    let address = if flags & AWT1_FLAG_ADDRESS != 0 {
        if bytes.len() < offset + 2 {
            return Err("truncated AWT1 address length".into());
        }
        let len = u16::from_be_bytes(bytes[offset..offset + 2].try_into().expect("len")) as usize;
        offset += 2;
        if len > MAX_AWT1_ADDRESS_LEN {
            return Err(format!("AWT1 address longer than {MAX_AWT1_ADDRESS_LEN} bytes"));
        }
        if bytes.len() < offset + len {
            return Err("truncated AWT1 address".into());
        }
        let addr = String::from_utf8(bytes[offset..offset + len].to_vec())
            .map_err(|_| "address not utf-8")?;
        offset += len;
        Some(addr)
    } else {
        None
    };
    let replicate_key = if flags & AWT1_FLAG_REPLICATE_KEY != 0 {
        if bytes.len() < offset + AWT1_REPLICATE_KEY_LEN {
            return Err("truncated AWT1 replicate key".into());
        }
        let key = bytes[offset..offset + AWT1_REPLICATE_KEY_LEN].to_vec();
        offset += AWT1_REPLICATE_KEY_LEN;
        Some(key)
    } else {
        None
    };
    if bytes.len() < offset + 4 {
        return Err("truncated AWT1 payload length".into());
    }
    let len = u32::from_be_bytes(bytes[offset..offset + 4].try_into().expect("len")) as usize;
    offset += 4;
    if len > max_payload {
        return Err(format!("frame payload larger than {max_payload} bytes"));
    }
    if bytes.len() < offset + len {
        return Err("truncated AWT1 payload".into());
    }
    Ok(WtFrame {
        route,
        bus_token,
        address,
        replicate_key,
        payload: Bytes::copy_from_slice(&bytes[offset..offset + len]),
    })
}

async fn read_frame<R: AsyncRead + Unpin>(
    reader: &mut R,
    max_payload: usize,
) -> Result<WtFrame, String> {
    let mut magic = [0u8; 4];
    read_exact(reader, &mut magic).await?;
    if &magic != MAGIC {
        return Err("expected AWT1 frame magic".into());
    }
    let mut route_buf = [0u8; 1];
    read_exact(reader, &mut route_buf).await?;
    let route = WtAetherRoute::from_tag(route_buf[0])
        .ok_or_else(|| format!("unknown route tag {}", route_buf[0]))?;
    let mut flags = [0u8; 1];
    read_exact(reader, &mut flags).await?;
    let bus_token = if flags[0] & AWT1_FLAG_BUS_TOKEN != 0 {
        let mut len_buf = [0u8; 2];
        read_exact(reader, &mut len_buf).await?;
        let len = u16::from_be_bytes(len_buf) as usize;
        if len > MAX_AWT1_BUS_TOKEN_LEN {
            return Err(format!("AWT1 bus token longer than {MAX_AWT1_BUS_TOKEN_LEN} bytes"));
        }
        let mut token = vec![0u8; len];
        read_exact(reader, &mut token).await?;
        Some(String::from_utf8(token).map_err(|_| "bus token not utf-8")?)
    } else {
        None
    };
    let address = if flags[0] & AWT1_FLAG_ADDRESS != 0 {
        let mut len_buf = [0u8; 2];
        read_exact(reader, &mut len_buf).await?;
        let len = u16::from_be_bytes(len_buf) as usize;
        if len > MAX_AWT1_ADDRESS_LEN {
            return Err(format!("AWT1 address longer than {MAX_AWT1_ADDRESS_LEN} bytes"));
        }
        let mut addr = vec![0u8; len];
        read_exact(reader, &mut addr).await?;
        Some(String::from_utf8(addr).map_err(|_| "address not utf-8")?)
    } else {
        None
    };
    let replicate_key = if flags[0] & AWT1_FLAG_REPLICATE_KEY != 0 {
        let mut key = vec![0u8; AWT1_REPLICATE_KEY_LEN];
        read_exact(reader, &mut key).await?;
        Some(key)
    } else {
        None
    };
    let mut len_buf = [0u8; 4];
    read_exact(reader, &mut len_buf).await?;
    let len = u32::from_be_bytes(len_buf) as usize;
    if len > max_payload {
        return Err(format!("frame payload larger than {max_payload} bytes"));
    }
    let mut payload = vec![0u8; len];
    read_exact(reader, &mut payload).await?;
    Ok(WtFrame {
        route,
        bus_token,
        address,
        replicate_key,
        payload: Bytes::from(payload),
    })
}

async fn read_exact<R: AsyncRead + Unpin>(reader: &mut R, buf: &mut [u8]) -> Result<(), String> {
    reader
        .read_exact(buf)
        .await
        .map(|_| ())
        .map_err(|err| err.to_string())
}

async fn write_bytes<W: AsyncWrite + Unpin>(writer: &mut W, bytes: &[u8]) -> Result<(), String> {
    writer
        .write_all(bytes)
        .await
        .map_err(|err| err.to_string())
}

fn encode_response_frame(route: WtAetherRoute, payload: Bytes) -> Bytes {
    let mut out = BytesMut::with_capacity(10 + payload.len());
    out.extend_from_slice(MAGIC);
    out.put_u8(route as u8);
    out.put_u8(0); // ok
    out.put_u32(payload.len() as u32);
    out.extend_from_slice(&payload);
    out.freeze()
}

/// Encode an `AWT1` request frame (integration tests and tooling).
#[doc(hidden)]
pub fn encode_request_frame(
    route: WtAetherRoute,
    address: Option<&str>,
    payload: Bytes,
) -> Result<Bytes, String> {
    encode_request_frame_with_options(route, address, None, None, payload)
}

/// Encode an `AWT1` request frame with optional bus token and replicate key proof.
#[doc(hidden)]
pub fn encode_request_frame_with_options(
    route: WtAetherRoute,
    address: Option<&str>,
    bus_token: Option<&str>,
    replicate_key: Option<&[u8]>,
    payload: Bytes,
) -> Result<Bytes, String> {
    if payload.len() > MAX_AWT1_FRAME_PAYLOAD {
        return Err(format!("frame payload larger than {MAX_AWT1_FRAME_PAYLOAD} bytes"));
    }
    let mut out = BytesMut::with_capacity(12 + payload.len());
    out.extend_from_slice(MAGIC);
    out.put_u8(route as u8);
    let mut flags = 0u8;
    if bus_token.is_some() {
        flags |= AWT1_FLAG_BUS_TOKEN;
    }
    if address.is_some() {
        flags |= AWT1_FLAG_ADDRESS;
    }
    if replicate_key.is_some() {
        flags |= AWT1_FLAG_REPLICATE_KEY;
    }
    out.put_u8(flags);
    if let Some(token) = bus_token {
        let bytes = token.as_bytes();
        if bytes.len() > MAX_AWT1_BUS_TOKEN_LEN {
            return Err(format!("bus token longer than {MAX_AWT1_BUS_TOKEN_LEN} bytes"));
        }
        let len = u16::try_from(bytes.len())
            .map_err(|_| format!("bus token longer than {MAX_AWT1_BUS_TOKEN_LEN} bytes"))?;
        out.put_u16(len);
        out.extend_from_slice(bytes);
    }
    if let Some(addr) = address {
        let bytes = addr.as_bytes();
        if bytes.len() > MAX_AWT1_ADDRESS_LEN {
            return Err(format!("address longer than {MAX_AWT1_ADDRESS_LEN} bytes"));
        }
        let len = u16::try_from(bytes.len())
            .map_err(|_| format!("address longer than {MAX_AWT1_ADDRESS_LEN} bytes"))?;
        out.put_u16(len);
        out.extend_from_slice(bytes);
    }
    if let Some(key) = replicate_key {
        if key.len() != AWT1_REPLICATE_KEY_LEN {
            return Err(format!(
                "replicate key must be exactly {AWT1_REPLICATE_KEY_LEN} bytes"
            ));
        }
        out.extend_from_slice(key);
    }
    out.put_u32(payload.len() as u32);
    out.extend_from_slice(&payload);
    Ok(out.freeze())
}

/// Decode an `AWT1` ok response frame (`flags == 0`).
#[doc(hidden)]
pub fn decode_ok_response_frame(bytes: &[u8]) -> Result<(WtAetherRoute, Bytes), String> {
    if bytes.len() < 10 {
        return Err("AWT1 response too short".into());
    }
    if &bytes[..4] != MAGIC {
        return Err("expected AWT1 frame magic".into());
    }
    let flags = bytes[5];
    if flags != 0 {
        let len = u32::from_be_bytes(bytes[6..10].try_into().expect("len")) as usize;
        if len > MAX_AWT1_FRAME_PAYLOAD {
            return Err(format!("AWT1 error payload larger than {MAX_AWT1_FRAME_PAYLOAD} bytes"));
        }
        let end = 10 + len;
        if bytes.len() < end {
            return Err("truncated AWT1 error payload".into());
        }
        let msg = std::str::from_utf8(&bytes[10..end]).unwrap_or("wt aether error");
        return Err(msg.to_string());
    }
    let route = WtAetherRoute::from_tag(bytes[4]).ok_or_else(|| "unknown route tag".to_string())?;
    let len = u32::from_be_bytes(bytes[6..10].try_into().expect("len")) as usize;
    if len > MAX_AWT1_FRAME_PAYLOAD {
        return Err(format!("AWT1 response payload larger than {MAX_AWT1_FRAME_PAYLOAD} bytes"));
    }
    let end = 10 + len;
    if bytes.len() < end {
        return Err("truncated AWT1 response payload".into());
    }
    Ok((route, Bytes::copy_from_slice(&bytes[10..end])))
}

fn encode_error_frame(message: &str) -> Bytes {
    let mut out = BytesMut::with_capacity(10 + message.len());
    out.extend_from_slice(MAGIC);
    out.put_u8(255);
    out.put_u8(1); // err
    out.put_u32(message.len() as u32);
    out.extend_from_slice(message.as_bytes());
    out.freeze()
}

async fn dispatch_frame(frame: WtFrame, auth: &WtAetherSessionAuth) -> Result<Bytes, String> {
    let cfg = config_snapshot();
    if !cfg.enabled {
        return Err("WT aether bridge disabled".into());
    }
    #[cfg(feature = "filter")]
    if wt_http_kernel_denies(auth.peer_ip, &wt_frame_path(frame.route)) {
        return Err("WT aether frame denied by filter".into());
    }
    if !cfg.routes.contains(&frame.route) {
        return Err(format!("route `{}` not enabled on WT bridge", frame.route.name()));
    }
    authorize_wt_route(
        frame.route,
        auth,
        &cfg,
        WtRouteAuth {
            bus_token: frame.bus_token.as_deref(),
            replicate_key: frame.replicate_key.as_deref(),
        },
    )?;
    match cfg.backend {
        WtAetherBackend::Local => dispatch_local(frame).await,
        WtAetherBackend::Wire { ref upstream } => {
            dispatch_wire(upstream.clone(), frame).await
        }
    }
}

/// Enforce mTLS, route policy, bus token (`control`), and replicate key proof (`replicate`).
pub fn authorize_wt_route(
    route: WtAetherRoute,
    auth: &WtAetherSessionAuth,
    cfg: &WtAetherConfig,
    creds: WtRouteAuth<'_>,
) -> Result<(), String> {
    if route == WtAetherRoute::Replicate && !cfg.allow_replicate_route {
        return Err(
            "WT replicate route disabled (set NAUTALID_WEBTRANSPORT_AETHER_ALLOW_REPLICATE=1)"
                .into(),
        );
    }
    if cfg.require_mtls_for_privileged
        && matches!(route, WtAetherRoute::Control | WtAetherRoute::Replicate)
        && !auth.peer_authenticated()
    {
        return Err("WT control/replicate require mTLS client certificate".into());
    }
    if cfg.require_mtls_for_data
        && matches!(route, WtAetherRoute::Command | WtAetherRoute::Dma)
        && !auth.peer_authenticated()
    {
        return Err("WT command/dma require mTLS client certificate".into());
    }
    verify_client_cert_for_route(route, auth, cfg)?;
    if route == WtAetherRoute::Control {
        verify_control_bus_token(creds.bus_token)?;
    }
    if route == WtAetherRoute::Replicate {
        verify_replicate_key(creds.replicate_key)?;
    }
    Ok(())
}

fn verify_control_bus_token(provided: Option<&str>) -> Result<(), String> {
    let token_required = crate::security::wt_control_token_required();
    let expected = crate::security::bus_token();
    if !token_required && expected.is_none() {
        return Ok(());
    }
    let Some(expected) = expected else {
        return Err(
            "WT control requires bus token (configure security.bus_token, NAUTALID_BUS_TOKEN, or NAUTALID_BUS_TOKEN_FILE)"
                .into(),
        );
    };
    let Some(provided) = provided else {
        return Err("WT control requires bus token in AWT1 frame".into());
    };
    if !constant_time_eq(expected.as_bytes(), provided.as_bytes()) {
        return Err("invalid bus token".into());
    }
    Ok(())
}

fn verify_replicate_key(provided: Option<&[u8]>) -> Result<(), String> {
    let Some(expected) = replicate_key_from_env() else {
        return Ok(());
    };
    let Some(provided) = provided else {
        return Err("WT replicate requires replicate key in AWT1 frame".into());
    };
    if provided.len() != expected.len() {
        return Err("invalid replicate key proof".into());
    }
    if !constant_time_eq(provided, &expected) {
        return Err("invalid replicate key proof".into());
    }
    Ok(())
}

fn replicate_key_from_env() -> Option<Vec<u8>> {
    std::env::var("AETHERDB_HOST_REPLICATE_KEY_HEX")
        .ok()
        .or_else(|| std::env::var("AETHERDB_HUB_REPLICATE_KEY_HEX").ok())
        .filter(|s| !s.trim().is_empty())
        .and_then(|hex| aetherdb::decode_master_key_hex(hex.trim()).ok().map(|k| k.to_vec()))
}

fn allow_replicate_route_from_env() -> bool {
    crate::surfaces::env_enabled("NAUTALID_WEBTRANSPORT_AETHER_ALLOW_REPLICATE")
}

fn restrict_client_certs_from_env() -> bool {
    match std::env::var("NAUTALID_WEBTRANSPORT_AETHER_RESTRICT_CERTS")
        .ok()
        .map(|v| v.trim().to_ascii_lowercase())
        .as_deref()
    {
        Some("0") | Some("false") | Some("no") | Some("off") => false,
        Some("1") | Some("true") | Some("yes") | Some("on") => true,
        _ => crate::tls::client_auth_required_from_env(),
    }
}

fn cert_sha256_allow_from_env(var: &str) -> HashSet<String> {
    std::env::var(var)
        .ok()
        .filter(|s| !s.trim().is_empty())
        .map(|s| parse_cert_sha256_list(&s))
        .unwrap_or_default()
}

fn parse_cert_sha256_list(raw: &str) -> HashSet<String> {
    raw.split(',')
        .flat_map(normalize_cert_sha256)
        .collect()
}

fn normalize_cert_sha256(entry: &str) -> Option<String> {
    let trimmed = entry.trim();
    if trimmed.is_empty() {
        return None;
    }
    let hex = trimmed
        .strip_prefix("sha256:")
        .or_else(|| trimmed.strip_prefix("SHA256:"))
        .unwrap_or(trimmed)
        .replace(':', "");
    if hex.len() != 64 || !hex.chars().all(|c| c.is_ascii_hexdigit()) {
        warn!(fingerprint = trimmed, "ignoring invalid WT cert SHA-256 allowlist entry");
        return None;
    }
    Some(hex.to_ascii_lowercase())
}

fn verify_client_cert_for_route(
    route: WtAetherRoute,
    auth: &WtAetherSessionAuth,
    cfg: &WtAetherConfig,
) -> Result<(), String> {
    if !cfg.restrict_client_certs {
        return Ok(());
    }
    let needs_cert = (cfg.require_mtls_for_privileged
        && matches!(route, WtAetherRoute::Control | WtAetherRoute::Replicate))
        || (cfg.require_mtls_for_data
            && matches!(route, WtAetherRoute::Command | WtAetherRoute::Dma));
    if !needs_cert {
        return Ok(());
    }
    let allowlist = cfg.effective_cert_allowlist(route);
    if allowlist.is_empty() {
        return Err(
            "WT client certificate allowlist is empty; set NAUTALID_WEBTRANSPORT_AETHER_CERT_SHA256_*"
                .into(),
        );
    }
    let Some(fp) = auth.peer_cert.leaf_sha256_hex() else {
        return Err("WT route requires client certificate".into());
    };
    if !allowlist.contains(&fp) {
        return Err("client certificate not permitted for this WT route".into());
    }
    Ok(())
}

fn default_wt_routes() -> HashSet<WtAetherRoute> {
    let mut routes = HashSet::from([
        WtAetherRoute::Command,
        WtAetherRoute::Dma,
        WtAetherRoute::Control,
    ]);
    if allow_replicate_route_from_env() {
        routes.insert(WtAetherRoute::Replicate);
    }
    routes
}

fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

fn privileged_routes_require_mtls_from_env() -> bool {
    match std::env::var("NAUTALID_WEBTRANSPORT_AETHER_REQUIRE_MTLS")
        .ok()
        .map(|v| v.trim().to_ascii_lowercase())
        .as_deref()
    {
        Some("0") | Some("false") | Some("no") | Some("off") => false,
        _ => true,
    }
}

fn data_routes_require_mtls_from_env() -> bool {
    match std::env::var("NAUTALID_WEBTRANSPORT_AETHER_REQUIRE_MTLS_DATA")
        .ok()
        .map(|v| v.trim().to_ascii_lowercase())
        .as_deref()
    {
        Some("0") | Some("false") | Some("no") | Some("off") => false,
        Some("1") | Some("true") | Some("yes") | Some("on") => true,
        _ => crate::tls::client_auth_required_from_env(),
    }
}

fn max_bidi_streams_from_env() -> usize {
    std::env::var("NAUTALID_WEBTRANSPORT_AETHER_MAX_BIDI")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(DEFAULT_MAX_BIDI_STREAMS)
        .max(1)
}

fn frame_timeout_secs_from_env() -> u64 {
    std::env::var("NAUTALID_WEBTRANSPORT_AETHER_FRAME_TIMEOUT_SECS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(DEFAULT_FRAME_TIMEOUT_SECS)
        .max(1)
}

fn max_frame_payload_from_env() -> usize {
    std::env::var("NAUTALID_WEBTRANSPORT_AETHER_MAX_FRAME_BYTES")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(DEFAULT_AWT1_FRAME_PAYLOAD)
        .clamp(1024, MAX_AWT1_FRAME_PAYLOAD)
}

async fn dispatch_local(frame: WtFrame) -> Result<Bytes, String> {
    let runtime = WT_AETHER
        .get()
        .and_then(|slot| {
            slot.read()
                .ok()
                .and_then(|guard| guard.runtime.upgrade())
        })
        .ok_or_else(|| "Aether runtime not available".to_string())?;
    let route = frame.route;
    let address = frame.address;
    let body = frame.payload;
    let payload = runtime.dispatch_wt_wire(route as u8, address.as_deref(), body)?;
    Ok(encode_response_frame(route, payload))
}

async fn dispatch_wire(upstream: String, frame: WtFrame) -> Result<Bytes, String> {
    if frame.route == WtAetherRoute::Control {
        return Err(
            "WT control route is not forwarded over a wire backend; use local aether backend"
                .into(),
        );
    }
    let route = frame.route;
    let payload = frame.payload;
    let response = wire_request(&upstream, route, payload)?;
    Ok(encode_response_frame(route, response))
}

fn wire_request(
    upstream: &str,
    route: WtAetherRoute,
    payload: Bytes,
) -> Result<Bytes, String> {
    let runtime = WT_AETHER
        .get()
        .and_then(|slot| {
            slot.read()
                .ok()
                .and_then(|guard| guard.runtime.upgrade())
        })
        .ok_or_else(|| "Aether runtime not available".to_string())?;

    let tls = crate::aether_tunnel::wire_client_zmq_tls_for_endpoint(upstream)
        .map_err(|err| err.to_string())?;

    match route {
        WtAetherRoute::Command | WtAetherRoute::Dma | WtAetherRoute::Control => {
            let engine = runtime
                .host_engine_runtime()
                .or_else(|_| aetherdb::EngineRuntime::with_defaults().map_err(|e| e.to_string()))?;
            let pool = aetherdb::WireSessionPool::with_tls(engine, tls);
            match route {
                WtAetherRoute::Command => pool
                    .command(upstream, payload)
                    .map(|resp| resp.encode())
                    .map_err(|err| err.to_string()),
                WtAetherRoute::Dma => pool
                    .dma(upstream, payload)
                    .map(|resp| resp.encode())
                    .map_err(|err| err.to_string()),
                WtAetherRoute::Control => pool
                    .control(upstream, payload)
                    .map(|resp| resp.encode())
                    .map_err(|err| err.to_string()),
                WtAetherRoute::Replicate => unreachable!(),
            }
        }
        WtAetherRoute::Replicate => {
            let engine = runtime
                .host_engine_runtime()
                .or_else(|_| aetherdb::EngineRuntime::with_defaults().map_err(|e| e.to_string()))?;
            let pool = crate::aether_wire::WireRoutePool::with_tls(engine, tls);
            pool.request(upstream, route.name(), payload)
        }
    }
}

fn parse_path_list(raw: &str) -> Vec<String> {
    raw.split(',')
        .map(str::trim)
        .filter(|s| !s.is_empty() && s.starts_with('/'))
        .map(str::to_string)
        .collect()
}

fn parse_routes(raw: &str) -> HashSet<WtAetherRoute> {
    let mut routes = HashSet::new();
    for part in raw.split(',') {
        match part.trim().to_ascii_lowercase().as_str() {
            "command" | "cmd" => {
                routes.insert(WtAetherRoute::Command);
            }
            "dma" => {
                routes.insert(WtAetherRoute::Dma);
            }
            "control" | "cluster" => {
                routes.insert(WtAetherRoute::Control);
            }
            "replicate" | "replication" => {
                routes.insert(WtAetherRoute::Replicate);
            }
            "" => {}
            other => warn!(route = other, "ignoring unknown WT aether route name"),
        }
    }
    if routes.is_empty() {
        return default_wt_routes();
    }
    routes
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn awt1_frame_roundtrip_helpers() {
        let payload = Bytes::from_static(b"payload");
        let frame =
            encode_request_frame(WtAetherRoute::Command, None, payload.clone()).expect("encode");
        let response = encode_response_frame(WtAetherRoute::Command, payload.clone());
        let (route, body) = decode_ok_response_frame(&response).expect("decode");
        assert_eq!(route, WtAetherRoute::Command);
        assert_eq!(body, payload);
        assert_eq!(&frame[..4], b"AWT1");
    }

    #[test]
    fn privileged_route_requires_mtls_by_default() {
        let cfg = WtAetherConfig {
            require_mtls_for_privileged: true,
            require_mtls_for_data: false,
            restrict_client_certs: false,
            ..WtAetherConfig::default()
        };
        assert!(cfg.require_mtls_for_privileged);
        assert!(authorize_wt_route(
            WtAetherRoute::Control,
            &WtAetherSessionAuth::anonymous(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_err());
        assert!(authorize_wt_route(
            WtAetherRoute::Command,
            &WtAetherSessionAuth::anonymous(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_ok());
        assert!(authorize_wt_route(
            WtAetherRoute::Control,
            &WtAetherSessionAuth::authenticated(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_ok());
    }

    #[test]
    fn data_route_requires_mtls_when_configured() {
        unsafe {
            std::env::set_var("NAUTALID_WEBTRANSPORT_AETHER_REQUIRE_MTLS_DATA", "1");
            std::env::set_var("NAUTALID_WEBTRANSPORT_AETHER_REQUIRE_MTLS", "0");
        }
        let cfg = WtAetherConfig::default();
        assert!(authorize_wt_route(
            WtAetherRoute::Command,
            &WtAetherSessionAuth::anonymous(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_err());
        assert!(authorize_wt_route(
            WtAetherRoute::Command,
            &WtAetherSessionAuth::authenticated(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_ok());
        unsafe {
            std::env::remove_var("NAUTALID_WEBTRANSPORT_AETHER_REQUIRE_MTLS_DATA");
            std::env::remove_var("NAUTALID_WEBTRANSPORT_AETHER_REQUIRE_MTLS");
        }
    }

    #[test]
    fn control_requires_bus_token_when_configured() {
        unsafe {
            std::env::set_var("NAUTALID_BUS_TOKEN", "wt-secret");
        }
        let cfg = WtAetherConfig {
            require_mtls_for_privileged: false,
            require_mtls_for_data: false,
            restrict_client_certs: false,
            ..WtAetherConfig::default()
        };
        assert!(authorize_wt_route(
            WtAetherRoute::Control,
            &WtAetherSessionAuth::anonymous(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_err());
        assert!(authorize_wt_route(
            WtAetherRoute::Control,
            &WtAetherSessionAuth::anonymous(),
            &cfg,
            WtRouteAuth {
                bus_token: Some("wrong"),
                ..WtRouteAuth::default()
            },
        )
        .is_err());
        assert!(authorize_wt_route(
            WtAetherRoute::Control,
            &WtAetherSessionAuth::anonymous(),
            &cfg,
            WtRouteAuth {
                bus_token: Some("wt-secret"),
                ..WtRouteAuth::default()
            },
        )
        .is_ok());
        unsafe {
            std::env::remove_var("NAUTALID_BUS_TOKEN");
        }
    }

    #[test]
    fn awt1_bus_token_roundtrip_in_frame_bytes() {
        let payload = Bytes::from_static(b"ping");
        let frame = encode_request_frame_with_options(
            WtAetherRoute::Control,
            None,
            Some("tok"),
            None,
            payload.clone(),
        )
        .expect("encode");
        assert_eq!(frame[5], AWT1_FLAG_BUS_TOKEN);
        assert_eq!(&frame[..4], b"AWT1");
    }

    #[test]
    fn session_limits_read_env() {
        unsafe {
            std::env::set_var("NAUTALID_WEBTRANSPORT_AETHER_MAX_BIDI", "8");
            std::env::set_var("NAUTALID_WEBTRANSPORT_AETHER_FRAME_TIMEOUT_SECS", "12");
        }
        let cfg = WtAetherConfig::from_env();
        assert_eq!(cfg.max_bidi_streams, 8);
        assert_eq!(cfg.frame_timeout_secs, 12);
        unsafe {
            std::env::remove_var("NAUTALID_WEBTRANSPORT_AETHER_MAX_BIDI");
            std::env::remove_var("NAUTALID_WEBTRANSPORT_AETHER_FRAME_TIMEOUT_SECS");
        }
    }

    #[test]
    fn decode_rejects_oversized_response_length() {
        let len = (MAX_AWT1_FRAME_PAYLOAD + 1) as u32;
        let mut frame = vec![b'A', b'W', b'T', b'1', 0, 0];
        frame.extend_from_slice(&len.to_be_bytes());
        let err = decode_ok_response_frame(&frame).expect_err("oversize");
        assert!(err.contains("larger than"));
    }

    #[test]
    fn encode_rejects_oversized_request_payload() {
        let huge = Bytes::from(vec![0u8; MAX_AWT1_FRAME_PAYLOAD + 1]);
        let err =
            encode_request_frame(WtAetherRoute::Command, None, huge).expect_err("oversize");
        assert!(err.contains("larger than"));
    }

    #[test]
    fn default_aether_path() {
        unsafe {
            std::env::remove_var("NAUTALID_WEBTRANSPORT_AETHER_PATH");
        }
        let cfg = WtAetherConfig::from_env();
        assert_eq!(cfg.session_paths, vec!["/wt/aether"]);
    }

    #[test]
    fn parses_route_list() {
        let routes = parse_routes("command,dma");
        assert!(routes.contains(&WtAetherRoute::Command));
        assert!(routes.contains(&WtAetherRoute::Dma));
        assert!(!routes.contains(&WtAetherRoute::Control));
    }

    #[test]
    fn nested_http_response_hides_metadata_without_privileged_mtls() {
        let cfg = wt_aether_config_privileged_mtls();
        let (status, body) = aether_nested_http_response(
            WtAetherSessionAuth::anonymous(),
            &cfg,
            "GET",
            "/nested",
        );
        assert_eq!(status, http::StatusCode::FORBIDDEN);
        assert!(body.get("enabled").is_none());
        assert_eq!(
            body.get("error").and_then(|v| v.as_str()),
            Some("mTLS client certificate required")
        );
    }

    #[test]
    fn nested_http_response_includes_metadata_for_authenticated_peer() {
        let cfg = wt_aether_config_permissive();
        let (status, body) = aether_nested_http_response(
            WtAetherSessionAuth::authenticated(),
            &cfg,
            "GET",
            "/nested",
        );
        assert_eq!(status, http::StatusCode::OK);
        assert_eq!(body.get("ok").and_then(|v| v.as_bool()), Some(true));
        assert!(body.get("enabled").is_none());
        assert!(body.get("service").is_none());
    }

    #[test]
    fn replicate_route_disabled_by_default() {
        let cfg = WtAetherConfig::default();
        assert!(!cfg.allow_replicate_route);
        assert!(!cfg.routes.contains(&WtAetherRoute::Replicate));
        assert!(authorize_wt_route(
            WtAetherRoute::Replicate,
            &WtAetherSessionAuth::authenticated(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_err());
    }

    #[test]
    fn replicate_requires_frame_key_when_env_configured() {
        let key_hex = "4242424242424242424242424242424242424242424242424242424242424242";
        let key = aetherdb::decode_master_key_hex(key_hex).expect("key");
        unsafe {
            std::env::set_var("AETHERDB_HOST_REPLICATE_KEY_HEX", key_hex);
        }
        let cfg = WtAetherConfig {
            allow_replicate_route: true,
            require_mtls_for_privileged: false,
            require_mtls_for_data: false,
            ..WtAetherConfig::default()
        };
        assert!(authorize_wt_route(
            WtAetherRoute::Replicate,
            &WtAetherSessionAuth::authenticated(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_err());
        assert!(authorize_wt_route(
            WtAetherRoute::Replicate,
            &WtAetherSessionAuth::authenticated(),
            &cfg,
            WtRouteAuth {
                replicate_key: Some(&key),
                ..WtRouteAuth::default()
            },
        )
        .is_ok());
        unsafe {
            std::env::remove_var("AETHERDB_HOST_REPLICATE_KEY_HEX");
        }
    }

    fn wt_aether_config_privileged_mtls() -> WtAetherConfig {
        WtAetherConfig {
            require_mtls_for_privileged: true,
            require_mtls_for_data: false,
            enabled: true,
            restrict_client_certs: false,
            ..WtAetherConfig::default()
        }
    }

    fn wt_aether_config_permissive() -> WtAetherConfig {
        WtAetherConfig {
            enabled: true,
            require_mtls_for_privileged: false,
            require_mtls_for_data: false,
            restrict_client_certs: false,
            ..WtAetherConfig::default()
        }
    }

    #[test]
    fn cert_allowlist_denies_unlisted_fingerprint_when_restricted() {
        let auth = WtAetherSessionAuth::authenticated();
        let fp = auth
            .peer_cert
            .leaf_sha256_hex()
            .expect("test cert fingerprint");
        let cfg = WtAetherConfig {
            enabled: true,
            require_mtls_for_privileged: false,
            require_mtls_for_data: true,
            restrict_client_certs: true,
            cert_sha256_allow_data: HashSet::from(["0000000000000000000000000000000000000000000000000000000000000000".into()]),
            ..WtAetherConfig::default()
        };
        assert!(authorize_wt_route(
            WtAetherRoute::Command,
            &auth,
            &cfg,
            WtRouteAuth::default(),
        )
        .is_err());
        let mut allowed = cfg;
        allowed.cert_sha256_allow_data = HashSet::from([fp]);
        assert!(authorize_wt_route(
            WtAetherRoute::Command,
            &auth,
            &allowed,
            WtRouteAuth::default(),
        )
        .is_ok());
    }

    #[test]
    fn replicate_route_rejected_without_privileged_mtls_peer() {
        let mut routes = default_wt_routes();
        routes.insert(WtAetherRoute::Replicate);
        let cfg = WtAetherConfig {
            allow_replicate_route: true,
            routes,
            require_mtls_for_privileged: true,
            require_mtls_for_data: false,
            ..WtAetherConfig::default()
        };
        assert!(authorize_wt_route(
            WtAetherRoute::Replicate,
            &WtAetherSessionAuth::anonymous(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_err());
    }

    #[test]
    fn route_subset_rejects_disabled_dma_route() {
        let cfg = WtAetherConfig {
            enabled: true,
            routes: HashSet::from([WtAetherRoute::Command]),
            require_mtls_for_privileged: false,
            require_mtls_for_data: false,
            ..WtAetherConfig::default()
        };
        assert!(authorize_wt_route(
            WtAetherRoute::Command,
            &WtAetherSessionAuth::anonymous(),
            &cfg,
            WtRouteAuth::default(),
        )
        .is_ok());
        // DMA not in routes — checked in dispatch_frame via cfg.routes.contains
        assert!(!cfg.routes.contains(&WtAetherRoute::Dma));
    }

    #[test]
    fn decode_request_frame_matches_encode_helper() {
        let payload = Bytes::from_static(b"payload");
        let frame =
            encode_request_frame(WtAetherRoute::Command, None, payload.clone()).expect("encode");
        let decoded =
            decode_request_frame(&frame, MAX_AWT1_FRAME_PAYLOAD).expect("decode");
        assert_eq!(decoded.route, WtAetherRoute::Command);
        assert_eq!(decoded.payload, payload);
    }

    #[test]
    fn session_kind_prefers_aether_path() {
        unsafe {
            std::env::set_var("NAUTALID_WEBTRANSPORT", "1");
            std::env::set_var("NAUTALID_WEBTRANSPORT_AETHER", "1");
        }
        install(
            Arc::new(AetherRuntime::bootstrap_from_env().expect("runtime")),
            WtAetherConfig {
                enabled: true,
                session_paths: vec!["/lid".into()],
                ..WtAetherConfig::default()
            },
        );
        assert_eq!(
            session_kind_for_path("/lid"),
            Some(WtSessionKind::Aether)
        );
        unsafe {
            std::env::remove_var("NAUTALID_WEBTRANSPORT");
            std::env::remove_var("NAUTALID_WEBTRANSPORT_AETHER");
        }
    }

    #[test]
    fn control_rejected_on_wire_backend() {
        unsafe {
            std::env::remove_var("NAUTALID_BUS_TOKEN");
            std::env::remove_var("NAUTALID_BUS_TOKEN_FILE");
            std::env::remove_var("NAUTALID_SECURITY_REQUIRE_WT_CONTROL_TOKEN");
        }
        let runtime = Arc::new(AetherRuntime::bootstrap_from_env().expect("runtime"));
        let mut cfg = wt_aether_config_permissive();
        cfg.backend = WtAetherBackend::Wire {
            upstream: "tcp://127.0.0.1:5555".into(),
        };
        cfg.routes.insert(WtAetherRoute::Control);
        install(runtime, cfg);
        let frame = encode_request_frame(WtAetherRoute::Control, None, Bytes::from_static(b"x"))
            .expect("encode");
        let rt = tokio::runtime::Runtime::new().expect("tokio runtime");
        let response = rt
            .block_on(dispatch_test_awt1_request(
                frame,
                WtAetherSessionAuth::anonymous(),
            ))
            .expect("dispatch");
        let err = decode_ok_response_frame(&response).expect_err("wire control");
        assert!(err.contains("wire backend"));
    }
}