boatramp_node/

node.rs

//! The node-graph assembly: given a built store (blobs + KV), a configured
//! [`Auth`](boatramp_server::Auth), and resolved
//! [`ServerOptions`](boatramp_server::ServerOptions), wire the deploy store,
//! handler runtime, compute reconcile loop, and domain-verify reconcile loop
//! into a [`RunningNode`] ready to hand to a transport (`serve_with` & friends).
//!
//! This is the headline extraction of `PLAN-node-library`: the binary's
//! `serve::run` used to inline this wiring, so no embedder or in-process test
//! could exercise the same graph the `boatramp serve` binary runs. `run` now
//! resolves the *environment* (args -> backends -> store, signal handlers,
//! migration, auth) and calls [`assemble`]; the cluster path keeps its own inline
//! copy until a later step converges it here.

use std::path::Path;
use std::sync::Arc;

use boatramp_core::Storage;
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::KvStore;

use crate::config::ServerConfig;
use crate::error::{Error, Result};

/// How often the compute reconcile loop converges desired vs actual workloads.
/// Defaults to 30s; override with `BOATRAMP_COMPUTE_RECONCILE_TICK_MS` (milliseconds)
/// so compute-backed tests can converge in a fraction of a second instead of
/// waiting a full tick for the launch/scale reconcile.
pub fn compute_reconcile_tick() -> std::time::Duration {
    std::env::var("BOATRAMP_COMPUTE_RECONCILE_TICK_MS")
        .ok()
        .and_then(|s| s.parse::<u64>().ok())
        .filter(|&ms| ms > 0)
        .map(std::time::Duration::from_millis)
        .unwrap_or(std::time::Duration::from_secs(30))
}
/// How often the domain-verify reconcile loop re-checks pending challenges.
pub const DOMAIN_VERIFY_RECONCILE_TICK: std::time::Duration = std::time::Duration::from_secs(60);
/// How long a compute workload may be idle before scale-to-zero sleeps it.
pub const COMPUTE_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);

/// The built store + resolved config handed to [`assemble`]. Owns the blob/KV
/// backends and the auth/options the caller already resolved; borrows the parsed
/// config and data directory.
pub struct NodeInput<'a> {
    /// The full parsed server config (the handler + compute sections are read here).
    pub config: &'a ServerConfig,
    /// The node data directory (per-site SQL, handler state).
    pub data_dir: &'a Path,
    /// The object store built by [`crate::blobs::build_blobs`].
    pub storage: Arc<dyn Storage>,
    /// Per-project blob backends (PLAN-per-project-blob-backend): a `project → RAW backend` map the
    /// serve wiring built (each with its own sealed S3 credential, fail-closed on an unresolvable one).
    /// RAW (undecorated) — the handler runtime wraps each in the SAME MarkerHeadCache + opt-in
    /// HedgedStorage decorator stack as the global `storage` and assembles a `CompositeBlobBackends`
    /// over the global default. Empty ⇒ no overrides (every project resolves the global; an all-global
    /// deployment is byte-for-byte unchanged). The project a backend binds to is host-resolved at serve
    /// time, never guest input (invariant 1).
    pub per_project_blob_backends: std::collections::HashMap<String, Arc<dyn Storage>>,
    /// The metadata KV, already cache-fronted, built by [`crate::backends::build_kv`].
    pub kv: Arc<dyn KvStore>,
    /// The control-plane auth built by [`crate::auth::configure_auth`].
    pub auth: boatramp_server::Auth,
    /// Server options, already carrying the resolved posture, daemon runtime, and
    /// (post-`configure_auth`/`configure_oidc`) issuer / OIDC verifier.
    pub options: boatramp_server::ServerOptions,
    /// The public HTTP serve bind address, if known — used (under
    /// `allow_guest_self_egress`) to let a handler guest's `wasi:http` reach this
    /// instance's own front door over loopback. `None` (an in-process embedder with no
    /// listener) disables self-egress.
    pub serve_addr: Option<std::net::SocketAddr>,
    /// The cloud blob-change watch provider (FA-5b2), if the backend is a cloud one.
    pub watch_provider: Option<Arc<dyn boatramp_core::blob_provision::WatchProvider>>,
    /// The provisioning tier for the watch provider.
    pub provision_tier: boatramp_core::blob_notify::ProvisionTier,
    /// The `wasi:messaging` substrate override for the handler runtime. `None` uses
    /// the single-node default (`LogMessaging` over the same backends); the cluster
    /// path passes its Raft-backed coordinator.
    pub messaging: Option<Arc<dyn boatramp_core::messaging::Messaging>>,
    /// The single leader gate for cron firing + the compute / domain-verify reconcile
    /// loops. Single-node passes an always-true gate (there is one node); the cluster
    /// passes its Raft `is_leader` check so a single node drives each sweep.
    pub is_leader: boatramp_server::CronLeaderGate,
    /// This node's compute scheduler id (`0` single-node; the cluster node id in a
    /// fleet, so replicas are tagged to the right node).
    pub node_id: u64,
    /// The binary the re-exec'd compute workers run as — the container backend's
    /// `__sandbox` jailer and the microVM backends' `__vmm-run`/`__vz-run` VM hosts.
    /// `None` uses this process's own executable (`current_exe`), which is what
    /// `boatramp serve` wants (the child *is* boatramp). An **embedding harness**
    /// whose own binary doesn't implement those subcommands should point this at a
    /// built `boatramp` binary, so it can drive the real container/microVM backends
    /// in-process (only the per-workload worker re-execs; the serving plane stays
    /// embedded). The docker backend needs neither — it talks to a daemon.
    pub worker_exe: Option<std::path::PathBuf>,
}

/// A fully wired node: the deploy store, handler runtime, auth, and options a
/// transport consumes, plus the detached reconcile loops kept alive for the
/// node's serving life. Destructure it and hold `reconcile` across the serve
/// await so the loops outlive assembly.
pub struct RunningNode {
    /// The deploy store (blob + KV) the router serves from.
    pub deploy: DeployStore,
    /// The handler runtime for wasm handlers (a disabled build ⇒ a no-op runtime).
    pub handlers: boatramp_server::HandlerRuntime,
    /// The control-plane auth.
    pub auth: boatramp_server::Auth,
    /// The resolved server options.
    pub options: boatramp_server::ServerOptions,
    /// The detached reconcile loops (compute + domain-verify). Tokio `JoinHandle`s
    /// do not abort on drop, so the loops run for the process life regardless; the
    /// handles are retained so an embedder can join/abort them on shutdown.
    pub reconcile: Vec<tokio::task::JoinHandle<()>>,
}

/// The instance's own serve socket(s) a guest self-call may reach, given the bind `addr` and
/// whether the posture (`allow_guest_self_egress`) permits it. A wildcard bind
/// (`0.0.0.0`/`::`) is reachable over loopback, so it normalizes to `127.0.0.1` **and** `::1`
/// on the serve port; a specific bind is reachable at itself. Empty when disabled or no
/// listener.
fn self_egress_addrs(
    addr: Option<std::net::SocketAddr>,
    enabled: bool,
) -> Vec<std::net::SocketAddr> {
    use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
    let Some(addr) = addr.filter(|_| enabled) else {
        return Vec::new();
    };
    if addr.ip().is_unspecified() {
        let port = addr.port();
        vec![
            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port),
            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), port),
        ]
    } else {
        vec![addr]
    }
}

/// Whether a node with **no configured control-plane issuer** should auto-provision an ephemeral
/// in-memory fleet signer (for host session cookies + delegable capabilities) from its serve bind.
///
/// The fleet signer is a distinct trust domain from control-plane admin auth, so a DEV / loopback
/// node with auth disabled should still be able to sign/verify session cookies and capabilities. It
/// is gated **strictly to a loopback bind** (`127.0.0.1`/`::1`) or an in-process embedder with no
/// bind address (`None`): NEVER a public or wildcard (`0.0.0.0`/`::`, reachable off-host) bind,
/// where an ephemeral key would silently invalidate live capabilities across a restart — a
/// public node that wants guest capabilities without control-plane auth must supply a persistent
/// key. (`is_loopback()` is already `false` for a wildcard/unspecified address, so a `0.0.0.0` bind
/// correctly does NOT auto-provision.)
#[cfg(feature = "handlers")]
fn should_autoprovision_fleet_signer(serve_addr: Option<std::net::SocketAddr>) -> bool {
    serve_addr.is_none_or(|a| a.ip().is_loopback())
}

