boatramp 0.2.2

//! The `serve` subcommand: select backends and run the server.

use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use boatramp_core::cache_coherence::Changelog;
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::{CachedKv, KvStore};
use boatramp_core::migrate;
use boatramp_node::backends::{BlobBackend, KvBackend};
use clap::ValueEnum;

use crate::config::ServerConfig;

/// The control-plane KV builder, reused by the standalone `boatramp migrate` command.
pub(crate) use boatramp_node::backends::build_kv as build_control_plane_kv;
use boatramp_node::blobs::{build_blobs, BlobArgs};

/// A failure running `boatramp serve`: selecting/initialising a backend, wiring
/// auth / OIDC / TLS, or the HTTP server itself exiting with an error. Most of
/// `serve` is behind a build feature (`tls` / `acme-dns` / `s3` / `slatedb` /
/// `cluster` / `handlers` / `http3` / `oidc` / `cloudflare-kv`), so each variant
/// is gated to match the `?` site / `bail!` it replaced — a variant is present
/// only when the code that produces it is compiled.
#[derive(Debug, thiserror::Error)]
pub enum Error {
    // ---- "rebuild with --features X" guards (the selected backend/mode is
    // not compiled into this binary) ----------------------------------------
    /// `[cluster]` config present but the binary lacks cluster support.
    #[cfg(not(feature = "cluster"))]
    #[error(
        "[cluster] config is present but this build has no cluster support; \
         rebuild with `--features cluster`"
    )]
    NoClusterSupport,
    /// A `--tls custom`/`acme` mode selected but the binary lacks TLS support.
    #[cfg(not(feature = "tls"))]
    #[error("this build has no TLS support; rebuild with `--features tls`")]
    NoTlsSupport,
    /// `--tls acme-dns` selected but the binary lacks ACME DNS-01 support.
    #[cfg(not(feature = "acme-dns"))]
    #[error("this build has no ACME DNS-01 support; rebuild with `--features acme-dns`")]
    NoAcmeDnsSupport,

    // ---- configuration / argument validation -------------------------------
    /// A token root **private** key (hex) failed to parse (cluster write-capability
    /// minting). The single-node auth path's key errors now live in
    /// [`boatramp_node::Error`].
    #[cfg(feature = "cluster")]
    #[error("invalid auth root private key: {0}")]
    AuthPrivKey(String),
    /// A raw-public-key TLS error — building the peer-mesh identity/config
    /// (`cluster`) or the `--tls rpk` bootstrap identity/config (`tls`). Both use
    /// the same RPK stack (`boatramp_rpktls`), so `mesh::MeshError` is an alias of
    /// `RpkError` and they share one `From` here (a second would collide).
    #[cfg(any(feature = "cluster", feature = "tls"))]
    #[error(transparent)]
    RpkTls(#[from] boatramp_rpktls::RpkError),
    /// Refusing to serve the peer mesh on a non-loopback address with no trust set
    /// configured — that would expose an unauthenticated control plane.
    /// The peer mesh has no trust anchor on a non-loopback bind.
    #[cfg(feature = "cluster")]
    #[error(
        "refusing to serve the peer mesh on {0} (non-loopback) with an empty trust \
         set: found with --cluster-init or join with --cluster-join <ticket>"
    )]
    MeshUnconfigured(std::net::SocketAddr),
    /// The cluster startup decision failed closed (F5) or a join could not be
    /// completed (no join token, seeds unreachable, root mismatch, …).
    #[cfg(feature = "cluster")]
    #[error("cluster startup: {0}")]
    ClusterStartup(String),
    /// Configuring the secrets-at-rest envelope failed.
    #[cfg(all(feature = "cluster", feature = "acme-dns"))]
    #[error("secrets envelope: {0}")]
    Envelope(String),
    /// Fetching the OIDC issuer's discovery document / JWKS failed.
    #[cfg(feature = "oidc")]
    #[error("OIDC setup failed: {0}")]
    OidcSetup(String),
    /// OIDC is enabled but no `--oidc-audience` is set, and the security posture
    /// requires one. Set an audience, or relax
    /// `oidc_require_audience` in `[security]` (e.g. the `dev` profile).
    #[cfg(feature = "oidc")]
    #[error(
        "OIDC is enabled without an audience, but the security posture requires one \
         (set --oidc-audience, or relax `oidc_require_audience`)"
    )]
    OidcAudienceRequired,
    /// `--tls custom` without `--tls-cert`.
    #[cfg(feature = "tls")]
    #[error("--tls-cert is required for --tls custom")]
    TlsCertRequired,
    /// `--tls custom` without `--tls-key`.
    #[cfg(feature = "tls")]
    #[error("--tls-key is required for --tls custom")]
    TlsKeyRequired,
    /// The `--tls-cert` PEM held no certificates (HTTP/3 cert loading).
    #[cfg(feature = "http3")]
    #[error("no certificates in {0}")]
    NoCert(String),
    /// The `--tls-key` PEM held no private key (HTTP/3 cert loading).
    #[cfg(feature = "http3")]
    #[error("no private key in {0}")]
    NoPrivateKey(String),
    /// `--tls acme` with no `--acme-domain`.
    #[cfg(feature = "tls")]
    #[error("at least one --acme-domain is required for --tls acme")]
    NoAcmeDomain,
    /// `--tls acme-dns` with no `--acme-domain`.
    #[cfg(feature = "acme-dns")]
    #[error("at least one --acme-domain is required for --tls acme-dns")]
    NoAcmeDomainDns,
    /// An unrecognised `--acme-dns-provider` value.
    #[cfg(feature = "acme-dns")]
    #[error("unknown --acme-dns-provider {0:?} (expected manual | cloudflare | route53 | oci)")]
    UnknownDnsProvider(String),
    /// No certificate is available yet — the cluster leader hasn't issued one.
    #[cfg(all(feature = "cluster", feature = "acme-dns"))]
    #[error(
        "no certificates available yet — awaiting the cluster leader to issue (retry shortly)"
    )]
    NoCertsYet,

    // ---- propagated library errors (`#[from]`) ------------------------------
    /// Node-library assembly (handler runtime / SQL binding) failed.
    #[error(transparent)]
    Assembly(#[from] boatramp_node::Error),
    /// Resolving the `[security]` posture (e.g. an unknown profile name) failed.
    #[error(transparent)]
    Security(#[from] boatramp_core::security::SecurityError),
    /// The HTTP server exited with an error.
    #[error(transparent)]
    Serve(#[from] boatramp_server::ServeError),
    /// A listener-bind / filesystem / axum-server I/O error on the serve path.
    #[cfg(any(feature = "tls", feature = "cluster"))]
    #[error(transparent)]
    Io(#[from] std::io::Error),
    /// An ACME DNS-01 issuance / cert-serving-config error.
    #[cfg(feature = "acme-dns")]
    #[error(transparent)]
    AcmeDns(#[from] crate::acme_dns::Error),
    /// An HTTP/3 (QUIC) endpoint / TLS-config error.
    #[cfg(feature = "http3")]
    #[error(transparent)]
    Http3(#[from] boatramp_server::Http3Error),
    /// A rustls error building the ACME client config (extra CA trust).
    #[cfg(feature = "tls")]
    #[error(transparent)]
    Rustls(#[from] rustls::Error),
    /// A cluster-managed-cert refresh error (replicated cert store).
    #[cfg(all(feature = "cluster", feature = "acme-dns"))]
    #[error(transparent)]
    ClusterTls(#[from] crate::cluster_tls::Error),
    /// Building / bootstrapping the embedded-Raft cluster node failed. Boxed: the
    /// openraft error types it wraps are ~230 bytes, and this variant is cold
    /// (constructed once, on a fatal bootstrap failure). Boxing it keeps `Error`
    /// — and `CliError` above it — under clippy's `result_large_err` threshold
    /// without a blanket `#[allow]`. `#[from]` can't box, so see the `From` below.
    #[cfg(feature = "cluster")]
    #[error(transparent)]
    Bootstrap(Box<boatramp_cluster::node::BootstrapError>),
    /// Opening the SlateDB / Cloudflare KV metadata store failed.
    #[cfg(any(feature = "slatedb", feature = "cloudflare-kv"))]
    #[error(transparent)]
    Kv(#[from] boatramp_core::kv::KvError),
    /// Building the WebAssembly handler engine failed.
    #[cfg(feature = "handlers")]
    #[error(transparent)]
    Handler(#[from] boatramp_handlers::HandlerError),
    /// The control-plane store holds pre-0.2.0 (layout 1) data and has not been
    /// migrated to the project-scoped layout. Refusing to serve so a half-read
    /// store can't silently drop sites/functions. Run `boatramp migrate` (or start
    /// with `--auto-migrate`).
    #[error(
        "the control-plane store is not migrated to the project-scoped (0.2.0) layout; \
         run `boatramp migrate` first, or start `serve --auto-migrate`"
    )]
    UnmigratedStore,
    /// Running the store migration failed.
    #[error("store migration failed: {0}")]
    Migrate(String),
}

impl From<boatramp_core::migrate::MigrateError> for Error {
    fn from(e: boatramp_core::migrate::MigrateError) -> Self {
        Self::Migrate(e.to_string())
    }
}

/// `serve` module result; `Err` is [`Error`].
type Result<T> = std::result::Result<T, Error>;

// Box the (large, openraft-backed) bootstrap error into [`Error`]: the variant is
// `Box<BootstrapError>` so `?` on a bare `BootstrapError` keeps working (thiserror's
// `#[from]` would generate `From<BootstrapError>`, not the boxing conversion).
#[cfg(feature = "cluster")]
impl From<boatramp_cluster::node::BootstrapError> for Error {
    fn from(e: boatramp_cluster::node::BootstrapError) -> Self {
        Self::Bootstrap(Box::new(e))
    }
}

// Guard the boxing decision: `Bootstrap` is boxed so this enum stays under clippy's
// `result_large_err` threshold (128 B) without a module-wide `#[allow]`. If a future
// variant grows past it, box that one too rather than re-adding the allow.
#[cfg(feature = "cluster")]
const _: () = assert!(std::mem::size_of::<Error>() <= 128);

/// TLS mode for the public listener.
#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
enum TlsMode {
    /// Plain HTTP (terminate TLS at an upstream proxy).
    Off,
    /// HTTPS with an operator-supplied cert/key (requires `--features tls`).
    Custom,
    /// HTTPS with automatic ACME certificates (requires `--features tls`).
    Acme,
    /// HTTPS with ACME **DNS-01** certificates, incl. wildcard preview certs
    /// (requires `--features acme-dns`).
    AcmeDns,
    /// HTTPS with a **raw-public-key** (RFC 7250) identity the client pins — an
    /// encrypted, server-authenticated control channel with no ACME, tunnel, or
    /// TLS-terminating proxy (requires `--features tls`). The client authenticates
    /// with a bearer token; the identity printed at startup is pinned client-side
    /// with `--server-pubkey`. For a first-boot / bare-metal control plane.
    Rpk,
}

/// Arguments for `boatramp serve`.
#[derive(Debug, clap::Args)]
pub struct ServeArgs {
    /// Address to bind the HTTP server to (flag/env > `serve.addr` >
    /// `127.0.0.1:8080`).
    #[arg(long, env = "BOATRAMP_ADDR")]
    addr: Option<SocketAddr>,

    /// Data directory for filesystem backends (blobs in `<dir>/blobs`,
    /// metadata in `<dir>/kv`). Flag/env > `serve.data_dir` > `./data`.
    #[arg(long, env = "BOATRAMP_DATA_DIR")]
    data_dir: Option<PathBuf>,

    /// Blob storage backend.
    #[arg(long, value_enum, default_value_t = BlobBackend::Fs)]
    blobs: BlobBackend,

    /// Metadata (KV) backend.
    #[arg(long, value_enum, default_value_t = KvBackend::Slatedb)]
    kv: KvBackend,

    /// Migrate a pre-0.2.0 (layout 1) control-plane store to the project-scoped
    /// layout at startup instead of refusing to serve. Off by default so the
    /// re-key is an explicit, one-time operator action (`boatramp migrate`).
    #[arg(long)]
    auto_migrate: bool,

    /// S3 bucket (required for `--blobs s3`).
    #[arg(long, env = "BOATRAMP_S3_BUCKET")]
    s3_bucket: Option<String>,

    /// S3 endpoint URL, e.g. a MinIO server (optional).
    #[arg(long, env = "BOATRAMP_S3_ENDPOINT")]
    s3_endpoint: Option<String>,

    /// S3 region (optional).
    #[arg(long, env = "BOATRAMP_S3_REGION")]
    s3_region: Option<String>,

    /// Use S3 path-style addressing (required by MinIO).
    #[arg(long, env = "BOATRAMP_S3_PATH_STYLE")]
    s3_path_style: bool,

    /// GCS bucket (required for `--blobs gcs`).
    #[arg(long, env = "BOATRAMP_GCS_BUCKET")]
    gcs_bucket: Option<String>,

    /// GCS storage endpoint URL, e.g. a `fake-gcs-server` emulator (optional;
    /// defaults to the public GCS JSON API).
    #[arg(long, env = "BOATRAMP_GCS_ENDPOINT")]
    gcs_endpoint: Option<String>,

    /// Skip GCS credential resolution (anonymous — the emulator). Real GCS uses
    /// Application Default Credentials.
    #[arg(long, env = "BOATRAMP_GCS_ANONYMOUS")]
    gcs_anonymous: bool,

    /// Azure storage account name (required for `--blobs azure`).
    #[arg(long, env = "BOATRAMP_AZURE_ACCOUNT")]
    azure_account: Option<String>,

    /// Azure container name (required for `--blobs azure`).
    #[arg(long, env = "BOATRAMP_AZURE_CONTAINER")]
    azure_container: Option<String>,

    /// Azure storage account access key (shared-key auth; required unless
    /// `--azure-emulator`). Prefer the env var over the flag.
    #[arg(long, env = "BOATRAMP_AZURE_ACCESS_KEY")]
    azure_access_key: Option<String>,

    /// Use the Azurite emulator (well-known dev credentials + local endpoint).
    #[arg(long, env = "BOATRAMP_AZURE_EMULATOR")]
    azure_emulator: bool,

    /// Number of deploy manifests/pointers to keep in the in-memory LRU.
    #[arg(long, default_value_t = 256)]
    cache_entries: usize,

    /// Token root **private** key (hex) — this node verifies tokens *and*
    /// issues them (`/api/tokens`, OIDC exchange). Enables control-plane auth.
    /// Generate with `boatramp auth init`.
    #[arg(long, env = "BOATRAMP_AUTH_ROOT_PRIVATE_KEY")]
    auth_root_private_key: Option<String>,

    /// Token root **public** key (hex) — verify-only node (cannot issue).
    /// Enables control-plane auth. Ignored if `--auth-root-private-key` is set.
    #[arg(long, env = "BOATRAMP_AUTH_ROOT_PUBLIC_KEY")]
    auth_root_public_key: Option<String>,

    /// Single-use **bootstrap secret** enabling `POST /api/tokens/bootstrap` — mint
    /// the first control-plane token by presenting this secret (no admin bearer).
    /// Set it on a fresh deploy, run `boatramp token bootstrap`, then unset it.
    /// Rotating it re-enables bootstrap (recovery). Flag/env > `serve.bootstrap_secret`.
    #[arg(long, env = "BOATRAMP_BOOTSTRAP_SECRET")]
    bootstrap_secret: Option<String>,

    /// TLS mode for the listener.
    #[arg(long, value_enum, default_value_t = TlsMode::Off)]
    tls: TlsMode,

    /// PEM certificate chain (for `--tls custom`).
    #[arg(long, requires = "tls_key")]
    tls_cert: Option<PathBuf>,

    /// PEM private key (for `--tls custom`).
    #[arg(long, requires = "tls_cert")]
    tls_key: Option<PathBuf>,

    /// Domain to obtain an ACME certificate for (repeatable; for `--tls acme`).
    #[arg(long = "acme-domain")]
    acme_domain: Vec<String>,

    /// ACME directory URL (defaults to Let's Encrypt production).
    #[arg(long, default_value = "https://acme-v02.api.letsencrypt.org/directory")]
    acme_directory: String,

    /// Contact email for the ACME account.
    #[arg(long)]
    acme_contact: Option<String>,

    /// Extra root CA (PEM) to trust for the ACME server (e.g. Pebble's CA).
    #[arg(long)]
    acme_ca_cert: Option<PathBuf>,

    /// Directory for the ACME certificate cache.
    #[arg(long, default_value = "./data/acme")]
    acme_cache: PathBuf,

    /// DNS provider for `--tls acme-dns` and `boatramp dns`
    /// (`manual` | `cloudflare` | `route53` | `oci`). Credentials come from the
    /// environment (see `boatramp dns --help`).
    #[arg(long, default_value = "manual")]
    acme_dns_provider: String,

    /// With `--tls acme-dns`, also issue a `*.deploy.<domain>` wildcard cert so
    /// the wildcard preview host form gets TLS.
    #[arg(long)]
    acme_wildcard_preview: bool,

    /// Reject blob uploads larger than this many bytes (default: unlimited).
    /// Flag/env > `serve.max_upload_bytes`.
    #[arg(long, env = "BOATRAMP_MAX_UPLOAD_BYTES")]
    max_upload_bytes: Option<u64>,

    /// Abort an upload whose body stalls (no bytes received) for longer than this
    /// many seconds — slowloris protection. Flag/env > `serve.upload_idle_timeout_secs`.
    #[arg(long, env = "BOATRAMP_UPLOAD_IDLE_TIMEOUT")]
    upload_idle_timeout_secs: Option<u64>,

    /// Cap simultaneous blob uploads; further uploads get 503 until a slot frees.
    /// Flag/env > `serve.max_concurrent_uploads`.
    #[arg(long, env = "BOATRAMP_MAX_CONCURRENT_UPLOADS")]
    max_concurrent_uploads: Option<usize>,

    /// In a TLS mode, also bind this plain-HTTP address (e.g. `0.0.0.0:80`) on a
    /// second listener that 308-redirects every request to HTTPS. Flag/env >
    /// `serve.http_redirect_addr`. Ignored when `--tls off`.
    #[arg(long, env = "BOATRAMP_HTTP_REDIRECT_ADDR")]
    http_redirect_addr: Option<SocketAddr>,

    /// Site to serve for a `Host` that matches no domain, instead of 404
    /// (catch-all). Flag/env > `serve.default_site`.
    #[arg(long, env = "BOATRAMP_DEFAULT_SITE")]
    default_site: Option<String>,

    /// The fleet's canonical public origin (e.g. `https://cp.example.com`) that a
    /// per-request proof-of-possession must bind to. Required for holder-bound
    /// (`cnf`/PoP) tokens; compared against a proof's origin, never a request
    /// header. Flag/env > `serve.pop_origin`.
    #[arg(long, env = "BOATRAMP_POP_ORIGIN")]
    pop_origin: Option<String>,

    /// Rate-limit cluster-wide via the control-plane KV (shared fixed-window)
    /// instead of per-node in-process buckets. Meaningful with a shared/
    /// replicated KV; adds a KV round-trip per limited request. Flag/env >
    /// `serve.cluster_rate_limit`.
    #[arg(long, env = "BOATRAMP_CLUSTER_RATE_LIMIT")]
    cluster_rate_limit: bool,

    /// **Found a brand-new cluster** from this node (the explicit, one-time
    /// genesis signal — F5). Required to bring up the first node with no seeds;
    /// refused if `[cluster].seeds` are set (a node either founds or joins, not
    /// both). A no-op once the node has durable state (restart resumes).
    #[arg(long, env = "BOATRAMP_CLUSTER_INIT")]
    cluster_init: bool,

    /// This node's own mesh base URL that peers should dial to reach it (e.g.
    /// `https://10.0.0.4:7000`). Advertised at join so the leader can replicate
    /// back. Defaults to `https://<cluster.listen>`; set it when the bind address
    /// isn't the reachable address (NAT / container / `0.0.0.0`).
    #[arg(long, env = "BOATRAMP_CLUSTER_ADVERTISE_ADDR")]
    cluster_advertise_addr: Option<String>,

    /// **Join an existing cluster** using a one-paste ticket from `cluster add`
    /// (bundles the seeds + root anchor + single-use token). Overrides
    /// `[cluster].seeds`/`root_pubkeys`/`join_token`. Mutually exclusive with
    /// `--cluster-init`.
    #[arg(long, env = "BOATRAMP_CLUSTER_JOIN")]
    cluster_join: Option<String>,

    /// Keep the local config cache coherent across processes sharing one KV
    /// (Cloudflare KV / shared SlateDB): publish each control-plane write to a
    /// changelog and poll it to invalidate just the keys peers changed.
    /// Turn on when running multiple stateless frontends
    /// over one shared store; unnecessary single-node or in a Raft cluster.
    /// Flag/env > `serve.shared_cache_coherence`.
    #[arg(long, env = "BOATRAMP_SHARED_CACHE_COHERENCE")]
    shared_cache_coherence: bool,

    /// Require a valid control-plane token to view deployment previews
    /// (`/_deploy/<id>` and `<id>.deploy.<host>`). Flag/env >
    /// `serve.protect_previews`.
    #[arg(long, env = "BOATRAMP_PROTECT_PREVIEWS")]
    protect_previews: bool,

    /// Also serve HTTP/3 (QUIC) on the same UDP port (with `--tls custom`).
    /// Requires the `http3` build feature.
    #[cfg(feature = "http3")]
    #[arg(long)]
    http3: bool,

    /// OIDC issuer URL for control-plane bearer-JWT auth (its JWKS is fetched at
    /// startup; tokens' scope claim must carry boatramp scopes). Requires the
    /// `oidc` build feature.
    #[cfg(feature = "oidc")]
    #[arg(long, env = "BOATRAMP_OIDC_ISSUER")]
    oidc_issuer: Option<String>,

    /// Expected JWT `aud` for OIDC auth (audience validation is skipped if unset).
    #[cfg(feature = "oidc")]
    #[arg(long, env = "BOATRAMP_OIDC_AUDIENCE")]
    oidc_audience: Option<String>,

    /// JWT claim carrying boatramp scopes for OIDC auth (default `scope`).
    #[cfg(feature = "oidc")]
    #[arg(long, env = "BOATRAMP_OIDC_SCOPE_CLAIM")]
    oidc_scope_claim: Option<String>,
}

impl ServeArgs {
    /// Merge upload limits from flags/env over the `serve` config defaults, then
    /// fall back to the security posture's default cap: an
    /// unconfigured `max_upload_bytes` is no longer unbounded. The posture's `0`
    /// means "explicitly unlimited" (e.g. the `dev` profile).
    fn server_limits(
        &self,
        serve_cfg: &crate::config::ServeConfig,
        posture: &boatramp_core::security::SecurityPosture,
    ) -> boatramp_server::ServerLimits {
        boatramp_server::ServerLimits {
            max_upload_bytes: self
                .max_upload_bytes
                .or(serve_cfg.max_upload_bytes)
                .or_else(|| (posture.max_upload_bytes != 0).then_some(posture.max_upload_bytes)),
            upload_idle_timeout: self
                .upload_idle_timeout_secs
                .or(serve_cfg.upload_idle_timeout_secs)
                .map(std::time::Duration::from_secs),
            max_concurrent_uploads: self
                .max_concurrent_uploads
                .or(serve_cfg.max_concurrent_uploads),
        }
    }
}

/// Entry point for `boatramp serve`. Resolution precedence for the overridable
/// settings is flag/env > `serve` in `boatramp.cfg` > built-in default.
pub async fn run(args: ServeArgs, config: &ServerConfig) -> Result<()> {
    let serve_cfg = config.serve.clone().unwrap_or_default();
    // Resolve the operator security posture once (profile preset + overrides);
    // absent `[security]` ⇒ the strict `multi-tenant` default. Threaded into
    // `ServerOptions` so it reaches the cluster path too.
    let posture = config.security.clone().unwrap_or_default().resolve()?;
    // Server-level options (flag/env > `serve` config). Resolved before the
    // `serve` fields below are consumed.
    let mut options = boatramp_server::ServerOptions {
        limits: args.server_limits(&serve_cfg, &posture),
        default_site: args.default_site.clone().or(serve_cfg.default_site.clone()),
        pop_origin: args.pop_origin.clone().or(serve_cfg.pop_origin.clone()),
        protect_previews: args.protect_previews || serve_cfg.protect_previews,
        posture,
        // The listener terminates TLS in any non-`Off` mode; used to derive the
        // request scheme when X-Forwarded-Proto isn't trusted.
        served_over_tls: !matches!(args.tls, TlsMode::Off),
        bootstrap_secret: args
            .bootstrap_secret
            .clone()
            .or(serve_cfg.bootstrap_secret.clone()),
        ..Default::default()
    };
    let cluster_rate_limit = args.cluster_rate_limit || serve_cfg.cluster_rate_limit;
    let addr = args
        .addr
        .or(serve_cfg.addr)
        .unwrap_or_else(|| "127.0.0.1:8080".parse().expect("valid default addr"));
    // Implicit host routing (first-label `<site>.host` / sole-site at root) is a
    // dev / single-operator convenience: enable it when the posture allows, or
    // unconditionally on a loopback bind (only local clients reach it, so there
    // is no host-spoofing exposure). Strict `multi-tenant` on a public bind keeps
    // it off, so an unmatched host resolves only to `default_site` or 404.
    options.implicit_routing = options.posture.allow_implicit_routing || addr.ip().is_loopback();
    // Embedded web console (`[serve.console]`): when the operator enabled it,
    // mount the baked-in SPA at the configured host+path. Needs the `console`
    // build feature; enabling it without the feature is a logged no-op.
    if let Some(console) = serve_cfg.console.as_ref().filter(|c| c.enabled) {
        #[cfg(feature = "console")]
        {
            options.console = Some(boatramp_server::console::ConsoleMount::resolve(
                console.host.clone(),
                console.path.clone(),
            ));
        }
        #[cfg(not(feature = "console"))]
        {
            let _ = console;
            tracing::warn!(
                "[serve.console] enabled but this build lacks the `console` feature — \
                 the console is not served"
            );
        }
    }
    let data_dir = args
        .data_dir
        .clone()
        .or(serve_cfg.data_dir)
        .unwrap_or_else(|| PathBuf::from("./data"));

    // Cloud blob-change notification provisioning config (FA-5b2): the tier + the
    // account id that scopes a provisioned queue policy. Read before `storage` so
    // the notify-enabled S3 backend + its provider share one AWS config.
    let notify_tier = serve_cfg.blob_notify_tier;
    let notify_account = serve_cfg.blob_notify_account_id.clone();
    let blob_args = BlobArgs {
        blobs: args.blobs,
        s3_bucket: args.s3_bucket.clone(),
        s3_endpoint: args.s3_endpoint.clone(),
        s3_region: args.s3_region.clone(),
        s3_path_style: args.s3_path_style,
        gcs_bucket: args.gcs_bucket.clone(),
        gcs_endpoint: args.gcs_endpoint.clone(),
        gcs_anonymous: args.gcs_anonymous,
        azure_account: args.azure_account.clone(),
        azure_container: args.azure_container.clone(),
        azure_access_key: args.azure_access_key.clone(),
        azure_emulator: args.azure_emulator,
    };
    let built_blobs = build_blobs(&blob_args, &data_dir, notify_tier, notify_account).await?;
    let storage = built_blobs.storage.clone();

    // Cluster mode: triggered by a `[cluster]` config section OR the founding/
    // joining flags (`--cluster-init` / `--cluster-join <ticket>`), so a node can
    // join with just a ticket and no config file. The control-plane KvStore +
    // messaging then come from the embedded-Raft cluster node.
    #[cfg(feature = "cluster")]
    if config.cluster.is_some() || args.cluster_init || args.cluster_join.is_some() {
        let cluster_cfg = config.cluster.clone().unwrap_or_else(|| {
            // No `[cluster]` section: synthesize defaults (a flag-only bring-up).
            // The mesh binds the default port on the same host as `serve.addr`.
            crate::config::ClusterConfig {
                listen: std::net::SocketAddr::new(addr.ip(), DEFAULT_MESH_PORT),
                root_pubkeys: Vec::new(),
                seeds: Vec::new(),
                join_token: None,
                store_dir: None,
                mesh: None,
            }
        });
        return run_cluster(
            args,
            config,
            cluster_cfg,
            addr,
            data_dir,
            built_blobs,
            options,
        )
        .await;
    }
    #[cfg(not(feature = "cluster"))]
    if config.cluster.is_some() {
        return Err(Error::NoClusterSupport);
    }

    let kv_backend = boatramp_node::backends::build_kv(args.kv, &data_dir).await?;
    // Shared-mode coherence: when several processes share
    // one KV, publish each write to a changelog over the *uncached* backend and
    // poll it to invalidate peer-changed keys.
    let shared_coherence = args.shared_cache_coherence || serve_cfg.shared_cache_coherence;
    let changelog = shared_coherence
        .then(|| Arc::new(Changelog::new(kv_backend.clone(), CHANGELOG_RETENTION_SECS)));
    // Front the metadata store with an LRU so hot reads stay in memory.
    let mut cached = CachedKv::new(kv_backend.clone(), args.cache_entries);
    if let Some(changelog) = &changelog {
        cached = cached.with_publisher(changelog.clone());
    }
    let kv: Arc<dyn KvStore> = Arc::new(cached);

    // Layout guard (0.2.0): refuse to serve a store still on the pre-project layout
    // 1 — a half-read store would silently drop sites/functions/compute. The operator
    // migrates explicitly (`boatramp migrate`); `--auto-migrate` opts into an in-place
    // one-shot migration here. A `2-dual` store serves fine (reads are off the new
    // keys) and only needs a later `boatramp migrate --finalize` to reclaim old keys.
    match migrate::status(kv.as_ref()).await? {
        migrate::Status::Ready => {}
        migrate::Status::Dual => tracing::warn!(
            "control-plane store is in the 2-dual soak window; \
             run `boatramp migrate --finalize` to reclaim the old-layout keys"
        ),
        migrate::Status::NeedsMigration => {
            if args.auto_migrate {
                tracing::warn!(
                    "control-plane store is on the pre-0.2.0 layout; running a one-shot \
                     project re-keying migration (--auto-migrate)"
                );
                let report =
                    migrate::migrate(kv.as_ref(), migrate::MigrateOptions::one_shot()).await?;
                tracing::info!(
                    rekeyed = report.total_rekeyed(),
                    owner_entries = report.owner_entries,
                    "control-plane store migrated to the project-scoped layout"
                );
            } else {
                return Err(Error::UnmigratedStore);
            }
        }
    }

    // Handle for a final flush on graceful shutdown (SHUT-1): `kv` is moved into
    // the deploy store below; this clone reaches its backing store's `flush`.
    let kv_handle = kv.clone();
    // Rate-limit windows are coordination state, not config: they must NOT be
    // cached (a stale window would count wrong), so the limiter uses the
    // *uncached* backend directly.
    if cluster_rate_limit {
        options.cluster_rate_limit_kv = Some(kv_backend.clone());
    }
    // The dynamic daemon-config runtime, built here so SIGHUP and the shared-store
    // changelog can **wake** an immediate reload (push-driven convergence) rather
    // than relying on the runtime's backstop tick.
    let daemon_runtime = Arc::new(boatramp_server::DaemonRuntime::new(
        boatramp_server::config_baseline(&options),
    ));
    options.daemon_runtime = Some(daemon_runtime.clone());
    spawn_sighup_reload(kv.clone(), Some(daemon_runtime.clone()));
    if let Some(changelog) = changelog {
        spawn_cache_poller(changelog, kv.clone(), Some(daemon_runtime.clone()));
    }
    let auth = boatramp_node::auth::configure_auth(
        serve_cfg.signer.as_ref(),
        args.auth_root_private_key
            .clone()
            .or(serve_cfg.auth_root_private_key.clone()),
        args.auth_root_public_key
            .clone()
            .or(serve_cfg.auth_root_public_key.clone()),
        &mut options,
        kv.clone(),
    )
    .await?;
    configure_oidc(&args, &mut options).await?;
    // Fail-closed: don't expose an unauthenticated control plane on a public bind.
    boatramp_node::auth::enforce_auth_bind(addr, &auth, &options.posture)?;
    // Wire the built store + configured auth/options into a running node graph:
    // handler runtime, deploy store, compute + domain-verify reconcile loops. This
    // is the same assembly `boatramp serve` exercises, now a library call so an
    // embedder / in-process test builds the identical graph (PLAN-node-library N2b.3).
    // `_reconcile` holds the detached reconcile loops for the server's serving life.
    let boatramp_node::RunningNode {
        deploy,
        handlers,
        auth,
        options,
        reconcile: _reconcile,
    } = boatramp_node::assemble(boatramp_node::NodeInput {
        config,
        data_dir: data_dir.as_path(),
        storage,
        kv,
        auth,
        options,
        watch_provider: built_blobs.watch_provider.clone(),
        provision_tier: built_blobs.provision_tier,
        // Single-node: default messaging, an always-true leader gate (one node), id 0.
        messaging: None,
        is_leader: Arc::new(|| true),
        node_id: 0,
    })
    .await?;

    tracing::info!(
        blobs = ?args.blobs, kv = ?args.kv, tls = ?args.tls,
        auth = !auth.is_disabled(), "starting boatramp"
    );
    // In a TLS mode, optionally bind a second plain-HTTP listener that redirects
    // to HTTPS. Ignored for `--tls off`.
    #[cfg(feature = "tls")]
    if !matches!(args.tls, TlsMode::Off) {
        if let Some(redirect_addr) = args.http_redirect_addr.or(serve_cfg.http_redirect_addr) {
            spawn_http_redirect(redirect_addr, deploy.clone(), posture);
        }
    }
    let serve_result = match args.tls {
        TlsMode::Off => boatramp_server::serve_with(addr, deploy, auth, handlers, options)
            .await
            .map_err(Error::Serve),
        TlsMode::Custom => serve_custom(&args, addr, deploy, auth, handlers, options).await,
        TlsMode::Acme => serve_acme(&args, addr, deploy, auth, handlers, options).await,
        TlsMode::AcmeDns => serve_acme_dns(&args, addr, deploy, auth, handlers, options).await,
        TlsMode::Rpk => serve_rpk(&args, addr, deploy, auth, handlers, options, &data_dir).await,
    };
    // Graceful shutdown: force a final flush of the metadata store (SHUT-1).
    if let Err(e) = kv_handle.flush().await {
        tracing::warn!(error = %e, "metadata store flush on shutdown failed");
    }
    serve_result
}

/// How long changelog feed entries are kept (comfortably larger than the poll
/// interval so a poller can't miss entries between polls).
const CHANGELOG_RETENTION_SECS: u64 = 60;

/// Drive the shared-mode cache-coherence poller: every
/// second, pop the keys peers changed; periodically trim the feed; and every few
/// minutes do a full flush as the gap backstop (rare, so no thundering herd).
/// Detached for the server's lifetime.
fn spawn_cache_poller(
    changelog: Arc<Changelog>,
    cache: Arc<dyn KvStore>,
    daemon: Option<Arc<boatramp_server::DaemonRuntime>>,
) {
    use std::time::Duration;
    tokio::spawn(async move {
        let poll = Duration::from_secs(1);
        let flush_every = Duration::from_secs(300);
        let mut cursor = changelog.current_cursor().await;
        let mut since_trim = Duration::ZERO;
        let mut since_flush = Duration::ZERO;
        loop {
            tokio::time::sleep(poll).await;
            let changed = changelog.poll(&mut cursor).await;
            if !changed.is_empty() {
                cache.invalidate_keys(&changed);
                // A peer wrote dynamic daemon config → wake an immediate reload.
                if let Some(daemon) = &daemon {
                    if changed.iter().any(|k| k.starts_with("daemon/")) {
                        daemon.notify_reload();
                    }
                }
            }
            since_trim += poll;
            if since_trim >= Duration::from_secs(30) {
                changelog.trim().await;
                since_trim = Duration::ZERO;
            }
            since_flush += poll;
            if since_flush >= flush_every {
                cache.invalidate_cache();
                cursor = changelog.current_cursor().await;
                since_flush = Duration::ZERO;
            }
        }
    });
}

/// Spawn a `SIGHUP` handler that drops the control-plane KV cache, so the next
/// reads pull fresh config from the backing store — the manual "reload config"
/// signal (e.g. after another node wrote new config to the shared/replicated
/// store). No-op on non-Unix. Detached for the server's lifetime.
#[cfg(unix)]
fn spawn_sighup_reload(kv: Arc<dyn KvStore>, daemon: Option<Arc<boatramp_server::DaemonRuntime>>) {
    tokio::spawn(async move {
        let mut hup = match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup()) {
            Ok(sig) => sig,
            Err(err) => {
                tracing::warn!(%err, "could not install SIGHUP handler");
                return;
            }
        };
        while hup.recv().await.is_some() {
            kv.invalidate_cache();
            // Wake an immediate daemon-config reload (push, not the backstop tick).
            if let Some(daemon) = &daemon {
                daemon.notify_reload();
            }
            tracing::info!("SIGHUP: invalidated config cache (next reads reload from the store)");
        }
    });
}

#[cfg(not(unix))]
fn spawn_sighup_reload(
    _kv: Arc<dyn KvStore>,
    _daemon: Option<Arc<boatramp_server::DaemonRuntime>>,
) {
}

/// In a TLS mode, spawn a detached plain-HTTP listener on `addr` that
/// 308-redirects every request to HTTPS (dual-listener) — except the HTTP
/// domain-ownership challenge, which it serves directly so an unattached host
/// can verify itself over plain `:80` before it has a cert. Fire and forget: it
/// dies with the process; bind failures are logged, not fatal, so a missing
/// privilege on `:80` doesn't take down the HTTPS server. Uses `axum_server`
/// (the same plain/TLS server stack the TLS modes use).
#[cfg(feature = "tls")]
fn spawn_http_redirect(
    addr: SocketAddr,
    deploy: DeployStore,
    posture: boatramp_core::security::SecurityPosture,
) {
    tokio::spawn(async move {
        tracing::info!(%addr, "serving HTTP→HTTPS redirect listener");
        let service = boatramp_server::http_redirect_router(deploy, posture).into_make_service();
        if let Err(err) = axum_server::bind(addr).serve(service).await {
            tracing::error!(%addr, %err, "HTTP redirect listener failed");
        }
    });
}

/// Run in **self-hosted cluster mode**: the control-plane
/// `KvStore` and the `wasi:messaging` coordinator come from an embedded-Raft
/// cluster node instead of the local backends. This node serves its peer mesh
/// (`/raft/*` + `/stream/*`) on `[cluster].listen`, runs `DeployStore` over
/// `RaftKv` (writes→leader, reads→local durable state) and the dispatcher over
/// `RaftMessaging`, and fires crons only while it is the leader. Live multi-host
/// behavior needs live-platform validation; every component is gate-tested in-process.
/// How long a rotation waits for `K_new` to propagate before presenting it.
/// Only minimises the transient-rejection window — the live
/// verifier + dialer retry make a shorter/absent wait safe, not incorrect.
#[cfg(feature = "cluster")]
const MESH_ROTATION_PROPAGATION: std::time::Duration = std::time::Duration::from_secs(2);

/// Default Raft peer-mesh port when a node joins/founds with only flags (no
/// `[cluster]` section). Distinct from the public `serve.addr` port.
#[cfg(feature = "cluster")]
const DEFAULT_MESH_PORT: u16 = 7000;

/// Parse a mesh key-rotation cadence like `"30d"`, `"12h"`, `"90m"`, `"3600s"`
/// into a `Duration`. `None` for an empty/invalid value (⇒ no scheduled
/// rotation). Only the `s`/`m`/`h`/`d` suffixes are accepted.
#[cfg(feature = "cluster")]
fn parse_rotation_interval(spec: &str) -> Option<std::time::Duration> {
    let spec = spec.trim();
    let split = spec.find(|c: char| !c.is_ascii_digit())?;
    let (num, unit) = spec.split_at(split);
    let n: u64 = num.parse().ok()?;
    let secs = match unit {
        "s" => n,
        "m" => n.checked_mul(60)?,
        "h" => n.checked_mul(3600)?,
        "d" => n.checked_mul(86_400)?,
        _ => return None,
    };
    (secs > 0).then(|| std::time::Duration::from_secs(secs))
}

/// Clock skew tolerated on a join possession proof: `proof_iat` must be within
/// this window of the admitting node's clock, so a captured proof cannot be
/// replayed indefinitely (the single-use `jti` is the primary anti-replay; this
/// bounds the pre-spend window). Symmetric to cover both directions of skew.
#[cfg(feature = "cluster")]
const JOIN_PROOF_MAX_SKEW_SECS: u64 = 300;

/// TTL on the root-signed member assertions handed back in a join response: long
/// enough for the joiner to verify + adopt each key within the round-trip, short
/// enough that a captured response can't seed a node much later. The joiner
/// verifies each against the root anchor before trusting it (PLAN-cluster-join F3).
#[cfg(feature = "cluster")]
const MEMBER_ASSERTION_TTL_SECS: u64 = 300;

/// Bridges the server's `/api/cluster/*` control routes to the cluster runtime
/// (join admission + key rotation) over [`ClusterNode`]. `issuer` is the
/// control-plane **root signer**: a join admits only if this node can mint
/// root-signed member assertions for the joiner to adopt.
#[cfg(feature = "cluster")]
struct ClusterMeshControl {
    node: Arc<boatramp_cluster::node::ClusterNode>,
    issuer: Option<Arc<dyn boatramp_core::cose::Signer>>,
}

#[cfg(feature = "cluster")]
#[async_trait::async_trait]
impl boatramp_server::MeshControl for ClusterMeshControl {
    async fn admit(
        &self,
        mesh_pubkey_hex: &str,
        jti: &str,
        possession_proof: &[u8],
        proof_iat: u64,
        now: u64,
        advertise_addr: Option<&str>,
    ) -> std::result::Result<boatramp_server::JoinOutcome, String> {
        use boatramp_server::JoinOutcome;

        // (1) Freshness: the proof must be stamped within the skew window. This
        // bounds how long a captured (pre-spend) proof stays presentable.
        let fresh = proof_iat <= now.saturating_add(JOIN_PROOF_MAX_SKEW_SECS)
            && now <= proof_iat.saturating_add(JOIN_PROOF_MAX_SKEW_SECS);
        if !fresh {
            return Ok(JoinOutcome::ProofInvalid);
        }

        // (2) The joiner is identified by the key it claims; parse it to SPKI.
        let Ok(spki) = boatramp_cluster::mesh::parse_public_key(mesh_pubkey_hex) else {
            return Ok(JoinOutcome::ProofInvalid);
        };

        // (3) Possession: the proof must be a signature by that very key over the
        // domain-separated join challenge — so a bearer token alone (without the
        // private key) cannot join in another key's name.
        let challenge = boatramp_core::cose::join_challenge(jti, mesh_pubkey_hex, proof_iat);
        if !boatramp_rpktls::verify_signature(&spki, &challenge, possession_proof) {
            return Ok(JoinOutcome::ProofInvalid);
        }

        // (4) Single-use + re-admit-proof: the state machine spends the `jti`
        // (idempotent replay ⇒ already-spent) and refuses a revoked key (F6). A
        // stale/spent token or a revoked key that survived (3) stops here.
        use boatramp_cluster::raft::AdmitOutcome;
        match self
            .node
            .admit(mesh_pubkey_hex, jti, advertise_addr)
            .await
            .map_err(|e| e.to_string())?
        {
            AdmitOutcome::Admitted => {}
            AdmitOutcome::Spent => return Ok(JoinOutcome::TokenSpent),
            AdmitOutcome::Revoked => return Ok(JoinOutcome::Revoked),
        }

        // (5) Vouch for the current members with root-signed assertions so the
        // joiner adopts each only after verifying it against the root anchor.
        let Some(issuer) = self.issuer.as_ref() else {
            return Err("cluster node has no root signing key to vouch for members".to_string());
        };
        let mut members = Vec::new();
        for (node_id, pubkey) in self.node.trusted_member_keys().await {
            let assertion = boatramp_core::cose::mint_member_assertion(
                node_id,
                &pubkey,
                MEMBER_ASSERTION_TTL_SECS,
                now,
                issuer.as_ref(),
            )
            .await
            .map_err(|e| e.to_string())?;
            members.push(assertion);
        }
        // Advisory routing so the joiner can dial every member (each dial is still
        // key-authenticated). Only entries for members it also verified above are
        // usable to it.
        let addrs = self.node.peer_addrs();
        Ok(JoinOutcome::Admitted { members, addrs })
    }

    async fn rotate_key(&self) -> std::result::Result<String, String> {
        let new_pub = self
            .node
            .rotate_key(MESH_ROTATION_PROPAGATION)
            .await
            .map_err(|e| e.to_string())?;
        Ok(new_pub.iter().map(|b| format!("{b:02x}")).collect())
    }

    async fn revoke(&self, node: u64) -> std::result::Result<(), String> {
        self.node.revoke(node).await.map_err(|e| e.to_string())
    }

    async fn members(&self) -> std::result::Result<Vec<boatramp_server::MeshMember>, String> {
        let addrs = self.node.peer_addrs();
        Ok(self
            .node
            .members()
            .into_iter()
            .map(|m| boatramp_server::MeshMember {
                node: m.node,
                voter: m.voter,
                caught_up: m.caught_up,
                leader: m.leader,
                addr: addrs.get(&m.node).cloned(),
            })
            .collect())
    }

    async fn promote(&self, node: u64) -> std::result::Result<(), String> {
        self.node.promote(node).await.map_err(|e| e.to_string())
    }
}

/// Verifies a mesh client-write **cluster-write capability**: the
/// presented bearer must be a token signed by the control-plane root that grants
/// the `cluster-write` role. This trust root is separate from the mesh transport
/// key, so a mesh-key holder without a control-plane capability can't inject
/// writes.
#[cfg(feature = "cluster")]
struct MeshWriteAuthz {
    public: boatramp_core::cose::TokenPublicKey,
}

#[cfg(feature = "cluster")]
impl boatramp_cluster::http::ClientWriteAuthz for MeshWriteAuthz {
    fn authorize(&self, capability: Option<&str>) -> bool {
        let Some(token) = capability else {
            return false;
        };
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        let Ok(verified) = boatramp_core::cose::verify(token, &self.public, now) else {
            return false;
        };
        verified.roles.iter().any(|r| r.name == "cluster-write")
    }
}

/// Build the mesh write gate from config: when `mesh.gate_client_writes` is
/// set and the token signer (root **private** key) is available, mint this node's
/// cluster-write capability and an authorizer for incoming forwards. Returns
/// `(None, None)` when disabled or the root key is absent (gating then off —
/// defense-in-depth is opt-in and must not break a keyless cluster).
#[cfg(feature = "cluster")]
#[allow(clippy::type_complexity)]
async fn build_mesh_write_gate(
    args: &ServeArgs,
    config: &ServerConfig,
    mesh_cfg: &crate::config::MeshConfig,
) -> Result<(
    Option<String>,
    Option<Arc<dyn boatramp_cluster::http::ClientWriteAuthz>>,
)> {
    use boatramp_core::authz::GrantedRole;
    use boatramp_core::cose::{self, Claims, LocalSigner, Signer};

    if !mesh_cfg.gate_client_writes.unwrap_or(false) {
        return Ok((None, None));
    }
    let priv_hex = args.auth_root_private_key.clone().or_else(|| {
        config
            .serve
            .as_ref()
            .and_then(|s| s.auth_root_private_key.clone())
    });
    let Some(priv_hex) = priv_hex else {
        tracing::warn!(
            "cluster.mesh.gate_client_writes is set but no token root private key is \
             configured — mesh client-write gating is disabled"
        );
        return Ok((None, None));
    };
    let signer =
        LocalSigner::from_private_hex(&priv_hex).map_err(|e| Error::AuthPrivKey(e.to_string()))?;
    // No TTL: the capability lives for this node's process (now_unix unused).
    let claims = Claims {
        roles: vec![GrantedRole::global("cluster-write")],
        kind: cose::KIND_CLUSTER_WRITE.to_string(),
        ttl_secs: None,
        now_unix: 0,
    };
    let capability = cose::mint(&claims, &signer)
        .await
        .map_err(|e| Error::AuthPrivKey(format!("minting cluster-write capability: {e}")))?;
    let authz: Arc<dyn boatramp_cluster::http::ClientWriteAuthz> = Arc::new(MeshWriteAuthz {
        public: signer.public_key(),
    });
    Ok((Some(capability), Some(authz)))
}

/// Build the configured secrets-at-rest envelope from `[secrets]`,
/// resolving a Vault token from the environment. `None` ⇒ store cleartext.
#[cfg(all(feature = "cluster", feature = "acme-dns"))]
fn build_cert_envelope(
    secrets: Option<&crate::config::SecretsConfig>,
    data_dir: &Path,
) -> Result<Option<Arc<dyn boatramp_core::envelope::KeyEnvelope>>> {
    use boatramp_server::envelope::{build_envelope, EnvelopeSpec};
    let Some(cfg) = secrets else {
        return Ok(None);
    };
    let spec = match cfg.envelope.as_str() {
        "" => EnvelopeSpec::None,
        "local" => EnvelopeSpec::Local {
            kek_file: cfg
                .kek_file
                .clone()
                .unwrap_or_else(|| data_dir.join("secrets/kek")),
        },
        "vault" => {
            let v = cfg.vault.as_ref().ok_or_else(|| {
                Error::Envelope(
                    "secrets.envelope = \"vault\" needs a [secrets.vault] section".into(),
                )
            })?;
            let token = std::env::var(&v.token_env).map_err(|_| {
                Error::Envelope(format!("Vault token env `{}` is not set", v.token_env))
            })?;
            EnvelopeSpec::Vault {
                addr: v.addr.clone(),
                key: v.key.clone(),
                token,
            }
        }
        other => {
            return Err(Error::Envelope(format!(
                "unknown secrets.envelope {other:?} (want \"local\" or \"vault\")"
            )))
        }
    };
    build_envelope(spec).map_err(|e| Error::Envelope(e.to_string()))
}

/// Reloads this node's dynamic daemon-config runtime whenever a replicated
/// `daemon/*` write is applied to the Raft state machine — push convergence for
/// cluster followers and the leader through ordinary log replication, no polling.
#[cfg(feature = "cluster")]
struct DaemonConfigObserver(Arc<boatramp_server::DaemonRuntime>);

#[cfg(feature = "cluster")]
impl boatramp_cluster::raft::ApplyObserver for DaemonConfigObserver {
    fn on_apply(&self, muts: &[boatramp_core::kv::WriteOp]) {
        use boatramp_core::kv::WriteOp;
        let touched = muts.iter().any(|m| match m {
            WriteOp::Put(k, _) | WriteOp::Delete(k) => k.starts_with("daemon/"),
        });
        if touched {
            self.0.notify_reload();
        }
    }

    fn on_reset(&self, data: &std::collections::BTreeMap<String, Vec<u8>>) {
        if data.keys().any(|k| k.starts_with("daemon/")) {
            self.0.notify_reload();
        }
    }
}

#[cfg(feature = "cluster")]
#[allow(clippy::too_many_arguments)]
async fn run_cluster(
    args: ServeArgs,
    config: &ServerConfig,
    mut cluster_cfg: crate::config::ClusterConfig,
    addr: SocketAddr,
    data_dir: PathBuf,
    built_blobs: boatramp_node::blobs::BuiltBlobs,
    mut options: boatramp_server::ServerOptions,
) -> Result<()> {
    use boatramp_cluster::node::{build_node, ClusterParams};

    // The blob backend; `built_blobs` also carries the optional FA-5b2 blob-change
    // watch provider + tier the handler runtime is wired with below.
    let storage = built_blobs.storage.clone();

    // Node-local durable Raft log/state store (distinct from the *replicated*
    // control plane the cluster serves).
    let store_dir = cluster_cfg
        .store_dir
        .clone()
        .unwrap_or_else(|| data_dir.join("raft"));
    // Whether this node's durable store dir already exists, captured BEFORE opening
    // (which creates it). A weak "has booted before" signal — NOT the resume gate:
    // the dir is created just by opening the KV, before a first join completes.
    let store_dir_existed = store_dir.exists();
    let durable_kv: Arc<dyn KvStore> = Arc::new(
        boatramp_storage::SlateKv::open_local_with_flush(
            store_dir,
            boatramp_node::backends::CONTROL_PLANE_FLUSH,
        )
        .await?,
    );
    // The real resume-vs-found/join signal (F5): whether the store holds COMMITTED
    // cluster state (persisted mesh trust). A store dir that exists but has no
    // committed trust means an earlier boot opened the KV but never finished its
    // join — such a node must re-derive its action (rejoin), not resume into an
    // empty-trust, fail-closed mesh. Checked once, over the raw store.
    let has_committed_state = boatramp_cluster::persist::has_committed_trust(&durable_kv)
        .await
        .map_err(|e| Error::ClusterStartup(e.to_string()))?;
    // Keep a handle to force a final flush on graceful shutdown (SHUT-1): the
    // store is moved into the Raft stores below, so this clone is how we reach
    // its `flush` after serving stops.
    let durable_kv_handle = durable_kv.clone();

    use boatramp_cluster::mesh::{self, MeshIdentity, MeshTls, TrustSet};

    // No static peer map: the peer directory + mesh trust set start empty and are
    // populated by founding (self), joining (adopted members), or replication.
    let mut peers = std::collections::BTreeMap::new();
    let mut genesis_trust = std::collections::BTreeMap::new();

    // Load (or generate + persist `0600`) this node's Ed25519 mesh identity.
    let mesh_cfg = cluster_cfg.mesh.clone().unwrap_or_default();
    let key_file = mesh_cfg
        .key_file
        .clone()
        .unwrap_or_else(|| data_dir.join("mesh/identity.key"));
    let identity = MeshIdentity::load_or_generate(&key_file)?;

    // This node's Raft id is DERIVED from its mesh key (dynamic-join self-identity
    // — no config id).
    let node_id = boatramp_cluster::raft::derive_node_id(identity.public_key());
    tracing::info!(
        node_id,
        pubkey = %identity.public_key_hex(),
        "cluster: mesh identity"
    );

    // A one-paste `--cluster-join <ticket>` overrides the `[cluster]` seeds/root/
    // token: decode it and fold it into the config so the rest of the flow is
    // ticket-vs-config agnostic.
    if let Some(blob) = args.cluster_join.as_deref() {
        let ticket = crate::join::JoinTicket::decode(blob)
            .map_err(|e| Error::ClusterStartup(e.to_string()))?;
        cluster_cfg.seeds = ticket.seeds;
        cluster_cfg.root_pubkeys = ticket.root_pubkeys;
        cluster_cfg.join_token = Some(ticket.token);
    }

    // Decide founding vs joining vs resuming (F5): the single source of truth.
    let seeds_present = !cluster_cfg.seeds.is_empty();
    // The Kubernetes operator designates its **ordinal-0** StatefulSet pod as the
    // founder (via the downward-API pod name) — every other ordinal joins. This
    // designates the founder only; the node *identity* is still derived from the
    // mesh key, so it is not the reverted per-pod-identity coupling.
    let is_operator_founder =
        std::env::var("BOATRAMP_POD_NAME").is_ok_and(|name| name.rsplit('-').next() == Some("0"));
    let init_requested = args.cluster_init || is_operator_founder;
    let action = crate::join::decide_startup(&crate::join::StartupInputs {
        // Resume ONLY on committed state (persisted trust) — not on a store dir that
        // merely exists, which would wedge a joiner whose first join never finished.
        has_committed_state,
        // The dir already existing means this node booted before, but possibly never
        // committed any state; kept as the weaker "ever booted" signal for messaging.
        ever_member: store_dir_existed,
        seeds_present,
        init_requested,
    });
    // A resuming node seeds no genesis trust here — it rehydrates its trust set
    // from durable state inside `build_node`, so its empty-trust fail-closed check
    // is deferred until after that (below). Captured now: the `match action` moves.
    let is_resume = matches!(action, crate::join::StartupAction::Resume);

    // This node's own reachable mesh URL (advertised at join so the leader can
    // dial it), defaulting to the bind address.
    let self_advertise = args
        .cluster_advertise_addr
        .clone()
        .unwrap_or_else(|| format!("https://{}", cluster_cfg.listen));

    // A dynamic joiner starts with NO static membership; the seed admits it and
    // membership + trust arrive via replication. Bootstrap only when founding.
    let mut do_bootstrap = false;
    match action {
        crate::join::StartupAction::FailClosed(reason) => {
            return Err(Error::ClusterStartup(reason));
        }
        crate::join::StartupAction::Found => {
            // Genesis: this node is the sole founding member. List itself so
            // `build_node` takes the genesis path and `bootstrap` initializes it.
            peers.insert(node_id, self_advertise.clone());
            genesis_trust.insert(node_id, identity.public_key().to_vec());
            do_bootstrap = true;
            tracing::info!(node_id, "cluster: founding a new cluster (init)");
        }
        crate::join::StartupAction::Join => {
            // Redeem the join ticket against the seeds and adopt the returned,
            // root-verified members — seeding the trust set + peer directory with
            // NO static peer map. This node is a joiner: it does NOT list itself,
            // so `build_node` starts it with empty membership (the seed admits it).
            let roots = if cluster_cfg.root_pubkeys.is_empty() {
                config
                    .serve
                    .as_ref()
                    .and_then(|s| s.auth_root_public_key.clone())
                    .into_iter()
                    .collect()
            } else {
                cluster_cfg.root_pubkeys.clone()
            };
            let token = cluster_cfg
                .join_token
                .as_deref()
                .and_then(|s| crate::join::resolve_join_token(s).transpose())
                .transpose()
                .map_err(|e| Error::ClusterStartup(e.to_string()))?
                .ok_or_else(|| {
                    Error::ClusterStartup(
                        "joining requires [cluster].join_token (env:/path:/inline)".into(),
                    )
                })?;
            let ticket = crate::join::JoinTicket {
                seeds: cluster_cfg.seeds.clone(),
                root_pubkeys: roots,
                token,
            };
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0);
            let adopted = crate::join::join_cluster(&ticket, &identity, Some(&self_advertise), now)
                .await
                .map_err(|e| Error::ClusterStartup(e.to_string()))?;
            tracing::info!(
                node_id,
                members = adopted.len(),
                "cluster: joined via seeds"
            );
            for m in adopted {
                if let Ok(spki) = mesh::parse_public_key(&m.mesh_pubkey_hex) {
                    genesis_trust.insert(m.node_id, spki);
                }
                if let Some(addr) = m.mesh_addr {
                    peers.insert(m.node_id, addr);
                }
            }
        }
        crate::join::StartupAction::Resume => {
            // Durable state is authoritative; `build_node` hydrates trust + the
            // peer directory from it. Never re-bootstraps.
            tracing::info!(node_id, "cluster: resuming from durable state");
        }
    }

    // Fail closed: never bring up a non-loopback mesh with no trusted peers.
    // Found/Join seed `genesis_trust` right here, so they are checked now; a Resume
    // leaves it empty on purpose and is checked after `build_node` rehydrates its
    // trust from durable state.
    if !cluster_cfg.listen.ip().is_loopback() && genesis_trust.is_empty() && !is_resume {
        return Err(Error::MeshUnconfigured(cluster_cfg.listen));
    }

    let mesh_tls = Arc::new(MeshTls::new(
        Arc::new(identity),
        TrustSet::from_map(genesis_trust),
    ));

    // Optionally gate mesh client-writes behind a control-plane cluster-write
    // capability (this node's capability + the authorizer for incoming forwards).
    let (write_capability, write_authz) = build_mesh_write_gate(&args, config, &mesh_cfg).await?;
    if write_authz.is_some() {
        tracing::info!("cluster: mesh client-write gating enabled");
    }

    // Dynamic daemon-config runtime: a cluster ApplyObserver wakes an immediate
    // reload on every replicated `daemon/*` apply, so leader and followers converge
    // by push (through ordinary log replication) with no polling.
    let daemon_runtime = Arc::new(boatramp_server::DaemonRuntime::new(
        boatramp_server::config_baseline(&options),
    ));
    options.daemon_runtime = Some(daemon_runtime.clone());
    let daemon_observer: Arc<dyn boatramp_cluster::raft::ApplyObserver> =
        Arc::new(DaemonConfigObserver(daemon_runtime));

    let node = Arc::new(
        build_node(ClusterParams {
            node_id,
            peers,
            // Empty ⇒ every peer votes; otherwise the listed ids are the voting
            // quorum and the rest join as read-only learners (multi-region).
            // Founding uses the self-only peer map ⇒ this node is the sole voter;
            // a joiner's membership arrives from the seed. No static voter list.
            voters: std::collections::BTreeSet::new(),
            durable_kv,
            storage: storage.clone(),
            mesh: mesh_tls.clone(),
            cluster_write_capability: write_capability,
            extra_observers: vec![daemon_observer],
        })
        .await?,
    );

    // A resuming node rehydrated its trust set from durable state inside
    // `build_node`. Only now can it fail closed on a genuinely empty trust (a wiped
    // or corrupt volume) — otherwise the mesh would come up trusting no peer.
    if is_resume && !cluster_cfg.listen.ip().is_loopback() && mesh_tls.trust().snapshot().is_empty()
    {
        return Err(Error::MeshUnconfigured(cluster_cfg.listen));
    }

    // Serve this node's peer mesh over RFC 7250 raw-public-key mutual TLS 1.3:
    // every `/raft/*` + `/stream/*` request must present a trusted peer key. The
    // application `client-write` is additionally gated by the authorizer.
    let mesh_router = match write_authz {
        Some(authz) => node.router.clone().layer(axum::Extension::<
            boatramp_cluster::http::WriteAuthz,
        >(Some(authz))),
        None => node.router.clone(),
    };
    let mesh_config =
        axum_server::tls_rustls::RustlsConfig::from_config(Arc::new(mesh_tls.server()?));
    let listen = cluster_cfg.listen;
    tracing::info!(
        node_id, %listen,
        "cluster: serving peer mesh (mutual TLS)"
    );
    tokio::spawn(async move {
        if let Err(err) = axum_server::bind_rustls(listen, mesh_config)
            .serve(mesh_router.into_make_service())
            .await
        {
            tracing::error!(%err, "cluster: peer mesh server exited");
        }
    });

    // Initialize a brand-new cluster from this node (once) when founding.
    if do_bootstrap {
        node.bootstrap().await?;
        // A founder is never admitted, so nothing else records its address —
        // publish it into the replicated directory so joiners (and restarts) can
        // dial it with no peer map.
        node.advertise_addr(&self_advertise).await?;
        tracing::info!("cluster: bootstrapped membership");
    }

    // Scheduled mesh key rotation. Node-local, NOT leader-gated:
    // each node rotates its OWN key (only it holds/mints its private key), and
    // make-before-break is per-node + fail-safe, so nodes rotating independently
    // (even concurrently) is harmless. Absent cadence ⇒ manual rotation only.
    if let Some(interval) = mesh_cfg
        .key_rotation
        .as_deref()
        .and_then(parse_rotation_interval)
    {
        let rotate_node = node.clone();
        // Stagger the first rotation by node id (seconds) so a fleet booted
        // together doesn't rotate in lockstep; then every `interval`.
        let stagger = std::time::Duration::from_secs(node_id % 60);
        tokio::spawn(async move {
            loop {
                tokio::time::sleep(interval + stagger).await;
                match rotate_node.rotate_key(MESH_ROTATION_PROPAGATION).await {
                    Ok(pubkey) => tracing::info!(
                        pubkey = %pubkey.iter().map(|b| format!("{b:02x}")).collect::<String>(),
                        "cluster: rotated mesh key on schedule"
                    ),
                    Err(err) => {
                        tracing::error!(%err, "cluster: scheduled mesh key rotation failed");
                    }
                }
            }
        });
    }

    // The control-plane KvStore + messaging are the cluster facades.
    let kv: Arc<dyn KvStore> = node.kv.clone();

    // Layout guard (0.2.0), cluster path — the same fail-closed gate the single-node
    // path applies: refuse to serve a store still on the pre-project layout 1, since a
    // half-read store would silently drop sites/functions/compute. `node.kv` is the
    // replicated RaftKv facade, so a `--auto-migrate` here writes through Raft consensus
    // (a follower's writes forward to the leader) and every migration step is idempotent
    // + re-verifying — so even a concurrent racer converges rather than corrupts, no
    // CAS/leader-election needed. The load-bearing case is the founder (the leader,
    // authoritative for its own store) booting on legacy data; a joiner only ever
    // reaches an already-running cluster, which this guard kept from serving unmigrated.
    match migrate::status(kv.as_ref()).await? {
        migrate::Status::Ready => {}
        migrate::Status::Dual => tracing::warn!(
            "control-plane store is in the dual soak window; \
             run `boatramp migrate --finalize` to reclaim the old-layout keys"
        ),
        migrate::Status::NeedsMigration => {
            if args.auto_migrate {
                tracing::warn!(
                    "control-plane store is below the current schema version; running a \
                     one-shot migration through the cluster (--auto-migrate)"
                );
                let report =
                    migrate::migrate(kv.as_ref(), migrate::MigrateOptions::one_shot()).await?;
                tracing::info!(
                    rekeyed = report.total_rekeyed(),
                    owner_entries = report.owner_entries,
                    "control-plane store migrated to the project-scoped layout"
                );
            } else {
                return Err(Error::UnmigratedStore);
            }
        }
    }

    // Cluster-wide rate limiting shares the *replicated* RaftKv across nodes.
    if args.cluster_rate_limit || config.serve.as_ref().is_some_and(|s| s.cluster_rate_limit) {
        options.cluster_rate_limit_kv = Some(kv.clone());
    }
    // SIGHUP also force-reloads the daemon config (the ApplyObserver already
    // handles replicated `daemon/*` writes; this is the manual override).
    spawn_sighup_reload(kv.clone(), options.daemon_runtime.clone());
    let cluster_serve_cfg = config.serve.clone().unwrap_or_default();
    let auth = boatramp_node::auth::configure_auth(
        cluster_serve_cfg.signer.as_ref(),
        args.auth_root_private_key
            .clone()
            .or(cluster_serve_cfg.auth_root_private_key.clone()),
        args.auth_root_public_key
            .clone()
            .or(cluster_serve_cfg.auth_root_public_key),
        &mut options,
        kv.clone(),
    )
    .await?;
    configure_oidc(&args, &mut options).await?;
    // Fail-closed: don't expose an unauthenticated control plane on a public bind.
    boatramp_node::auth::enforce_auth_bind(addr, &auth, &options.posture)?;

    // The mesh control hook: `POST /api/cluster/join` + `/rotate-key` reach the
    // cluster runtime through it. Constructed after `configure_auth` so it carries
    // the control-plane root signer (`options.issuer`) — the join flow mints
    // root-signed member assertions with it.
    options.mesh_control = Some(Arc::new(ClusterMeshControl {
        node: node.clone(),
        issuer: options.issuer.clone(),
    }));
    // Node-graph assembly, shared with the single-node path (`boatramp_node::assemble`):
    // handler runtime, deploy store (+ reserved `default` project), compute + domain-
    // verify reconcile loops. The cluster differences are threaded as `NodeInput`
    // fields: the Raft messaging substrate, a Raft `is_leader` gate (cron firing +
    // both reconcile loops run only on the leader), and this node's compute id.
    let leader_raft = node.raft.clone();
    let leader_node_id = node.node_id;
    let is_leader: boatramp_server::CronLeaderGate =
        Arc::new(move || boatramp_cluster::raft::is_leader(&leader_raft, leader_node_id));
    let boatramp_node::RunningNode {
        deploy,
        handlers,
        auth,
        options,
        reconcile: _reconcile,
    } = boatramp_node::assemble(boatramp_node::NodeInput {
        config,
        data_dir: data_dir.as_path(),
        storage,
        kv,
        auth,
        options,
        watch_provider: built_blobs.watch_provider.clone(),
        provision_tier: built_blobs.provision_tier,
        messaging: Some(node.messaging.clone()),
        is_leader,
        node_id: node.node_id,
    })
    .await?;

    tracing::info!(tls = ?args.tls, "cluster: serving public traffic");
    #[cfg(feature = "tls")]
    if !matches!(args.tls, TlsMode::Off) {
        let redirect = args
            .http_redirect_addr
            .or_else(|| config.serve.as_ref().and_then(|s| s.http_redirect_addr));
        if let Some(redirect_addr) = redirect {
            spawn_http_redirect(redirect_addr, deploy.clone(), options.posture);
        }
    }
    let serve_result = match args.tls {
        TlsMode::Off => boatramp_server::serve_with(addr, deploy, auth, handlers, options)
            .await
            .map_err(Error::Serve),
        TlsMode::Custom => serve_custom(&args, addr, deploy, auth, handlers, options).await,
        TlsMode::Acme => serve_acme(&args, addr, deploy, auth, handlers, options).await,
        // Raw-public-key control channel: a self-signed RPK identity the client
        // pins — no cluster cert management needed, so it serves like single-node.
        TlsMode::Rpk => serve_rpk(&args, addr, deploy, auth, handlers, options, &data_dir).await,
        // Cluster-managed certs: the leader issues + stores in the
        // replicated control plane; every node serves the replicated cert.
        #[cfg(feature = "acme-dns")]
        TlsMode::AcmeDns => {
            // Wrap replicated cert private keys at rest when `[secrets]` is set.
            let cert_store: Arc<dyn boatramp_core::cert::CertStore> =
                match build_cert_envelope(config.secrets.as_ref(), &data_dir)? {
                    Some(envelope) => Arc::new(boatramp_core::cert::KvCertStore::with_envelope(
                        node.kv.clone(),
                        envelope,
                    )),
                    None => Arc::new(boatramp_core::cert::KvCertStore::new(node.kv.clone())),
                };
            let cert_raft = node.raft.clone();
            let cert_node_id = node.node_id;
            serve_cluster_acme_dns(
                &args,
                addr,
                deploy,
                auth,
                handlers,
                options,
                cert_store,
                move || boatramp_cluster::raft::is_leader(&cert_raft, cert_node_id),
            )
            .await
        }
        #[cfg(not(feature = "acme-dns"))]
        TlsMode::AcmeDns => serve_acme_dns(&args, addr, deploy, auth, handlers, options).await,
    };

    // Graceful shutdown: force a final flush of the durable Raft store so no
    // committed log/state write is lost to the flush timer (SHUT-1).
    if let Err(e) = durable_kv_handle.flush().await {
        tracing::warn!(error = %e, "cluster: durable Raft store flush on shutdown failed");
    } else {
        tracing::info!("cluster: durable Raft store flushed on shutdown");
    }
    serve_result
}

/// Serve HTTPS with **cluster-managed** ACME DNS-01 certs:
/// the leader issues each cert (sole writer of the DNS-01 TXT — no races) and
/// stores it in the replicated control plane; every node loads the stored cert
/// and serves it, hot-swapping on renewal. The live CA round-trip needs
/// live-platform validation; the store↔serve bridge + leader-gating are unit-tested
/// (`crate::cluster_tls`).
#[cfg(all(feature = "cluster", feature = "acme-dns"))]
#[allow(clippy::too_many_arguments)]
async fn serve_cluster_acme_dns(
    args: &ServeArgs,
    addr: SocketAddr,
    deploy: DeployStore,
    auth: boatramp_server::Auth,
    handlers: boatramp_server::HandlerRuntime,
    options: boatramp_server::ServerOptions,
    cert_store: Arc<dyn boatramp_core::cert::CertStore>,
    is_leader: impl Fn() -> bool + Send + Sync + Clone + 'static,
) -> Result<()> {
    use boatramp_acme::acme::CertRequest;
    use std::time::Duration;

    if args.acme_domain.is_empty() {
        return Err(Error::NoAcmeDomainDns);
    }
    install_crypto_provider();

    let kind = parse_dns_provider(&args.acme_dns_provider)?;
    let provider: Arc<dyn boatramp_acme::dns::DnsProvider> =
        crate::acme_dns::build_provider(kind).await?.into();
    let base = CertRequest {
        directory_url: args.acme_directory.clone(),
        contact_email: args.acme_contact.clone(),
        domains: Vec::new(),
        dns_ttl: 60,
        propagation_delay: Duration::from_secs(15),
        timeout: Duration::from_secs(120),
    };
    let domains = crate::acme_dns::server_domains(&args.acme_domain, args.acme_wildcard_preview);
    let cache = args.acme_cache.clone();

    // Initial pass: the leader issues any missing cert + stores it; all nodes
    // load whatever is in the replicated store.
    let entries =
        cluster_refresh_certs(&cert_store, &domains, is_leader(), &provider, &base, &cache).await?;
    if entries.is_empty() {
        return Err(Error::NoCertsYet);
    }
    // HTTP/3: when enabled, stand up a QUIC endpoint sharing the same
    // ACME certs (build_server_configs gives a `h3`-ALPN config off one resolver);
    // its cert is hot-swapped on renewal below, exactly as the TCP path reloads.
    #[cfg(feature = "http3")]
    let (config, h3_endpoint) = if args.http3 {
        let (tcp, h3) = crate::acme_dns::build_server_configs(entries)?;
        let endpoint =
            boatramp_server::http3_endpoint(addr, boatramp_server::quinn_server_config(h3)?)?;
        (tcp, Some(endpoint))
    } else {
        (crate::acme_dns::build_server_config(entries)?, None)
    };
    #[cfg(not(feature = "http3"))]
    let config = crate::acme_dns::build_server_config(entries)?;
    let tls = axum_server::tls_rustls::RustlsConfig::from_config(Arc::new(config));

    // Background renewal: re-run the leader-gated pass and hot-swap (TCP + h3).
    {
        let (tls, cert_store, provider, base, cache, domains, is_leader) = (
            tls.clone(),
            cert_store.clone(),
            provider.clone(),
            base.clone(),
            cache.clone(),
            domains.clone(),
            is_leader.clone(),
        );
        #[cfg(feature = "http3")]
        let h3_renew = h3_endpoint.clone();
        tokio::spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_secs(6 * 3600)).await;
                match cluster_refresh_certs(
                    &cert_store,
                    &domains,
                    is_leader(),
                    &provider,
                    &base,
                    &cache,
                )
                .await
                {
                    Ok(entries) if !entries.is_empty() => {
                        #[cfg(feature = "http3")]
                        if let Some(endpoint) = &h3_renew {
                            match crate::acme_dns::build_server_configs(entries) {
                                Ok((tcp, h3)) => {
                                    tls.reload_from_config(Arc::new(tcp));
                                    match boatramp_server::quinn_server_config(h3) {
                                        Ok(qc) => endpoint.set_server_config(Some(qc)),
                                        Err(err) => {
                                            tracing::error!(%err, "cluster acme-dns: rebuilding h3 config failed");
                                        }
                                    }
                                }
                                Err(err) => {
                                    tracing::error!(%err, "cluster acme-dns: rebuilding TLS config failed");
                                }
                            }
                        } else {
                            match crate::acme_dns::build_server_config(entries) {
                                Ok(config) => tls.reload_from_config(Arc::new(config)),
                                Err(err) => {
                                    tracing::error!(%err, "cluster acme-dns: rebuilding TLS config failed");
                                }
                            }
                        }
                        #[cfg(not(feature = "http3"))]
                        match crate::acme_dns::build_server_config(entries) {
                            Ok(config) => tls.reload_from_config(Arc::new(config)),
                            Err(err) => {
                                tracing::error!(%err, "cluster acme-dns: rebuilding TLS config failed");
                            }
                        }
                    }
                    Ok(_) => {} // nothing stored yet (follower awaiting the leader)
                    Err(err) => tracing::error!(%err, "cluster acme-dns: renewal failed"),
                }
            }
        });
    }

    tracing::info!(%addr, domains = ?domains, "cluster: serving HTTPS (cluster-managed ACME DNS-01)");
    #[cfg(feature = "handlers")]
    let _scheduler = handlers.spawn_scheduler(deploy.clone());
    let handle = spawn_tls_shutdown();
    let app = boatramp_server::router_with(deploy, auth, handlers, options);
    // Serve h3 over the QUIC endpoint + advertise it on the HTTPS responses.
    #[cfg(feature = "http3")]
    let app = if let Some(endpoint) = h3_endpoint {
        let app_h3 = app.clone();
        tokio::spawn(async move {
            if let Err(err) = boatramp_server::serve_http3_endpoint(endpoint, app_h3).await {
                tracing::error!(%err, "cluster acme-dns: HTTP/3 listener failed");
            }
        });
        boatramp_server::advertise_http3(app, addr.port())
    } else {
        app
    };
    axum_server::bind_rustls(addr, tls)
        .handle(handle)
        .serve(app.into_make_service_with_connect_info::<SocketAddr>())
        .await?;
    Ok(())
}

/// One leader-gated refresh pass: per domain, the leader issues (live CA, via
/// `obtain_or_load`) + stores; every node loads the stored cert. Returns the
/// `(domain, cert)` entries to serve.
#[cfg(all(feature = "cluster", feature = "acme-dns"))]
async fn cluster_refresh_certs(
    cert_store: &Arc<dyn boatramp_core::cert::CertStore>,
    domains: &[String],
    is_leader: bool,
    provider: &Arc<dyn boatramp_acme::dns::DnsProvider>,
    base: &boatramp_acme::acme::CertRequest,
    cache: &Path,
) -> Result<Vec<(String, boatramp_acme::acme::IssuedCert)>> {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    // The `issue` closure yields a *typed* error (`acme_dns::Error`); the
    // refresh itself fails with `cluster_tls::Error`, propagated via `?` into
    // our `ClusterTls` variant. (See `cluster_tls::refresh_entries` — its `Fut`
    // output bound must accept this typed error, not a boxed dynamic one.)
    let entries = crate::cluster_tls::refresh_entries(
        cert_store.as_ref(),
        domains,
        is_leader,
        now,
        |domain| {
            let (provider, base, cache) = (provider.clone(), base.clone(), cache.to_path_buf());
            async move {
                let issued =
                    crate::acme_dns::obtain_or_load(&domain, &base, provider.as_ref(), &cache)
                        .await?;
                Ok::<_, crate::acme_dns::Error>(crate::cluster_tls::issued_to_stored(&issued, now))
            }
        },
    )
    .await?;
    Ok(entries)
}

/// Parse the `--acme-dns-provider` value into a provider kind, using the
/// `ValueEnum` spellings + aliases so **every** built-in provider (all ten) is
/// selectable at serve time — exactly as for the `dns` subcommand (the old
/// hand-rolled match knew only four).
#[cfg(feature = "acme-dns")]
fn parse_dns_provider(value: &str) -> Result<crate::acme_dns::DnsProviderKind> {
    use clap::ValueEnum;
    crate::acme_dns::DnsProviderKind::from_str(value, true)
        .map_err(|_| Error::UnknownDnsProvider(value.to_string()))
}

/// Serve HTTPS with ACME **DNS-01** certificates (wildcards included). Obtains
/// each `--acme-domain` (and, with `--acme-wildcard-preview`, its
/// `*.deploy.<domain>`) via the configured DNS provider, serves them by SNI,
/// and renews in the background. The live CA + DNS round-trip is the
/// integration seam (validated against a Pebble/staging directory + real zone).
#[cfg(feature = "acme-dns")]
async fn serve_acme_dns(
    args: &ServeArgs,
    addr: SocketAddr,
    deploy: DeployStore,
    auth: boatramp_server::Auth,
    handlers: boatramp_server::HandlerRuntime,
    options: boatramp_server::ServerOptions,
) -> Result<()> {
    use std::time::Duration;

    use boatramp_acme::acme::CertRequest;

    if args.acme_domain.is_empty() {
        return Err(Error::NoAcmeDomainDns);
    }
    install_crypto_provider();

    let kind = parse_dns_provider(&args.acme_dns_provider)?;
    let provider = crate::acme_dns::build_provider(kind).await?;
    let base = CertRequest {
        directory_url: args.acme_directory.clone(),
        contact_email: args.acme_contact.clone(),
        domains: Vec::new(),
        dns_ttl: 60,
        propagation_delay: Duration::from_secs(15),
        timeout: Duration::from_secs(120),
    };
    let domains = crate::acme_dns::server_domains(&args.acme_domain, args.acme_wildcard_preview);

    // Obtain (or load cached) certs for every domain up front.
    let entries = obtain_all(&domains, &base, provider.as_ref(), &args.acme_cache).await?;
    // HTTP/3: a QUIC endpoint sharing the same ACME certs, hot-swapped on
    // renewal below.
    #[cfg(feature = "http3")]
    let (config, h3_endpoint) = if args.http3 {
        let (tcp, h3) = crate::acme_dns::build_server_configs(entries)?;
        let endpoint =
            boatramp_server::http3_endpoint(addr, boatramp_server::quinn_server_config(h3)?)?;
        (tcp, Some(endpoint))
    } else {
        (crate::acme_dns::build_server_config(entries)?, None)
    };
    #[cfg(not(feature = "http3"))]
    let config = crate::acme_dns::build_server_config(entries)?;
    let tls = axum_server::tls_rustls::RustlsConfig::from_config(Arc::new(config));

    // Background renewal: re-load (reissuing any near-expiry cert) and hot-swap
    // the served config (TCP + h3), so the process never needs a restart to renew.
    {
        let (tls, base, cache) = (tls.clone(), base.clone(), args.acme_cache.clone());
        let domains = domains.clone();
        let provider = crate::acme_dns::build_provider(kind).await?;
        #[cfg(feature = "http3")]
        let h3_renew = h3_endpoint.clone();
        tokio::spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_secs(6 * 3600)).await;
                match obtain_all(&domains, &base, provider.as_ref(), &cache).await {
                    Ok(entries) => {
                        #[cfg(feature = "http3")]
                        if let Some(endpoint) = &h3_renew {
                            match crate::acme_dns::build_server_configs(entries) {
                                Ok((tcp, h3)) => {
                                    tls.reload_from_config(Arc::new(tcp));
                                    match boatramp_server::quinn_server_config(h3) {
                                        Ok(qc) => endpoint.set_server_config(Some(qc)),
                                        Err(err) => {
                                            tracing::error!(%err, "acme-dns: rebuilding h3 config failed");
                                        }
                                    }
                                }
                                Err(err) => {
                                    tracing::error!(%err, "acme-dns: rebuilding TLS config failed");
                                }
                            }
                        } else {
                            match crate::acme_dns::build_server_config(entries) {
                                Ok(config) => tls.reload_from_config(Arc::new(config)),
                                Err(err) => {
                                    tracing::error!(%err, "acme-dns: rebuilding TLS config failed");
                                }
                            }
                        }
                        #[cfg(not(feature = "http3"))]
                        match crate::acme_dns::build_server_config(entries) {
                            Ok(config) => tls.reload_from_config(Arc::new(config)),
                            Err(err) => {
                                tracing::error!(%err, "acme-dns: rebuilding TLS config failed");
                            }
                        }
                    }
                    Err(err) => tracing::error!(%err, "acme-dns: renewal failed"),
                }
            }
        });
    }

    tracing::info!(%addr, domains = ?domains, "serving HTTPS (ACME DNS-01)");
    // Background scheduler (consumers/crons) — must run under TLS too, not only
    // `--tls off`; in cluster mode its cron tick is gated on `is_leader`. The
    // handle detaches for the server's lifetime.
    #[cfg(feature = "handlers")]
    let _scheduler = handlers.spawn_scheduler(deploy.clone());
    let handle = spawn_tls_shutdown();
    let app = boatramp_server::router_with(deploy, auth, handlers, options);
    #[cfg(feature = "http3")]
    let app = if let Some(endpoint) = h3_endpoint {
        let app_h3 = app.clone();
        tokio::spawn(async move {
            if let Err(err) = boatramp_server::serve_http3_endpoint(endpoint, app_h3).await {
                tracing::error!(%err, "acme-dns: HTTP/3 listener failed");
            }
        });
        boatramp_server::advertise_http3(app, addr.port())
    } else {
        app
    };
    axum_server::bind_rustls(addr, tls)
        .handle(handle)
        .serve(app.into_make_service_with_connect_info::<SocketAddr>())
        .await?;
    Ok(())
}

