hop-relay 2.0.0

use std::collections::HashMap;
use std::convert::Infallible;
use std::fs::{self, File, OpenOptions};
use std::future::Future;
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex as StdMutex};
use std::time::{Duration, Instant};

use hop_actions::{apply_action, Action};
use hop_channels::store::{ChannelSnapshot, ChannelStore};
use hop_channels::store_fs::FsStore;
use hop_channels::{Channel, ChannelError};
use hop_core::{CompanionMatrix, Field};
use once_cell::sync::Lazy;
use prometheus::{
    Encoder, HistogramOpts, HistogramVec, IntCounterVec, IntGauge, IntGaugeVec, Opts, Registry,
    TextEncoder,
};
use serde::{Deserialize, Serialize};
use tokio::sync::{Mutex, OwnedSemaphorePermit, RwLock, Semaphore};
use warp::http::{header, StatusCode};
use warp::{Filter, Reply};

// Optional projection helpers
#[cfg(feature = "torsor_masking")]
use hop_channels::mask::try_apply_twist_row;
use hop_channels::proj::try_row0_of_poly;

// NEW: receipts
use base64::engine::general_purpose::STANDARD as B64;
use base64::engine::general_purpose::URL_SAFE_NO_PAD as B64URL;
use base64::Engine;
use ed25519_dalek::{Signature, Signer, SigningKey, VerifyingKey};
use rand_core::{OsRng, RngCore};

// ------------------------ Shared + helpers ------------------------

#[derive(Debug)]
struct InFlightGuard {
    gauge: IntGauge,
}

impl InFlightGuard {
    fn new(gauge: IntGauge) -> Self {
        gauge.inc();
        Self { gauge }
    }
}

impl Drop for InFlightGuard {
    fn drop(&mut self) {
        self.gauge.dec();
    }
}

#[derive(Debug)]
struct RequestMeta {
    request_id: String,
    method: String,
    path: String,
    forwarded_for: Option<String>,
    user_agent: Option<String>,
    origin: Option<String>,
    start: Instant,
    _in_flight: InFlightGuard,
}

#[derive(Clone, Debug)]
pub struct ServerConfig {
    pub data_dir: PathBuf,
    /// If set, enable authentication.
    ///
    /// Admin endpoints require `Authorization: Bearer <admin token>`.
    /// Vault-scoped endpoints accept either the admin token or a per-vault capability token.
    pub auth_token: Option<String>,
    /// Maximum accepted JSON body size for POST endpoints.
    pub max_body_bytes: u64,
    /// Maximum length of `coeffs` accepted by endpoints that bootstrap channels/vaults.
    pub max_coeffs_len: usize,
    /// Maximum number of actions accepted in a single request.
    pub max_actions_per_request: usize,
    /// Maximum estimated field operations for one action bundle.
    pub max_action_work: u64,
    /// Maximum number of transcript entries returned by `/vault/prove`.
    pub max_prove_entries: usize,
    /// Maximum number of in-flight requests (excluding GET /health and OPTIONS).
    ///
    /// If set to 0, the limit is disabled (not recommended for production).
    pub max_in_flight_requests: usize,
    /// Per-request processing timeout in milliseconds (0 disables).
    pub request_timeout_ms: u64,
    /// Global rate limit in requests/second (0 disables).
    pub rate_limit_rps: u32,
    /// Global rate limit burst capacity (0 disables).
    pub rate_limit_burst: u32,
    /// If true, enable permissive CORS for dev (`Access-Control-Allow-Origin: *`).
    /// Prefer `cors_allow_origins` for production.
    pub cors_allow_any_origin: bool,
    /// If non-empty, allow CORS for these exact origins.
    pub cors_allow_origins: Vec<String>,
    /// Maximum time to wait for in-flight requests after receiving SIGINT/SIGTERM (0 waits forever).
    pub shutdown_grace_ms: u64,
}

impl Default for ServerConfig {
    fn default() -> Self {
        Self {
            data_dir: PathBuf::from("./data"),
            auth_token: None,
            max_body_bytes: 1024 * 1024, // 1 MiB
            max_coeffs_len: 1024,
            max_actions_per_request: 4096,
            max_action_work: 10_000_000,
            max_prove_entries: 10_000,
            max_in_flight_requests: 128,
            request_timeout_ms: 30_000,
            rate_limit_rps: 0,
            rate_limit_burst: 0,
            cors_allow_any_origin: false,
            cors_allow_origins: Vec::new(),
            shutdown_grace_ms: 10_000,
        }
    }
}

#[derive(Debug)]
struct Metrics {
    registry: Registry,
    requests_total: IntCounterVec,
    request_duration_seconds: HistogramVec,
    in_flight: IntGauge,
    authz_checks_total: IntCounterVec,
    authz_cache_total: IntCounterVec,
    authz_cache_entries: IntGauge,
}

fn route_label(path: &str, method: &str) -> &'static str {
    if method == "OPTIONS" {
        return "options";
    }

    match path {
        "/health" => "health",
        "/ready" => "ready",
        "/metrics" => "metrics",
        "/bundle" => "bundle",
        "/state" => "state",
        "/vault/new" => "vault_new",
        "/vault/token/rotate" => "vault_token_rotate",
        "/vault/append" => "vault_append",
        "/vault/get" => "vault_get",
        "/vault/prove" => "vault_prove",
        _ => "other",
    }
}

fn status_class(status: StatusCode) -> &'static str {
    if status.is_success() {
        "2xx"
    } else if status.is_client_error() {
        "4xx"
    } else if status.is_server_error() {
        "5xx"
    } else {
        "other"
    }
}

impl Metrics {
    fn new(cfg: &ServerConfig) -> Result<Self, prometheus::Error> {
        let registry = Registry::new();

        let requests_total = IntCounterVec::new(
            Opts::new(
                "hop_relay_http_requests_total",
                "Total number of HTTP requests processed by hop-relay.",
            ),
            &["method", "route", "status"],
        )?;
        registry.register(Box::new(requests_total.clone()))?;

        let request_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                "hop_relay_http_request_duration_seconds",
                "HTTP request latency in seconds.",
            )
            .buckets(vec![
                0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
            ]),
            &["method", "route"],
        )?;
        registry.register(Box::new(request_duration_seconds.clone()))?;

        let in_flight = IntGauge::with_opts(Opts::new(
            "hop_relay_http_in_flight_requests",
            "Number of HTTP requests currently being processed by hop-relay.",
        ))?;
        registry.register(Box::new(in_flight.clone()))?;

        let build_info = IntGaugeVec::new(
            Opts::new("hop_relay_build_info", "Build information for hop-relay."),
            &["version"],
        )?;
        registry.register(Box::new(build_info.clone()))?;
        build_info
            .with_label_values(&[env!("CARGO_PKG_VERSION")])
            .set(1);

        let config_max_in_flight = IntGauge::with_opts(Opts::new(
            "hop_relay_config_max_in_flight",
            "Configured maximum in-flight requests (0 disables the limit).",
        ))?;
        config_max_in_flight.set(cfg.max_in_flight_requests as i64);
        registry.register(Box::new(config_max_in_flight.clone()))?;

        let config_request_timeout_ms = IntGauge::with_opts(Opts::new(
            "hop_relay_config_request_timeout_ms",
            "Configured per-request timeout in milliseconds (0 disables the timeout).",
        ))?;
        config_request_timeout_ms.set(cfg.request_timeout_ms as i64);
        registry.register(Box::new(config_request_timeout_ms.clone()))?;

        let config_rate_limit_rps = IntGauge::with_opts(Opts::new(
            "hop_relay_config_rate_limit_rps",
            "Configured global rate limit in requests/second (0 disables).",
        ))?;
        config_rate_limit_rps.set(cfg.rate_limit_rps as i64);
        registry.register(Box::new(config_rate_limit_rps.clone()))?;

        let config_rate_limit_burst = IntGauge::with_opts(Opts::new(
            "hop_relay_config_rate_limit_burst",
            "Configured global rate limit burst capacity (0 disables).",
        ))?;
        config_rate_limit_burst.set(cfg.rate_limit_burst as i64);
        registry.register(Box::new(config_rate_limit_burst.clone()))?;

        let authz_checks_total = IntCounterVec::new(
            Opts::new(
                "hop_relay_authz_checks_total",
                "Total number of authorization checks performed by hop-relay.",
            ),
            &["perm", "result"],
        )?;
        registry.register(Box::new(authz_checks_total.clone()))?;

        let authz_cache_total = IntCounterVec::new(
            Opts::new(
                "hop_relay_authz_cache_total",
                "Authz cache operations (hits, misses, refreshes, evictions).",
            ),
            &["result"],
        )?;
        registry.register(Box::new(authz_cache_total.clone()))?;

        let authz_cache_entries = IntGauge::with_opts(Opts::new(
            "hop_relay_authz_cache_entries",
            "Current number of in-memory vault authz cache entries.",
        ))?;
        registry.register(Box::new(authz_cache_entries.clone()))?;

        Ok(Self {
            registry,
            requests_total,
            request_duration_seconds,
            in_flight,
            authz_checks_total,
            authz_cache_total,
            authz_cache_entries,
        })
    }

    fn observe(&self, meta: &RequestMeta, status: StatusCode, latency: Duration) {
        let route = route_label(&meta.path, &meta.method);
        let class = status_class(status);

        self.requests_total
            .with_label_values(&[meta.method.as_str(), route, class])
            .inc();
        self.request_duration_seconds
            .with_label_values(&[meta.method.as_str(), route])
            .observe(latency.as_secs_f64());
    }
}

#[derive(Debug)]
struct ApiError {
    status: StatusCode,
    message: String,
}

impl warp::reject::Reject for ApiError {}

#[derive(Debug)]
struct Unauthorized;

impl warp::reject::Reject for Unauthorized {}

#[derive(Clone)]
enum Authn {
    Anonymous,
    Admin,
    Token(String),
}

impl std::fmt::Debug for Authn {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Anonymous => f.write_str("Anonymous"),
            Self::Admin => f.write_str("Admin"),
            Self::Token(_) => f.write_str("Token(<redacted>)"),
        }
    }
}

impl Authn {
    fn bearer_token(&self) -> Option<&str> {
        match self {
            Self::Token(t) => Some(t.as_str()),
            _ => None,
        }
    }
}

#[derive(Debug)]
struct Overloaded;

impl warp::reject::Reject for Overloaded {}

#[derive(Debug)]
struct RateLimited {
    retry_after_secs: u64,
}

impl warp::reject::Reject for RateLimited {}

#[derive(Debug)]
struct TimedOut;

impl warp::reject::Reject for TimedOut {}

#[derive(Debug)]
struct TokenBucket {
    capacity: u64,
    tokens: u64,
    refill_per_sec: u64,
    last: Instant,
}

impl TokenBucket {
    fn new(refill_per_sec: u64, burst: u64) -> Self {
        Self {
            capacity: burst,
            tokens: burst,
            refill_per_sec,
            last: Instant::now(),
        }
    }

    fn refill(&mut self, now: Instant) {
        if self.capacity == 0 || self.refill_per_sec == 0 {
            self.tokens = self.capacity;
            self.last = now;
            return;
        }

        let elapsed = now.duration_since(self.last);
        let add = elapsed
            .as_micros()
            .saturating_mul(self.refill_per_sec as u128)
            .saturating_div(1_000_000) as u64;
        if add > 0 {
            self.tokens = self.capacity.min(self.tokens.saturating_add(add));
            self.last = now;
        }
    }

    fn try_take(&mut self, now: Instant) -> bool {
        self.refill(now);
        if self.tokens > 0 {
            self.tokens -= 1;
            true
        } else {
            false
        }
    }

    fn retry_after_secs(&self) -> u64 {
        // A conservative hint for clients. We keep this simple on purpose.
        1
    }
}

fn timeout_dur(ms: u64) -> Option<Duration> {
    if ms == 0 {
        None
    } else {
        Some(Duration::from_millis(ms))
    }
}