/// Wire [`NodeInput`] into a [`RunningNode`]: build the handler runtime, the
/// deploy store (materializing the reserved `default` project), the compute
/// backends + reconcile loop, and the domain-verify reconcile loop.
///
/// The caller has already built the store and configured auth/OIDC on `options`;
/// this is the pure node-graph wiring, identical to what `boatramp serve` runs.
pub async fn assemble(input: NodeInput<'_>) -> Result<RunningNode> {
    let NodeInput {
        config,
        data_dir,
        storage,
        per_project_blob_backends,
        kv,
        auth,
        options,
        serve_addr,
        watch_provider,
        provision_tier,
        messaging,
        is_leader,
        node_id,
        worker_exe,
    } = input;
    // Keep a clone of the node blob storage for the project-export driver (it copies the
    // project's blob prefix in/out via `blob_migrate`), captured before `storage` is
    // moved into the handler runtime below.
    #[cfg(feature = "project-export")]
    let export_blobs = storage.clone();
    // Copy out the posture scalars up front so `options` can be moved into the
    // returned `RunningNode` without a lingering borrow.
    let max_handler_blob_bytes = options.posture.max_handler_blob_bytes;
    let max_component_bytes = options.posture.max_component_bytes;
    let allow_guest_private_egress = options.posture.allow_guest_private_egress;
    let allow_guest_tcp_egress = options.posture.allow_guest_tcp_egress;
    let allow_guest_tcp_listener = options.posture.allow_guest_tcp_listener;
    let allow_env_secret_refs = options.posture.allow_env_secret_refs;
    let allow_guest_email = options.posture.allow_guest_email;
    // The instance's own serve socket(s) a guest self-call may reach, when the posture allows
    // it: a wildcard bind (`0.0.0.0`/`::`) is reachable on loopback, so normalize to
    // `127.0.0.1`/`::1`; a specific bind is itself.
    let self_egress_addrs = self_egress_addrs(serve_addr, options.posture.allow_guest_self_egress);
    let allow_shared_kernel = options.posture.allow_shared_kernel_compute;
    let domain_verify_allow_private = options.posture.domain_verify_allow_private;

    // The deploy store the router serves from — built up front so the handler
    // runtime's managed compute-backed `sql` binding can resolve DB endpoints from
    // the same store the reconcile writes.
    let compute_storage = storage.clone();
    let deploy = DeployStore::new(storage, kv.clone());
    // **Crown-jewel frontier-sync (v0.9.0 KV-recovery, C2).** The sealed-secret / sealed-credential
    // stores below hold IRREPLACEABLE data (a dropped write = a lost secret with no client copy), so
    // rather than rely on per-call-site discipline we route them ALL through a `CheckpointKv`: every
    // write advances the durable frontier before it acks, so a crash's self-heal-on-open trailing-
    // tail quarantine can never drop one. Wrapping ONCE here is the can't-miss guarantee. (On the
    // cluster `RaftKv` / `MemoryKv` the added checkpoint is a no-op, so this is inert there.) The
    // mixed `DeployStore` above keeps the RAW `kv` and selects the frontier-sync per method instead
    // (its `current/*` pointers, history, invocation records and metering stay async).
    let secrets_kv: Arc<dyn KvStore> = Arc::new(boatramp_core::kv::CheckpointKv::new(kv.clone()));
    // The `[secrets]` envelope (local KEK / Vault) that seals a managed SQL
    // credential at rest. `None` ⇒ no wrapping (a managed DB then fails closed).
    let secrets_envelope = build_secrets_envelope(config.secrets.as_ref(), data_dir)?;
    // The project-scoped internal secret store, built from the same KV + `[secrets]`
    // envelope that seal managed-DB credentials. Backs both the `boatramp:<name>`
    // resolver (wired into the handler runtime below, when that feature is present)
    // and the admin secrets API (threaded into `ServerOptions` unconditionally, so it
    // works on a lean node too). `None` when no envelope is configured — the admin
    // endpoints then fail closed with a clear 501, never a panic.
    let secret_store = secrets_envelope.clone().map(|envelope| {
        Arc::new(boatramp_core::secret_store::SecretStore::new(
            secrets_kv.clone(),
            envelope,
        ))
    });
    // The per-TENANT sealed secret store (task #493), sealed with the SAME `[secrets]` envelope.
    // Backs both the control-plane CRUD (threaded into `ServerOptions` below) and — when the
    // `tenant-secrets` feature is compiled — the runtime guest binding (the SAME `Arc` handed to
    // `set_tenant_secret_store`). `None` when no envelope is configured, so the endpoints fail
    // closed with a clear 501 and the guest binding is not built.
    let tenant_secret_store = secrets_envelope.clone().map(|envelope| {
        Arc::new(boatramp_core::secret_store::TenantSecretStore::new(
            secrets_kv.clone(),
            envelope,
        ))
    });
    // The project-scoped SMTP email-profile store, built from the same KV + envelope
    // (the password is sealed at rest). Backs the admin API (`options` below,
    // unconditionally, so it works on a lean node) and — when the `email` feature +
    // `allow_guest_email` posture permit — the runtime's host-side profile
    // resolution (wired inside `build_handler_runtime`). `None` with no envelope, so
    // the admin email endpoints fail closed with a clear 501.
    let email_profile_store = secrets_envelope.clone().map(|envelope| {
        Arc::new(boatramp_core::email_config::EmailProfileStore::new(
            secrets_kv.clone(),
            envelope,
        ))
    });

    // Dev-posture guest-egress extra CA(s): when the posture permits (off/refused under
    // multi-tenant) AND the operator pointed `BOATRAMP_GUEST_EGRESS_EXTRA_CA_FILE` at a PEM, parse
    // it into trust anchors the guest's outbound `wasi:http` TLS client trusts on TOP of the webpki
    // roots (for a hermetic HTTPS test double). Empty otherwise. A configured-but-unreadable/
    // unparsable file is a hard config error (fail closed), never a silent no-trust.
    let guest_egress_extra_roots =
        load_guest_egress_extra_roots(options.posture.allow_guest_egress_extra_ca)?;

    // The handler runtime reuses the same blob/KV backends (per-site prefixed)
    // for its wasi:blobstore/keyvalue bindings; the sql binding is selected by
    // `[handlers.bindings.sql]` (default: per-site libsql files under <data-dir>).
    let handlers = crate::handlers::build_handler_runtime(
        kv.clone(),
        compute_storage.clone(),
        per_project_blob_backends,
        data_dir,
        config.handlers.as_ref(),
        messaging,
        max_handler_blob_bytes,
        max_component_bytes,
        allow_guest_private_egress,
        allow_guest_tcp_egress,
        allow_guest_tcp_listener,
        self_egress_addrs,
        guest_egress_extra_roots,
        allow_env_secret_refs,
        allow_guest_email,
        options.posture.require_tenancy_declaration,
        options.posture.allow_cross_tenant_db,
        &deploy,
        secrets_envelope.clone(),
    )
    .await?;
    // JWKS pre-warm (federated-gateway overhead fix): fetch the configured first-party (`own`-tier)
    // JWKS up front so even the FIRST request per first-party issuer is instant, not a cold fetch.
    // Best-effort + spawned (a slow/unreachable IdP must never delay node readiness); the token
    // cache's single-flight lock dedups a pre-warm fetch against a concurrent first request.
    // `prewarm_own_jwks` internally honors the `[handlers] jwks_prewarm` flag (no-op when off).
    // Sites deployed AFTER startup simply warm lazily on first request (today's behavior).
    #[cfg(feature = "handlers")]
    {
        let prewarm_deploy = deploy.clone();
        tokio::spawn(async move {
            let entries = collect_own_jwks_prewarm(&prewarm_deploy).await;
            if !entries.is_empty() {
                boatramp_server::prewarm_own_jwks(&entries).await;
            }
        });
    }
    // Hand the runtime the SAME per-tenant secret store `Arc` the control-plane routes hold (task
    // #493), so a guest `tenant-secrets` `get` and a control-plane `PUT` seal/unseal against ONE
    // store. Unset when no `[secrets]` envelope, so the guest binding is not built (fail-closed).
    // Handlers-gated: `set_tenant_secret_store` lives on the handler runtime, so a lean (no-handlers)
    // node has no guest binding to wire (the control-plane routes still work via `ServerOptions`).
    #[cfg(feature = "handlers")]
    if let Some(store) = tenant_secret_store.clone() {
        handlers.set_tenant_secret_store(store);
    }
    // Record the node data directory so the node-health endpoint can report `/data` filesystem
    // headroom (a node-health-alerting request — the disk-full incident signal).
    #[cfg(feature = "handlers")]
    handlers.set_data_dir(data_dir.to_path_buf());
    // Wire the fleet session-cookie signer (R3, PLAN-tenancy-principal): the same issuer that mints
    // control-plane tokens signs + verifies the host-issued anonymous session cookie AND the
    // delegable capabilities (PLAN-delegable-capabilities). Handlers-gated: the session-cookie
    // machinery lives on the handler runtime, so a lean (no-handlers) build has nothing to wire.
    //
    // The fleet signer is a DIFFERENT trust domain from control-plane admin auth (signing a
    // customer's session cookie / an embed capability is not the authority to admit an operator to
    // the control plane), but production derives it from the control-plane issuer for convenience.
    // For a DEV / loopback node with control-plane auth disabled (`options.issuer` is `None`),
    // auto-provision an EPHEMERAL in-memory Ed25519 fleet key so the guest-facing signer just works
    // — session cookies + capability mint/verify — WITHOUT turning on control-plane auth. Strictly
    // gated to a loopback bind (or an in-process embedder with no bind address): never on a public
    // bind, where an ephemeral key would silently invalidate live capabilities across a restart (a
    // public node that wants guest capabilities without control-plane auth must supply a persistent
    // key). Ephemeral = issue + verify within one process run; nothing persisted, no cross-process
    // or cross-deploy trust. Production is byte-identical: a real deploy supplies a control-plane key
    // ⇒ `issuer` is `Some` ⇒ this fallback is never taken.
    #[cfg(feature = "handlers")]
    {
        let fleet_signer = options.issuer.clone().or_else(|| {
            should_autoprovision_fleet_signer(serve_addr).then(|| {
                tracing::warn!(
                    "control-plane auth is disabled and no signer is configured; auto-provisioning \
                     an EPHEMERAL in-memory fleet signer (Ed25519) for host session cookies + \
                     delegable capabilities on this loopback/dev node — regenerated each start, \
                     never persisted. Configure a control-plane key (or a dedicated signer) for \
                     production."
                );
                Arc::new(boatramp_core::cose::LocalSigner::generate(
                    boatramp_core::cose::TokenAlg::Ed25519,
                )) as Arc<dyn boatramp_core::cose::Signer>
            })
        });
        if let Some(issuer) = fleet_signer {
            handlers.set_session_signer(issuer);
        }
    }
    // Enable guest capability minting (`boatramp:handlers/capability`, PLAN-delegable-capabilities)
    // when the operator posture allows it. A minted capability is verified against the same fleet
    // signer as the session cookie (wired just above), so this only enables the mint path + the TTL
    // ceiling; posture-off (or a zero ceiling) ⇒ not offered (a guest `mint` is access-denied).
    #[cfg(feature = "capability")]
    if options.posture.allow_guest_mint_capability {
        handlers.set_capability_minting(options.posture.max_guest_capability_ttl_secs);
    }
    // Per-project tenancy/capability posture overrides (Gap 4a): resolve each
    // `[security.projects.<p>]` override against the fleet base so one serve process can run a
    // strict-isolation project beside a looser one on a shared, multi-project machine. Empty ⇒
    // every project uses the node base wired just above. Only these four in-project knobs are
    // per-project; cross-project isolation stays structural (project = database).
    #[cfg(feature = "handlers")]
    {
        let base = &options.posture;
        let overrides: std::collections::BTreeMap<
            String,
            boatramp_core::security::ResolvedProjectTenancy,
        > = config
            .security
            .as_ref()
            .map(|s| {
                s.projects
                    .iter()
                    .map(|(project, ovr)| (project.clone(), base.project_tenancy(ovr)))
                    .collect()
            })
            .unwrap_or_default();
        handlers.set_project_tenancy_overrides(overrides);
    }
    // Wire the guest project self-config capability (`boatramp:handlers/admin`) when the
    // operator posture enables at least one surface. The controller reuses the same in-process
    // domain-verify / email-profile / secret / site-config subsystems + the real domain probe;
    // it's project-scoped per grant and rate-limited + audited. Posture-off ⇒ not offered.
    #[cfg(feature = "admin")]
    {
        use boatramp_handlers::AdminSurface;
        let p = &options.posture;
        let mut surfaces = std::collections::BTreeSet::new();
        if p.allow_guest_admin_domains {
            surfaces.insert(AdminSurface::Domains);
        }
        if p.allow_guest_admin_email {
            surfaces.insert(AdminSurface::Email);
        }
        if p.allow_guest_admin_site {
            surfaces.insert(AdminSurface::Site);
        }
        if p.allow_guest_admin_secrets {
            surfaces.insert(AdminSurface::Secrets);
        }
        if !surfaces.is_empty() {
            let controller = Arc::new(boatramp_server::ServerAdminController::with_server_probe(
                deploy.clone(),
                email_profile_store.clone(),
                secret_store.clone(),
                p.domain_verify_allow_private,
            ));
            handlers.set_admin(controller, surfaces);
        }
    }
    // Leader-gate cron firing (cluster: only the Raft leader fires; single-node: an
    // always-true gate, equivalent to the unset default). The same gate drives the
    // reconcile loops below, so all three converge on one leader per fleet. Only the
    // handler runtime has a scheduler, so this is a no-op without the `handlers` feature.
    #[cfg(feature = "handlers")]
    handlers.set_cron_leader_gate(is_leader.clone());
    // FA-5b2: on a cloud backend, wire the blob-change notification provisioner +
    // its tier so adding a `blob` trigger provisions (and removing it retracts).
    #[cfg(feature = "handlers")]
    if let Some(provider) = watch_provider {
        handlers.set_watch_provider(provider);
        handlers.set_provision_tier(provision_tier);
    }
    #[cfg(not(feature = "handlers"))]
    let _ = (watch_provider, provision_tier);

    // Materialize the reserved `default` project so `project ls` / `project show
    // default` reflect it on a fresh install, not only after a migration. Best
    // effort: the reader backstop keeps listings correct even if this write can't
    // land, so a transient failure must never block serving.
    match deploy.ensure_default_project().await {
        Ok(true) => tracing::info!("materialized the reserved `default` project record"),
        Ok(false) => {}
        Err(e) => tracing::warn!(
            error = %e,
            "could not materialize the `default` project record; readers use the synthesized default"
        ),
    }
    // Wire the function-to-function invoke resolver now the deploy store exists,
    // so a function granted `invoke` can call a sibling in-process (FI).
    #[cfg(feature = "handlers")]
    handlers.set_invoker(deploy.clone());

    // Compute reconcile loop. Single-node is always the "leader". Backends are
    // built from the `[compute]` config + capability detection; a no-op when none
    // are registered. Detached for the server's life.
    // Decide which single workload (if any) may adopt the pre-v0.46 shared legacy `data`
    // volume on this node, from the `databases` config — so the container backend migrates a
    // reused deployment's managed-DB data into the per-(project, workload) scoped layout
    // safely and a non-owner fails closed. `None` when there is no managed DB or ownership is
    // ambiguous (≥2 co-located managed DBs, none designated via `adopt_legacy_data_volume`).
    // The node-trusted legacy-`data` claimant (+ a credential probe reused by the F4 migration
    // below). Computed from the node's OWN config + control-plane records (the deploy store),
    // NEVER a workload spec — so it recognizes the node-default (turnkey) managed pg as the owner
    // of the bare `<volumes>/data` cluster even when the `databases` config is EMPTY (the common
    // ENV-configured turnkey deployment). Needs the `[secrets]` envelope that seals managed-DB
    // credentials; without it there are no managed DBs, so the claimant is `None`.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let (legacy_data_claimant, f4_creds) = match secrets_envelope.clone() {
        Some(envelope) => {
            let creds =
                crate::managed_sql::ManagedSqlCredentials::new(secrets_kv.clone(), envelope);
            let empty = std::collections::BTreeMap::new();
            let databases = config
                .handlers
                .as_ref()
                .and_then(|h| h.bindings.sql.as_ref())
                .map(|sql| &sql.databases)
                .unwrap_or(&empty);
            let claimant =
                crate::managed_sql::legacy_data_volume_claimant(databases, &deploy, &creds).await;
            (claimant, Some(creds))
        }
        None => (None, None),
    };
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let legacy_data_claimant: Option<(String, String)> = None;
    let (compute_backends, compute_node) = crate::compute::build_compute(
        config.compute.as_ref(),
        compute_storage,
        data_dir,
        node_id,
        !allow_shared_kernel,
        options.daemon_runtime.clone(),
        worker_exe.as_deref(),
        legacy_data_claimant.clone(),
    )
    .await;
    // Adopt the IPs of already-running replicas into each backend's fresh-on-boot
    // IP pool BEFORE the reconcile loop starts allocating. A backend with a per-node
    // pool (the native container backend) rebuilds it empty each process start; without
    // this the boot reconcile could re-hand a live address to a different workload —
    // the container-IP collision — or move a replica's endpoint on relaunch. Feeds
    // every persisted replica's `(workload, replica, endpoint-ip)`; each backend keeps
    // only the IPs in its own subnet (a cheap no-op for docker/cloudflare/VMM).
    crate::compute::adopt_running_replica_ips(&deploy, &compute_backends).await;
    // F4 — node-side BULK migration of managed-tenant legacy volume dirs into the per-(project,
    // workload) scoped layout, BEFORE the reconcile loop launches anything. Authorized ENTIRELY
    // from node-trusted state (the deploy store + node-minted sealed credentials + the legacy-
    // `data` claimant), never a workload spec. This proactively prevents the launch-time
    // fail-closed for node-provisioned managed tenants on upgrade — the node-default pg adopts its
    // `<volumes>/data` cluster, and each per-tenant Single dir moves to its scoped path — while the
    // launch-time adoption remains the safety backstop for anything not covered here. Idempotent.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    if let Some(creds) = &f4_creds {
        crate::managed_sql::migrate_managed_tenant_legacy_volumes(
            &deploy,
            creds,
            &legacy_data_claimant,
            &compute_backends,
        )
        .await;
    }
    // Capture a run_ephemeral-capable compute backend for the project-export/import + `sql exec`
    // managed-SQL seam BEFORE `compute_backends` is consumed by the reconcile loop below (a cheap
    // `Arc` clone). Prefer the Docker-daemon backend, but FALL BACK to the native `container`
    // backend — both implement `run_ephemeral` (the native one since v0.32.0), so the seam wires on
    // a fly/firecracker (container-only) node too, not only a Docker host. `None` ⇒ neither backend
    // is present, so the managed-SQL export seam is not wired (the `501 operator-SQL seam not
    // configured`). This was the gap: it previously required `"docker"` and silently no-op'd on every
    // container node.
    #[cfg(feature = "project-export")]
    let export_sql_backend: Option<Arc<dyn boatramp_core::compute::ComputeBackend>> =
        compute_backends
            .get("docker")
            .or_else(|| compute_backends.get("container"))
            .cloned();
    // Per-project internal DNS (service discovery): start the resolver on the bridge
    // gateway so a guest resolves peers by name within its project. On by default;
    // starts only when the container backend + bridge are up (Linux). Detached for
    // the node's serving life (pushed into `reconcile` below). Started before the
    // reconcile loop consumes `compute_backends` — it borrows the registry to check
    // the container backend is present.
    let internal_dns =
        crate::compute::spawn_internal_dns(config.compute.as_ref(), &compute_backends, &deploy);
    // Activate the compute sql-shim (PLAN-compute-bindings): bind its listener +
    // build the resolver when a sql provider and `compute.sql_shim_url` are both present.
    #[cfg(feature = "handlers")]
    let sql_resolver = boatramp_server::sql_shim::spawn_sql_shim(
        handlers.sql_backends(),
        config.compute.as_ref().and_then(|c| c.sql_shim_url.clone()),
    )
    .await;
    #[cfg(not(feature = "handlers"))]
    let sql_resolver: Option<Arc<dyn boatramp_core::compute::ComputeBindingResolver>> = None;

    // Activate the compute blob-shim (PLAN-compute-bindings): the symmetric sibling of the
    // sql-shim — bind its listener + build the resolver when a blob provider and
    // `compute.blob_shim_url` are both present, so a workload's `--bind blob` reaches the
    // SAME per-site `hblob/{qualified-site}/…` namespace a `wasi:blobstore` handler sees.
    #[cfg(feature = "handlers")]
    let blob_resolver = boatramp_server::blob_shim::spawn_blob_shim(
        handlers.blob_backends(),
        config
            .compute
            .as_ref()
            .and_then(|c| c.blob_shim_url.clone()),
        config
            .compute
            .as_ref()
            .and_then(|c| c.blob_shim_max_object_bytes),
    )
    .await;
    #[cfg(not(feature = "handlers"))]
    let blob_resolver: Option<Arc<dyn boatramp_core::compute::ComputeBindingResolver>> = None;

    // Activate the compute messaging-shim (PLAN-compute-bindings): the symmetric sibling of the
    // sql/blob shims — bind its listener + build the resolver when a messaging substrate and
    // `compute.messaging_shim_url` are both present, so a workload's `--bind messaging` publishes
    // to the SAME `{project}/bus/<topic>` project bus a `wasi:messaging` producer's `bus:` publish
    // reaches (the native-MX-edge → `bus:deliver` → consumer-function round-trip, M28 28b).
    #[cfg(feature = "handlers")]
    let messaging_resolver = boatramp_server::messaging_shim::spawn_messaging_shim(
        handlers.messaging(),
        config
            .compute
            .as_ref()
            .and_then(|c| c.messaging_shim_url.clone()),
    )
    .await;
    #[cfg(not(feature = "handlers"))]
    let messaging_resolver: Option<Arc<dyn boatramp_core::compute::ComputeBindingResolver>> = None;

    // Retain a clone of the blob resolver (before it is moved into the reconcile chain) so
    // the dev/CI affordance `POST /api/compute/dev-bind-blob` can mint + register a blob
    // credential for a non-managed local process through the EXACT managed `resolve` path.
    // `None` when the blob-shim is inactive ⇒ that endpoint returns `501`.
    let dev_blob_bind_resolver = blob_resolver.clone();
    // The symmetric clone for `POST /api/compute/dev-bind-messaging` (the messaging dev-bind
    // affordance). `None` when the messaging-shim is inactive ⇒ that endpoint returns `501`.
    let dev_messaging_bind_resolver = messaging_resolver.clone();

    // One resolver for the reconcile: each shim filters to its own binding kind, so chaining
    // them lets a workload declaring `sql`, `blob`, and/or `messaging` bindings be served by all
    // three from a single `resolve` call.
    let binding_resolver = boatramp_core::compute::ChainedBindingResolver::chain([
        sql_resolver,
        blob_resolver,
        messaging_resolver,
    ]);

    // Managed compute-backed SQL (PLAN-managed-compute-sql P2-b): if the handler
    // `sql` config declares any managed database, inject its `POSTGRES_*`/`MYSQL_*`
    // server-init env into the DB workload at launch from the sealed credential.
    // Reaching here with a managed DB implies an envelope (build_handler_runtime
    // fails closed otherwise), so the credential store always has one to seal with.
    // Keep a clone of the secrets envelope for the operator-SQL capability below
    // (the managed_db_resolver match moves the original).
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let operator_envelope = secrets_envelope.clone();
    // …and a second clone for the tenant-deprovision capability (drops a deleted
    // tenant's managed DB/role/credential on project/site delete). It needs a real
    // envelope to seal/unseal + delete per-tenant credentials, so it is wired only
    // when one is present (same fail-closed gating as the managed-DB paths).
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let deprovision_envelope = secrets_envelope.clone();
    // …and a clone for the provisioning drift-repair capability (owner-model retrofit /
    // reconcile). Like the migrate path it connects as the sealed owner role and re-seals
    // credentials, so it needs a real envelope; wired only when one is present.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let repair_envelope = secrets_envelope.clone();
    // …and a third clone for the soft-delete tombstone reaper (the leader-gated task
    // that hard-drops a Shared-Postgres tenant once its grace window elapses). It, too,
    // needs a real envelope to unseal the superuser credential + delete the per-tenant
    // one on hard-drop.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let reaper_envelope = secrets_envelope.clone();
    // …and a clone for the declarative managed-database capability (#501 Stage B): it
    // provisions the declared DB + seals its credential server-side, so it needs a real
    // envelope; wired only when one is present (fail-closed like every managed-DB path).
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let declare_envelope = secrets_envelope.clone();
    // …and a clone for the project-export driver (it unseals managed-DB credentials to
    // assemble the dump/restore connection). Captured here before the resolver match
    // moves the original.
    #[cfg(feature = "project-export")]
    let export_envelope = secrets_envelope.clone();
    // Build the managed-DB server-init credential injector whenever a secrets envelope is
    // present — NOT only when the node-static `[databases]` map is non-empty. A turnkey,
    // ENV-configured node with EMPTY node-static `[databases]` still has APPLIED managed dbs
    // (declared via `boatramp apply`, stored in the deploy store, launched as a derived
    // `bramp-db-<digest>` workload); the resolver resolves those from the deploy store at launch
    // (`ManagedDbEnv::resolve_spec`), so without this the fresh applied db would launch with no
    // `POSTGRES_*`/`MYSQL_*` env and the stock image would refuse to `initdb`. The resolver is a
    // no-op for every non-managed workload, so building it unconditionally (given an envelope) is
    // safe. (Same relaxation the declarative capability already applies — a node with no
    // node-static databases can still serve a project declaration.) No envelope ⇒ `None`: a
    // managed DB requires one to seal its credential, so there is nothing to inject.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let managed_db_resolver: Option<Arc<dyn boatramp_core::compute::ManagedDbEnvResolver>> =
        match secrets_envelope {
            Some(envelope) => {
                let static_dbs = config
                    .handlers
                    .as_ref()
                    .and_then(|h| h.bindings.sql.as_ref())
                    .map(|sql| sql.databases.clone())
                    .unwrap_or_default();
                // Crown-jewel (C2): sealed managed-DB credentials go through the frontier-syncing
                // `secrets_kv`, so a freshly-minted DB credential is past the durable frontier before
                // the workload sees it — never droppable by the self-heal-on-open trailing-tail quarantine.
                let creds =
                    crate::managed_sql::ManagedSqlCredentials::new(secrets_kv.clone(), envelope);
                let privilege = config
                    .compute
                    .as_ref()
                    .map(|c| c.managed_db_privilege)
                    .unwrap_or_default();
                let env = crate::managed_sql::ManagedDbEnv::from_config(
                    &static_dbs,
                    creds,
                    privilege,
                    deploy.clone(),
                );
                Some(Arc::new(env) as Arc<_>)
            }
            None => None,
        };
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let managed_db_resolver: Option<Arc<dyn boatramp_core::compute::ManagedDbEnvResolver>> = None;

    // Turnkey managed DB: auto-register the compute workload(s) backing each managed
    // co-located database that has none yet, so declaring the `databases` binding is
    // enough to boot the DB (no separate `compute set` / apply). Tenant-aware — a
    // `Shared` binding registers its one shared server; a `Single` binding registers
    // nothing at boot (its per-tenant `<compute>-<ident>` is created durably by the lazy
    // resolve on first `sql` use and relaunched by the reconcile, so a project that never
    // uses `sql` — e.g. a static-only site — never gets a spurious DB). Non-clobbering +
    // idempotent; runs before the reconcile loop so its first tick can launch what it
    // registered.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    if let Some(sql) = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .filter(|sql| !sql.databases.is_empty())
    {
        crate::managed_sql::auto_register_managed_db_workloads(&deploy, &sql.databases).await;
    }

    // Operator SQL capability (managed-DB migrations/queries via the sealed credential, resolved
    // server-side) — backs `POST /api/sql/{db}/{exec,query}`. The SAME concrete NodeOperatorSql
    // also backs the owner-gated schema-migration runner (which reuses its owner + superuser
    // backends), so build it ONCE and share it.
    // Build the SAME concrete NodeOperatorSql once (when a managed DB is configured) and share it:
    // it backs both `operator_sql` (the sql exec/query cap) and the migration runner (which reuses
    // its owner + superuser backends). Two separate bindings so neither annotation is a complex type.
    // The migration substrate is a single dispatcher (crate::managed_sql::DispatchMigrationRunner)
    // routing each `(project, db)` to its engine's substrate: the sqlx NodeMigrationRunner for a
    // Postgres/MySQL binding, the LibsqlMigrationRunner for a `libsql` binding. It compiles whenever a
    // sqlx engine OR `migrate` (⇒ libsql) is on, so the embedded-libsql default can migrate even on a
    // node with no external sqlx engine. `operator_sql` (the `POST /api/sql/{db}/{exec,query}` cap)
    // stays sqlx-only — a libsql file has no operator-SQL/credential seam.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let operator_sql: Option<Arc<dyn boatramp_core::sql::OperatorSql>>;
    // Late-init: the `if want_sql_substrate` branches below assign it (and, under sqlx,
    // `operator_sql` in the same block), and the bodies differ by feature-cfg — so a direct
    // `let … = if {…}` would need cfg'd branch bodies. The late-init keeps that readable; the
    // value is always assigned before use.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql", feature = "migrate"))]
    #[allow(clippy::needless_late_init)]
    let migration_substrate: Option<Arc<dyn boatramp_core::sql::MigrationSubstrate>>;
    // Build the operator-SQL / migration substrate when the node has a node-static `databases`
    // map OR — under a sqlx engine — a secrets envelope. The envelope case is the turnkey,
    // ENV-configured node with EMPTY node-static `[databases]`: it can still have APPLIED managed
    // dbs (declared via `boatramp apply`, in the deploy store) that `sql exec/query` must reach,
    // and `NodeOperatorSql` resolves those through the SAME merge point the declare capability
    // uses. (A `migrate`-only node has no managed-credential seam, so it still builds the libsql
    // substrate only when a `libsql` binding is actually declared in the static config.)
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql", feature = "migrate"))]
    let sql_static_dbs = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .map(|sql| sql.databases.clone())
        .unwrap_or_default();
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let want_sql_substrate = !sql_static_dbs.is_empty() || operator_envelope.is_some();
    #[cfg(all(
        not(any(feature = "sql-postgres", feature = "sql-mysql")),
        feature = "migrate"
    ))]
    let want_sql_substrate = !sql_static_dbs.is_empty();
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql", feature = "migrate"))]
    if want_sql_substrate {
        // The sqlx (Postgres/MySQL) arm — the shared NodeOperatorSql backs both the operator-SQL
        // cap and the sqlx migration runner. Only built when a sqlx engine is compiled in.
        #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
        let node_op = Arc::new(crate::managed_sql::NodeOperatorSql::new(
            sql_static_dbs.clone(),
            kv.clone(),
            operator_envelope,
            deploy.clone(),
        ));
        // The operator's trusted-extension allowlist — the only extensions a migration may
        // enable (empty ⇒ none). See ExternalSqlConfig::migrate_trusted_extensions.
        #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
        let trusted: std::collections::BTreeSet<String> = config
            .handlers
            .as_ref()
            .and_then(|h| h.bindings.sql.as_ref())
            .and_then(|sql| sql.migrate_trusted_extensions.clone())
            .unwrap_or_default()
            .into_iter()
            .collect();
        migration_substrate = Some(Arc::new(crate::managed_sql::DispatchMigrationRunner::new(
            #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
            node_op.clone(),
            #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
            trusted,
            #[cfg(feature = "migrate")]
            sql_static_dbs.clone(),
        ))
            as Arc<dyn boatramp_core::sql::MigrationSubstrate>);
        #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
        {
            operator_sql = Some(node_op as Arc<dyn boatramp_core::sql::OperatorSql>);
        }
    } else {
        #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
        {
            operator_sql = None;
        }
        migration_substrate = None;
    }
    // When only `migrate` (no sqlx) is compiled, the operator-SQL cap does not exist.
    #[cfg(all(
        not(any(feature = "sql-postgres", feature = "sql-mysql")),
        feature = "migrate"
    ))]
    let operator_sql: Option<Arc<dyn boatramp_core::sql::OperatorSql>> = None;
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql", feature = "migrate")))]
    let migration_substrate: Option<Arc<dyn boatramp_core::sql::MigrationSubstrate>> = None;
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql", feature = "migrate")))]
    let operator_sql: Option<Arc<dyn boatramp_core::sql::OperatorSql>> = None;

    // Tenant-deprovision capability (drop a deleted tenant's managed DB/role/sealed
    // credential on project/site delete). Wired only when a compute-backed managed
    // database + a secrets envelope are both present — same gating as operator_sql,
    // plus the envelope requirement (it must seal/unseal per-tenant credentials).
    // The soft-delete grace window for a Shared-Postgres managed tenant
    // (`handlers.bindings.sql.deprovision_grace_secs`, env-settable). Default 7 days;
    // `0` disables the soft path (immediate hard drop). Threaded to the deprovisioner
    // (which soft-deletes) and implicitly honored by the reaper (which only ever finds
    // tombstones a >0 grace produced).
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let deprovision_grace_secs = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .and_then(|sql| sql.deprovision_grace_secs)
        .unwrap_or(crate::tenant_sql::DEFAULT_DEPROVISION_GRACE_SECS);
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let tenant_deprovisioner: Option<Arc<dyn boatramp_core::sql::TenantDeprovisioner>> = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .filter(|sql| !sql.databases.is_empty())
        .zip(deprovision_envelope)
        .map(|(sql, envelope)| {
            Arc::new(crate::tenant_sql::NodeTenantDeprovisioner::new(
                deploy.clone(),
                kv.clone(),
                envelope,
                sql.databases.clone(),
                deprovision_grace_secs,
            )) as Arc<_>
        });
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let tenant_deprovisioner: Option<Arc<dyn boatramp_core::sql::TenantDeprovisioner>> = None;

    // Provisioning drift-repair capability (owner-model retrofit / reconcile) — backs the
    // `Project·Admin`-gated `/api/repair/{db}` + `/dry-run`. Same gating as operator_sql plus
    // the envelope requirement (it re-seals the owner credential + connects as it). The
    // envelope is threaded as `Some(_)` so a lean-but-managed node still gets a clear
    // per-check error rather than a panic if none is configured.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let tenant_repair: Option<Arc<dyn boatramp_core::sql::TenantRepair>> = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .filter(|sql| !sql.databases.is_empty())
        .map(|sql| {
            Arc::new(crate::repair::NodeTenantRepair::new(
                sql.databases.clone(),
                deploy.clone(),
                kv.clone(),
                repair_envelope.clone(),
            )) as Arc<_>
        });
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let tenant_repair: Option<Arc<dyn boatramp_core::sql::TenantRepair>> = None;

    // Declarative managed-database capability (#501 Stage B) — backs the `Project·Admin`-gated
    // `PUT /api/projects/{proj}/databases/{name}` + `POST …/ensure`. It persists a manifest
    // `databases:` entry to `project/{project}/database/{name}`, enforces daemon-config-wins at
    // the merge point (against the node-static `sql.databases`), refuses an identity change,
    // binds provisioning to the caller's project, and eagerly provisions via `provision_tenant`.
    // Requires a sqlx engine (it provisions a Postgres/MySQL server) AND a `[secrets]` envelope
    // (a managed DB seals its credential). UNLIKE repair/operator_sql it is NOT gated on
    // `!sql.databases.is_empty()` — a node with NO node-static databases can still accept a
    // project declaration (the whole point of the declarative front door). The node-static map
    // (possibly empty) is threaded so the daemon-wins conflict check has both sources.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let managed_db_declare: Option<Arc<dyn boatramp_core::sql::ManagedDbDeclare>> =
        declare_envelope.map(|envelope| {
            let sql_cfg = config
                .handlers
                .as_ref()
                .and_then(|h| h.bindings.sql.as_ref());
            let static_dbs = sql_cfg.map(|sql| sql.databases.clone()).unwrap_or_default();
            // Per-project declaration ceilings (#501 Stage B MEDIUM-1 — the disk-exhaustion
            // guard): the operator's `max_declared_databases`/`max_declared_volume_mib`, or
            // the built-in `apply_db_caps` defaults (16 DBs / 512 GiB) when unset.
            let quota = crate::managed_db_declare::DeclareQuota::from_config(
                sql_cfg.and_then(|sql| sql.max_declared_databases),
                sql_cfg.and_then(|sql| sql.max_declared_volume_mib),
            );
            Arc::new(crate::managed_db_declare::NodeManagedDbDeclare::new(
                static_dbs,
                deploy.clone(),
                kv.clone(),
                envelope,
                quota,
                compute_backends.clone(),
            )) as Arc<_>
        });
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let managed_db_declare: Option<Arc<dyn boatramp_core::sql::ManagedDbDeclare>> = None;

    // Operator compute-exec capability (run a command inside a running workload) —
    // backs `POST /api/compute/{name}/exec`, gated by the `allow_compute_exec`
    // posture. Clone the backend registry before the reconcile loop consumes it.
    let compute_exec: Option<Arc<dyn boatramp_core::compute::ComputeExec>> = Some(Arc::new(
        crate::compute::NodeComputeExec::new(compute_backends.clone(), deploy.clone()),
    ) as Arc<_>);

    // Operator volume-reclamation capability (list + remove persistent volumes) —
    // backs `GET /api/compute/volumes` + `DELETE /api/compute/volumes/{name}`.
    // Same admin-scoped `/api/compute/*` gate; clone the registry before the
    // reconcile loop consumes the original below.
    let compute_volumes: Option<Arc<dyn boatramp_core::compute::ComputeVolumes>> = Some(Arc::new(
        crate::compute::NodeComputeVolumes::new(compute_backends.clone(), deploy.clone()),
    )
        as Arc<_>);

    // Operator reconcile-plane control capability (restart a replica) — backs
    // `POST /api/compute/maintenance/restart` (admin-scoped). Clone the registry
    // before the reconcile loop consumes the original below.
    let compute_control: Option<Arc<dyn boatramp_core::compute::ComputeControl>> = Some(Arc::new(
        crate::compute::NodeComputeControl::new(compute_backends.clone(), deploy.clone()),
    )
        as Arc<_>);

    let compute_reconcile = boatramp_server::spawn_compute_reconcile(
        deploy.clone(),
        compute_backends,
        vec![compute_node],
        boatramp_core::compute::BackendPolicy::from_shared_kernel_allowed(allow_shared_kernel),
        is_leader.clone(),
        compute_reconcile_tick(),
        COMPUTE_IDLE_TIMEOUT,
        binding_resolver,
        managed_db_resolver,
    );

    // Tenant tombstone reaper: leader-gated hard-drop of soft-deleted Shared-Postgres
    // tenants past their grace window (safe deprovision — see `tenant_sql`). Wired only
    // when a compute-backed managed database + a secrets envelope are both present
    // (same gating as the deprovisioner); each tombstone carries its own server +
    // superuser, so the reaper needs no per-binding config. A `0` grace never writes a
    // tombstone, so the sweep is simply inert then.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let tombstone_reaper: Option<tokio::task::JoinHandle<()>> = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .filter(|sql| !sql.databases.is_empty())
        .zip(reaper_envelope)
        .map(|(_sql, envelope)| {
            crate::tenant_sql::spawn_tenant_tombstone_reaper(
                deploy.clone(),
                kv.clone(),
                envelope,
                is_leader.clone(),
                crate::tenant_sql::TOMBSTONE_REAPER_TICK,
            )
        });
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let tombstone_reaper: Option<tokio::task::JoinHandle<()>> = None;

    // Domain-verify auto-complete: periodically re-check every site's pending
    // ownership challenges and attach any that now pass — a published token (e.g.
    // via `domain add --provider`) converges without a manual `domain verify`.
    let dv_reconcile = boatramp_server::spawn_domain_verify_reconcile(
        deploy.clone(),
        domain_verify_allow_private,
        is_leader,
        DOMAIN_VERIFY_RECONCILE_TICK,
    );

    // Native project export/import capability (env-clone). The node ALWAYS has the
    // deploy + KV + blob deps, so the driver is always wired (a node with no SQL still
    // exports its KV slice + blobs); the engine contexts attach optionally — the managed
    // pg/mysql ephemeral-dump context when a docker backend + envelope + operator-SQL are
    // present, and the LOCAL libsql backend when the `sql` binding is embedded libsql
    // (not remote sqld).
    #[cfg(feature = "project-export")]
    let project_export_cap: Option<Arc<dyn boatramp_core::project_export::ProjectExport>> = {
        let sql_cfg = config
            .handlers
            .as_ref()
            .and_then(|h| h.bindings.sql.as_ref());
        let databases = sql_cfg.map(|s| s.databases.clone()).unwrap_or_default();
        // The operator trusted-extension allowlist the superuser pre-creates in a pg shadow before
        // the NON-superuser apply, so a legit `CREATE EXTENSION IF NOT EXISTS <ext>` (pgvector) in an
        // untrusted dump no-ops instead of failing under the non-superuser apply (HIGH-1 pgvector
        // wrinkle). Same operator-controlled allowlist the owner-gated migration surface uses.
        let trusted_extensions = sql_cfg
            .and_then(|s| s.migrate_trusted_extensions.clone())
            .unwrap_or_default();
        let mut driver = crate::project_export::NodeProjectExport::new(
            databases,
            deploy.clone(),
            kv.clone(),
            export_blobs,
            data_dir.join("project-export"),
        )
        .with_trusted_extensions(trusted_extensions)
        // T3: the configured base for resumable export/import dirs (None ⇒ surface disabled).
        .with_resumable_base(
            config
                .serve
                .as_ref()
                .and_then(|s| s.resumable_export_dir.clone()),
        );
        // Operator digest-pin / override for the ephemeral SQL-client images (the `pg_dump`/
        // `mysqldump`/`psql` one-shots), so a `pgvector/pgvector:pg16@sha256:<hex>` ref can be
        // supplied by env and accepted under the multi-tenant posture without rootfs staging. `None`
        // ⇒ the version-computed tag.
        if let Some(c) = config.compute.as_ref() {
            driver =
                driver.with_client_images(c.pg_client_image.clone(), c.mysql_client_image.clone());
        }
        if let (Some(backend), Some(envelope), Some(op)) =
            (export_sql_backend, export_envelope, operator_sql.clone())
        {
            let published = config
                .compute
                .as_ref()
                .map(|c| {
                    matches!(
                        c.docker_endpoint,
                        boatramp_docker::DockerEndpoint::Published
                    )
                })
                .unwrap_or(true);
            driver = driver.with_managed_pg(backend, envelope, op, published);
        }
        // Local libsql backend (embedded per-site DBs) — skipped for a remote-sqld project
        // (no local file to snapshot). Mirrors `build_sql_backends`' dir derivation.
        if sql_cfg.map(|c| c.url.is_none()).unwrap_or(true) {
            let dir = sql_cfg
                .and_then(|c| c.dir.clone())
                .unwrap_or_else(|| data_dir.join("handlers-sql"));
            let libsql_backends = boatramp_storage::LibsqlSqlBackends::local(dir);
            // inc2: sweep any stale `*.shadow_*` left by a crash mid-import — the atomic
            // shadow-restore's RAII drop-guard only fires in-process, so a crash would otherwise
            // orphan the in-flight shadow DB. Safe at startup (no import is in flight yet).
            match libsql_backends.sweep_stale_shadows() {
                Ok(n) if n > 0 => {
                    tracing::info!(
                        swept = n,
                        "project-export: removed stale libsql import shadows"
                    )
                }
                Ok(_) => {}
                Err(e) => tracing::warn!(error = %e, "project-export: stale-shadow sweep failed"),
            }
            driver = driver.with_libsql(std::sync::Arc::new(libsql_backends));
        }
        let driver = Arc::new(driver);
        // inc2 Stage C/D: best-effort startup sweep of stale `*__shadow_*` / `*__old_*` managed
        // pg/mysql databases left by a crash mid-import (the in-process RAII drop-on-abort only
        // fires in-process). DETACHED so a not-yet-ready / unreachable managed server never blocks
        // or fails node boot; it covers a Shared server's whole fleet via the default-project
        // maintenance connection (a non-default Single instance relies on the in-process guard).
        {
            let sweeper = driver.clone();
            tokio::spawn(async move {
                let n = sweeper
                    .sweep_stale_sql_shadows(boatramp_core::project::DEFAULT_PROJECT)
                    .await;
                if n > 0 {
                    tracing::info!(
                        swept = n,
                        "project-export: removed stale managed-SQL import shadow databases"
                    );
                }
            });
        }
        Some(driver as Arc<dyn boatramp_core::project_export::ProjectExport>)
    };
    #[cfg(not(feature = "project-export"))]
    let project_export_cap: Option<Arc<dyn boatramp_core::project_export::ProjectExport>> = None;

    // Wire the operator capabilities onto the options the router is built from.
    let mut options = options;
    options.operator_sql = operator_sql;
    options.project_export = project_export_cap;
    options.migration_substrate = migration_substrate;
    options.tenant_repair = tenant_repair;
    options.managed_db_declare = managed_db_declare;
    options.tenant_deprovisioner = tenant_deprovisioner;
    options.compute_exec = compute_exec;
    options.dev_blob_bind = dev_blob_bind_resolver;
    options.dev_messaging_bind = dev_messaging_bind_resolver;
    options.compute_volumes = compute_volumes;
    options.compute_control = compute_control;
    // The internal secret store backs the admin secrets API (set/list/delete). Not
    // handlers-gated — it must be reachable even on a lean node.
    options.secret_store = secret_store;
    // The per-tenant sealed secret store backs the control-plane CRUD (task #493). Like the secret
    // store it is not handlers-gated, so the endpoints work on a lean node.
    options.tenant_secret_store = tenant_secret_store;
    // The email-profile store backs the admin API (`/api/email/profiles`); like the
    // secret store it is not handlers-gated, so it works on a lean node.
    options.email_profile_store = email_profile_store;

    // The detached reconcile loops: the always-present compute + domain-verify ones,
    // plus the optional tenant-tombstone reaper (only when a managed DB is configured).
    let mut reconcile = vec![compute_reconcile, dv_reconcile];
    if let Some(reaper) = tombstone_reaper {
        reconcile.push(reaper);
    }
    if let Some(dns) = internal_dns {
        reconcile.push(dns);
    }

    Ok(RunningNode {
        deploy,
        handlers,
        auth,
        options,
        reconcile,
    })
}