/// Obtain (or load) every domain's cert, returning `(SNI-pattern, cert)` pairs.
#[cfg(feature = "acme-dns")]
async fn obtain_all(
    domains: &[String],
    base: &boatramp_acme::acme::CertRequest,
    provider: &dyn boatramp_acme::dns::DnsProvider,
    cache: &Path,
) -> Result<Vec<(String, boatramp_acme::acme::IssuedCert)>> {
    let mut entries = Vec::with_capacity(domains.len());
    for domain in domains {
        let cert = crate::acme_dns::obtain_or_load(domain, base, provider, cache).await?;
        entries.push((domain.clone(), cert));
    }
    Ok(entries)
}

#[cfg(not(feature = "acme-dns"))]
async fn serve_acme_dns(
    _args: &ServeArgs,
    _addr: SocketAddr,
    _deploy: DeployStore,
    _auth: boatramp_server::Auth,
    _handlers: boatramp_server::HandlerRuntime,
    _options: boatramp_server::ServerOptions,
) -> Result<()> {
    Err(Error::NoAcmeDnsSupport)
}

/// Construct the OIDC verifier for `/api/auth/exchange` when `--oidc-issuer` is
/// set (fetching the issuer's JWKS now — the live network step) and
/// stash it in `options`. No-op without the `oidc` feature or the flag.
#[cfg(feature = "oidc")]
async fn configure_oidc(
    args: &ServeArgs,
    options: &mut boatramp_server::ServerOptions,
) -> Result<()> {
    let Some(issuer) = args.oidc_issuer.clone() else {
        return Ok(());
    };
    // Without an audience, a JWT minted for a different client at the
    // same issuer could be exchanged for a token. The posture can require one.
    if options.posture.oidc_require_audience && args.oidc_audience.is_none() {
        return Err(Error::OidcAudienceRequired);
    }
    let mut config = boatramp_server::OidcConfig::new(issuer);
    config.audience = args.oidc_audience.clone();
    if let Some(claim) = args.oidc_scope_claim.clone() {
        config.scope_claim = claim;
    }
    let http = reqwest::Client::new();
    let verifier = Arc::new(
        boatramp_server::OidcVerifier::from_discovery(&http, &config)
            .await
            .map_err(|err| Error::OidcSetup(err.to_string()))?,
    );
    tracing::info!(issuer = %config.issuer, "OIDC → token exchange enabled");
    // Periodically re-fetch the JWKS so an IdP key rollover is picked up without
    // a restart. Detached for the server's lifetime; fetch failures are logged.
    {
        let verifier = verifier.clone();
        tokio::spawn(async move {
            loop {
                tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
                if let Err(err) = verifier.refresh().await {
                    tracing::warn!(%err, "OIDC JWKS refresh failed (keeping current keys)");
                }
            }
        });
    }
    options.oidc_verifier = Some(verifier);
    Ok(())
}