async fn maybe_timeout<T, F>(dur: Option<Duration>, fut: F) -> Result<T, warp::Rejection>
where
    F: Future<Output = Result<T, warp::Rejection>>,
{
    let Some(dur) = dur else {
        return fut.await;
    };

    match tokio::time::timeout(dur, fut).await {
        Ok(res) => res,
        Err(_) => Err(warp::reject::custom(TimedOut)),
    }
}

#[derive(Debug)]
struct ReadinessCache {
    last_check: Instant,
    ok: bool,
    message: Option<String>,
}

impl Default for ReadinessCache {
    fn default() -> Self {
        let last_check = Instant::now()
            .checked_sub(Duration::from_secs(3600))
            .unwrap_or_else(Instant::now);
        Self {
            last_check,
            ok: false,
            message: Some("not yet checked".to_string()),
        }
    }
}

fn readiness_probe(data_dir: &Path) -> std::io::Result<()> {
    fs::create_dir_all(data_dir)?;

    let mut suffix = [0u8; 8];
    OsRng.fill_bytes(&mut suffix);
    let probe = data_dir.join(format!(".ready-probe.{}", hex::encode(suffix)));

    let res = (|| {
        let mut f = OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&probe)?;
        f.write_all(b"ok\n")?;
        f.sync_all()?;
        drop(f);
        fs::remove_file(&probe)?;
        Ok::<(), std::io::Error>(())
    })();

    if res.is_err() {
        let _ = fs::remove_file(&probe);
    }
    res?;
    Ok(())
}

#[derive(Serialize)]
struct ReadyResp {
    ok: bool,
}

async fn handle_ready(
    cfg: Arc<ServerConfig>,
    cache: Arc<Mutex<ReadinessCache>>,
) -> Result<impl warp::Reply, warp::Rejection> {
    let now = Instant::now();
    {
        let c = cache.lock().await;
        if now.duration_since(c.last_check) < Duration::from_secs(1) {
            if c.ok {
                return Ok(warp::reply::with_status(
                    warp::reply::json(&ReadyResp { ok: true }),
                    StatusCode::OK,
                ));
            }
            return Err(service_unavailable(
                c.message.clone().unwrap_or_else(|| "not ready".to_string()),
            ));
        }
    }

    let data_dir = cfg.data_dir.clone();
    let probe = tokio::task::spawn_blocking(move || readiness_probe(&data_dir))
        .await
        .map_err(|_| internal("readiness probe task failed"))?;

    let mut c = cache.lock().await;
    c.last_check = Instant::now();
    match probe {
        Ok(()) => {
            c.ok = true;
            c.message = None;
            Ok(warp::reply::with_status(
                warp::reply::json(&ReadyResp { ok: true }),
                StatusCode::OK,
            ))
        }
        Err(e) => {
            // Don't leak raw OS errors to clients; keep details in logs.
            tracing::warn!(error = %e, "readiness probe failed");
            c.ok = false;
            c.message = Some(format!("not ready: {:?}", e.kind()));
            Err(service_unavailable("not ready"))
        }
    }
}

fn metrics_reply(metrics: Arc<Metrics>) -> warp::reply::Response {
    let encoder = TextEncoder::new();
    let mfs = metrics.registry.gather();
    let mut buf = Vec::new();
    if let Err(e) = encoder.encode(&mfs, &mut buf) {
        tracing::error!(error = %e, "metrics encode failed");
        return warp::reply::with_status(
            "metrics encode failed",
            StatusCode::INTERNAL_SERVER_ERROR,
        )
        .into_response();
    }

    let mut res = warp::reply::with_status(buf, StatusCode::OK).into_response();
    if let Ok(v) = header::HeaderValue::from_str(encoder.format_type()) {
        res.headers_mut().insert(header::CONTENT_TYPE, v);
    }
    res
}

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

fn sanitize_request_id(id: &str) -> Option<String> {
    // Avoid log/header injection by restricting to a conservative character set.
    let id = id.trim();
    if id.is_empty() || id.len() > 64 {
        return None;
    }
    if !id
        .bytes()
        .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
    {
        return None;
    }
    Some(id.to_string())
}

fn generate_request_id() -> String {
    let mut bytes = [0u8; 16];
    OsRng.fill_bytes(&mut bytes);
    hex::encode(bytes)
}

fn request_meta(
    metrics: Arc<Metrics>,
) -> impl Filter<Extract = (RequestMeta,), Error = Infallible> + Clone {
    warp::any()
        .and(warp::method())
        .and(warp::path::full())
        .and(warp::header::headers_cloned())
        .map(
            move |method: warp::http::Method,
                  path: warp::path::FullPath,
                  headers: warp::http::HeaderMap| {
                // Keep an in-flight gauge guard for the duration of request processing.
                let in_flight = InFlightGuard::new(metrics.in_flight.clone());

                let get_header = |name: &str| -> Option<String> {
                    headers
                        .get(name)
                        .and_then(|v| v.to_str().ok())
                        .map(|s| s.to_string())
                };

                let user_agent = get_header("user-agent");
                let origin = get_header("origin");
                let forwarded_for = get_header("x-forwarded-for");

                let request_id = headers
                    .get("x-request-id")
                    .and_then(|v| v.to_str().ok())
                    .and_then(sanitize_request_id)
                    .unwrap_or_else(generate_request_id);

                RequestMeta {
                    request_id,
                    method: method.as_str().to_string(),
                    path: path.as_str().to_string(),
                    forwarded_for,
                    user_agent,
                    origin,
                    start: Instant::now(),
                    _in_flight: in_flight,
                }
            },
        )
}

fn bad_request(msg: impl Into<String>) -> warp::Rejection {
    warp::reject::custom(ApiError {
        status: StatusCode::BAD_REQUEST,
        message: msg.into(),
    })
}

fn not_found(msg: impl Into<String>) -> warp::Rejection {
    warp::reject::custom(ApiError {
        status: StatusCode::NOT_FOUND,
        message: msg.into(),
    })
}

fn conflict(msg: impl Into<String>) -> warp::Rejection {
    warp::reject::custom(ApiError {
        status: StatusCode::CONFLICT,
        message: msg.into(),
    })
}

fn forbidden(msg: impl Into<String>) -> warp::Rejection {
    warp::reject::custom(ApiError {
        status: StatusCode::FORBIDDEN,
        message: msg.into(),
    })
}

fn internal(msg: impl Into<String>) -> warp::Rejection {
    warp::reject::custom(ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        message: msg.into(),
    })
}

fn service_unavailable(msg: impl Into<String>) -> warp::Rejection {
    warp::reject::custom(ApiError {
        status: StatusCode::SERVICE_UNAVAILABLE,
        message: msg.into(),
    })
}

async fn handle_rejection(err: warp::Rejection) -> Result<warp::reply::Response, Infallible> {
    // auth
    if err.find::<Unauthorized>().is_some() {
        let r = warp::reply::with_status(
            warp::reply::json(&serde_json::json!({"error":"unauthorized"})),
            StatusCode::UNAUTHORIZED,
        );
        return Ok(r.into_response());
    }

    // timeouts
    if err.find::<TimedOut>().is_some() {
        let r = warp::reply::with_status(
            warp::reply::json(&serde_json::json!({"error":"timeout"})),
            StatusCode::GATEWAY_TIMEOUT,
        );
        return Ok(r.into_response());
    }

    // load-shedding
    if err.find::<Overloaded>().is_some() {
        let mut res = warp::reply::with_status(
            warp::reply::json(&serde_json::json!({"error":"overloaded"})),
            StatusCode::SERVICE_UNAVAILABLE,
        )
        .into_response();
        res.headers_mut()
            .insert(header::RETRY_AFTER, header::HeaderValue::from_static("1"));
        return Ok(res);
    }

    if let Some(e) = err.find::<RateLimited>() {
        let mut res = warp::reply::with_status(
            warp::reply::json(&serde_json::json!({"error":"rate limited"})),
            StatusCode::TOO_MANY_REQUESTS,
        )
        .into_response();
        let secs = e.retry_after_secs.max(1);
        if let Ok(v) = header::HeaderValue::from_str(&secs.to_string()) {
            res.headers_mut().insert(header::RETRY_AFTER, v);
        }
        return Ok(res);
    }

    // well-formed API errors
    if let Some(e) = err.find::<ApiError>() {
        let r = warp::reply::with_status(
            warp::reply::json(&serde_json::json!({"error": e.message})),
            e.status,
        );
        return Ok(r.into_response());
    }

    // payload too large
    if err.find::<warp::reject::PayloadTooLarge>().is_some() {
        let r = warp::reply::with_status(
            warp::reply::json(&serde_json::json!({"error":"payload too large"})),
            StatusCode::PAYLOAD_TOO_LARGE,
        );
        return Ok(r.into_response());
    }

    // invalid JSON
    if err
        .find::<warp::filters::body::BodyDeserializeError>()
        .is_some()
    {
        let r = warp::reply::with_status(
            warp::reply::json(&serde_json::json!({"error":"invalid JSON body"})),
            StatusCode::BAD_REQUEST,
        );
        return Ok(r.into_response());
    }

    // method / route
    if err.is_not_found() {
        let r = warp::reply::with_status(
            warp::reply::json(&serde_json::json!({"error":"not found"})),
            StatusCode::NOT_FOUND,
        );
        return Ok(r.into_response());
    }
    if err.find::<warp::reject::MethodNotAllowed>().is_some() {
        let r = warp::reply::with_status(
            warp::reply::json(&serde_json::json!({"error":"method not allowed"})),
            StatusCode::METHOD_NOT_ALLOWED,
        );
        return Ok(r.into_response());
    }

    // fallback
    let r = warp::reply::with_status(
        warp::reply::json(&serde_json::json!({"error":"internal server error"})),
        StatusCode::INTERNAL_SERVER_ERROR,
    );
    Ok(r.into_response())
}

type PerIdLocks = HashMap<[u8; 32], Arc<Mutex<()>>>;

static LOCKS: Lazy<Mutex<PerIdLocks>> = Lazy::new(|| Mutex::new(HashMap::new()));

async fn cleanup_lock(id: [u8; 32], lock_arc: &Arc<Mutex<()>>) {
    // If only this handler and the map still reference the lock Arc, remove it
    // to prevent unbounded growth under many distinct IDs.
    let mut m = LOCKS.lock().await;
    if Arc::strong_count(lock_arc) == 2 {
        m.remove(&id);
    }
}

fn parse_id_hex64_checked(id_hex: &str) -> Option<[u8; 32]> {
    if id_hex.len() != 64 {
        return None;
    }
    let bytes = hex::decode(id_hex).ok()?;
    if bytes.len() < 32 {
        return None;
    }
    let mut id = [0u8; 32];
    for (i, b) in bytes.iter().take(32).enumerate() {
        id[i] = *b;
    }
    Some(id)
}

fn id_hex(id: &[u8; 32]) -> String {
    hex::encode(id)
}

fn coefficients_are_canonical(coefficients: &[u64]) -> bool {
    !coefficients.is_empty() && coefficients.iter().all(|value| *value < Field::MOD)
}

fn action_inputs_are_valid(actions: &[ActionIn], maximum_coefficients: usize) -> bool {
    actions.iter().all(|action| {
        action.g.len() <= maximum_coefficients && coefficients_are_canonical(&action.g)
    })
}

fn action_work(actions: &[ActionIn], dimension: usize) -> Option<u64> {
    actions.iter().try_fold(0_u64, |total, action| {
        let coefficients = u64::try_from(action.g.len()).ok()?;
        let dimension = u64::try_from(dimension).ok()?;
        total.checked_add(coefficients.checked_mul(dimension)?)
    })
}

fn hash_state(fields: &[Field]) -> String {
    use sha2::{Digest, Sha256};
    let mut h = Sha256::new();
    h.update(b"HOP-STATE\0\x02");
    h.update((fields.len() as u128).to_le_bytes());
    for f in fields {
        h.update(f.to_le_bytes());
    }
    format!("{:x}", h.finalize())
}