/// Build the `[secrets]` envelope (secrets-at-rest wrapping) from `boatramp.cfg`'s
/// `[secrets]` section: `local` (a machine-local AES-256-GCM KEK) or `vault` (Vault
/// Env var an operator points at a PEM file of extra CA(s) the guest's outbound `wasi:http` TLS
/// client should trust on top of the webpki roots — honored only under the
/// `allow_guest_egress_extra_ca` posture (a hermetic HTTPS test double lever).
const GUEST_EGRESS_EXTRA_CA_ENV: &str = "BOATRAMP_GUEST_EGRESS_EXTRA_CA_FILE";

/// Parse the operator's guest-egress extra-CA PEM ([`GUEST_EGRESS_EXTRA_CA_ENV`]) into rustls trust
/// Collect the first-party (`own`-tier) JWKS URLs to pre-warm at startup: every site's gateway-level
/// `[handlers.graphql.data].claims_from_token` that is single-issuer (`jwks_url` set) and NOT
/// multi-issuer (`issuer_trust` unset) — foreign/multi-issuer issuers can't be pre-warmed (unknown
/// until a token arrives), and `jwks_env` is an env var, not a fetch. Returns deduped
/// `(url, issuer, audience)` tuples. Best-effort: a per-site read error is logged and skipped, never
/// propagated (pre-warm must not fail node startup). Per-route `HandlerConfig`/consumer/session/
/// function `token_claims` live in content-addressed deployment manifests not enumerated here; those
/// warm lazily on first request (today's behavior).
#[cfg(feature = "handlers")]
async fn collect_own_jwks_prewarm(deploy: &DeployStore) -> Vec<(String, String, Option<String>)> {
    use boatramp_core::project::ProjectRef;
    let sites = match deploy.list_sites_all().await {
        Ok(s) => s,
        Err(e) => {
            tracing::warn!(error = %e, "JWKS pre-warm: could not list sites; skipping pre-warm");
            return Vec::new();
        }
    };
    let mut seen = std::collections::HashSet::new();
    let mut out = Vec::new();
    for (project, site) in sites {
        let sc = match deploy
            .get_site_config(ProjectRef::new(&project), &site)
            .await
        {
            Ok(Some(sc)) => sc,
            Ok(None) => continue,
            Err(e) => {
                tracing::warn!(project = %project, site = %site, error = %e,
                    "JWKS pre-warm: could not load site config; skipping");
                continue;
            }
        };
        let Some(claims) = sc
            .handlers
            .as_ref()
            .and_then(|h| h.graphql.as_ref())
            .and_then(|g| g.data.as_ref())
            .and_then(|d| d.claims_from_token.as_ref())
        else {
            continue;
        };
        // OWN tier only: operator-fixed single-issuer `jwks_url`, never multi-issuer discovery.
        if claims.issuer_trust.is_some() {
            continue;
        }
        if let Some(url) = claims.jwks_url.as_ref() {
            let entry = (url.clone(), claims.issuer.clone(), claims.audience.clone());
            if seen.insert(entry.clone()) {
                out.push(entry);
            }
        }
    }
    out
}