#[cfg(not(feature = "oidc"))]
async fn configure_oidc(
    _args: &ServeArgs,
    _options: &mut boatramp_server::ServerOptions,
) -> Result<()> {
    Ok(())
}

#[cfg(feature = "tls")]
async fn serve_custom(
    args: &ServeArgs,
    addr: SocketAddr,
    deploy: DeployStore,
    auth: boatramp_server::Auth,
    handlers: boatramp_server::HandlerRuntime,
    options: boatramp_server::ServerOptions,
) -> Result<()> {
    install_crypto_provider();
    let cert = args.tls_cert.clone().ok_or(Error::TlsCertRequired)?;
    let key = args.tls_key.clone().ok_or(Error::TlsKeyRequired)?;

    let config = axum_server::tls_rustls::RustlsConfig::from_pem_file(&cert, &key).await?;
    tracing::info!(%addr, "serving HTTPS (custom certificate)");
    // Background scheduler (consumers/crons) — must run under TLS too, not only
    // `--tls off`; in cluster mode its cron tick is gated on `is_leader`. The
    // handle detaches for the server's lifetime.
    #[cfg(feature = "handlers")]
    let _scheduler = handlers.spawn_scheduler(deploy.clone());
    let handle = spawn_tls_shutdown();
    let app = boatramp_server::router_with(deploy, auth, handlers, options);

    // Optionally serve HTTP/3 on the same UDP port, feeding the same router, and
    // advertise it (`Alt-Svc`) on the HTTPS responses so clients upgrade to h3 —
    // without the header the h3 listener is never discovered.
    #[cfg(feature = "http3")]
    let app = if args.http3 {
        let (certs, key) = load_cert_chain_and_key(&cert, &key)?;
        let app_h3 = app.clone();
        tokio::spawn(async move {
            if let Err(err) = boatramp_server::serve_http3(addr, certs, key, app_h3).await {
                tracing::error!(%err, "HTTP/3 listener failed");
            }
        });
        boatramp_server::advertise_http3(app, addr.port())
    } else {
        app
    };

    axum_server::bind_rustls(addr, config)
        .handle(handle)
        .serve(app.into_make_service_with_connect_info::<SocketAddr>())
        .await?;
    Ok(())
}