fn ensure_vault_dir(base: &Path, id_hex: &str) -> std::io::Result<PathBuf> {
    let d = base.join(id_hex);
    fs::create_dir_all(&d)?;
    Ok(d)
}

fn write_private_create_new(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
    let mut opts = OpenOptions::new();
    opts.write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        opts.mode(0o600);
    }
    let mut f = opts.open(path)?;
    f.write_all(bytes)?;
    f.sync_all()?;

    #[cfg(unix)]
    if let Some(dir) = path.parent() {
        // Ensure directory entry is durable (best effort for crash safety).
        let _ = File::open(dir).and_then(|d| d.sync_all());
    }

    Ok(())
}

fn atomic_write_replace(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
    #[cfg(unix)]
    {
        let dir = path.parent().unwrap_or_else(|| Path::new("."));
        let name = path
            .file_name()
            .and_then(|s| s.to_str())
            .unwrap_or("atomic");

        let mut suffix = [0u8; 8];
        OsRng.fill_bytes(&mut suffix);
        let tmp = dir.join(format!(".{}.tmp.{}", name, hex::encode(suffix)));

        let mut f = OpenOptions::new().write(true).create_new(true).open(&tmp)?;
        f.write_all(bytes)?;
        f.sync_all()?;

        match fs::rename(&tmp, path) {
            Ok(()) => {
                // Ensure directory entry is durable (best effort for crash safety).
                let _ = File::open(dir).and_then(|d| d.sync_all());
                Ok(())
            }
            Err(e) => {
                let _ = fs::remove_file(&tmp);
                Err(e)
            }
        }
    }
    #[cfg(not(unix))]
    {
        std::fs::write(path, bytes)
    }
}

fn atomic_write_replace_private(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;

        let dir = path.parent().unwrap_or_else(|| Path::new("."));
        let name = path
            .file_name()
            .and_then(|s| s.to_str())
            .unwrap_or("atomic");

        let mut suffix = [0u8; 8];
        OsRng.fill_bytes(&mut suffix);
        let tmp = dir.join(format!(".{}.tmp.{}", name, hex::encode(suffix)));

        let mut f = OpenOptions::new()
            .write(true)
            .create_new(true)
            .mode(0o600)
            .open(&tmp)?;
        f.write_all(bytes)?;
        f.sync_all()?;

        match fs::rename(&tmp, path) {
            Ok(()) => {
                // Ensure directory entry is durable (best effort for crash safety).
                let _ = File::open(dir).and_then(|d| d.sync_all());
                Ok(())
            }
            Err(e) => {
                let _ = fs::remove_file(&tmp);
                Err(e)
            }
        }
    }
    #[cfg(not(unix))]
    {
        std::fs::write(path, bytes)
    }
}

// --- Receipt helpers (keypair + transcript hash) ---

fn server_keypair_path(base: &Path) -> PathBuf {
    base.join("server_keypair.json")
}

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct ServerKeypairDisk {
    pk: String,
    sk: String,
}

static SERVER_KEYPAIR_LOCK: Lazy<StdMutex<()>> = Lazy::new(|| StdMutex::new(()));

fn load_or_init_server_keypair(base: &Path) -> Result<(SigningKey, VerifyingKey), std::io::Error> {
    let _guard = SERVER_KEYPAIR_LOCK
        .lock()
        .map_err(|_| std::io::Error::other("server keypair lock poisoned"))?;

    fs::create_dir_all(base)?;
    let p = server_keypair_path(base);
    if p.exists() {
        let raw = std::fs::read_to_string(&p)?;

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            // Best-effort tighten perms on existing key files.
            let _ = fs::set_permissions(&p, fs::Permissions::from_mode(0o600));
        }

        let disk: ServerKeypairDisk = serde_json::from_str(&raw)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;

        let sk_bytes_vec = B64
            .decode(disk.sk)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
        let pk_bytes_vec = B64
            .decode(disk.pk)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;

        let sk_bytes: [u8; 32] = sk_bytes_vec.try_into().map_err(|_| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "invalid signing key length",
            )
        })?;
        let pk_bytes: [u8; 32] = pk_bytes_vec.try_into().map_err(|_| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "invalid verifying key length",
            )
        })?;

        let sk = SigningKey::from_bytes(&sk_bytes);
        let vk = VerifyingKey::from_bytes(&pk_bytes)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
        Ok((sk, vk))
    } else {
        let sk = SigningKey::generate(&mut OsRng);
        let vk = sk.verifying_key();
        let disk = ServerKeypairDisk {
            pk: B64.encode(vk.as_bytes()),
            sk: B64.encode(sk.to_bytes()),
        };
        let payload = serde_json::to_vec_pretty(&disk).map_err(std::io::Error::other)?;
        match write_private_create_new(&p, &payload) {
            Ok(()) => Ok((sk, vk)),
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                // Another process may have initialized it; read it.
                let raw = std::fs::read_to_string(&p)?;
                let disk: ServerKeypairDisk = serde_json::from_str(&raw)
                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;

                let sk_bytes_vec = B64
                    .decode(disk.sk)
                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
                let pk_bytes_vec = B64
                    .decode(disk.pk)
                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;

                let sk_bytes: [u8; 32] = sk_bytes_vec.try_into().map_err(|_| {
                    std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "invalid signing key length",
                    )
                })?;
                let pk_bytes: [u8; 32] = pk_bytes_vec.try_into().map_err(|_| {
                    std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "invalid verifying key length",
                    )
                })?;

                let sk = SigningKey::from_bytes(&sk_bytes);
                let vk = VerifyingKey::from_bytes(&pk_bytes)
                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
                Ok((sk, vk))
            }
            Err(e) => Err(e),
        }
    }
}

fn transcript_hash_path(base: &Path, id_hex: &str) -> PathBuf {
    base.join(id_hex).join("actions.blake3")
}

fn write_transcript_hash(base: &Path, id_hex: &str, h: [u8; 32]) -> std::io::Result<()> {
    atomic_write_replace(&transcript_hash_path(base, id_hex), &h)
}

fn recompute_transcript_hash(base: &Path, id_hex: &str) -> std::io::Result<[u8; 32]> {
    let path = transcript_path(base, id_hex);
    let mut hasher = blake3::Hasher::new();
    if path.exists() {
        let mut f = File::open(&path)?;
        let mut buf = [0u8; 64 * 1024];
        loop {
            let n = f.read(&mut buf)?;
            if n == 0 {
                break;
            }
            hasher.update(&buf[..n]);
        }
    }
    Ok(*hasher.finalize().as_bytes())
}

// ------------------------ Legacy /bundle + /state ------------------------