/// anchors, gated by the `allow_guest_egress_extra_ca` posture (`allow`). No env set ⇒ empty (the
/// default). `allow == false` (e.g. multi-tenant) with a file set ⇒ empty + a warning (the posture
/// refuses it). Set + readable + ≥1 cert ⇒ those certs. Set-but-unreadable / no valid cert ⇒ a hard
/// error (fail closed — a configured-but-broken CA must not silently degrade to no-trust).
fn load_guest_egress_extra_roots(
    allow: bool,
) -> Result<Vec<rustls::pki_types::CertificateDer<'static>>> {
    let path = match std::env::var(GUEST_EGRESS_EXTRA_CA_ENV) {
        Ok(p) if !p.is_empty() => p,
        _ => return Ok(Vec::new()),
    };
    if !allow {
        tracing::warn!(
            env = GUEST_EGRESS_EXTRA_CA_ENV,
            "ignoring a guest-egress extra CA: the security posture forbids it \
             (allow_guest_egress_extra_ca is off — e.g. under multi-tenant)"
        );
        return Ok(Vec::new());
    }
    let pem =
        std::fs::read(&path).map_err(|e| Error::GuestEgressCa(format!("reading {path:?}: {e}")))?;
    let certs = rustls_pemfile::certs(&mut std::io::BufReader::new(&pem[..]))
        .collect::<std::result::Result<Vec<_>, _>>()
        .map_err(|e| Error::GuestEgressCa(format!("parsing {path:?}: {e}")))?;
    if certs.is_empty() {
        return Err(Error::GuestEgressCa(format!(
            "{path:?} contained no PEM certificate"
        )));
    }
    tracing::info!(
        env = GUEST_EGRESS_EXTRA_CA_ENV,
        count = certs.len(),
        path = %path,
        "guest egress trusts operator extra CA(s) (dev-posture; webpki roots still apply)"
    );
    Ok(certs)
}