/// Serve the control-plane over **RFC 7250 raw-public-key TLS** (`--tls rpk`):
/// present a persisted control-plane RPK identity the client pins, with the
/// client authenticating via a bearer token (a server-authenticated channel). No
/// ACME, tunnel, or TLS-terminating proxy — an encrypted first-boot / bare-metal
/// control plane.
///
/// The identity is a dedicated `<data-dir>/controlplane-tls.key` (Ed25519,
/// `0600`), **not** the root auth key: the root key may be remote/async
/// (KMS/HSM) while rustls needs a local synchronous signing key, and
/// cross-protocol key reuse is poor hygiene. The public-key fingerprint is
/// logged + printed at startup so the operator can pin it (`--server-pubkey`).
#[cfg(feature = "tls")]
async fn serve_rpk(
    _args: &ServeArgs,
    addr: SocketAddr,
    deploy: DeployStore,
    auth: boatramp_server::Auth,
    handlers: boatramp_server::HandlerRuntime,
    mut options: boatramp_server::ServerOptions,
    data_dir: &Path,
) -> Result<()> {
    install_crypto_provider();

    let key_file = data_dir.join("controlplane-tls.key");
    let identity = boatramp_rpktls::RpkIdentity::load_or_generate(&key_file)?;
    let fingerprint = identity.public_key_hex();

    // If this node holds the root signing key, mint a root-signed attestation of
    // this TLS identity and serve it at `/.well-known/boatramp-bootstrap-identity`
    // so a client can pin *only* the root key and learn the TLS key from the
    // attestation. A verify-only node (no issuer) skips it; the client then pins
    // the printed identity directly with `--server-pubkey`.
    if let Some(signer) = options.issuer.clone() {
        // A year: the attested key is stable across restarts (persisted key file);
        // rotating the identity re-mints a fresh attestation on next boot.
        const ATTESTATION_TTL_SECS: u64 = 365 * 24 * 60 * 60;
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        match boatramp_core::cose::mint_attestation(
            &fingerprint,
            ATTESTATION_TTL_SECS,
            now,
            signer.as_ref(),
        )
        .await
        {
            Ok(att) => options.bootstrap_attestation = Some(att),
            Err(err) => {
                tracing::warn!(%err, "could not mint the bootstrap-TLS attestation; --root-pubkey pinning unavailable");
            }
        }
    }

    // No client-auth trust set: the client authenticates with a bearer token, not
    // a client cert (that is the mutual-`cnf` binding of a later stage).
    let rpk =
        boatramp_rpktls::RpkTls::new(Arc::new(identity), boatramp_rpktls::TrustSet::default());
    let config = axum_server::tls_rustls::RustlsConfig::from_config(Arc::new(rpk.server_auth()?));

    tracing::info!(%addr, pubkey = %fingerprint, "serving HTTPS (RPK bootstrap TLS)");
    // The identity is public (not a secret); print it so the operator can copy it
    // to the client's `--server-pubkey`.
    println!(
        "control-plane RPK TLS identity — pin the client with:\n  --server-pubkey {fingerprint}"
    );

    #[cfg(feature = "handlers")]
    let _scheduler = handlers.spawn_scheduler(deploy.clone());
    let handle = spawn_tls_shutdown();
    let app = boatramp_server::router_with(deploy, auth, handlers, options);
    axum_server::bind_rustls(addr, config)
        .handle(handle)
        .serve(app.into_make_service_with_connect_info::<SocketAddr>())
        .await?;
    Ok(())
}