#[derive(Serialize, Deserialize, Clone)]
#[serde(deny_unknown_fields)]
struct ActionIn {
    g: Vec<u64>,
    nonce: u64,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct BundleReq {
    id: String,
    expected_height: u64,
    /// Legacy field (ignored). Server storage is configured at startup.
    #[allow(dead_code)]
    #[serde(default)]
    dir: Option<String>,
    coeffs: Vec<u64>,
    actions: Vec<ActionIn>,

    // Optional projection params
    #[serde(default)]
    subs_g: Option<Vec<u64>>,
    #[serde(default)]
    mask_degree: Option<usize>,
    #[serde(default)]
    mask_nonce: Option<u64>,
}

#[derive(Serialize)]
struct BundleResp {
    height: u64,
    state_hash: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    proj_unmasked: Option<u64>,
    #[cfg(feature = "torsor_masking")]
    #[serde(skip_serializing_if = "Option::is_none")]
    proj_masked: Option<u64>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct StateReq {
    id: String,
    /// Legacy field (ignored). Server storage is configured at startup.
    #[allow(dead_code)]
    #[serde(default)]
    dir: Option<String>,
    #[serde(default)]
    subs_g: Option<Vec<u64>>,
    #[serde(default)]
    mask_degree: Option<usize>,
    #[serde(default)]
    mask_nonce: Option<u64>,
}

#[derive(Serialize)]
struct StateResp {
    ok: bool,
    height: u64,
    state_hash: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    proj_unmasked: Option<u64>,
    #[cfg(feature = "torsor_masking")]
    #[serde(skip_serializing_if = "Option::is_none")]
    proj_masked: Option<u64>,
}

async fn handle_bundle(
    cfg: Arc<ServerConfig>,
    req: BundleReq,
) -> Result<impl warp::Reply, warp::Rejection> {
    let id = match parse_id_hex64_checked(&req.id) {
        Some(x) => x,
        None => return Err(bad_request("invalid id; expected 64 hex chars")),
    };
    let idh = id_hex(&id);
    if req.actions.len() > cfg.max_actions_per_request {
        return Err(bad_request("too many actions"));
    }
    if !action_inputs_are_valid(&req.actions, cfg.max_coeffs_len) {
        return Err(bad_request(
            "action polynomial is empty, too long, or noncanonical",
        ));
    }
    if req.coeffs.len() > cfg.max_coeffs_len {
        return Err(bad_request("coeffs too long"));
    }
    if !req.coeffs.is_empty() && !coefficients_are_canonical(&req.coeffs) {
        return Err(bad_request("coeffs must be canonical field values"));
    }
    if manifest_path(&cfg.data_dir, &idh).exists() {
        return Err(conflict("id is managed as a vault; use /vault/* endpoints"));
    }

    // lock per ID to prevent concurrent mutation races
    let lock_arc = {
        let mut m = LOCKS.lock().await;
        m.entry(id)
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    };

    let dur = timeout_dur(cfg.request_timeout_ms);
    let lock_arc2 = lock_arc.clone();
    let cfg2 = cfg.clone();
    let res = maybe_timeout(dur, async move {
        let _guard = lock_arc2.lock().await;
        // Restore or bootstrap
        let store = FsStore::new(&cfg2.data_dir).map_err(|_| internal("failed to init store"))?;
        let mut ch = match Channel::restore(id, Box::new(store)) {
            Ok(ch) => {
                if !req.coeffs.is_empty() {
                    let requested = CompanionMatrix::from_coeffs(&req.coeffs);
                    if requested.matrix() != ch.companion().matrix() {
                        return Err(conflict("coeffs do not match the existing channel"));
                    }
                }
                ch
            }
            Err(ChannelError::SnapshotNotFound) => {
                if req.coeffs.is_empty() {
                    return Err(bad_request("coeffs must be non-empty when bootstrapping"));
                }
                let c = CompanionMatrix::from_coeffs(&req.coeffs);
                let k = c.dimension();
                let mut s = vec![Field::zero(); k];
                s[0] = Field::new(1);
                Channel::try_open(c, s, id).map_err(|_| internal("failed to initialize channel"))?
            }
            Err(_) => return Err(internal("existing channel state is corrupt or unreadable")),
        };
        if ch.height() != req.expected_height {
            return Err(conflict("expected_height does not match channel height"));
        }
        if !matches!(
            action_work(&req.actions, ch.dimension()),
            Some(work) if work <= cfg2.max_action_work
        ) {
            return Err(bad_request("action bundle exceeds the computation budget"));
        }
        if !ch.has_store() {
            let runtime_store =
                FsStore::new(&cfg2.data_dir).map_err(|_| internal("failed to init store"))?;
            ch = ch.with_store(Box::new(runtime_store));
        }

        // Validate the complete bundle before committing its ordered actions.
        let actions: Vec<Action> = req
            .actions
            .iter()
            .map(|a| Action {
                g_coeffs: a.g.clone(),
                nonce: a.nonce,
                mask_row0: None,
            })
            .collect();
        crate::apply_bundle(&mut ch, &actions).map_err(|error| {
            if error.committed == 0 {
                bad_request("invalid action bundle")
            } else {
                internal("bundle storage failed after a committed prefix")
            }
        })?;

        // Optional projections
        let mut proj_unmasked: Option<u64> = None;
        #[cfg(feature = "torsor_masking")]
        let mut proj_masked: Option<u64> = None;

        if let Some(g) = &req.subs_g {
            if g.len() > cfg2.max_coeffs_len {
                return Err(bad_request("projection polynomial too long"));
            }
            let row = try_row0_of_poly(ch.companion(), g)
                .map_err(|_| bad_request("invalid projection polynomial"))?;
            let mut acc = Field::zero();
            for (w, x) in row.iter().zip(ch.state().iter()) {
                acc += *w * *x;
            }
            proj_unmasked = Some(acc.value());

            #[cfg(feature = "torsor_masking")]
            if let (Some(deg), Some(nonce)) = (req.mask_degree, req.mask_nonce) {
                if deg > cfg2.max_coeffs_len {
                    return Err(bad_request("mask degree too large"));
                }
                let row_t = try_apply_twist_row(ch.companion(), &row, nonce, deg)
                    .map_err(|_| bad_request("invalid projection twist"))?;
                let mut acc2 = Field::zero();
                for (w, x) in row_t.iter().zip(ch.state().iter()) {
                    acc2 += *w * *x;
                }
                proj_masked = Some(acc2.value());
            }
        }

        // Persist & reply
        ch.save_now()
            .map_err(|_| internal("failed to persist snapshot"))?;
        let resp = BundleResp {
            height: ch.height(),
            state_hash: hash_state(ch.state()),
            proj_unmasked,
            #[cfg(feature = "torsor_masking")]
            proj_masked,
        };
        Ok(warp::reply::with_status(
            warp::reply::json(&resp),
            StatusCode::OK,
        ))
    })
    .await;
    cleanup_lock(id, &lock_arc).await;
    res
}

async fn handle_state(
    cfg: Arc<ServerConfig>,
    req: StateReq,
) -> Result<impl warp::Reply, warp::Rejection> {
    let id = match parse_id_hex64_checked(&req.id) {
        Some(x) => x,
        None => return Err(bad_request("invalid id; expected 64 hex chars")),
    };
    let idh = id_hex(&id);
    if manifest_path(&cfg.data_dir, &idh).exists() {
        return Err(conflict("id is managed as a vault; use /vault/* endpoints"));
    }

    // lock per ID (consistent with /bundle)
    let lock_arc = {
        let mut m = LOCKS.lock().await;
        m.entry(id)
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    };

    let dur = timeout_dur(cfg.request_timeout_ms);
    let lock_arc2 = lock_arc.clone();
    let cfg2 = cfg.clone();
    let res = maybe_timeout(dur, async move {
        let _guard = lock_arc2.lock().await;
        let store = FsStore::new(&cfg2.data_dir).map_err(|_| internal("failed to init store"))?;
        let ch = match Channel::restore(id, Box::new(store)) {
            Ok(ch) => ch,
            Err(_) => return Err(not_found("channel not found")),
        };

        let mut proj_unmasked: Option<u64> = None;
        #[cfg(feature = "torsor_masking")]
        let mut proj_masked: Option<u64> = None;

        if let Some(g) = &req.subs_g {
            if g.len() > cfg2.max_coeffs_len {
                return Err(bad_request("projection polynomial too long"));
            }
            let row = try_row0_of_poly(ch.companion(), g)
                .map_err(|_| bad_request("invalid projection polynomial"))?;
            let mut acc = Field::zero();
            for (w, x) in row.iter().zip(ch.state().iter()) {
                acc += *w * *x;
            }
            proj_unmasked = Some(acc.value());

            #[cfg(feature = "torsor_masking")]
            if let (Some(deg), Some(nonce)) = (req.mask_degree, req.mask_nonce) {
                if deg > cfg2.max_coeffs_len {
                    return Err(bad_request("mask degree too large"));
                }
                let row_t = try_apply_twist_row(ch.companion(), &row, nonce, deg)
                    .map_err(|_| bad_request("invalid projection twist"))?;
                let mut acc2 = Field::zero();
                for (w, x) in row_t.iter().zip(ch.state().iter()) {
                    acc2 += *w * *x;
                }
                proj_masked = Some(acc2.value());
            }
        }

        let resp = StateResp {
            ok: true,
            height: ch.height(),
            state_hash: hash_state(ch.state()),
            proj_unmasked,
            #[cfg(feature = "torsor_masking")]
            proj_masked,
        };
        Ok(warp::reply::with_status(
            warp::reply::json(&resp),
            StatusCode::OK,
        ))
    })
    .await;
    cleanup_lock(id, &lock_arc).await;
    res
}

// ------------------------ Vault API ------------------------

#[derive(Serialize, Deserialize, Clone, PartialEq, Eq, Default)]
#[serde(deny_unknown_fields)]
struct VaultPolicy {
    #[serde(default)]
    nonce_monotonic: bool,
    #[serde(default)]
    min_nonce: Option<u64>,
    #[serde(default)]
    projection_whitelist: Option<Vec<Vec<u64>>>,
}

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct VaultManifest {
    #[serde(default)]
    protocol_version: u32,
    id: String,
    coeffs: Vec<u64>,
    policy: VaultPolicy,
    created_unix: u64,
}

const VAULT_PROTOCOL_VERSION: u32 = 2;

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct VaultNewReq {
    id: String,
    /// Legacy field (ignored). Server storage is configured at startup.
    #[allow(dead_code)]
    #[serde(default)]
    dir: Option<String>,
    coeffs: Vec<u64>,
    #[serde(default)]
    policy: Option<VaultPolicy>,
}

#[derive(Serialize)]
struct VaultTokensResp {
    read: String,
    append: String,
}

#[derive(Serialize)]
struct VaultNewResp {
    ok: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    tokens: Option<VaultTokensResp>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct VaultTokenRotateReq {
    id: String,
}

#[derive(Serialize)]
struct VaultTokenRotateResp {
    ok: bool,
    tokens: VaultTokensResp,
}

const VAULT_AUTHZ_VERSION: u32 = 1;
const MAX_VAULT_TOKENS_PER_SCOPE: usize = 2;

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct VaultAuthzFile {
    version: u32,
    #[serde(default)]
    read: Vec<String>,
    #[serde(default)]
    append: Vec<String>,
    updated_unix: u64,
}

#[derive(Clone, Default)]
struct VaultAuthz {
    read: Vec<[u8; 32]>,
    append: Vec<[u8; 32]>,
}

const VAULT_AUTHZ_CACHE_TTL: Duration = Duration::from_secs(5);
const VAULT_AUTHZ_CACHE_MAX_ENTRIES: usize = 50_000;

#[derive(Clone)]
struct VaultAuthzCacheEntry {
    authz: Arc<VaultAuthz>,
    mtime: Option<std::time::SystemTime>,
    expires_at: Instant,
    last_access: Instant,
}

type VaultAuthzCache = RwLock<HashMap<String, VaultAuthzCacheEntry>>;

#[derive(Copy, Clone, Debug)]
enum VaultPermission {
    Read,
    Append,
}

fn vault_authz_path(base: &Path, id_hex: &str) -> PathBuf {
    base.join(id_hex).join("authz.json")
}

fn unix_now_secs() -> Result<u64, warp::Rejection> {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_err(|_| internal("system time before unix epoch"))
        .map(|d| d.as_secs())
}

fn generate_vault_token() -> String {
    let mut bytes = [0u8; 32];
    OsRng.fill_bytes(&mut bytes);
    B64URL.encode(bytes)
}

fn vault_token_hash(token: &str) -> [u8; 32] {
    use sha2::{Digest, Sha256};
    let mut h = Sha256::new();
    h.update(b"HOP_VAULT_TOKEN_V1\0");
    h.update(token.as_bytes());
    h.finalize().into()
}

fn vault_token_hash_hex(token: &str) -> String {
    hex::encode(vault_token_hash(token))
}

fn parse_hash_hex32(s: &str) -> Option<[u8; 32]> {
    if s.len() != 64 {
        return None;
    }
    let bytes = hex::decode(s).ok()?;
    bytes.try_into().ok()
}

fn parse_vault_authz(file: &VaultAuthzFile) -> Option<VaultAuthz> {
    if file.version != VAULT_AUTHZ_VERSION {
        return None;
    }

    let read = file
        .read
        .iter()
        .filter_map(|x| parse_hash_hex32(x))
        .collect();
    let append = file
        .append
        .iter()
        .filter_map(|x| parse_hash_hex32(x))
        .collect();
    Some(VaultAuthz { read, append })
}

#[derive(Copy, Clone, Debug)]
enum VaultAuthzReadError {
    Io(std::io::ErrorKind),
    Invalid,
}

async fn read_vault_authz_file(
    base: &Path,
    id_hex: &str,
) -> Result<Option<(VaultAuthz, Option<std::time::SystemTime>)>, VaultAuthzReadError> {
    let p = vault_authz_path(base, id_hex);

    let mtime = match tokio::fs::metadata(&p).await {
        Ok(md) => md.modified().ok(),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(VaultAuthzReadError::Io(e.kind())),
    };

    let raw = match tokio::fs::read(&p).await {
        Ok(raw) => raw,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(VaultAuthzReadError::Io(e.kind())),
    };

    let file: VaultAuthzFile =
        serde_json::from_slice(&raw).map_err(|_| VaultAuthzReadError::Invalid)?;
    let Some(authz) = parse_vault_authz(&file) else {
        return Err(VaultAuthzReadError::Invalid);
    };
    Ok(Some((authz, mtime)))
}

fn evict_vault_authz_cache_if_needed(
    map: &mut HashMap<String, VaultAuthzCacheEntry>,
    metrics: &Metrics,
) {
    if map.len() <= VAULT_AUTHZ_CACHE_MAX_ENTRIES {
        return;
    }

    // Remove least-recently-used entries to bound memory usage under many vault IDs.
    let mut items: Vec<(String, Instant)> = map
        .iter()
        .map(|(k, v)| (k.clone(), v.last_access))
        .collect();
    items.sort_by_key(|(_, t)| *t);

    let remove = map.len().saturating_sub(VAULT_AUTHZ_CACHE_MAX_ENTRIES);
    for (k, _) in items.into_iter().take(remove) {
        map.remove(&k);
    }
    if remove > 0 {
        metrics
            .authz_cache_total
            .with_label_values(&["evict"])
            .inc_by(remove as u64);
    }
}

async fn get_vault_authz_cached(
    cfg: &ServerConfig,
    cache: &Arc<VaultAuthzCache>,
    metrics: &Metrics,
    id_hex: &str,
) -> Option<Arc<VaultAuthz>> {
    let now = Instant::now();

    // Fast path: unexpired cache entry.
    {
        let map = cache.read().await;
        if let Some(e) = map.get(id_hex) {
            if now < e.expires_at {
                metrics.authz_cache_total.with_label_values(&["hit"]).inc();
                return Some(e.authz.clone());
            }
        }
    }

    metrics.authz_cache_total.with_label_values(&["miss"]).inc();

    let p = vault_authz_path(&cfg.data_dir, id_hex);

    // If we have an entry and the authz file mtime is unchanged, just extend the TTL.
    let cached_mtime = {
        let map = cache.read().await;
        map.get(id_hex).and_then(|e| e.mtime)
    };
    let current_mtime = match tokio::fs::metadata(&p).await {
        Ok(md) => md.modified().ok(),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
        Err(e) => {
            tracing::warn!(error = %e, "authz metadata read failed");
            metrics
                .authz_cache_total
                .with_label_values(&["meta_error"])
                .inc();
            None
        }
    };

    if let (Some(a), Some(b)) = (cached_mtime, current_mtime) {
        if a == b {
            let mut map = cache.write().await;
            let authz = if let Some(e) = map.get_mut(id_hex) {
                e.mtime = Some(b);
                e.expires_at = now + VAULT_AUTHZ_CACHE_TTL;
                e.last_access = now;
                Some(e.authz.clone())
            } else {
                None
            };
            if let Some(authz) = authz {
                metrics
                    .authz_cache_total
                    .with_label_values(&["revalidated"])
                    .inc();
                metrics.authz_cache_entries.set(map.len() as i64);
                return Some(authz);
            }
        }
    }

    // Slow path: reload from disk.
    match read_vault_authz_file(&cfg.data_dir, id_hex).await {
        Ok(Some((authz, mtime))) => {
            let authz = Arc::new(authz);
            let mut map = cache.write().await;
            map.insert(
                id_hex.to_string(),
                VaultAuthzCacheEntry {
                    authz: authz.clone(),
                    mtime,
                    expires_at: now + VAULT_AUTHZ_CACHE_TTL,
                    last_access: now,
                },
            );
            evict_vault_authz_cache_if_needed(&mut map, metrics);
            metrics
                .authz_cache_total
                .with_label_values(&["reload"])
                .inc();
            metrics.authz_cache_entries.set(map.len() as i64);
            Some(authz)
        }
        Ok(None) => {
            let mut map = cache.write().await;
            map.remove(id_hex);
            metrics
                .authz_cache_total
                .with_label_values(&["notfound"])
                .inc();
            metrics.authz_cache_entries.set(map.len() as i64);
            None
        }
        Err(VaultAuthzReadError::Invalid) => {
            tracing::warn!("authz file invalid");
            let mut map = cache.write().await;
            map.remove(id_hex);
            metrics
                .authz_cache_total
                .with_label_values(&["invalid"])
                .inc();
            metrics.authz_cache_entries.set(map.len() as i64);
            None
        }
        Err(VaultAuthzReadError::Io(kind)) => {
            tracing::warn!(error_kind = ?kind, "authz read failed");
            metrics
                .authz_cache_total
                .with_label_values(&["io_error"])
                .inc();
            None
        }
    }
}

fn token_hash_matches_any(hash: &[u8; 32], list: &[[u8; 32]]) -> bool {
    list.iter()
        .any(|h| constant_time_eq(h.as_slice(), hash.as_slice()))
}

fn perm_label(perm: VaultPermission) -> &'static str {
    match perm {
        VaultPermission::Read => "read",
        VaultPermission::Append => "append",
    }
}

async fn authorize_vault(
    cfg: &ServerConfig,
    cache: &Arc<VaultAuthzCache>,
    metrics: &Metrics,
    authn: &Authn,
    id_hex: &str,
    perm: VaultPermission,
) -> Result<(), warp::Rejection> {
    let p = perm_label(perm);

    if cfg.auth_token.is_none() {
        metrics
            .authz_checks_total
            .with_label_values(&[p, "auth_disabled"])
            .inc();
        return Ok(());
    }

    if matches!(authn, Authn::Admin) {
        metrics
            .authz_checks_total
            .with_label_values(&[p, "allow_admin"])
            .inc();
        return Ok(());
    }

    let Some(token) = authn.bearer_token() else {
        metrics
            .authz_checks_total
            .with_label_values(&[p, "unauthorized"])
            .inc();
        return Err(warp::reject::custom(Unauthorized));
    };

    let Some(authz) = get_vault_authz_cached(cfg, cache, metrics, id_hex).await else {
        metrics
            .authz_checks_total
            .with_label_values(&[p, "unauthorized_missing"])
            .inc();
        return Err(warp::reject::custom(Unauthorized));
    };

    let hash = vault_token_hash(token);
    let is_append = token_hash_matches_any(&hash, &authz.append);
    let is_read = token_hash_matches_any(&hash, &authz.read);

    match perm {
        VaultPermission::Read => {
            if is_append || is_read {
                metrics
                    .authz_checks_total
                    .with_label_values(&[p, "allow_token"])
                    .inc();
                Ok(())
            } else {
                metrics
                    .authz_checks_total
                    .with_label_values(&[p, "unauthorized"])
                    .inc();
                Err(warp::reject::custom(Unauthorized))
            }
        }
        VaultPermission::Append => {
            if is_append {
                metrics
                    .authz_checks_total
                    .with_label_values(&[p, "allow_token"])
                    .inc();
                Ok(())
            } else if is_read {
                metrics
                    .authz_checks_total
                    .with_label_values(&[p, "forbidden"])
                    .inc();
                Err(forbidden("forbidden"))
            } else {
                metrics
                    .authz_checks_total
                    .with_label_values(&[p, "unauthorized"])
                    .inc();
                Err(warp::reject::custom(Unauthorized))
            }
        }
    }
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct VaultAppendReq {
    id: String,
    expected_height: u64,
    /// Legacy field (ignored). Server storage is configured at startup.
    #[allow(dead_code)]
    #[serde(default)]
    dir: Option<String>,
    actions: Vec<ActionIn>,
}

#[derive(Serialize)]
struct VaultAppendResp {
    ok: bool,
    height: u64,
    state_hash: String,
    transcript_hash: String, // base64
    pubkey: String,          // base64
    sig: String,             // base64
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct VaultGetReq {
    id: String,
    /// Legacy field (ignored). Server storage is configured at startup.
    #[allow(dead_code)]
    #[serde(default)]
    dir: Option<String>,
    #[serde(default)]
    subs_g: Option<Vec<u64>>,
    #[serde(default)]
    mask_degree: Option<usize>,
    #[serde(default)]
    mask_nonce: Option<u64>,
}

#[derive(Serialize)]
struct VaultGetResp {
    ok: bool,
    height: u64,
    state_hash: String,
    manifest: VaultManifest,
    #[serde(skip_serializing_if = "Option::is_none")]
    proj_unmasked: Option<u64>,
    #[cfg(feature = "torsor_masking")]
    #[serde(skip_serializing_if = "Option::is_none")]
    proj_masked: Option<u64>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct VaultProveReq {
    id: String,
    /// Legacy field (ignored). Server storage is configured at startup.
    #[allow(dead_code)]
    #[serde(default)]
    dir: Option<String>,
    #[serde(default)]
    since_height: Option<u64>,
    #[serde(default)]
    limit: Option<usize>,
}

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct TranscriptEntry {
    height_after: u64,
    state_hash_after: String,
    action: ActionIn,
}

#[derive(Serialize)]
struct VaultProveResp {
    ok: bool,
    entries: Vec<TranscriptEntry>,
}

fn manifest_path(base: &Path, id_hex: &str) -> PathBuf {
    // Standardize on manifest.json so CLI tools can find it
    base.join(id_hex).join("manifest.json")
}
fn transcript_path(base: &Path, id_hex: &str) -> PathBuf {
    base.join(id_hex).join("actions.jsonl")
}

fn vault_genesis_state(c: &CompanionMatrix) -> Vec<Field> {
    let mut s = vec![Field::zero(); c.dimension()];
    if c.dimension() > 0 {
        s[0] = Field::new(1);
    }
    s
}

fn restore_vault_state(
    base: &Path,
    id: [u8; 32],
    id_hex: &str,
    c: &CompanionMatrix,
) -> Result<(Vec<Field>, u64, Option<u64>), warp::Rejection> {
    let store = FsStore::new(base).map_err(|_| internal("failed to init store"))?;

    let mut state = vault_genesis_state(c);
    let mut height: u64 = 0;
    if let Some(snap) = store
        .load(&id)
        .map_err(|_| internal("failed to load snapshot"))?
    {
        if snap.c.dimension() != c.dimension() || snap.state.len() != c.dimension() {
            return Err(internal("snapshot incompatible with manifest coeffs"));
        }
        state = snap.state;
        height = snap.height;
    }

    let tpath = transcript_path(base, id_hex);
    let f = match File::open(&tpath) {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            if height == 0 {
                return Ok((state, height, None));
            }
            return Err(internal("transcript missing"));
        }
        Err(_) => return Err(internal("failed to open transcript")),
    };

    let mut max_height_seen: u64 = 0;
    let mut expected_transcript_height = 1_u64;
    let mut last_nonce = None;
    let reader = BufReader::new(f);
    for line in reader.lines() {
        let line = line.map_err(|_| internal("failed to read transcript"))?;
        if line.trim().is_empty() {
            continue;
        }
        let e: TranscriptEntry =
            serde_json::from_str(&line).map_err(|_| internal("corrupt transcript"))?;
        if e.height_after != expected_transcript_height {
            return Err(internal("transcript height discontinuity"));
        }
        expected_transcript_height = expected_transcript_height
            .checked_add(1)
            .ok_or_else(|| internal("transcript height overflow"))?;
        last_nonce = Some(e.action.nonce);
        max_height_seen = max_height_seen.max(e.height_after);

        if e.height_after <= height {
            continue;
        }

        if e.height_after != height.saturating_add(1) {
            return Err(internal("transcript height discontinuity"));
        }

        let action = Action {
            g_coeffs: e.action.g.clone(),
            nonce: e.action.nonce,
            mask_row0: None,
        };
        state = apply_action(c, &state, &action)
            .map_err(|_| internal("transcript contains invalid action"))?;
        height = height
            .checked_add(1)
            .ok_or_else(|| internal("transcript height overflow"))?;

        let computed = hash_state(&state);
        if computed != e.state_hash_after {
            return Err(internal("transcript state hash mismatch"));
        }
    }

    if max_height_seen < height {
        return Err(internal("snapshot ahead of transcript"));
    }

    Ok((state, height, last_nonce))
}

async fn vault_new(
    cfg: Arc<ServerConfig>,
    req: VaultNewReq,
) -> Result<impl warp::Reply, warp::Rejection> {
    let id = match parse_id_hex64_checked(&req.id) {
        Some(x) => x,
        None => return Err(bad_request("invalid id; expected 64 hex chars")),
    };
    if req.coeffs.is_empty() {
        return Err(bad_request("coeffs must be non-empty"));
    }
    if req.coeffs.len() > cfg.max_coeffs_len {
        return Err(bad_request("coeffs too long"));
    }
    if !coefficients_are_canonical(&req.coeffs) {
        return Err(bad_request("coeffs must be canonical field values"));
    }
    let idh = id_hex(&id);

    // lock per ID
    let lock_arc = {
        let mut m = LOCKS.lock().await;
        m.entry(id)
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    };

    let dur = timeout_dur(cfg.request_timeout_ms);
    let lock_arc2 = lock_arc.clone();
    let cfg2 = cfg.clone();
    let res = maybe_timeout(dur, async move {
        let _guard = lock_arc2.lock().await;
        let mut tokens: Option<VaultTokensResp> = None;

        // create directory
        let vault_dir = ensure_vault_dir(&cfg2.data_dir, &idh)
            .map_err(|_| internal("failed to create vault dir"))?;
        let man_path = manifest_path(&cfg2.data_dir, &idh);

        if man_path.exists() {
            // Existing vault: ensure request is consistent.
            let raw = std::fs::read(&man_path).map_err(|_| internal("failed to read manifest"))?;
            let existing: VaultManifest = serde_json::from_slice(&raw)
                .map_err(|_| bad_request("manifest is invalid JSON"))?;

            if existing.protocol_version != VAULT_PROTOCOL_VERSION {
                return Err(conflict("vault uses an unsupported protocol version"));
            }
            if parse_id_hex64_checked(&existing.id) != Some(id)
                || !coefficients_are_canonical(&existing.coeffs)
            {
                return Err(internal("vault manifest violates protocol invariants"));
            }

            if existing.coeffs != req.coeffs {
                return Err(conflict("vault already exists with different coeffs"));
            }
            if let Some(p) = &req.policy {
                if *p != existing.policy {
                    return Err(conflict("vault already exists with different policy"));
                }
            }
        } else {
            if req.coeffs.is_empty() {
                return Err(bad_request("coeffs must be non-empty"));
            }

            let policy = req.policy.unwrap_or_default();
            if policy.projection_whitelist.as_ref().is_some_and(|entries| {
                entries.iter().any(|entry| {
                    entry.len() > cfg2.max_coeffs_len || !coefficients_are_canonical(entry)
                })
            }) {
                return Err(bad_request(
                    "projection whitelist contains an invalid polynomial",
                ));
            }
            let created_unix = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_err(|_| internal("system time before unix epoch"))?
                .as_secs();
            let manifest = VaultManifest {
                protocol_version: VAULT_PROTOCOL_VERSION,
                id: req.id.clone(),
                coeffs: req.coeffs.clone(),
                policy,
                created_unix,
            };

            let manifest_bytes = serde_json::to_vec_pretty(&manifest)
                .map_err(|_| internal("failed to serialize manifest"))?;
            atomic_write_replace(&man_path, &manifest_bytes)
                .map_err(|_| internal("failed to write manifest"))?;

            // Bootstrap channel
            let c = CompanionMatrix::from_coeffs(&manifest.coeffs);
            let k = c.dimension();
            let mut s = vec![Field::zero(); k];
            s[0] = Field::new(1);
            let store =
                FsStore::new(&cfg2.data_dir).map_err(|_| internal("failed to init store"))?;
            let ch = Channel::try_open(c, s, id)
                .map_err(|_| internal("failed to initialize vault channel"))?
                .with_store(Box::new(store));
            ch.save_now()
                .map_err(|_| internal("failed to save snapshot"))?;

            // Touch transcript
            OpenOptions::new()
                .create(true)
                .append(true)
                .open(vault_dir.join("actions.jsonl"))
                .map_err(|_| internal("failed to open transcript"))?;

            // Initialize transcript hash
            if let Ok(init_h) = recompute_transcript_hash(&cfg2.data_dir, &idh) {
                let _ = write_transcript_hash(&cfg2.data_dir, &idh, init_h);
            }

            // Bootstrap per-vault capability tokens when auth is enabled.
            if cfg2.auth_token.is_some() {
                let read_token = generate_vault_token();
                let append_token = generate_vault_token();

                let authz = VaultAuthzFile {
                    version: VAULT_AUTHZ_VERSION,
                    read: vec![vault_token_hash_hex(&read_token)],
                    append: vec![vault_token_hash_hex(&append_token)],
                    updated_unix: unix_now_secs()?,
                };

                let authz_bytes = serde_json::to_vec_pretty(&authz)
                    .map_err(|_| internal("failed to serialize authz"))?;
                let p = vault_authz_path(&cfg2.data_dir, &idh);
                match write_private_create_new(&p, &authz_bytes) {
                    Ok(()) => {
                        tokens = Some(VaultTokensResp {
                            read: read_token,
                            append: append_token,
                        });
                    }
                    Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {}
                    Err(_) => return Err(internal("failed to write vault authz")),
                }
            }
        }

        Ok(warp::reply::with_status(
            warp::reply::json(&VaultNewResp { ok: true, tokens }),
            StatusCode::OK,
        ))
    })
    .await;

    cleanup_lock(id, &lock_arc).await;
    res
}

async fn vault_token_rotate(
    cfg: Arc<ServerConfig>,
    authz_cache: Arc<VaultAuthzCache>,
    metrics: Arc<Metrics>,
    req: VaultTokenRotateReq,
) -> Result<impl warp::Reply, warp::Rejection> {
    if cfg.auth_token.is_none() {
        return Err(bad_request("auth is disabled"));
    }

    let id = match parse_id_hex64_checked(&req.id) {
        Some(x) => x,
        None => return Err(bad_request("invalid id; expected 64 hex chars")),
    };
    let idh = id_hex(&id);

    // lock per ID
    let lock_arc = {
        let mut m = LOCKS.lock().await;
        m.entry(id)
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    };

    let dur = timeout_dur(cfg.request_timeout_ms);
    let lock_arc2 = lock_arc.clone();
    let cfg2 = cfg.clone();
    let authz_cache2 = authz_cache.clone();
    let metrics2 = metrics.clone();
    let res = maybe_timeout(dur, async move {
        let _guard = lock_arc2.lock().await;

        let man_path = manifest_path(&cfg2.data_dir, &idh);
        if !man_path.exists() {
            return Err(not_found("vault not found"));
        }

        let p = vault_authz_path(&cfg2.data_dir, &idh);
        let mut authz = std::fs::read(&p)
            .ok()
            .and_then(|raw| serde_json::from_slice::<VaultAuthzFile>(&raw).ok())
            .filter(|f| f.version == VAULT_AUTHZ_VERSION)
            .unwrap_or(VaultAuthzFile {
                version: VAULT_AUTHZ_VERSION,
                read: Vec::new(),
                append: Vec::new(),
                updated_unix: unix_now_secs()?,
            });

        // Keep the file tidy even if previous entries were malformed.
        authz.read.retain(|x| parse_hash_hex32(x).is_some());
        authz.append.retain(|x| parse_hash_hex32(x).is_some());

        let read_token = generate_vault_token();
        let append_token = generate_vault_token();
        authz.read.insert(0, vault_token_hash_hex(&read_token));
        authz.append.insert(0, vault_token_hash_hex(&append_token));
        authz.read.truncate(MAX_VAULT_TOKENS_PER_SCOPE);
        authz.append.truncate(MAX_VAULT_TOKENS_PER_SCOPE);
        authz.updated_unix = unix_now_secs()?;

        let authz_bytes =
            serde_json::to_vec_pretty(&authz).map_err(|_| internal("failed to serialize authz"))?;
        atomic_write_replace_private(&p, &authz_bytes)
            .map_err(|_| internal("failed to write vault authz"))?;

        // Update in-memory cache immediately so freshly-rotated tokens work without waiting
        // for the cache TTL to expire.
        if let Some(parsed) = parse_vault_authz(&authz) {
            let now = Instant::now();
            let mtime = std::fs::metadata(&p).ok().and_then(|md| md.modified().ok());
            let mut map = authz_cache2.write().await;
            map.insert(
                idh.clone(),
                VaultAuthzCacheEntry {
                    authz: Arc::new(parsed),
                    mtime,
                    expires_at: now + VAULT_AUTHZ_CACHE_TTL,
                    last_access: now,
                },
            );
            evict_vault_authz_cache_if_needed(&mut map, metrics2.as_ref());
            metrics2
                .authz_cache_total
                .with_label_values(&["store"])
                .inc();
            metrics2.authz_cache_entries.set(map.len() as i64);
        }

        let resp = VaultTokenRotateResp {
            ok: true,
            tokens: VaultTokensResp {
                read: read_token,
                append: append_token,
            },
        };

        Ok(warp::reply::with_status(
            warp::reply::json(&resp),
            StatusCode::OK,
        ))
    })
    .await;

    cleanup_lock(id, &lock_arc).await;
    res
}

async fn vault_append(
    authn: Authn,
    cfg: Arc<ServerConfig>,
    authz_cache: Arc<VaultAuthzCache>,
    metrics: Arc<Metrics>,
    req: VaultAppendReq,
) -> Result<impl warp::Reply, warp::Rejection> {
    let id = match parse_id_hex64_checked(&req.id) {
        Some(x) => x,
        None => return Err(bad_request("invalid id; expected 64 hex chars")),
    };
    let idh = id_hex(&id);
    authorize_vault(
        cfg.as_ref(),
        &authz_cache,
        metrics.as_ref(),
        &authn,
        &idh,
        VaultPermission::Append,
    )
    .await?;

    // lock per vault
    let lock_arc = {
        let mut m = LOCKS.lock().await;
        m.entry(id)
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    };

    let dur = timeout_dur(cfg.request_timeout_ms);
    let lock_arc2 = lock_arc.clone();
    let cfg2 = cfg.clone();
    let res = maybe_timeout(dur, async move {
        let _guard = lock_arc2.lock().await;
        if req.actions.is_empty() {
            return Err(bad_request("actions must be non-empty"));
        }
        if req.actions.len() > cfg2.max_actions_per_request {
            return Err(bad_request("too many actions"));
        }
        if !action_inputs_are_valid(&req.actions, cfg2.max_coeffs_len) {
            return Err(bad_request(
                "action polynomial is empty, too long, or noncanonical",
            ));
        }

        // server keypair
        let (sk, vk) = load_or_init_server_keypair(&cfg2.data_dir)
            .map_err(|_| internal("failed to init server keypair"))?;

        // read manifest
        let man_path = manifest_path(&cfg2.data_dir, &idh);
        if !man_path.exists() {
            return Err(not_found("vault not found; create first"));
        }
        let manifest_raw =
            std::fs::read(&man_path).map_err(|_| internal("failed to read manifest"))?;
        let manifest: VaultManifest = serde_json::from_slice(&manifest_raw)
            .map_err(|_| bad_request("manifest is invalid JSON"))?;

        if manifest.protocol_version != VAULT_PROTOCOL_VERSION {
            return Err(conflict("vault uses an unsupported protocol version"));
        }
        if parse_id_hex64_checked(&manifest.id) != Some(id)
            || !coefficients_are_canonical(&manifest.coeffs)
        {
            return Err(internal("vault manifest violates protocol invariants"));
        }

        if manifest.coeffs.len() > cfg2.max_coeffs_len {
            return Err(internal("manifest coeffs exceed server limit"));
        }

        let c = CompanionMatrix::from_coeffs(&manifest.coeffs);
        if !matches!(
            action_work(&req.actions, c.dimension()),
            Some(work) if work <= cfg2.max_action_work
        ) {
            return Err(bad_request("action bundle exceeds the computation budget"));
        }
        let (mut state, mut height, last_nonce) =
            restore_vault_state(&cfg2.data_dir, id, &idh, &c)?;
        if height != req.expected_height {
            return Err(conflict("expected_height does not match vault height"));
        }

        // Enforce persistent vault policy before computing any new state.
        if manifest.policy.nonce_monotonic || manifest.policy.min_nonce.is_some() {
            let min_n = manifest.policy.min_nonce.unwrap_or(0);
            for a in &req.actions {
                if a.nonce < min_n {
                    return Err(bad_request("nonce below policy minimum"));
                }
            }
            if manifest.policy.nonce_monotonic {
                // Enforce nondecreasing order from the persisted transcript onward.
                let mut prev = last_nonce;
                for a in &req.actions {
                    if let Some(p) = prev {
                        if a.nonce < p {
                            return Err(bad_request("nonce not monotonic in batch"));
                        }
                    }
                    prev = Some(a.nonce);
                }
            }
        }

        if let Some(allow) = &manifest.policy.projection_whitelist {
            if req
                .actions
                .iter()
                .any(|action| !allow.iter().any(|allowed| allowed == &action.g))
            {
                return Err(bad_request("action polynomial is not permitted by policy"));
            }
        }

        // Apply actions in-memory and buffer transcript lines.
        let mut lines: Vec<String> = Vec::with_capacity(req.actions.len());
        for a in req.actions.iter() {
            let action = Action {
                g_coeffs: a.g.clone(),
                nonce: a.nonce,
                mask_row0: None,
            };
            state = apply_action(&c, &state, &action).map_err(|_| bad_request("invalid action"))?;
            height = height
                .checked_add(1)
                .ok_or_else(|| internal("vault height overflow"))?;

            let entry = TranscriptEntry {
                height_after: height,
                state_hash_after: hash_state(&state),
                action: ActionIn {
                    g: a.g.clone(),
                    nonce: a.nonce,
                },
            };
            let line = serde_json::to_string(&entry)
                .map_err(|_| internal("failed to serialize transcript entry"))?;
            lines.push(line);
        }

        // Transactionally append transcript entries; on failure, truncate back.
        let transcript_p = transcript_path(&cfg2.data_dir, &idh);
        let mut tf = OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .truncate(false)
            .open(&transcript_p)
            .map_err(|_| internal("failed to open transcript"))?;
        let start_len = tf
            .metadata()
            .map_err(|_| internal("failed to stat transcript"))?
            .len();
        tf.seek(SeekFrom::End(0))
            .map_err(|_| internal("failed to seek transcript"))?;

        for line in &lines {
            tf.write_all(line.as_bytes())
                .map_err(|_| internal("failed to append transcript"))?;
            tf.write_all(b"\n")
                .map_err(|_| internal("failed to append transcript"))?;
        }

        if let Err(e) = tf.flush().and_then(|_| tf.sync_data()) {
            let _ = tf.set_len(start_len);
            let _ = tf.sync_data();
            let _ = e; // avoid leaking OS errors over the wire
            return Err(internal("failed to sync transcript"));
        }

        let th = match recompute_transcript_hash(&cfg2.data_dir, &idh) {
            Ok(h) => h,
            Err(_) => {
                let _ = tf.set_len(start_len);
                let _ = tf.sync_data();
                return Err(internal("failed to hash transcript"));
            }
        };

        // Persist snapshot once at end.
        let store = FsStore::new(&cfg2.data_dir).map_err(|_| internal("failed to init store"))?;
        let state_h_hex = hash_state(&state);
        let snap = ChannelSnapshot {
            id,
            c: c.clone(),
            state,
            height,
        };
        if let Err(e) = store.save(&snap) {
            let _ = tf.set_len(start_len);
            let _ = tf.sync_data();
            let _ = e;
            return Err(internal("failed to save snapshot"));
        }

        // Best-effort cache of transcript hash (receipt includes it regardless).
        let _ = write_transcript_hash(&cfg2.data_dir, &idh, th);

        let ctx = b"HOP_VAULT_APPEND_V2";
        let msg = [
            ctx.as_slice(),
            idh.as_bytes(),
            &height.to_le_bytes(),
            state_h_hex.as_bytes(),
            &th,
        ]
        .concat();
        let sig: Signature = sk.sign(&msg);

        let resp = VaultAppendResp {
            ok: true,
            height,
            state_hash: state_h_hex,
            transcript_hash: B64.encode(th),
            pubkey: B64.encode(vk.as_bytes()),
            sig: B64.encode(sig.to_bytes()),
        };
        Ok(warp::reply::with_status(
            warp::reply::json(&resp),
            StatusCode::OK,
        ))
    })
    .await;

    cleanup_lock(id, &lock_arc).await;
    res
}

async fn vault_get(
    authn: Authn,
    cfg: Arc<ServerConfig>,
    authz_cache: Arc<VaultAuthzCache>,
    metrics: Arc<Metrics>,
    req: VaultGetReq,
) -> Result<impl warp::Reply, warp::Rejection> {
    let id = match parse_id_hex64_checked(&req.id) {
        Some(x) => x,
        None => return Err(bad_request("invalid id; expected 64 hex chars")),
    };
    let idh = id_hex(&id);
    authorize_vault(
        cfg.as_ref(),
        &authz_cache,
        metrics.as_ref(),
        &authn,
        &idh,
        VaultPermission::Read,
    )
    .await?;

    // lock
    let lock_arc = {
        let mut m = LOCKS.lock().await;
        m.entry(id)
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    };

    let dur = timeout_dur(cfg.request_timeout_ms);
    let lock_arc2 = lock_arc.clone();
    let cfg2 = cfg.clone();
    let res = maybe_timeout(dur, async move {
        let _guard = lock_arc2.lock().await;
        // manifest
        let man_path = manifest_path(&cfg2.data_dir, &idh);
        if !man_path.exists() {
            return Err(not_found("vault not found"));
        }
        let manifest_raw =
            std::fs::read(&man_path).map_err(|_| internal("failed to read manifest"))?;
        let manifest: VaultManifest = serde_json::from_slice(&manifest_raw)
            .map_err(|_| bad_request("manifest is invalid JSON"))?;

        if manifest.protocol_version != VAULT_PROTOCOL_VERSION {
            return Err(conflict("vault uses an unsupported protocol version"));
        }
        if parse_id_hex64_checked(&manifest.id) != Some(id)
            || !coefficients_are_canonical(&manifest.coeffs)
        {
            return Err(internal("vault manifest violates protocol invariants"));
        }

        let c = CompanionMatrix::from_coeffs(&manifest.coeffs);
        let (state, height, _) = restore_vault_state(&cfg2.data_dir, id, &idh, &c)?;

        // projections (optional)
        let mut proj_unmasked: Option<u64> = None;
        #[cfg(feature = "torsor_masking")]
        let mut proj_masked: Option<u64> = None;

        if let Some(g) = &req.subs_g {
            if g.len() > cfg2.max_coeffs_len {
                return Err(bad_request("projection polynomial too long"));
            }
            let row = try_row0_of_poly(&c, g)
                .map_err(|_| bad_request("invalid projection polynomial"))?;
            let mut acc = Field::zero();
            for (w, x) in row.iter().zip(state.iter()) {
                acc += *w * *x;
            }
            proj_unmasked = Some(acc.value());

            #[cfg(feature = "torsor_masking")]
            if let (Some(deg), Some(nonce)) = (req.mask_degree, req.mask_nonce) {
                if deg > cfg2.max_coeffs_len {
                    return Err(bad_request("mask degree too large"));
                }
                let row_t = try_apply_twist_row(&c, &row, nonce, deg)
                    .map_err(|_| bad_request("invalid projection twist"))?;
                let mut acc2 = Field::zero();
                for (w, x) in row_t.iter().zip(state.iter()) {
                    acc2 += *w * *x;
                }
                proj_masked = Some(acc2.value());
            }
        }

        let resp = VaultGetResp {
            ok: true,
            height,
            state_hash: hash_state(&state),
            manifest,
            proj_unmasked,
            #[cfg(feature = "torsor_masking")]
            proj_masked,
        };
        Ok(warp::reply::with_status(
            warp::reply::json(&resp),
            StatusCode::OK,
        ))
    })
    .await;

    cleanup_lock(id, &lock_arc).await;
    res
}

async fn vault_prove(
    authn: Authn,
    cfg: Arc<ServerConfig>,
    authz_cache: Arc<VaultAuthzCache>,
    metrics: Arc<Metrics>,
    req: VaultProveReq,
) -> Result<impl warp::Reply, warp::Rejection> {
    let id = match parse_id_hex64_checked(&req.id) {
        Some(x) => x,
        None => return Err(bad_request("invalid id; expected 64 hex chars")),
    };
    let idh = id_hex(&id);
    authorize_vault(
        cfg.as_ref(),
        &authz_cache,
        metrics.as_ref(),
        &authn,
        &idh,
        VaultPermission::Read,
    )
    .await?;

    // lock
    let lock_arc = {
        let mut m = LOCKS.lock().await;
        m.entry(id)
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    };

    let dur = timeout_dur(cfg.request_timeout_ms);
    let lock_arc2 = lock_arc.clone();
    let cfg2 = cfg.clone();
    let res = maybe_timeout(dur, async move {
        let _guard = lock_arc2.lock().await;
        let man_path = manifest_path(&cfg2.data_dir, &idh);
        if !man_path.exists() {
            return Err(not_found("vault not found"));
        }

        let mut limit = req.limit.unwrap_or(cfg2.max_prove_entries);
        if limit > cfg2.max_prove_entries {
            limit = cfg2.max_prove_entries;
        }

        if limit == 0 {
            let resp = VaultProveResp {
                ok: true,
                entries: Vec::new(),
            };
            return Ok(warp::reply::with_status(
                warp::reply::json(&resp),
                StatusCode::OK,
            ));
        }

        let path = transcript_path(&cfg2.data_dir, &idh);
        let f = match File::open(&path) {
            Ok(f) => f,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                let resp = VaultProveResp {
                    ok: true,
                    entries: Vec::new(),
                };
                return Ok(warp::reply::with_status(
                    warp::reply::json(&resp),
                    StatusCode::OK,
                ));
            }
            Err(_) => return Err(internal("failed to open transcript")),
        };

        let mut entries = Vec::<TranscriptEntry>::new();
        let reader = BufReader::new(f);
        for line in reader.lines() {
            let line = line.map_err(|_| internal("failed to read transcript"))?;
            if line.trim().is_empty() {
                continue;
            }
            let e: TranscriptEntry =
                serde_json::from_str(&line).map_err(|_| internal("corrupt transcript"))?;
            if let Some(since) = req.since_height {
                if e.height_after <= since {
                    continue;
                }
            }
            entries.push(e);
            if entries.len() >= limit {
                break;
            }
        }

        let resp = VaultProveResp { ok: true, entries };
        Ok(warp::reply::with_status(
            warp::reply::json(&resp),
            StatusCode::OK,
        ))
    })
    .await;

    cleanup_lock(id, &lock_arc).await;
    res
}

// ------------------------ Server ------------------------

fn with_cfg(
    cfg: Arc<ServerConfig>,
) -> impl Filter<Extract = (Arc<ServerConfig>,), Error = Infallible> + Clone {
    warp::any().map(move || cfg.clone())
}

fn with_metrics(
    metrics: Arc<Metrics>,
) -> impl Filter<Extract = (Arc<Metrics>,), Error = Infallible> + Clone {
    warp::any().map(move || metrics.clone())
}

fn with_readiness_cache(
    cache: Arc<Mutex<ReadinessCache>>,
) -> impl Filter<Extract = (Arc<Mutex<ReadinessCache>>,), Error = Infallible> + Clone {
    warp::any().map(move || cache.clone())
}

fn with_authz_cache(
    cache: Arc<VaultAuthzCache>,
) -> impl Filter<Extract = (Arc<VaultAuthzCache>,), Error = Infallible> + Clone {
    warp::any().map(move || cache.clone())
}

fn bearer_token_from_header(authz: &str) -> Option<&str> {
    let h = authz.trim();
    let rest = h.strip_prefix("Bearer ")?;
    let token = rest.trim();
    if token.is_empty() || token.len() > 4096 {
        return None;
    }
    Some(token)
}

fn auth_admin(
    cfg: Arc<ServerConfig>,
) -> impl Filter<Extract = (), Error = warp::Rejection> + Clone {
    warp::header::optional::<String>("authorization")
        .and_then(move |authz: Option<String>| {
            let cfg = cfg.clone();
            async move {
                let Some(expected) = cfg.auth_token.as_deref() else {
                    return Ok::<(), warp::Rejection>(());
                };
                let Some(token) = authz.as_deref().and_then(|h| bearer_token_from_header(h)) else {
                    return Err(warp::reject::custom(Unauthorized));
                };

                if constant_time_eq(token.as_bytes(), expected.as_bytes()) {
                    Ok(())
                } else {
                    Err(warp::reject::custom(Unauthorized))
                }
            }
        })
        .untuple_one()
}

fn authn(
    cfg: Arc<ServerConfig>,
) -> impl Filter<Extract = (Authn,), Error = warp::Rejection> + Clone {
    warp::header::optional::<String>("authorization").and_then(move |authz: Option<String>| {
        let cfg = cfg.clone();
        async move {
            let Some(expected) = cfg.auth_token.as_deref() else {
                return Ok::<Authn, warp::Rejection>(Authn::Anonymous);
            };
            let Some(token) = authz.as_deref().and_then(|h| bearer_token_from_header(h)) else {
                return Err(warp::reject::custom(Unauthorized));
            };

            if constant_time_eq(token.as_bytes(), expected.as_bytes()) {
                Ok(Authn::Admin)
            } else {
                Ok(Authn::Token(token.to_string()))
            }
        }
    })
}

fn json_body<T: serde::de::DeserializeOwned + Send>(
    max_body_bytes: u64,
) -> impl Filter<Extract = (T,), Error = warp::Rejection> + Clone {
    warp::body::content_length_limit(max_body_bytes).and(warp::body::json())
}

fn cors_allowed_origin(cfg: &ServerConfig, origin: Option<&str>) -> Option<String> {
    if cfg.cors_allow_any_origin {
        return Some("*".to_string());
    }
    if cfg.cors_allow_origins.is_empty() {
        return None;
    }
    let o = origin?;
    if cfg.cors_allow_origins.iter().any(|x| x == o) {
        Some(o.to_string())
    } else {
        None
    }
}

fn apply_cors_headers(cfg: &ServerConfig, origin: Option<&str>, headers: &mut header::HeaderMap) {
    let Some(allow_origin) = cors_allowed_origin(cfg, origin) else {
        return;
    };

    if allow_origin == "*" {
        headers.insert(
            header::ACCESS_CONTROL_ALLOW_ORIGIN,
            header::HeaderValue::from_static("*"),
        );
    } else if let Ok(v) = header::HeaderValue::from_str(&allow_origin) {
        headers.insert(header::ACCESS_CONTROL_ALLOW_ORIGIN, v);
        // Ensure caches don't mix responses across different Origins.
        headers.insert(header::VARY, header::HeaderValue::from_static("Origin"));
    }

    // Allow browser clients to read correlation and protocol metadata.
    headers.insert(
        header::ACCESS_CONTROL_EXPOSE_HEADERS,
        header::HeaderValue::from_static("x-request-id, x-hop-protocol-version"),
    );
}

fn cors_preflight_response(cfg: &ServerConfig, origin: Option<&str>) -> warp::reply::Response {
    let mut res = warp::reply::with_status("", StatusCode::NO_CONTENT).into_response();
    apply_cors_headers(cfg, origin, res.headers_mut());

    // If origin isn't allowed, omit the rest; browsers will block.
    if res
        .headers()
        .get(header::ACCESS_CONTROL_ALLOW_ORIGIN)
        .is_none()
    {
        return res;
    }

    res.headers_mut().insert(
        header::ACCESS_CONTROL_ALLOW_METHODS,
        header::HeaderValue::from_static("GET, POST, OPTIONS"),
    );
    res.headers_mut().insert(
        header::ACCESS_CONTROL_ALLOW_HEADERS,
        header::HeaderValue::from_static("content-type, authorization, x-request-id"),
    );
    res
}

pub fn routes(
    cfg: Arc<ServerConfig>,
) -> Result<warp::filters::BoxedFilter<(warp::reply::Response,)>, prometheus::Error> {
    let max = cfg.max_body_bytes;

    let metrics = Arc::new(Metrics::new(cfg.as_ref())?);
    let with_metrics = with_metrics(metrics.clone());

    let readiness_cache = Arc::new(Mutex::new(ReadinessCache::default()));
    let with_ready = with_readiness_cache(readiness_cache);

    let authz_cache = Arc::new(RwLock::new(HashMap::new()));
    let with_authz_cache = with_authz_cache(authz_cache);

    let with_cfg = with_cfg(cfg.clone());
    let auth_admin = auth_admin(cfg.clone());
    let authn = authn(cfg.clone());

    // OPTIONS /* (CORS preflight)
    let preflight_cfg = cfg.clone();
    let options = warp::options()
        .and(warp::path::full())
        .and(warp::header::optional::<String>("origin"))
        .map(move |_path: warp::path::FullPath, origin: Option<String>| {
            cors_preflight_response(&preflight_cfg, origin.as_deref())
        });

    // POST /bundle (legacy)
    let bundle = warp::path("bundle")
        .and(warp::post())
        .and(auth_admin.clone())
        .and(with_cfg.clone())
        .and(json_body::<BundleReq>(max))
        .and_then(handle_bundle)
        .map(Reply::into_response);

    // POST /state (legacy)
    let state = warp::path("state")
        .and(warp::post())
        .and(auth_admin.clone())
        .and(with_cfg.clone())
        .and(json_body::<StateReq>(max))
        .and_then(handle_state)
        .map(Reply::into_response);

    // Vault endpoints
    let v_new = warp::path!("vault" / "new")
        .and(warp::post())
        .and(auth_admin.clone())
        .and(with_cfg.clone())
        .and(json_body::<VaultNewReq>(max))
        .and_then(vault_new)
        .map(Reply::into_response);

    let v_token_rotate = warp::path!("vault" / "token" / "rotate")
        .and(warp::post())
        .and(auth_admin.clone())
        .and(with_cfg.clone())
        .and(with_authz_cache.clone())
        .and(with_metrics.clone())
        .and(json_body::<VaultTokenRotateReq>(max))
        .and_then(vault_token_rotate)
        .map(Reply::into_response);

    let v_append = warp::path!("vault" / "append")
        .and(warp::post())
        .and(authn.clone())
        .and(with_cfg.clone())
        .and(with_authz_cache.clone())
        .and(with_metrics.clone())
        .and(json_body::<VaultAppendReq>(max))
        .and_then(vault_append)
        .map(Reply::into_response);

    let v_get = warp::path!("vault" / "get")
        .and(warp::post())
        .and(authn.clone())
        .and(with_cfg.clone())
        .and(with_authz_cache.clone())
        .and(with_metrics.clone())
        .and(json_body::<VaultGetReq>(max))
        .and_then(vault_get)
        .map(Reply::into_response);

    let v_prove = warp::path!("vault" / "prove")
        .and(warp::post())
        .and(authn.clone())
        .and(with_cfg.clone())
        .and(with_authz_cache.clone())
        .and(with_metrics.clone())
        .and(json_body::<VaultProveReq>(max))
        .and_then(vault_prove)
        .map(Reply::into_response);

    // GET /health (always public)
    let health = warp::path("health")
        .and(warp::get())
        .map(|| warp::reply::json(&serde_json::json!({"ok": true})).into_response());

    // GET /ready (readiness: basic storage writability check)
    let ready = warp::path("ready")
        .and(warp::get())
        .and(with_cfg.clone())
        .and(with_ready.clone())
        .and_then(handle_ready)
        .map(Reply::into_response);

    // GET /metrics (Prometheus exposition format)
    let metrics_route = warp::path("metrics")
        .and(warp::get())
        .and(with_metrics.clone())
        .map(metrics_reply);

    // Unify to a single reply type (`Response`) to keep compilation fast and
    // make it easy to apply global middleware (CORS, etc).
    let routes = options
        .or(bundle)
        .unify()
        .or(state)
        .unify()
        .or(v_new)
        .unify()
        .or(v_token_rotate)
        .unify()
        .or(v_append)
        .unify()
        .or(v_get)
        .unify()
        .or(v_prove)
        .unify()
        .or(health)
        .unify()
        .or(ready)
        .unify()
        .or(metrics_route)
        .unify();

    let cors_cfg = cfg.clone();
    let metrics2 = metrics.clone();
    let meta = request_meta(metrics.clone());

    // Availability controls: global concurrency cap + optional rate limiting.
    let sem = if cfg.max_in_flight_requests == 0 {
        None
    } else {
        Some(Arc::new(Semaphore::new(cfg.max_in_flight_requests)))
    };
    let rate_limiter = if cfg.rate_limit_rps > 0 && cfg.rate_limit_burst > 0 {
        Some(Arc::new(Mutex::new(TokenBucket::new(
            cfg.rate_limit_rps as u64,
            cfg.rate_limit_burst as u64,
        ))))
    } else {
        None
    };

    let guard = warp::method().and(warp::path::full()).and_then(
        move |method: warp::http::Method, path: warp::path::FullPath| {
            let sem = sem.clone();
            let rate_limiter = rate_limiter.clone();
            async move {
                // Allow health checks and browser preflights to proceed even when saturated.
                if method == warp::http::Method::GET
                    && matches!(path.as_str(), "/health" | "/ready" | "/metrics")
                {
                    return Ok::<Option<OwnedSemaphorePermit>, warp::Rejection>(None);
                }
                if method == warp::http::Method::OPTIONS {
                    return Ok::<Option<OwnedSemaphorePermit>, warp::Rejection>(None);
                }

                // Concurrency/load-shedding.
                let permit = if let Some(sem) = sem {
                    match sem.try_acquire_owned() {
                        Ok(p) => Some(p),
                        Err(_) => return Err(warp::reject::custom(Overloaded)),
                    }
                } else {
                    None
                };

                // Process-local global token-bucket rate limit.
                if let Some(rl) = rate_limiter {
                    let mut rl = rl.lock().await;
                    if !rl.try_take(Instant::now()) {
                        return Err(warp::reject::custom(RateLimited {
                            retry_after_secs: rl.retry_after_secs(),
                        }));
                    }
                }

                Ok(permit)
            }
        },
    );

    let guarded = guard
        .and(routes)
        .map(|_permit: Option<OwnedSemaphorePermit>, res: warp::reply::Response| res)
        .recover(handle_rejection)
        .unify();

    let routes = meta
        .and(guarded)
        .map(move |meta: RequestMeta, mut res: warp::reply::Response| {
            apply_cors_headers(&cors_cfg, meta.origin.as_deref(), res.headers_mut());

            // Always echo or generate an ID that can be used to correlate server logs with clients.
            if let Ok(v) = header::HeaderValue::from_str(&meta.request_id) {
                res.headers_mut()
                    .insert(header::HeaderName::from_static("x-request-id"), v);
            }
            res.headers_mut().insert(
                header::HeaderName::from_static("x-hop-protocol-version"),
                header::HeaderValue::from_static("2"),
            );

            let status = res.status();
            let latency = meta.start.elapsed();
            metrics2.observe(&meta, status, latency);
            let latency_ms = latency.as_millis() as u64;

            let forwarded_for = meta.forwarded_for.as_deref();
            let ua = meta.user_agent.as_deref();

            if status.is_success() {
                tracing::info!(
                    request_id = %meta.request_id,
                    method = %meta.method,
                    path = %meta.path,
                    status = status.as_u16(),
                    latency_ms,
                    forwarded_for,
                    user_agent = ua,
                    "request"
                );
            } else if status.is_client_error() {
                tracing::warn!(
                    request_id = %meta.request_id,
                    method = %meta.method,
                    path = %meta.path,
                    status = status.as_u16(),
                    latency_ms,
                    forwarded_for,
                    user_agent = ua,
                    "request"
                );
            } else if status.is_server_error() {
                tracing::error!(
                    request_id = %meta.request_id,
                    method = %meta.method,
                    path = %meta.path,
                    status = status.as_u16(),
                    latency_ms,
                    forwarded_for,
                    user_agent = ua,
                    "request"
                );
            } else {
                tracing::info!(
                    request_id = %meta.request_id,
                    method = %meta.method,
                    path = %meta.path,
                    status = status.as_u16(),
                    latency_ms,
                    forwarded_for,
                    user_agent = ua,
                    "request"
                );
            }

            res
        })
        .boxed();

    Ok(routes)
}

pub async fn serve(
    addr: &str,
    cfg: ServerConfig,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    fs::create_dir_all(&cfg.data_dir)?;
    let sock: SocketAddr = addr.parse().map_err(|_| "bad listen addr")?;

    let shutdown_grace_ms = cfg.shutdown_grace_ms;
    let shutdown_grace = Duration::from_millis(shutdown_grace_ms);
    let cfg = Arc::new(cfg);

    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
    tokio::spawn(async move {
        if let Err(error) = shutdown_signal().await {
            tracing::error!(%error, "failed to install shutdown signal handler");
        }
        let _ = shutdown_tx.send(true);
    });

    let mut shutdown_rx1 = shutdown_rx.clone();
    let shutdown = async move {
        while !*shutdown_rx1.borrow() {
            if shutdown_rx1.changed().await.is_err() {
                break;
            }
        }
    };

    let listener = tokio::net::TcpListener::bind(sock).await?;
    let bound = listener.local_addr()?;

    let server = warp::serve(routes(cfg)?)
        .incoming(listener)
        .graceful(shutdown);
    tracing::info!(listen = %bound, "hop-relay listening");

    if shutdown_grace.is_zero() {
        server.run().await;
        return Ok(());
    }

    let mut shutdown_rx2 = shutdown_rx.clone();
    let force = async move {
        while !*shutdown_rx2.borrow() {
            if shutdown_rx2.changed().await.is_err() {
                return;
            }
        }
        tokio::time::sleep(shutdown_grace).await;
    };

    tokio::select! {
        _ = server.run() => Ok(()),
        _ = force => {
            tracing::warn!(shutdown_grace_ms, "graceful shutdown timed out; forcing exit");
            Ok(())
        }
    }
}

async fn shutdown_signal() -> std::io::Result<()> {
    let ctrl_c = tokio::signal::ctrl_c();

    #[cfg(unix)]
    {
        use tokio::signal::unix::{signal, SignalKind};
        let mut term = signal(SignalKind::terminate())?;
        tokio::select! {
            result = ctrl_c => result?,
            _ = term.recv() => {},
        }
    }
    #[cfg(not(unix))]
    {
        ctrl_c.await?;
    }
    tracing::info!("shutdown signal received");
    Ok(())
}