/// Transit). `None`/empty ⇒ no wrapping. The Vault token is read from the
/// environment (`token_env`), never a file. This seals a managed SQL credential at
/// rest; a managed database fails closed without it.
fn build_secrets_envelope(
    secrets: Option<&crate::config::SecretsConfig>,
    data_dir: &Path,
) -> Result<Option<Arc<dyn boatramp_core::envelope::KeyEnvelope>>> {
    use boatramp_server::envelope::{EnvelopeSpec, build_envelope};
    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()))
}

#[cfg(all(test, feature = "fs"))]
mod tests {
    use super::*;
    use boatramp_core::kv::MemoryKv;
    use boatramp_core::security::SecurityProfile;

    /// The dev/loopback ephemeral fleet-signer auto-provision is gated STRICTLY to a loopback bind
    /// (or an in-process embedder with no bind): a public / wildcard / private-network bind must
    /// NOT silently provision an ephemeral signing key (it would invalidate live capabilities on a
    /// restart — such a node must supply a persistent key). This is the security-critical boundary.
    #[cfg(feature = "handlers")]
    #[test]
    fn ephemeral_fleet_signer_auto_provisions_only_on_loopback_or_in_process() {
        use std::net::SocketAddr;
        let sa = |s: &str| s.parse::<SocketAddr>().unwrap();
        // In-process (no listener) and loopback → auto-provision the dev fleet signer.
        assert!(should_autoprovision_fleet_signer(None));
        assert!(should_autoprovision_fleet_signer(Some(sa(
            "127.0.0.1:8080"
        ))));
        assert!(should_autoprovision_fleet_signer(Some(sa("[::1]:8080"))));
        // Off-host-reachable binds MUST NOT auto-provision an ephemeral key: a public IP, a
        // private-network IP, and a wildcard bind (`0.0.0.0`/`::`, reachable off-host).
        assert!(!should_autoprovision_fleet_signer(Some(sa(
            "203.0.113.5:8080"
        ))));
        assert!(!should_autoprovision_fleet_signer(Some(sa(
            "10.0.0.4:8080"
        ))));
        assert!(!should_autoprovision_fleet_signer(Some(sa("0.0.0.0:8080"))));
        assert!(!should_autoprovision_fleet_signer(Some(sa("[::]:8080"))));
    }