/// Load a PEM cert chain + private key as DER, for the HTTP/3 (quinn) listener.
#[cfg(feature = "http3")]
fn load_cert_chain_and_key(
    cert: &Path,
    key: &Path,
) -> Result<(
    Vec<rustls::pki_types::CertificateDer<'static>>,
    rustls::pki_types::PrivateKeyDer<'static>,
)> {
    let cert_pem = std::fs::read(cert)?;
    let certs =
        rustls_pemfile::certs(&mut &cert_pem[..]).collect::<std::result::Result<Vec<_>, _>>()?;
    if certs.is_empty() {
        return Err(Error::NoCert(cert.display().to_string()));
    }
    let key_pem = std::fs::read(key)?;
    let key = rustls_pemfile::private_key(&mut &key_pem[..])?
        .ok_or_else(|| Error::NoPrivateKey(key.display().to_string()))?;
    Ok((certs, key))
}

#[cfg(feature = "tls")]
async fn serve_acme(
    args: &ServeArgs,
    addr: SocketAddr,
    deploy: DeployStore,
    auth: boatramp_server::Auth,
    handlers: boatramp_server::HandlerRuntime,
    options: boatramp_server::ServerOptions,
) -> Result<()> {
    use futures::StreamExt;
    use rustls_acme::{caches::DirCache, AcmeConfig};

    if args.acme_domain.is_empty() {
        return Err(Error::NoAcmeDomain);
    }
    install_crypto_provider();

    let mut config = AcmeConfig::new(args.acme_domain.clone())
        .cache(DirCache::new(args.acme_cache.clone()))
        .directory(args.acme_directory.clone());
    if let Some(contact) = &args.acme_contact {
        config = config.contact_push(format!("mailto:{contact}"));
    }
    if let Some(ca) = &args.acme_ca_cert {
        config = config.client_tls_config(acme_client_config(ca)?);
    }

    let mut state = config.state();
    let acceptor = state.axum_acceptor(state.default_rustls_config());
    tokio::spawn(async move {
        loop {
            match state.next().await {
                Some(Ok(event)) => tracing::info!("acme: {event:?}"),
                Some(Err(err)) => tracing::error!("acme error: {err}"),
                None => break,
            }
        }
    });

    tracing::info!(%addr, domains = ?args.acme_domain, "serving HTTPS (ACME)");
    // Background scheduler (consumers/crons) — must run under TLS too, not only
    // `--tls off`; in cluster mode its cron tick is gated on `is_leader`. The
    // handle detaches for the server's lifetime.
    #[cfg(feature = "handlers")]
    let _scheduler = handlers.spawn_scheduler(deploy.clone());
    let handle = spawn_tls_shutdown();
    axum_server::bind(addr)
        .handle(handle)
        .acceptor(acceptor)
        .serve(
            boatramp_server::router_with(deploy, auth, handlers, options)
                .into_make_service_with_connect_info::<SocketAddr>(),
        )
        .await?;
    Ok(())
}

/// Install a process-wide default rustls crypto provider (rustls 0.23 requires
/// one before building any TLS config). Idempotent.
#[cfg(feature = "tls")]
fn install_crypto_provider() {
    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
}

/// An `axum_server::Handle` that triggers graceful shutdown (10s drain) when a
/// Ctrl-C / SIGTERM signal arrives, matching the plain-HTTP listener.
#[cfg(feature = "tls")]
fn spawn_tls_shutdown() -> axum_server::Handle<SocketAddr> {
    let handle = axum_server::Handle::new();
    let trigger = handle.clone();
    tokio::spawn(async move {
        boatramp_server::shutdown_signal().await;
        trigger.graceful_shutdown(Some(std::time::Duration::from_secs(10)));
    });
    handle
}

/// Build a rustls client config that trusts an extra root CA (for a test ACME
/// server like Pebble whose directory uses a self-signed certificate).
#[cfg(feature = "tls")]
fn acme_client_config(ca_path: &std::path::Path) -> Result<Arc<rustls::ClientConfig>> {
    let pem = std::fs::read(ca_path)?;
    let mut roots = rustls::RootCertStore::empty();
    for cert in rustls_pemfile::certs(&mut &pem[..]) {
        roots.add(cert?)?;
    }
    let config = rustls::ClientConfig::builder()
        .with_root_certificates(roots)
        .with_no_client_auth();
    Ok(Arc::new(config))
}