    /// The headline in-process fidelity check (PLAN-node-library N2b.3): `assemble`
    /// over a temp `FsStorage` + `MemoryKv` produces a `RunningNode` whose deploy
    /// store is live (the reserved `default` project was materialized during
    /// assembly) and whose router — the exact one `boatramp serve` builds — answers
    /// `/healthz`. No listener is bound: the request is driven through the router
    /// via `tower::oneshot`, so the whole assembly runs in-process.
    #[tokio::test]
    async fn assemble_produces_a_serving_node_over_a_temp_store() {
        use axum::body::Body;
        use axum::http::{Request, StatusCode};
        use tower::ServiceExt;

        let tmp = tempfile::tempdir().unwrap();
        let storage: Arc<dyn Storage> = Arc::new(boatramp_storage::FsStorage::new(tmp.path()));
        let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
        let config = ServerConfig::default();
        let options = boatramp_server::ServerOptions {
            // The strict `multi-tenant` posture, as an unconfigured `serve` resolves.
            posture: SecurityProfile::MultiTenant.preset(),
            ..Default::default()
        };

        let node = assemble(NodeInput {
            config: &config,
            data_dir: tmp.path(),
            storage,
            per_project_blob_backends: std::collections::HashMap::new(),
            kv,
            auth: boatramp_server::Auth::disabled(),
            options,
            serve_addr: None,
            watch_provider: None,
            provision_tier: boatramp_core::blob_notify::ProvisionTier::default(),
            messaging: None,
            is_leader: Arc::new(|| true),
            node_id: 0,
            worker_exe: None,
        })
        .await
        .expect("assemble a node over a temp store");

        // The deploy store is live: `assemble` already materialized the reserved
        // `default` project, so a second ensure reports "already present" (`false`).
        assert!(
            !node
                .deploy
                .ensure_default_project()
                .await
                .expect("read the default project"),
            "assemble should have materialized the default project"
        );

        // The assembled router (the same wiring `serve` binds) answers /healthz.
        let router =
            boatramp_server::router_with(node.deploy, node.auth, node.handlers, node.options);
        let response = router
            .oneshot(
                Request::builder()
                    .uri("/healthz")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .expect("route /healthz");
        assert_eq!(response.status(), StatusCode::OK);
    }
}