#[cfg(not(feature = "tls"))]
async fn serve_custom(
    _args: &ServeArgs,
    _addr: SocketAddr,
    _deploy: DeployStore,
    _auth: boatramp_server::Auth,
    _handlers: boatramp_server::HandlerRuntime,
    _options: boatramp_server::ServerOptions,
) -> Result<()> {
    Err(Error::NoTlsSupport)
}

#[cfg(not(feature = "tls"))]
async fn serve_acme(
    _args: &ServeArgs,
    _addr: SocketAddr,
    _deploy: DeployStore,
    _auth: boatramp_server::Auth,
    _handlers: boatramp_server::HandlerRuntime,
    _options: boatramp_server::ServerOptions,
) -> Result<()> {
    Err(Error::NoTlsSupport)
}

#[cfg(not(feature = "tls"))]
async fn serve_rpk(
    _args: &ServeArgs,
    _addr: SocketAddr,
    _deploy: DeployStore,
    _auth: boatramp_server::Auth,
    _handlers: boatramp_server::HandlerRuntime,
    _options: boatramp_server::ServerOptions,
    _data_dir: &Path,
) -> Result<()> {
    Err(Error::NoTlsSupport)
}

#[cfg(test)]
mod tests {
    // Every test remaining in this module is `cluster`-gated; the import is unused
    // in a lean build (the single-node auth tests moved to `boatramp_node::auth`).
    #[cfg(feature = "cluster")]
    use super::*;

    /// The mesh write authorizer accepts only a token from the control-plane
    /// root granting `cluster-write` — no token, a wrong-role token, garbage, or a
    /// foreign-root capability are all refused.
    #[cfg(feature = "cluster")]
    #[tokio::test]
    async fn mesh_write_authz_accepts_only_a_cluster_write_capability() {
        use boatramp_cluster::http::ClientWriteAuthz;
        use boatramp_core::authz::GrantedRole;
        use boatramp_core::cose::{self, Claims, LocalSigner, Signer, TokenAlg};

        async fn cap(signer: &dyn Signer, role: &str) -> String {
            let claims = Claims {
                roles: vec![GrantedRole::global(role)],
                kind: cose::KIND_CLUSTER_WRITE.to_string(),
                ttl_secs: None,
                now_unix: 0,
            };
            cose::mint(&claims, signer).await.unwrap()
        }

        let signer = LocalSigner::generate(TokenAlg::Es256);
        let authz = MeshWriteAuthz {
            public: signer.public_key(),
        };

        assert!(
            authz.authorize(Some(&cap(&signer, "cluster-write").await)),
            "a real cluster-write capability"
        );

        assert!(!authz.authorize(None), "no capability");
        assert!(
            !authz.authorize(Some(&cap(&signer, "admin").await)),
            "wrong role"
        );
        assert!(!authz.authorize(Some("not-a-token")), "garbage");

        let other = LocalSigner::generate(TokenAlg::Es256);
        assert!(
            !authz.authorize(Some(&cap(&other, "cluster-write").await)),
            "foreign root key"
        );
    }

    #[cfg(feature = "cluster")]
    #[test]
    fn rotation_interval_parses_units_and_rejects_junk() {
        use std::time::Duration;
        assert_eq!(
            parse_rotation_interval("30d"),
            Some(Duration::from_secs(30 * 86_400))
        );
        assert_eq!(
            parse_rotation_interval("12h"),
            Some(Duration::from_secs(12 * 3600))
        );
        assert_eq!(
            parse_rotation_interval("90m"),
            Some(Duration::from_secs(90 * 60))
        );
        assert_eq!(
            parse_rotation_interval(" 45s "),
            Some(Duration::from_secs(45))
        );
        // No unit, unknown unit, zero, and empty are all rejected (⇒ no schedule).
        assert_eq!(parse_rotation_interval("30"), None);
        assert_eq!(parse_rotation_interval("5w"), None);
        assert_eq!(parse_rotation_interval("0d"), None);
        assert_eq!(parse_rotation_interval(""), None);
    }
}