// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! `OpenShell` Server library.
//!
//! This crate provides the server implementation for `OpenShell`, including:
//! - gRPC service implementation
//! - HTTP health endpoints
//! - Protocol multiplexing (gRPC + HTTP on same port)
//! - mTLS support
//!
//! Compiled-in compute drivers are installed into a registry at gateway
//! startup. Runtime selection only consults that registry or a configured
//! external endpoint; it does not switch on driver names.

mod auth;
pub mod certgen;
pub mod cli;
mod compute;
pub mod config_file;
mod config_update_operation;
mod credentials;
mod defaults;
mod gateway_listener;
mod gateway_ocsf;
mod grpc;
mod http;
mod middleware;
mod multiplex;
mod ocsf_log;
mod otel_tracing;
mod pagination;
mod persistence;
pub(crate) mod policy_store;
mod provider_profile_sources;
mod provider_refresh;
mod readiness;
mod sandbox_index;
mod sandbox_watch;
mod service_routing;
mod ssh_sessions;
mod storage_proto;
mod supervisor_owner;
pub mod supervisor_session;
mod telemetry;
#[cfg(any(test, feature = "test-support"))]
pub mod test_support;
mod tls;
#[cfg(test)]
pub(crate) mod tls_test_utils;
pub mod tracing_bus;
mod tracing_setup;
mod watch_cursor;
mod ws_tunnel;

use metrics_exporter_prometheus::PrometheusBuilder;
use openshell_core::net::set_tcp_nodelay_best_effort;
use openshell_core::telemetry::TelemetryComputeDriver;
use openshell_core::{Config, Error, ObjectLabels, Result};
use openshell_extension_core::{
    BearerTokenSlot, ExtensionAudience, ExtensionCallerKind, ExtensionKind, MAX_EXTENSION_TOKEN_TTL,
};
use openshell_supervisor_middleware::MiddlewareRegistry;
use std::collections::{BTreeMap, HashMap};
use std::io::ErrorKind;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
#[cfg(test)]
use std::sync::LazyLock;
use std::sync::{
    Arc, Mutex,
    atomic::{AtomicBool, Ordering},
};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::watch;
use tracing::{debug, error, info, warn};

#[cfg(test)]
pub(crate) static TEST_ENV_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));

/// Serializes tests that assert on captured spans, which share one exporter.
#[cfg(test)]
pub(crate) static TEST_TRACING_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));

pub(crate) fn install_jsonwebtoken_crypto_provider() {
    let _ = jsonwebtoken::crypto::aws_lc::DEFAULT_PROVIDER.install_default();
}

use compute::ComputeRuntime;
use gateway_listener::{BoundGatewayListener, bind_gateway_listener};
pub use grpc::OpenShellService;
pub use http::{health_router, http_router, metrics_router, service_http_router};

/// Deriving `Debug` is safe here: `BearerTokenSlot` renders only its expiry,
/// and an extension audience is configuration rather than secret material.
#[derive(Debug)]
struct GatewayExtensionCredential {
    name: String,
    audience: ExtensionAudience,
    slot: BearerTokenSlot,
    ttl: Duration,
}

fn extension_token_ttl(issuer: &auth::sandbox_jwt::ExtensionJwtIssuer) -> Duration {
    issuer.token_ttl().min(MAX_EXTENSION_TOKEN_TTL)
}

/// Mint the gateway-caller credential for one extension registration.
///
/// Returns `Ok(None)` when the operator has explicitly opted the registration
/// out of extension authentication. The opt-out is deliberately loud: it
/// downgrades a security boundary, so it is reported once per registration at
/// startup rather than being silently tolerated.
fn mint_gateway_extension_credential(
    issuer: &Arc<auth::sandbox_jwt::ExtensionJwtIssuer>,
    kind: ExtensionKind,
    name: &str,
    audience: &str,
    endpoint: &str,
    allow_insecure_transport: bool,
) -> Result<Option<GatewayExtensionCredential>> {
    // A middleware endpoint must be reachable from sandbox supervisors, so a
    // gateway-local Unix socket is only an option for interceptors.
    let (accepted, supports_unix) = match kind {
        ExtensionKind::Middleware => ("https://", false),
        ExtensionKind::Interceptor => ("https:// or unix://", true),
        _ => {
            return Err(Error::config(format!(
                "extension kind '{kind}' is not supported by gateway authentication"
            )));
        }
    };
    if allow_insecure_transport {
        warn!(
            extension = %name,
            endpoint = %endpoint,
            "extension authentication is DISABLED for this registration by \
             allow_insecure_transport; OpenShell attaches no caller credential \
             and the service cannot distinguish OpenShell from any other \
             network client. Use {accepted} with the opt-out removed outside \
             trusted-network development deployments."
        );
        return Ok(None);
    }
    let transport_supported =
        endpoint.starts_with("https://") || (supports_unix && endpoint.starts_with("unix://"));
    if !transport_supported {
        return Err(Error::config(format!(
            "authenticated {kind} '{name}' must use {accepted}; set \
             allow_insecure_transport = true to opt this registration out of \
             extension authentication instead"
        )));
    }
    let audience = ExtensionAudience::new(audience.to_string()).map_err(|error| {
        Error::config(format!(
            "extension '{name}' has an invalid audience: {error}"
        ))
    })?;
    let ttl = extension_token_ttl(issuer);
    let minted = issuer
        .mint_extension_token(&audience, ExtensionCallerKind::Gateway, None, ttl)
        .map_err(|status| {
            Error::config(format!(
                "failed to mint credential for extension '{name}': {}",
                status.message()
            ))
        })?;
    let slot = BearerTokenSlot::new(&minted.token, minted.expires_at_ms).map_err(|error| {
        Error::config(format!(
            "failed to install credential for extension '{name}': {error}"
        ))
    })?;
    Ok(Some(GatewayExtensionCredential {
        name: name.to_string(),
        audience,
        slot,
        ttl,
    }))
}

fn spawn_gateway_extension_token_refresh(
    issuer: Arc<auth::sandbox_jwt::ExtensionJwtIssuer>,
    credentials: Vec<GatewayExtensionCredential>,
) {
    if credentials.is_empty() {
        return;
    }
    tokio::spawn(async move {
        loop {
            let now_ms = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map_or(0, |duration| {
                    i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
                });
            let remaining_ms = credentials
                .iter()
                .filter_map(|credential| credential.slot.expires_at_ms())
                .min()
                .map_or(60_000, |expiry_ms| expiry_ms.saturating_sub(now_ms));
            let refresh_delay = if remaining_ms <= 0 {
                Duration::from_millis(100)
            } else {
                Duration::from_millis(
                    u64::try_from(remaining_ms)
                        .unwrap_or(u64::MAX)
                        .saturating_mul(4)
                        .checked_div(5)
                        .unwrap_or(100)
                        .max(100),
                )
            };
            tokio::time::sleep(refresh_delay).await;
            for credential in &credentials {
                match issuer.mint_extension_token(
                    &credential.audience,
                    ExtensionCallerKind::Gateway,
                    None,
                    credential.ttl,
                ) {
                    Ok(minted) => {
                        if let Err(error) =
                            credential.slot.update(&minted.token, minted.expires_at_ms)
                        {
                            warn!(
                                extension = %credential.name,
                                error = %error,
                                "failed to rotate gateway extension credential"
                            );
                        }
                    }
                    Err(status) => warn!(
                        extension = %credential.name,
                        error = %status,
                        "failed to mint gateway extension credential"
                    ),
                }
            }
        }
    });
}
pub use multiplex::{MultiplexService, MultiplexedService};
pub use persistence::Store;
use sandbox_index::SandboxIndex;
use sandbox_watch::SandboxWatchBus;
pub use tls::TlsAcceptor;
use tracing_bus::TracingLogBus;

pub(crate) struct ServerStartupConfig {
    pub config: Config,
    pub config_file: Option<config_file::ConfigFile>,
    pub guest_tls: Option<compute::driver_config::GuestTlsPaths>,
    pub compute_driver: ComputeDriverSelection,
    pub legacy_compute_driver_env_seen: bool,
}

/// Server state shared across handlers.
#[derive(Debug)]
pub struct ServerState {
    /// Server configuration.
    pub config: Config,

    /// Persistence store.
    pub store: Arc<Store>,

    /// Compute orchestration over the configured driver.
    pub compute: ComputeRuntime,

    /// Credential-driver selection and resolution runtime.
    pub credentials: credentials::CredentialRuntime,

    /// In-memory sandbox correlation index.
    pub sandbox_index: SandboxIndex,

    /// In-memory bus for sandbox update notifications.
    pub sandbox_watch_bus: SandboxWatchBus,

    /// In-memory bus for server process logs.
    pub tracing_log_bus: TracingLogBus,

    /// In-memory anonymous telemetry accounting for active sandbox sessions.
    pub(crate) telemetry: telemetry::TelemetryState,

    /// Active SSH tunnel connection counts per session token.
    pub ssh_connections_by_token: Mutex<HashMap<String, u32>>,

    /// Active SSH tunnel connection counts per sandbox id.
    pub ssh_connections_by_sandbox: Mutex<HashMap<String, u32>>,

    /// Serializes settings mutations (global and sandbox) to prevent
    /// read-modify-write races. Held for the duration of any setting
    /// set/delete operation, including the precedence check on sandbox
    /// mutations that reads global state.
    pub settings_mutex: tokio::sync::Mutex<()>,

    /// Registry of active supervisor sessions and pending relay channels.
    ///
    /// Stored as `Arc` so compiled compute drivers can be constructed before
    /// `ServerState` and still
    /// query session state to surface supervisor readiness.
    pub supervisor_sessions: Arc<supervisor_session::SupervisorSessionRegistry>,

    /// Set once graceful gateway shutdown begins so stream handlers can
    /// distinguish expected transport closes from runtime failures.
    pub(crate) gateway_shutting_down: AtomicBool,

    /// Stable identity for this gateway process.
    pub replica_id: String,

    /// Internal endpoint other gateway replicas can dial for peer RPCs.
    pub peer_endpoint: Option<String>,

    /// Reused peer connections, peer token, and owner lookups for relay
    /// forwarding. Keeps per-relay cost off the connection and auth paths.
    pub peer_routes: Arc<supervisor_session::PeerRouteCache>,

    /// Idle HTTP/1 upstreams to sandbox services, so routed requests reuse a
    /// relay instead of opening one per request.
    pub service_upstreams: Arc<service_routing::ServiceUpstreamPool>,

    /// Validated built-in and operator-registered supervisor middleware.
    pub middleware_registry: Arc<MiddlewareRegistry>,

    /// OIDC JWKS cache for JWT validation. `None` when OIDC is not configured.
    pub oidc_cache: Option<Arc<auth::oidc::JwksCache>>,

    /// Typed extension JWT issuer and public verification metadata.
    pub extension_jwt_issuer: Option<Arc<auth::sandbox_jwt::ExtensionJwtIssuer>>,

    /// Launch-scoped gateway and Sandbox Protocol token authority.
    pub sandbox_session_jwt_authority: Option<Arc<auth::sandbox_jwt::SandboxSessionJwtAuthority>>,

    /// Optional selected-driver authenticator for the `IssueSandboxToken`
    /// bootstrap path.
    pub compute_driver_authenticator: Option<Arc<auth::compute_driver::ComputeDriverAuthenticator>>,

    /// Optional K8s `ServiceAccount` authenticator for gateway peer RPCs.
    pub peer_authenticator: Option<Arc<auth::peer::PeerServiceAccountAuthenticator>>,

    /// Gateway-wide gRPC request rate limiter shared by every multiplex path.
    pub(crate) grpc_rate_limiter: Option<multiplex::GrpcRateLimiter>,

    /// Per-sandbox bound on extension credential minting, which resolves the
    /// caller's effective policy on every request.
    pub(crate) extension_mint_limiter: auth::extension_mint_limit::ExtensionMintLimiter,

    /// Immutable gateway interceptor execution plan. `None` when disabled.
    pub(crate) gateway_interceptors:
        Option<openshell_gateway_interceptors::GatewayInterceptorRuntime>,

    /// Gateway-local provider profile sources. User-imported profiles are read
    /// on demand when the user source is configured.
    pub(crate) provider_profile_sources: provider_profile_sources::ProviderProfileSources,

    /// OIDC admin role name for workspace-level authorization.
    /// Empty when OIDC is not configured — `authorize_workspace()` treats
    /// every authenticated user as Platform Admin in that case.
    pub admin_role: String,
}

fn is_benign_tls_handshake_failure(error: &std::io::Error) -> bool {
    matches!(
        error.kind(),
        ErrorKind::UnexpectedEof | ErrorKind::ConnectionReset
    )
}

fn is_benign_connection_close(error: &(dyn std::error::Error + 'static)) -> bool {
    openshell_core::transport_errors::is_expected_transport_close_error(error)
}

impl ServerState {
    /// Create new server state.
    #[must_use]
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        config: Config,
        store: Arc<Store>,
        compute: ComputeRuntime,
        sandbox_index: SandboxIndex,
        sandbox_watch_bus: SandboxWatchBus,
        tracing_log_bus: TracingLogBus,
        supervisor_sessions: Arc<supervisor_session::SupervisorSessionRegistry>,
        oidc_cache: Option<Arc<auth::oidc::JwksCache>>,
    ) -> Self {
        let credentials =
            credentials::CredentialRuntime::from_config_with_store(&config, Arc::clone(&store))
                .expect("server config should be validated before ServerState::new");
        Self::new_with_credentials(
            config,
            store,
            compute,
            sandbox_index,
            sandbox_watch_bus,
            tracing_log_bus,
            supervisor_sessions,
            oidc_cache,
            credentials,
        )
    }

    /// Create new server state with an already-initialized credential runtime.
    #[must_use]
    #[allow(clippy::too_many_arguments)]
    pub fn new_with_credentials(
        config: Config,
        store: Arc<Store>,
        compute: ComputeRuntime,
        sandbox_index: SandboxIndex,
        sandbox_watch_bus: SandboxWatchBus,
        tracing_log_bus: TracingLogBus,
        supervisor_sessions: Arc<supervisor_session::SupervisorSessionRegistry>,
        oidc_cache: Option<Arc<auth::oidc::JwksCache>>,
        credentials: credentials::CredentialRuntime,
    ) -> Self {
        let replica_id = compute::lease::replica_id();
        let peer_endpoint = derive_peer_endpoint(&config);
        let grpc_rate_limiter = multiplex::GrpcRateLimiter::from_config(&config);
        let admin_role = config
            .oidc
            .as_ref()
            .map_or_else(String::new, |oidc| oidc.admin_role.clone());
        Self {
            config,
            store,
            compute,
            credentials,
            sandbox_index,
            sandbox_watch_bus,
            tracing_log_bus,
            telemetry: telemetry::TelemetryState::new(),
            ssh_connections_by_token: Mutex::new(HashMap::new()),
            ssh_connections_by_sandbox: Mutex::new(HashMap::new()),
            settings_mutex: tokio::sync::Mutex::new(()),
            supervisor_sessions,
            gateway_shutting_down: AtomicBool::new(false),
            replica_id,
            peer_endpoint,
            peer_routes: Arc::new(supervisor_session::PeerRouteCache::default()),
            service_upstreams: Arc::new(service_routing::ServiceUpstreamPool::default()),
            extension_mint_limiter: auth::extension_mint_limit::ExtensionMintLimiter::default(),
            middleware_registry: Arc::new(MiddlewareRegistry::default()),
            oidc_cache,
            extension_jwt_issuer: None,
            sandbox_session_jwt_authority: None,
            compute_driver_authenticator: None,
            peer_authenticator: None,
            grpc_rate_limiter,
            gateway_interceptors: None,
            provider_profile_sources:
                provider_profile_sources::ProviderProfileSources::with_default_sources(),
            admin_role,
        }
    }
}

fn derive_peer_endpoint(config: &Config) -> Option<String> {
    if let Ok(endpoint) = std::env::var("OPENSHELL_PEER_ENDPOINT")
        && !endpoint.trim().is_empty()
    {
        return Some(endpoint.trim().to_string());
    }

    let pod_name = std::env::var("OPENSHELL_POD_NAME").ok()?;
    let namespace = std::env::var("OPENSHELL_POD_NAMESPACE").ok()?;
    let service = std::env::var("OPENSHELL_PEER_SERVICE_NAME").ok()?;
    if pod_name.trim().is_empty() || namespace.trim().is_empty() || service.trim().is_empty() {
        return None;
    }

    let scheme = if config.tls.is_some() {
        "https"
    } else {
        "http"
    };
    Some(format!(
        "{scheme}://{pod}.{service}.{namespace}.svc.cluster.local:{port}",
        pod = pod_name.trim(),
        service = service.trim(),
        namespace = namespace.trim(),
        port = config.bind_address.port()
    ))
}

/// Reject a plaintext peer endpoint on a gateway that serves TLS.
///
/// Peer relay traffic carries whole supervisor sessions between replicas. The
/// chart renders a plaintext peer endpoint only when the gateway itself serves
fn validate_peer_endpoint_scheme(config: &Config, peer_endpoint: &str) -> Result<()> {
    if peer_endpoint.starts_with("https://") {
        return Ok(());
    }
    if config.tls.is_some() {
        return Err(Error::config(format!(
            "gateway peer endpoint {peer_endpoint} is plaintext but this gateway serves TLS; \
             set an https:// OPENSHELL_PEER_ENDPOINT so peer relay traffic is not downgraded"
        )));
    }
    warn!(
        peer_endpoint,
        "gateway peer relay traffic is plaintext because this gateway does not serve TLS"
    );
    Ok(())
}

/// Run the `OpenShell` server.
///
/// This starts a multiplexed gRPC/HTTP server on the configured bind address.
///
/// # Errors
///
/// Returns an error if the server fails to start or encounters a fatal error.
pub(crate) async fn run_server(
    startup: ServerStartupConfig,
    tracing_log_bus: TracingLogBus,
    compute_drivers: ComputeDriverRegistry,
) -> Result<()> {
    let ServerStartupConfig {
        config,
        config_file,
        guest_tls,
        compute_driver,
        legacy_compute_driver_env_seen: _,
    } = startup;
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    auth::descriptor_authz::init()
        .map_err(|error| Error::config(format!("invalid gRPC authorization metadata: {error}")))?;

    let database_url = config.database_url.trim();
    if database_url.is_empty() {
        return Err(Error::config("database_url is required"));
    }

    // Load signing material before connecting remote extensions so their
    // startup Describe calls can authenticate with gateway-caller tokens.
    let (extension_jwt_issuer, sandbox_session_jwt_authority) =
        if let Some(ref jwt) = config.gateway_jwt {
            let signing_pem = std::fs::read(&jwt.signing_key_path).map_err(|e| {
                Error::config(format!(
                    "failed to read sandbox JWT signing key from {}: {e}",
                    jwt.signing_key_path.display()
                ))
            })?;
            let public_pem = std::fs::read(&jwt.public_key_path).map_err(|e| {
                Error::config(format!(
                    "failed to read sandbox JWT public key from {}: {e}",
                    jwt.public_key_path.display()
                ))
            })?;
            let kid = std::fs::read_to_string(&jwt.kid_path)
                .map_err(|e| {
                    Error::config(format!(
                        "failed to read sandbox JWT kid from {}: {e}",
                        jwt.kid_path.display()
                    ))
                })?
                .trim()
                .to_string();
            if kid.is_empty() {
                return Err(Error::config(format!(
                    "sandbox JWT kid file {} is empty",
                    jwt.kid_path.display()
                )));
            }
            let issuer = Arc::new(
                auth::sandbox_jwt::ExtensionJwtIssuer::from_pem(
                    &signing_pem,
                    &public_pem,
                    kid.clone(),
                    &jwt.gateway_id,
                    jwt.token_ttl(),
                )
                .map_err(Error::config)?,
            );
            let session_authority = Arc::new(
                auth::sandbox_jwt::SandboxSessionJwtAuthority::from_pem(
                    &signing_pem,
                    &public_pem,
                    kid,
                    &jwt.gateway_id,
                    jwt.sandbox_token_ttl(),
                )
                .map_err(Error::config)?,
            );
            info!(
                gateway_id = %jwt.gateway_id,
                ttl_secs = jwt.ttl_secs.map(std::num::NonZeroU64::get),
                "gateway-minted sandbox JWT enabled"
            );
            (Some(issuer), Some(session_authority))
        } else {
            (None, None)
        };

    let middleware_registrations = config_file
        .as_ref()
        .map(|file| {
            file.openshell
                .supervisor
                .middleware
                .iter()
                .map(openshell_core::proto::SupervisorMiddlewareService::try_from)
                .collect::<std::result::Result<Vec<_>, _>>()
        })
        .transpose()
        .map_err(|error| Error::config(format!("middleware registration failed: {error}")))?
        .unwrap_or_default();
    let mut gateway_extension_credentials = Vec::new();
    let middleware_registry = Arc::new(
        if let Some(issuer) = extension_jwt_issuer.as_ref() {
            let mut slots = HashMap::new();
            for registration in &middleware_registrations {
                if let Some(credential) = mint_gateway_extension_credential(
                    issuer,
                    ExtensionKind::Middleware,
                    &registration.name,
                    &registration.audience,
                    &registration.grpc_endpoint,
                    registration.allow_insecure_transport,
                )? {
                    slots.insert(registration.name.clone(), credential.slot.clone());
                    gateway_extension_credentials.push(credential);
                }
            }
            MiddlewareRegistry::connect_services_authenticated(
                openshell_supervisor_middleware_builtins::services(),
                middleware_registrations,
                &slots,
            )
            .await
        } else {
            MiddlewareRegistry::connect_services(
                openshell_supervisor_middleware_builtins::services(),
                middleware_registrations,
            )
            .await
        }
        .map_err(|error| Error::config(format!("middleware registration failed: {error}")))?,
    );

    let store = Arc::new(Store::connect(database_url).await?);
    let credentials = credentials::CredentialRuntime::from_config_file_with_store(
        &config,
        config_file.as_ref(),
        Arc::clone(&store),
    )
    .await?;

    let oidc_cache = if let Some(ref oidc) = config.oidc {
        // Validate RBAC configuration before starting.
        let policy = auth::authz::AuthzPolicy {
            admin_role: oidc.admin_role.clone(),
            user_role: oidc.user_role.clone(),
            scopes_enabled: !oidc.scopes_claim.is_empty(),
        };
        policy.validate().map_err(Error::config)?;

        let cache = auth::oidc::JwksCache::new(oidc)
            .await
            .map_err(|e| Error::config(format!("OIDC initialization failed: {e}")))?;
        info!("OIDC JWT validation enabled (issuer: {})", oidc.issuer);
        Some(Arc::new(cache))
    } else {
        None
    };

    let sandbox_index = SandboxIndex::new();
    let sandbox_watch_bus = SandboxWatchBus::new();
    let supervisor_sessions = Arc::new(supervisor_session::SupervisorSessionRegistry::new());
    let driver_startup = compute::driver_config::DriverStartupContext {
        file: config_file.as_ref(),
        guest_tls: guest_tls.as_ref(),
        gateway_port: config.bind_address.port(),
        gateway_tls_enabled: config.tls.is_some(),
        endpoint_overrides: &config.compute_driver_endpoints,
    };
    let compute = build_compute_runtime(
        &compute_drivers,
        &compute_driver,
        &config,
        driver_startup,
        store.clone(),
        sandbox_index.clone(),
        sandbox_watch_bus.clone(),
        tracing_log_bus.clone(),
        supervisor_sessions.clone(),
        shutdown_rx.clone(),
    )
    .await?;
    let gateway_interceptors = if let Some(issuer) = extension_jwt_issuer.as_ref() {
        let mut slots = BTreeMap::new();
        for interceptor in &config.gateway_interceptors {
            let audience = interceptor.resolved_audience();
            if let Some(credential) = mint_gateway_extension_credential(
                issuer,
                ExtensionKind::Interceptor,
                &interceptor.name,
                audience.as_ref(),
                &interceptor.grpc_endpoint,
                interceptor.allow_insecure_transport,
            )? {
                slots.insert(interceptor.name.clone(), credential.slot.clone());
                gateway_extension_credentials.push(credential);
            }
        }
        openshell_gateway_interceptors::initialize_authenticated(
            config.gateway_interceptors.clone(),
            slots,
        )
        .await
    } else {
        openshell_gateway_interceptors::initialize(config.gateway_interceptors.clone()).await
    }
    .map_err(|e| Error::config(format!("gateway interceptor initialization failed: {e}")))?;
    let provider_profile_sources = provider_profile_sources::ProviderProfileSources::from_config(
        &config.provider_profile_sources,
        gateway_interceptors.as_ref(),
    )
    .map_err(|err| {
        Error::config(format!(
            "provider profile source configuration failed: {err}"
        ))
    })?;
    info!(
        sources = ?provider_profile_sources.source_ids(),
        "provider profile sources configured"
    );
    let mut state = ServerState::new_with_credentials(
        config.clone(),
        store.clone(),
        compute,
        sandbox_index,
        sandbox_watch_bus,
        tracing_log_bus,
        supervisor_sessions,
        oidc_cache,
        credentials,
    );
    state.middleware_registry = middleware_registry;
    state.gateway_interceptors = gateway_interceptors;
    state.provider_profile_sources = provider_profile_sources;
    state.extension_jwt_issuer = extension_jwt_issuer.clone();
    state.sandbox_session_jwt_authority = sandbox_session_jwt_authority;
    if let Some(issuer) = extension_jwt_issuer {
        spawn_gateway_extension_token_refresh(issuer, gateway_extension_credentials);
    }

    if state.sandbox_session_jwt_authority.is_some()
        && state.compute.supports_sandbox_authentication()
    {
        state.compute_driver_authenticator = Some(Arc::new(
            auth::compute_driver::ComputeDriverAuthenticator::new(state.compute.clone()),
        ));
        info!(
            driver = state.compute.configured_driver_name(),
            "compute-driver sandbox bootstrap authenticator enabled"
        );
    }

    let peer_routing_expected = state.peer_endpoint.is_some() && !state.store.is_single_replica();
    if let Some(peer_endpoint) = state.peer_endpoint.as_deref()
        && peer_routing_expected
    {
        validate_peer_endpoint_scheme(&state.config, peer_endpoint)?;
    }
    if state.peer_endpoint.is_none() && !state.store.is_single_replica() {
        warn!(
            "no gateway peer endpoint configured; this replica owns its supervisor sessions but \
             peers cannot reach it. Single-gateway deployments are unaffected; set \
             OPENSHELL_PEER_ENDPOINT on every replica when running more than one."
        );
    }

    if std::env::var_os("KUBERNETES_SERVICE_HOST").is_some() {
        let namespace = std::env::var("OPENSHELL_POD_NAMESPACE").ok();
        let service_account = std::env::var("OPENSHELL_SERVICE_ACCOUNT_NAME").ok();
        match (namespace, service_account) {
            (Some(namespace), Some(service_account))
                if !namespace.trim().is_empty() && !service_account.trim().is_empty() =>
            {
                let required_labels =
                    auth::peer::required_pod_labels_from_env().map_err(Error::config)?;
                match kube::Client::try_default().await {
                    Ok(client) => {
                        let audience = auth::peer::peer_token_audience_from_env();
                        let resolver = Arc::new(auth::peer::LiveGatewayPeerResolver::new(
                            client,
                            namespace.trim(),
                            audience.clone(),
                            service_account.trim().to_string(),
                            required_labels,
                        ));
                        let cache_ttl = auth::peer::peer_token_cache_ttl_from_env();
                        let resolver = Arc::new(auth::peer::CachingGatewayPeerResolver::new(
                            resolver, cache_ttl,
                        ));
                        let authenticator =
                            auth::peer::PeerServiceAccountAuthenticator::new(resolver);
                        state.peer_authenticator = Some(Arc::new(authenticator));
                        info!(
                            namespace = %namespace.trim(),
                            service_account = %service_account.trim(),
                            audience,
                            token_cache_ttl_secs = cache_ttl.as_secs(),
                            "gateway peer ServiceAccount TokenReview authentication enabled"
                        );
                    }
                    Err(err) if peer_routing_expected => {
                        return Err(Error::config(format!(
                            "in-cluster K8s client construction failed ({err}); \
                             gateway peer authentication is required because \
                             OPENSHELL_PEER_ENDPOINT is configured"
                        )));
                    }
                    Err(err) => warn!(
                        error = %err,
                        "in-cluster K8s client construction failed; \
                         gateway peer ServiceAccount authentication is disabled"
                    ),
                }
            }
            _ if peer_routing_expected => {
                return Err(Error::config(
                    "OPENSHELL_POD_NAMESPACE or OPENSHELL_SERVICE_ACCOUNT_NAME missing; \
                     both are required for gateway peer authentication because \
                     OPENSHELL_PEER_ENDPOINT is configured"
                        .to_string(),
                ));
            }
            _ => {
                debug!(
                    "OPENSHELL_POD_NAMESPACE or OPENSHELL_SERVICE_ACCOUNT_NAME missing; \
                     gateway peer ServiceAccount authentication disabled"
                );
            }
        }
    } else if peer_routing_expected {
        return Err(Error::config(
            "OPENSHELL_PEER_ENDPOINT is configured but the gateway is not running in a \
             Kubernetes cluster, so gateway peer authentication is unavailable"
                .to_string(),
        ));
    }

    let state = Arc::new(state);

    // Reconcile local-driver running intent before watchers spawn so their
    // first snapshots observe the post-start backend state. Explicitly stopped
    // sandboxes remain stopped.
    ensure_default_workspace(&store).await?;
    grpc::policy::validate_provider_composition_startup_preflight(&state)
        .await
        .map_err(|error| {
            Error::config(format!(
                "provider policy composition startup preflight failed: {}",
                error.message()
            ))
        })?;
    grpc::policy::invalidate_endpoint_status_on_startup(&state)
        .await
        .map_err(|error| {
            Error::execution(format!(
                "tool server endpoint-status startup reconciliation failed: {}",
                error.message()
            ))
        })?;

    let gateway_listener = bind_gateway_listener(config.bind_address).await?;

    // Create the multiplexed service
    let service = MultiplexService::new(state.clone());

    // Bind the unauthenticated health endpoint on a separate port when configured.
    if let Some(health_bind_address) = config.health_bind_address {
        let health_listener = TcpListener::bind(health_bind_address).await.map_err(|e| {
            Error::transport(format!(
                "failed to bind health port {health_bind_address}: {e}"
            ))
        })?;
        info!(address = %health_bind_address, "Health server listening");
        // `health_router` returns immediately; the listener serves
        // `Initializing → 503` until the background monitor publishes the
        // first real probe outcome, so the endpoint is always responsive.
        let router = health_router(store.clone());
        tokio::spawn(async move {
            if let Err(e) = axum::serve(health_listener, router.into_make_service()).await {
                error!("Health server error: {e}");
            }
        });
    } else {
        info!("Health server disabled");
    }

    // Bind the Prometheus metrics endpoint on a dedicated port when configured.
    if let Some(metrics_bind_address) = config.metrics_bind_address {
        let prometheus_handle = PrometheusBuilder::new()
            .install_recorder()
            .map_err(|e| Error::config(format!("failed to install metrics recorder: {e}")))?;
        let metrics_listener = TcpListener::bind(metrics_bind_address).await.map_err(|e| {
            Error::transport(format!(
                "failed to bind metrics port {metrics_bind_address}: {e}",
            ))
        })?;
        info!(address = %metrics_bind_address, "Metrics server listening");
        tokio::spawn(async move {
            if let Err(e) = axum::serve(
                metrics_listener,
                metrics_router(prometheus_handle).into_make_service(),
            )
            .await
            {
                error!("Metrics server error: {e}");
            }
        });
    } else {
        info!("Metrics server disabled");
    }

    // Build TLS acceptor when TLS is configured; otherwise serve plaintext.
    let tls_acceptor = if let Some(tls) = &config.tls {
        let acceptor = TlsAcceptor::from_files(
            &tls.cert_path,
            &tls.key_path,
            tls.client_ca_path.as_deref(),
            tls.require_client_auth,
            tls.external_cert_path.as_deref(),
            tls.external_key_path.as_deref(),
            tls.external_server_names.clone(),
        )?;

        // Spawn file-watcher-based TLS certificate reload worker.
        // Watches parent directories of cert/key/CA files and atomically
        // reloads when changes are detected.
        acceptor.spawn_reload_worker(shutdown_rx.clone());

        Some(acceptor)
    } else {
        info!("TLS disabled — accepting plaintext connections");
        None
    };

    let enable_loopback_service_http = config.service_routing.enable_loopback_service_http;
    let listener_task = tokio::spawn(serve_gateway_listener(
        gateway_listener,
        service.clone(),
        tls_acceptor.clone(),
        enable_loopback_service_http,
        shutdown_rx.clone(),
    ));

    // Deadlines must run while restored supervisors wait for policy repair.
    let (startup_tx, startup_rx) = watch::channel(false);
    state.compute.spawn_watchers(
        shutdown_rx.clone(),
        startup_rx,
        state.sandbox_session_jwt_authority.clone(),
    );

    // Serve the gateway before reconciling persisted sandboxes so restored
    // supervisors can fetch policy and register their sessions.
    if let Err(err) = state
        .compute
        .start_persisted_sandboxes_with_authentication(
            |sandbox| {
                let state = state.clone();
                let sandbox = sandbox.clone();
                async move {
                    if state.sandbox_session_jwt_authority.is_none() {
                        return Ok(Vec::new());
                    }
                    let authentication = grpc::mint_persisted_authentication(&state, &sandbox)
                        .map_err(|error| error.to_string())?;
                    serde_json::to_vec(&authentication)
                        .map_err(|error| format!("encode launch authentication: {error}"))
                }
            },
            |_| async { Ok(()) },
            |_| {},
        )
        .await
    {
        warn!(error = %err, "Failed to start persisted sandboxes during startup");
    }

    startup_tx.send_replace(true);
    // The poller exists to observe writes made by other replicas. Single-
    // replica backends have none, so it would only add load.
    if !store.is_single_replica() {
        sandbox_watch::spawn_store_poller(
            store.clone(),
            state.sandbox_watch_bus.clone(),
            sandbox_watch::DEFAULT_STORE_POLL_INTERVAL,
            shutdown_rx.clone(),
        );
    }
    ssh_sessions::spawn_session_reaper(store.clone(), Duration::from_hours(1));
    supervisor_session::spawn_relay_reaper(state.clone(), Duration::from_secs(30));
    provider_refresh::spawn_refresh_worker(state.clone(), Duration::from_mins(1));

    shutdown_signal().await;
    info!("Shutdown signal received; stopping gateway");
    state.gateway_shutting_down.store(true, Ordering::Release);
    state.supervisor_sessions.close_admission();
    let _ = shutdown_tx.send(true);

    if let Err(err) = listener_task.await {
        warn!(error = %err, "Gateway listener task failed during shutdown");
    }

    let compute_cleanup = state.compute.cleanup_on_shutdown().await;
    // A stopped supervisor may still have a detached task deleting its owner
    // record. Drain it even when compute cleanup failed before exiting Tokio.
    let session_cleanup = state
        .supervisor_sessions
        .shutdown(Duration::from_secs(10))
        .await;
    if let Err(err) = &session_cleanup {
        warn!(error = %err, "Gateway supervisor session cleanup incomplete");
    }
    compute_cleanup
        .map_err(|err| Error::execution(format!("gateway shutdown cleanup failed: {err}")))?;
    session_cleanup.map_err(Error::execution)?;

    Ok(())
}

async fn serve_gateway_listener(
    bound_listener: BoundGatewayListener,
    service: MultiplexService,
    tls_acceptor: Option<TlsAcceptor>,
    enable_loopback_service_http: bool,
    mut shutdown: watch::Receiver<bool>,
) {
    let BoundGatewayListener {
        listener,
        address: listen_addr,
    } = bound_listener;

    loop {
        let accepted = tokio::select! {
            changed = shutdown.changed() => {
                if changed.is_err() || *shutdown.borrow() {
                    break;
                }
                continue;
            }
            accepted = listener.accept() => accepted,
        };

        let (stream, addr) = match accepted {
            Ok(conn) => conn,
            Err(e) => {
                error!(error = %e, listen = %listen_addr, "Failed to accept connection");
                continue;
            }
        };
        set_tcp_nodelay_best_effort(&stream);

        spawn_gateway_connection(
            stream,
            addr,
            listen_addr,
            service.clone(),
            tls_acceptor.clone(),
            enable_loopback_service_http,
        );
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ConnectionProtocol {
    Tls,
    PlainHttp,
    Unknown,
}

async fn classify_connection_protocol(stream: &TcpStream) -> std::io::Result<ConnectionProtocol> {
    let mut prefix = [0_u8; 8];
    let read = stream.peek(&mut prefix).await?;
    Ok(classify_initial_bytes(&prefix[..read]))
}

fn classify_initial_bytes(prefix: &[u8]) -> ConnectionProtocol {
    if looks_like_tls(prefix) {
        ConnectionProtocol::Tls
    } else if looks_like_http(prefix) {
        ConnectionProtocol::PlainHttp
    } else {
        ConnectionProtocol::Unknown
    }
}

fn looks_like_tls(prefix: &[u8]) -> bool {
    prefix.len() >= 3 && prefix[0] == 0x16 && prefix[1] == 0x03
}

fn looks_like_http(prefix: &[u8]) -> bool {
    const METHODS: [&[u8]; 10] = [
        b"GET ",
        b"POST ",
        b"PUT ",
        b"PATCH ",
        b"DELETE ",
        b"HEAD ",
        b"OPTIONS ",
        b"TRACE ",
        b"CONNECT ",
        b"PRI ",
    ];

    if prefix.is_empty() {
        return false;
    }
    METHODS
        .iter()
        .any(|method| method.starts_with(prefix) || prefix.starts_with(method))
}

fn allow_plaintext_service_http(
    enabled: bool,
    listen_addr: SocketAddr,
    peer_addr: SocketAddr,
) -> bool {
    enabled && listen_addr.ip().is_loopback() && peer_addr.ip().is_loopback()
}

fn spawn_gateway_connection(
    stream: TcpStream,
    addr: SocketAddr,
    listen_addr: SocketAddr,
    service: MultiplexService,
    tls_acceptor: Option<TlsAcceptor>,
    enable_loopback_service_http: bool,
) {
    if let Some(acceptor) = tls_acceptor {
        tokio::spawn(async move {
            match classify_connection_protocol(&stream).await {
                Ok(ConnectionProtocol::PlainHttp)
                    if allow_plaintext_service_http(
                        enable_loopback_service_http,
                        listen_addr,
                        addr,
                    ) =>
                {
                    if let Err(e) = service.serve_service_http(stream).await {
                        if is_benign_connection_close(e.as_ref()) {
                            debug!(error = %e, client = %addr, listen = %listen_addr, "Plaintext service HTTP connection closed");
                        } else {
                            error!(error = %e, client = %addr, listen = %listen_addr, "Plaintext service HTTP connection error");
                        }
                    }
                }
                Ok(ConnectionProtocol::PlainHttp) => {
                    warn!(
                        client = %addr,
                        listen = %listen_addr,
                        "Rejected plaintext HTTP on gateway listener"
                    );
                }
                Ok(ConnectionProtocol::Tls | ConnectionProtocol::Unknown) => {
                    // acceptor.acceptor() snapshots the current TLS config;
                    // the returned acceptor owns an Arc that stays alive for
                    // the full duration of the handshake.
                    match acceptor.acceptor().accept(stream).await {
                        Ok(tls_stream) => {
                            let peer_identity = multiplex::extract_peer_identity(&tls_stream);
                            if let Err(e) = service
                                .serve_with_peer_identity(tls_stream, peer_identity)
                                .await
                            {
                                if is_benign_connection_close(e.as_ref()) {
                                    debug!(error = %e, client = %addr, "Connection closed");
                                } else {
                                    error!(error = %e, client = %addr, "Connection error");
                                }
                            }
                        }
                        Err(e) => {
                            if is_benign_tls_handshake_failure(&e) {
                                debug!(error = %e, client = %addr, "TLS handshake closed early");
                            } else {
                                error!(error = %e, client = %addr, "TLS handshake failed");
                            }
                        }
                    }
                }
                Err(e) => {
                    debug!(error = %e, client = %addr, "Failed to inspect connection preface");
                }
            }
        });
    } else {
        tokio::spawn(async move {
            if let Err(e) = service.serve(stream).await {
                if is_benign_connection_close(e.as_ref()) {
                    debug!(error = %e, client = %addr, "Connection closed");
                } else {
                    error!(error = %e, client = %addr, "Connection error");
                }
            }
        });
    }
}

async fn shutdown_signal() {
    #[cfg(unix)]
    {
        tokio::select! {
            () = ctrl_c_signal() => {}
            () = terminate_signal() => {}
        }
    }

    #[cfg(not(unix))]
    {
        ctrl_c_signal().await;
    }
}

async fn ctrl_c_signal() {
    if let Err(err) = tokio::signal::ctrl_c().await {
        warn!(error = %err, "Failed to install Ctrl-C signal handler");
        std::future::pending::<()>().await;
    }
}

#[cfg(unix)]
async fn terminate_signal() {
    let Ok(mut signal) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
    else {
        warn!("Failed to install SIGTERM signal handler");
        std::future::pending::<()>().await;
        return;
    };
    let _ = signal.recv().await;
}

pub use compute::{
    AcquiredRemoteDriverEndpoint, DriverWatchStream, ManagedDriverProcess, SharedComputeDriver,
};

/// Driver instance returned by a compiled compute-driver factory.
pub enum ComputeDriverInstance {
    /// A driver hosted in the gateway process.
    InProcess(SharedComputeDriver),
    /// A driver process launched and owned by the gateway.
    ManagedRemote(AcquiredRemoteDriverEndpoint),
}

/// Factory for a compute driver linked into a gateway binary.
#[async_trait::async_trait]
pub trait ComputeDriverFactory: Send + Sync {
    /// Validate selected-driver configuration without starting a driver,
    /// connecting a transport, or modifying runtime state.
    ///
    /// The default preserves source compatibility for existing out-of-tree
    /// factories during normal startup. Package preflight rejects a selected
    /// factory unless [`Self::supports_config_preflight`] is also overridden
    /// to return `true`.
    fn validate_config(&self, _context: ComputeDriverConfigContext<'_>) -> Result<()> {
        Ok(())
    }

    /// Whether [`Self::validate_config`] fully validates this factory's
    /// configuration without runtime side effects.
    fn supports_config_preflight(&self) -> bool {
        false
    }

    async fn build(&self, context: ComputeDriverBuildContext<'_>) -> Result<ComputeDriverInstance>;
}

/// One named compiled-driver registration.
#[derive(Clone)]
pub struct ComputeDriverRegistration {
    name: String,
    detection_priority: u16,
    detect: Option<fn() -> bool>,
    factory: Arc<dyn ComputeDriverFactory>,
    telemetry_category: TelemetryComputeDriver,
    in_process_tracing: Option<openshell_otel::ComputeDriverTracing>,
}

impl std::fmt::Debug for ComputeDriverRegistration {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ComputeDriverRegistration")
            .field("name", &self.name)
            .field("detection_priority", &self.detection_priority)
            .field("has_detection_probe", &self.detect.is_some())
            .finish_non_exhaustive()
    }
}

impl ComputeDriverRegistration {
    /// Define a compiled driver. Lower detection priorities are preferred.
    pub fn new(
        name: impl Into<String>,
        detection_priority: u16,
        detect: Option<fn() -> bool>,
        factory: impl ComputeDriverFactory + 'static,
    ) -> Result<Self> {
        let name = openshell_core::config::normalize_compute_driver_name(&name.into())
            .map_err(Error::config)?;
        Ok(Self {
            name,
            detection_priority,
            detect,
            factory: Arc::new(factory),
            telemetry_category: TelemetryComputeDriver::custom(),
            in_process_tracing: None,
        })
    }

    /// Compatibility no-op retained for source compatibility with schema-v1
    /// factory registrations. Schema v2 never inherits gateway keys into a
    /// driver table; move every driver setting under
    /// `[openshell.drivers.<name>]`.
    #[deprecated(
        since = "0.0.0",
        note = "schema v2 does not inherit gateway keys into driver configuration"
    )]
    #[must_use]
    pub fn with_inherited_config_keys(self, _keys: &'static [&'static str]) -> Self {
        self
    }

    /// Assign a bounded telemetry category chosen by the binary composition
    /// boundary. Runtime driver names are never used as telemetry values.
    #[must_use]
    pub fn with_telemetry_category(mut self, category: TelemetryComputeDriver) -> Self {
        self.telemetry_category = category;
        self
    }

    /// Compatibility no-op retained for existing factory registrations.
    /// Gateway mTLS user authentication is independent of compute drivers.
    #[must_use]
    pub fn with_local_singleplayer(self) -> Self {
        self
    }

    /// Attach process-wide tracing for this in-process compiled driver.
    #[must_use]
    pub fn with_in_process_tracing(
        mut self,
        tracing: openshell_otel::ComputeDriverTracing,
    ) -> Self {
        self.in_process_tracing = Some(tracing);
        self
    }

    #[must_use]
    pub fn in_process_tracing(&self) -> Option<openshell_otel::ComputeDriverTracing> {
        self.in_process_tracing
    }
}

/// Registry of compute drivers compiled into this gateway binary.
///
/// Like `SQLx`'s `Any` driver registry, installation is explicit at the binary
/// composition boundary while runtime selection is generic.
#[derive(Clone, Default)]
pub struct ComputeDriverRegistry {
    drivers: BTreeMap<String, ComputeDriverRegistration>,
}

#[derive(Clone, Debug)]
struct ComputeDriverDetection {
    available: Vec<String>,
}

impl ComputeDriverDetection {
    fn selected(&self) -> Option<&str> {
        self.available.first().map(String::as_str)
    }
}

#[derive(Clone, Debug)]
pub(crate) enum ComputeDriverSelection {
    Configured { name: String },
    AutoDetected(ComputeDriverDetection),
}

impl ComputeDriverSelection {
    pub(crate) fn name(&self) -> &str {
        match self {
            Self::Configured { name } => name,
            Self::AutoDetected(detection) => detection
                .selected()
                .expect("auto-detected selection has an available driver"),
        }
    }
}

impl ComputeDriverRegistry {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Install a compiled driver factory.
    pub fn install(&mut self, registration: ComputeDriverRegistration) -> Result<()> {
        let name = registration.name.clone();
        match self.drivers.entry(name.clone()) {
            std::collections::btree_map::Entry::Vacant(entry) => {
                entry.insert(registration);
                Ok(())
            }
            std::collections::btree_map::Entry::Occupied(_) => Err(Error::config(format!(
                "compute driver '{name}' registered twice"
            ))),
        }
    }

    /// Names installed into this gateway binary, in lexical order.
    pub fn installed_driver_names(&self) -> impl Iterator<Item = &str> {
        self.drivers.keys().map(String::as_str)
    }

    /// Names whose runtime registrations participate in auto-detection.
    ///
    /// This does not execute detection probes. Package preflight uses it to
    /// validate configured candidate tables without connecting local sockets
    /// or starting discovery commands.
    pub(crate) fn auto_detectable_driver_names(&self) -> impl Iterator<Item = &str> {
        self.drivers
            .values()
            .filter(|registration| registration.detect.is_some())
            .map(|registration| registration.name.as_str())
    }

    pub(crate) fn get(&self, name: &str) -> Option<&ComputeDriverRegistration> {
        self.drivers.get(name)
    }

    fn in_process_tracing(
        &self,
        selection: &ComputeDriverSelection,
        endpoint_overrides: &BTreeMap<String, PathBuf>,
    ) -> Option<openshell_otel::ComputeDriverTracing> {
        let name = selection.name();
        if endpoint_overrides.contains_key(name) {
            return None;
        }
        self.get(name)
            .and_then(ComputeDriverRegistration::in_process_tracing)
    }

    fn detect(&self) -> ComputeDriverDetection {
        let mut candidates = self
            .drivers
            .values()
            .filter(|registration| registration.detect.is_some())
            .collect::<Vec<_>>();
        candidates.sort_by(|left, right| {
            left.detection_priority
                .cmp(&right.detection_priority)
                .then_with(|| left.name.cmp(&right.name))
        });
        let available = candidates
            .into_iter()
            .filter(|registration| registration.detect.is_some_and(|detect| detect()))
            .map(|registration| registration.name.clone())
            .collect();
        ComputeDriverDetection { available }
    }

    pub(crate) fn select(&self, configured_driver: Option<&str>) -> Result<ComputeDriverSelection> {
        match configured_driver {
            None => {
                let detection = self.detect();
                if detection.selected().is_none() {
                    return Err(Error::config(
                        "no compute driver configured and auto-detection found no suitable installed \
                        driver; set --compute-driver <name> or OPENSHELL_COMPUTE_DRIVER=<name>",
                    ));
                }
                Ok(ComputeDriverSelection::AutoDetected(detection))
            }
            Some(driver) => {
                let name = openshell_core::config::normalize_compute_driver_name(driver)
                    .map_err(Error::config)?;
                Ok(ComputeDriverSelection::Configured { name })
            }
        }
    }
}

/// Read-only inputs available while validating a selected compute driver.
///
/// This context deliberately exposes no shutdown handle, runtime store, or
/// transport client. Implementations must remain deterministic and must not
/// start processes, connect sockets, or modify state.
#[derive(Clone, Copy)]
pub struct ComputeDriverConfigContext<'a> {
    driver_name: &'a str,
    gateway_name: &'a str,
    gateway_bind_address: SocketAddr,
    gateway_log_level: &'a str,
    driver_startup: compute::driver_config::DriverStartupContext<'a>,
}

impl ComputeDriverConfigContext<'_> {
    #[must_use]
    pub fn driver_name(&self) -> &str {
        self.driver_name
    }

    #[must_use]
    pub fn gateway_name(&self) -> &str {
        self.gateway_name
    }

    #[must_use]
    pub fn gateway_bind_address(&self) -> SocketAddr {
        self.gateway_bind_address
    }

    #[must_use]
    pub fn gateway_log_level(&self) -> &str {
        self.gateway_log_level
    }

    #[must_use]
    pub fn gateway_port(&self) -> u16 {
        self.driver_startup.gateway_port
    }

    #[must_use]
    pub fn gateway_tls_enabled(&self) -> bool {
        self.driver_startup.gateway_tls_enabled
    }

    /// Deserialize the selected driver's merged TOML table.
    pub fn driver_config<T>(&self) -> Result<T>
    where
        T: Default + serde::de::DeserializeOwned,
    {
        compute::driver_config::driver_config_from_context(self.driver_startup, self.driver_name)
    }
}

pub struct ComputeDriverBuildContext<'a> {
    config: ComputeDriverConfigContext<'a>,
    shutdown_rx: watch::Receiver<bool>,
}

impl ComputeDriverBuildContext<'_> {
    #[must_use]
    pub fn config_context(&self) -> ComputeDriverConfigContext<'_> {
        self.config
    }

    #[must_use]
    pub fn driver_name(&self) -> &str {
        self.config.driver_name()
    }

    #[must_use]
    pub fn gateway_name(&self) -> &str {
        self.config.gateway_name()
    }

    #[must_use]
    pub fn gateway_bind_address(&self) -> SocketAddr {
        self.config.gateway_bind_address()
    }

    #[must_use]
    pub fn gateway_log_level(&self) -> &str {
        self.config.gateway_log_level()
    }

    #[must_use]
    pub fn gateway_port(&self) -> u16 {
        self.config.gateway_port()
    }

    #[must_use]
    pub fn gateway_tls_enabled(&self) -> bool {
        self.config.gateway_tls_enabled()
    }

    /// Gateway CA certificate that a local driver may provide to supervisors.
    #[must_use]
    pub fn guest_tls_ca(&self) -> Option<&Path> {
        self.config
            .driver_startup
            .guest_tls
            .map(compute::driver_config::GuestTlsPaths::as_path)
    }

    /// Deserialize the selected driver's merged TOML table.
    pub fn driver_config<T>(&self) -> Result<T>
    where
        T: Default + serde::de::DeserializeOwned,
    {
        self.config.driver_config()
    }

    #[must_use]
    pub fn shutdown_receiver(&self) -> watch::Receiver<bool> {
        self.shutdown_rx.clone()
    }

    #[must_use]
    pub fn otlp_config(&self) -> Option<&config_file::OtlpConfig> {
        self.config
            .driver_startup
            .file
            .and_then(|file| file.openshell.gateway.otlp.as_ref())
    }
}

#[allow(clippy::too_many_arguments)]
async fn build_compute_runtime(
    registry: &ComputeDriverRegistry,
    selection: &ComputeDriverSelection,
    config: &Config,
    driver_startup: compute::driver_config::DriverStartupContext<'_>,
    store: Arc<Store>,
    sandbox_index: SandboxIndex,
    sandbox_watch_bus: SandboxWatchBus,
    tracing_log_bus: TracingLogBus,
    supervisor_sessions: Arc<supervisor_session::SupervisorSessionRegistry>,
    shutdown_rx: watch::Receiver<bool>,
) -> Result<ComputeRuntime> {
    let driver = validate_compute_driver_config(
        registry,
        selection.name(),
        &config.name,
        config.bind_address,
        &config.log_level,
        driver_startup,
        false,
    )?;
    let telemetry_compute_driver = driver.telemetry_compute_driver(registry);
    let admission =
        compute::driver_config::admission_config_from_context(driver_startup, driver.name())?;
    info!(driver = %driver.name(), "Using compute driver");
    let runtime = match driver {
        ConfiguredComputeDriver::Registered(registration) => {
            let build_context = ComputeDriverBuildContext {
                config: ComputeDriverConfigContext {
                    driver_name: &registration.name,
                    gateway_name: &config.name,
                    gateway_bind_address: config.bind_address,
                    gateway_log_level: &config.log_level,
                    driver_startup,
                },
                shutdown_rx,
            };
            let instance = registration.factory.build(build_context).await?;
            match instance {
                ComputeDriverInstance::InProcess(driver) => ComputeRuntime::from_driver(
                    registration.name,
                    driver,
                    None,
                    store,
                    sandbox_index,
                    sandbox_watch_bus,
                    tracing_log_bus,
                    supervisor_sessions,
                )
                .await
                .map_err(|error| {
                    Error::execution(format!("failed to create compute runtime: {error}"))
                })?,
                ComputeDriverInstance::ManagedRemote(mut endpoint) => {
                    endpoint.name = registration.name;
                    ComputeRuntime::new_remote_driver(
                        endpoint,
                        store,
                        sandbox_index,
                        sandbox_watch_bus,
                        tracing_log_bus,
                        supervisor_sessions,
                    )
                    .await
                    .map_err(|error| {
                        Error::execution(format!("failed to create compute runtime: {error}"))
                    })?
                }
            }
        }
        ConfiguredComputeDriver::Remote { name } => {
            let remote_config =
                compute::driver_config::remote_driver_config_from_context(driver_startup, &name)?;
            info!(
                driver = %name,
                socket = %remote_config.socket_path.display(),
                "Using remote compute driver endpoint"
            );
            let endpoint = compute::connect_remote_compute_driver(name, &remote_config.socket_path)
                .await
                .map_err(|e| Error::execution(format!("failed to create compute runtime: {e}")))?;
            ComputeRuntime::new_remote_driver(
                endpoint,
                store,
                sandbox_index,
                sandbox_watch_bus,
                tracing_log_bus,
                supervisor_sessions,
            )
            .await
            .map_err(|e| Error::execution(format!("failed to create compute runtime: {e}")))?
        }
    };

    let runtime = runtime
        .with_admission_policy(admission)
        .map_err(Error::config)?;
    Ok(runtime.with_telemetry_compute_driver(telemetry_compute_driver))
}

#[derive(Debug, Clone)]
enum ConfiguredComputeDriver {
    Registered(ComputeDriverRegistration),
    Remote { name: String },
}

impl ConfiguredComputeDriver {
    fn name(&self) -> &str {
        match self {
            Self::Registered(registration) => &registration.name,
            Self::Remote { name } => name,
        }
    }

    fn telemetry_compute_driver(&self, registry: &ComputeDriverRegistry) -> TelemetryComputeDriver {
        match self {
            Self::Registered(registration) => registration.telemetry_category,
            Self::Remote { name } => registry
                .get(name)
                .map_or_else(TelemetryComputeDriver::custom, |registration| {
                    registration.telemetry_category
                }),
        }
    }
}

#[cfg(test)]
fn configured_compute_driver(
    registry: &ComputeDriverRegistry,
    config: &Config,
    driver_startup: compute::driver_config::DriverStartupContext<'_>,
) -> Result<ConfiguredComputeDriver> {
    let selection = registry.select(config.compute_driver.as_deref())?;
    resolve_configured_compute_driver(registry, selection.name(), driver_startup)
}

#[allow(clippy::too_many_arguments)]
fn validate_compute_driver_config(
    registry: &ComputeDriverRegistry,
    driver_name: &str,
    gateway_name: &str,
    gateway_bind_address: SocketAddr,
    gateway_log_level: &str,
    driver_startup: compute::driver_config::DriverStartupContext<'_>,
    require_preflight_support: bool,
) -> Result<ConfiguredComputeDriver> {
    let driver = resolve_configured_compute_driver(registry, driver_name, driver_startup)?;
    match &driver {
        ConfiguredComputeDriver::Registered(registration) => {
            if require_preflight_support && !registration.factory.supports_config_preflight() {
                return Err(Error::config(format!(
                    "compute driver '{}' does not support side-effect-free configuration preflight",
                    registration.name
                )));
            }
            registration
                .factory
                .validate_config(ComputeDriverConfigContext {
                    driver_name: &registration.name,
                    gateway_name,
                    gateway_bind_address,
                    gateway_log_level,
                    driver_startup,
                })?;
        }
        ConfiguredComputeDriver::Remote { name } => {
            compute::driver_config::remote_driver_config_from_context(driver_startup, name)?;
        }
    }
    Ok(driver)
}

fn resolve_configured_compute_driver(
    registry: &ComputeDriverRegistry,
    driver_name: &str,
    driver_startup: compute::driver_config::DriverStartupContext<'_>,
) -> Result<ConfiguredComputeDriver> {
    let name = openshell_core::config::normalize_compute_driver_name(driver_name)
        .map_err(Error::config)?;
    // An operator-provided endpoint replaces normal construction for the
    // selected name, including a compiled registration with the same name.
    // The gateway connects to it; it does not provision the remote driver.
    if driver_startup.endpoint_overrides.contains_key(&name) {
        return Ok(ConfiguredComputeDriver::Remote { name });
    }

    if let Some(registration) = registry.get(&name) {
        return Ok(ConfiguredComputeDriver::Registered(registration.clone()));
    }

    Ok(ConfiguredComputeDriver::Remote { name })
}

pub(crate) async fn ensure_default_workspace(store: &Store) -> Result<()> {
    use grpc::workspace::{DEFAULT_WORKSPACE_NAME, WORKSPACE_OBJECT_TYPE};
    use openshell_core::proto::Workspace;
    use openshell_core::proto::datamodel::v1::ObjectMeta;
    use prost::Message;

    let id = uuid::Uuid::new_v4().to_string();
    let workspace = Workspace {
        metadata: Some(ObjectMeta {
            id: id.clone(),
            name: DEFAULT_WORKSPACE_NAME.to_string(),
            created_time: openshell_core::time::timestamp_from_millis(
                persistence::current_time_ms(),
            )
            .ok(),
            labels: HashMap::new(),
            annotations: HashMap::new(),
            resource_version: 0,
            workspace: String::new(),
            deletion_time: None,
        }),
        status: Some(openshell_core::proto::datamodel::v1::WorkspaceStatus {
            phase: openshell_core::proto::datamodel::v1::WorkspacePhase::Active.into(),
        }),
    };

    let labels_map = workspace.object_labels();
    let labels_json = if labels_map.as_ref().is_none_or(HashMap::is_empty) {
        None
    } else {
        Some(
            serde_json::to_string(&labels_map).map_err(|e| Error::Config {
                message: format!("failed to serialize labels: {e}"),
            })?,
        )
    };
    match store
        .put_if(
            WORKSPACE_OBJECT_TYPE,
            &id,
            DEFAULT_WORKSPACE_NAME,
            "",
            &workspace.encode_to_vec(),
            labels_json.as_deref(),
            persistence::WriteCondition::MustCreate,
        )
        .await
    {
        Ok(_) => {
            info!("Created default workspace");
            Ok(())
        }
        Err(persistence::PersistenceError::UniqueViolation { .. }) => {
            debug!("Default workspace already exists");
            Ok(())
        }
        Err(e) => Err(Error::config(format!(
            "failed to ensure default workspace: {e}"
        ))),
    }
}

#[cfg(test)]
mod tests {
    use super::{
        BoundGatewayListener, ConfiguredComputeDriver, ConnectionProtocol, ExtensionKind,
        MultiplexService, ServerState, TlsAcceptor, allow_plaintext_service_http,
        bind_gateway_listener, classify_initial_bytes, configured_compute_driver,
        extension_token_ttl, is_benign_tls_handshake_failure, mint_gateway_extension_credential,
        serve_gateway_listener, validate_peer_endpoint_scheme,
    };
    use openshell_core::{
        Config,
        proto::{HealthRequest, open_shell_client::OpenShellClient},
    };
    use std::io::{Error, ErrorKind};
    use std::net::SocketAddr;
    use std::sync::{
        Arc, LazyLock, Mutex,
        atomic::{AtomicBool, Ordering},
    };
    use std::time::Duration;
    use tempfile::{TempDir, tempdir};
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::{TcpListener, TcpStream};
    use tokio::sync::watch;

    use crate::tls_test_utils::generate_test_certs_with_ca;
    use axum::body::Body;
    use http::{Request, StatusCode};
    use tower::ServiceExt;

    fn tls_enabled_config() -> Config {
        Config::new(Some(openshell_core::TlsConfig {
            cert_path: "/tmp/cert.pem".into(),
            key_path: "/tmp/key.pem".into(),
            client_ca_path: None,
            require_client_auth: false,
            external_cert_path: None,
            external_key_path: None,
            external_server_names: Vec::new(),
        }))
    }

    #[test]
    fn plaintext_peer_endpoint_is_rejected_on_a_tls_gateway() {
        let error = validate_peer_endpoint_scheme(&tls_enabled_config(), "http://10.0.0.1:8080")
            .expect_err("plaintext peer endpoint must not be accepted alongside gateway TLS");
        assert!(
            error.to_string().contains("plaintext"),
            "error should name the downgrade: {error}"
        );
    }

    #[test]
    fn https_peer_endpoint_is_accepted_on_a_tls_gateway() {
        validate_peer_endpoint_scheme(&tls_enabled_config(), "https://10.0.0.1:8080").unwrap();
    }

    #[test]
    fn plaintext_peer_endpoint_is_allowed_on_a_plaintext_gateway() {
        let config = Config::new(None);
        assert!(config.tls.is_none());
        validate_peer_endpoint_scheme(&config, "http://10.0.0.1:8080").unwrap();
    }

    static DETECTION_PROBE_ORDER: LazyLock<Mutex<Vec<&'static str>>> =
        LazyLock::new(|| Mutex::new(Vec::new()));

    fn record_detection_probe(name: &'static str, available: bool) -> bool {
        DETECTION_PROBE_ORDER.lock().unwrap().push(name);
        available
    }

    fn unavailable_first_probe() -> bool {
        record_detection_probe("first", false)
    }

    fn available_second_probe() -> bool {
        record_detection_probe("second", true)
    }

    fn available_third_probe() -> bool {
        record_detection_probe("third", true)
    }

    fn extension_test_issuer() -> Arc<crate::auth::sandbox_jwt::ExtensionJwtIssuer> {
        extension_test_issuer_with_ttl(Duration::from_mins(15))
    }

    fn extension_test_issuer_with_ttl(
        ttl: Duration,
    ) -> Arc<crate::auth::sandbox_jwt::ExtensionJwtIssuer> {
        let material = openshell_bootstrap::jwt::generate_jwt_key().expect("jwt key");
        Arc::new(
            crate::auth::sandbox_jwt::ExtensionJwtIssuer::from_pem(
                material.signing_key_pem.as_bytes(),
                material.public_key_pem.as_bytes(),
                material.kid,
                "gateway-a",
                ttl,
            )
            .expect("issuer"),
        )
    }

    #[test]
    fn gateway_ttl_bounds_extension_ttl() {
        let issuer = extension_test_issuer_with_ttl(Duration::from_mins(15));
        assert_eq!(extension_token_ttl(&issuer), Duration::from_mins(15));
    }

    #[test]
    fn extension_token_ttl_is_capped_at_one_hour() {
        let issuer = extension_test_issuer_with_ttl(Duration::from_hours(24));
        assert_eq!(extension_token_ttl(&issuer), Duration::from_hours(1));

        let short = extension_test_issuer_with_ttl(Duration::from_mins(5));
        assert_eq!(extension_token_ttl(&short), Duration::from_mins(5));
    }

    #[test]
    fn plaintext_extension_endpoint_is_rejected_unless_explicitly_opted_out() {
        let issuer = extension_test_issuer();

        // Default posture: a plaintext endpoint cannot carry a bearer
        // credential, so startup fails and names the opt-out.
        let error = mint_gateway_extension_credential(
            &issuer,
            ExtensionKind::Middleware,
            "content-guard",
            "urn:openshell:extension:middleware:content-guard",
            "http://host.openshell.internal:50051",
            false,
        )
        .expect_err("plaintext endpoint must not silently downgrade");
        assert!(error.to_string().contains("allow_insecure_transport"));

        // Explicit opt-out starts the gateway with no credential attached.
        assert!(
            mint_gateway_extension_credential(
                &issuer,
                ExtensionKind::Middleware,
                "content-guard",
                "urn:openshell:extension:middleware:content-guard",
                "http://host.openshell.internal:50051",
                true,
            )
            .expect("opt-out must be permitted")
            .is_none()
        );
    }

    #[test]
    fn authenticated_extension_endpoints_mint_a_credential() {
        let issuer = extension_test_issuer();
        let credential = mint_gateway_extension_credential(
            &issuer,
            ExtensionKind::Middleware,
            "content-guard",
            "urn:openshell:extension:middleware:content-guard",
            "https://content-guard.example:50051",
            false,
        )
        .expect("credential")
        .expect("authenticated endpoint mints a credential");
        assert_eq!(credential.name, "content-guard");
        assert!(credential.slot.expires_at_ms().is_some_and(|ms| ms > 0));

        // Unix sockets are gateway-local, so only interceptors can use them.
        // A middleware endpoint must also be reachable from every supervisor.
        let error = mint_gateway_extension_credential(
            &issuer,
            ExtensionKind::Middleware,
            "content-guard",
            "urn:openshell:extension:middleware:content-guard",
            "unix:///run/openshell/content-guard.sock",
            false,
        )
        .expect_err("middleware cannot be reached over a gateway-local socket");
        assert!(error.to_string().contains("must use https://"));

        assert!(
            mint_gateway_extension_credential(
                &issuer,
                ExtensionKind::Interceptor,
                "quota",
                "urn:openshell:extension:interceptor:quota",
                "unix:///run/openshell/interceptors/quota.sock",
                false,
            )
            .expect("credential")
            .is_some()
        );
    }

    fn test_driver_startup<'a>(
        config: &'a Config,
        file: Option<&'a super::config_file::ConfigFile>,
    ) -> crate::compute::driver_config::DriverStartupContext<'a> {
        crate::compute::driver_config::DriverStartupContext {
            file,
            guest_tls: None,
            gateway_port: openshell_core::config::DEFAULT_SERVER_PORT,
            gateway_tls_enabled: false,
            endpoint_overrides: &config.compute_driver_endpoints,
        }
    }

    fn test_compute_drivers() -> super::ComputeDriverRegistry {
        let mut registry = super::ComputeDriverRegistry::new();
        for (name, priority) in [("alpha", 100), ("beta", 200), ("gamma", 300)] {
            registry
                .install(
                    super::ComputeDriverRegistration::new(
                        name,
                        priority,
                        None,
                        TestComputeDriverFactory,
                    )
                    .unwrap()
                    .with_telemetry_category(
                        openshell_core::telemetry::TelemetryComputeDriver::anonymous_category(
                            "registered",
                        ),
                    ),
                )
                .unwrap();
        }
        registry
    }

    #[derive(Clone, Copy)]
    struct TestComputeDriverFactory;

    // Omitting validate_config exercises source compatibility for out-of-tree
    // factories written before package preflight introduced that hook.
    #[async_trait::async_trait]
    impl super::ComputeDriverFactory for TestComputeDriverFactory {
        async fn build(
            &self,
            _context: super::ComputeDriverBuildContext<'_>,
        ) -> openshell_core::Result<super::ComputeDriverInstance> {
            unreachable!("selection tests do not construct the driver")
        }
    }

    fn test_tls_acceptor() -> (TempDir, TlsAcceptor) {
        let dir = tempdir().expect("failed to create tempdir");
        generate_test_certs_with_ca(dir.path());

        let acceptor = TlsAcceptor::from_files(
            &dir.path().join("server-cert.pem"),
            &dir.path().join("server-key.pem"),
            Some(&dir.path().join("ca.pem")),
            false,
            None,
            None,
            Vec::new(),
        )
        .expect("failed to build tls acceptor");

        (dir, acceptor)
    }

    async fn test_state(
        bind_addr: SocketAddr,
        enable_loopback_service_http: bool,
    ) -> Arc<ServerState> {
        let store = Arc::new(
            crate::persistence::Store::connect("sqlite::memory:?cache=shared")
                .await
                .expect("failed to create test store"),
        );
        let compute = crate::compute::new_test_runtime(store.clone()).await;
        Arc::new(ServerState::new(
            Config::new(None)
                .with_database_url("sqlite::memory:?cache=shared")
                .with_bind_address(bind_addr)
                .with_server_sans(["*.dev.openshell.localhost"])
                .with_loopback_service_http(enable_loopback_service_http)
                .with_credential_drivers(["test-static"]),
            store,
            compute,
            crate::sandbox_index::SandboxIndex::new(),
            crate::sandbox_watch::SandboxWatchBus::new(),
            crate::tracing_bus::TracingLogBus::new(),
            Arc::new(crate::supervisor_session::SupervisorSessionRegistry::new()),
            None,
        ))
    }

    #[tokio::test]
    async fn websocket_tunnel_is_mounted_only_when_enabled() {
        let state = test_state("127.0.0.1:17670".parse().unwrap(), true).await;
        let response = super::http_router(state.clone())
            .oneshot(Request::get("/_ws_tunnel").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::NOT_FOUND);

        let mut enabled = state;
        Arc::get_mut(&mut enabled)
            .unwrap()
            .config
            .enable_websocket_tunnel = true;
        let response = super::http_router(enabled)
            .oneshot(Request::get("/_ws_tunnel").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_ne!(response.status(), StatusCode::NOT_FOUND);
    }

    async fn start_tls_gateway_listener(
        bind_addr: &str,
        enable_loopback_service_http: bool,
    ) -> (
        SocketAddr,
        watch::Sender<bool>,
        tokio::task::JoinHandle<()>,
        TempDir,
    ) {
        let listener = TcpListener::bind(bind_addr)
            .await
            .expect("failed to bind test listener");
        let listen_addr = listener.local_addr().expect("failed to read local addr");
        let state = test_state(listen_addr, enable_loopback_service_http).await;
        let service = MultiplexService::new(state);
        let (tls_dir, tls_acceptor) = test_tls_acceptor();
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let handle = tokio::spawn(serve_gateway_listener(
            BoundGatewayListener {
                listener,
                address: listen_addr,
            },
            service,
            Some(tls_acceptor),
            enable_loopback_service_http,
            shutdown_rx,
        ));
        (listen_addr, shutdown_tx, handle, tls_dir)
    }

    async fn send_plain_http(addr: SocketAddr, request: String) -> String {
        let connect_addr: SocketAddr = format!("127.0.0.1:{}", addr.port())
            .parse()
            .expect("failed to build loopback connect addr");
        let mut stream = TcpStream::connect(connect_addr)
            .await
            .expect("failed to connect to test listener");
        stream
            .write_all(request.as_bytes())
            .await
            .expect("failed to write request");

        let mut response = Vec::new();
        let read_result =
            tokio::time::timeout(Duration::from_secs(2), stream.read_to_end(&mut response))
                .await
                .expect("timed out reading response");
        if let Err(err) = read_result
            && err.kind() != ErrorKind::ConnectionReset
        {
            panic!("failed to read response: {err}");
        }
        String::from_utf8_lossy(&response).into_owned()
    }

    fn service_request(addr: SocketAddr, extra_headers: &[(&str, &str)]) -> String {
        let mut request = format!(
            "GET / HTTP/1.1\r\nHost: default--my-sandbox--web.dev.openshell.localhost:{}\r\nConnection: close\r\n",
            addr.port()
        );
        for (name, value) in extra_headers {
            request.push_str(name);
            request.push_str(": ");
            request.push_str(value);
            request.push_str("\r\n");
        }
        request.push_str("\r\n");
        request
    }

    async fn stop_listener(shutdown: watch::Sender<bool>, handle: tokio::task::JoinHandle<()>) {
        let _ = shutdown.send(true);
        let _ = tokio::time::timeout(Duration::from_secs(2), handle).await;
    }

    #[test]
    fn classifies_probe_style_tls_disconnects_as_benign() {
        for kind in [ErrorKind::UnexpectedEof, ErrorKind::ConnectionReset] {
            let error = Error::new(kind, "probe disconnected");
            assert!(is_benign_tls_handshake_failure(&error));
        }
    }

    #[test]
    fn preserves_real_tls_failures_as_errors() {
        for kind in [
            ErrorKind::InvalidData,
            ErrorKind::PermissionDenied,
            ErrorKind::Other,
        ] {
            let error = Error::new(kind, "real tls failure");
            assert!(!is_benign_tls_handshake_failure(&error));
        }
    }

    #[test]
    fn classifies_tls_and_plain_http_prefaces() {
        assert_eq!(
            classify_initial_bytes(&[0x16, 0x03, 0x01, 0x00]),
            ConnectionProtocol::Tls
        );
        assert_eq!(
            classify_initial_bytes(b"GET / HTTP/1.1\r\n"),
            ConnectionProtocol::PlainHttp
        );
        assert_eq!(classify_initial_bytes(b"G"), ConnectionProtocol::PlainHttp);
        assert_eq!(
            classify_initial_bytes(b"\x00\x01\x02"),
            ConnectionProtocol::Unknown
        );
    }

    #[test]
    fn plaintext_service_http_requires_loopback_listener_and_peer() {
        let loopback: SocketAddr = "127.0.0.1:8080".parse().unwrap();
        let peer: SocketAddr = "127.0.0.1:54000".parse().unwrap();
        let wildcard: SocketAddr = "0.0.0.0:8080".parse().unwrap();
        let remote_peer: SocketAddr = "192.0.2.10:54000".parse().unwrap();
        assert!(allow_plaintext_service_http(true, loopback, peer));
        assert!(!allow_plaintext_service_http(false, loopback, peer));
        assert!(!allow_plaintext_service_http(true, wildcard, peer));
        assert!(!allow_plaintext_service_http(true, loopback, remote_peer));
    }

    #[tokio::test]
    async fn plaintext_service_http_listener_rejects_non_loopback_bind() {
        let (addr, shutdown, handle, _tls_dir) =
            start_tls_gateway_listener("0.0.0.0:0", true).await;

        let response = send_plain_http(addr, service_request(addr, &[])).await;

        assert!(
            response.is_empty(),
            "non-loopback gateway listener should drop plaintext service HTTP, got: {response:?}"
        );
        stop_listener(shutdown, handle).await;
    }

    #[tokio::test]
    async fn plaintext_service_http_rejects_cross_origin_browser_contexts() {
        let (addr, shutdown, handle, _tls_dir) =
            start_tls_gateway_listener("127.0.0.1:0", true).await;
        let cases = [
            (
                "cross-site fetch metadata",
                vec![("Sec-Fetch-Site", "cross-site")],
            ),
            (
                "same-site sibling fetch metadata",
                vec![("Sec-Fetch-Site", "same-site")],
            ),
            (
                "mismatched origin",
                vec![(
                    "Origin",
                    "http://other-sandbox--web.dev.openshell.localhost:8080",
                )],
            ),
            (
                "mismatched referer",
                vec![(
                    "Referer",
                    "http://other-sandbox--web.dev.openshell.localhost:8080/page",
                )],
            ),
        ];

        for (name, headers) in cases {
            let response = send_plain_http(addr, service_request(addr, &headers)).await;

            assert!(
                response.starts_with("HTTP/1.1 403 Forbidden"),
                "{name} should be rejected before service lookup, got: {response:?}"
            );
            assert!(
                response.contains("Cross-origin service request rejected"),
                "{name} should explain the service rejection, got: {response:?}"
            );
        }
        stop_listener(shutdown, handle).await;
    }

    #[tokio::test]
    async fn plaintext_service_http_allows_same_origin_browser_context_to_reach_service_lookup() {
        let (addr, shutdown, handle, _tls_dir) =
            start_tls_gateway_listener("127.0.0.1:0", true).await;
        let origin = format!(
            "http://default--my-sandbox--web.dev.openshell.localhost:{}",
            addr.port()
        );
        let response = send_plain_http(
            addr,
            service_request(
                addr,
                &[("Sec-Fetch-Site", "same-origin"), ("Origin", &origin)],
            ),
        )
        .await;

        assert!(
            response.starts_with("HTTP/1.1 404 Not Found"),
            "same-origin browser context should pass CSRF guard and miss only because no endpoint exists, got: {response:?}"
        );
        assert!(
            !response.contains("Cross-origin service request rejected"),
            "same-origin browser context should not be rejected as cross-origin, got: {response:?}"
        );
        stop_listener(shutdown, handle).await;
    }

    #[tokio::test]
    async fn plaintext_service_http_does_not_expose_grpc_gateway() {
        let (addr, shutdown, handle, _tls_dir) =
            start_tls_gateway_listener("127.0.0.1:0", true).await;
        let grpc_endpoint = format!("http://127.0.0.1:{}", addr.port());
        let grpc_succeeded = tokio::time::timeout(Duration::from_secs(2), async {
            match OpenShellClient::connect(grpc_endpoint).await {
                Ok(mut client) => client.health(HealthRequest {}).await.is_ok(),
                Err(_) => false,
            }
        })
        .await
        .expect("timed out checking plaintext gRPC exposure");

        assert!(
            !grpc_succeeded,
            "plaintext service HTTP must not expose successful gateway gRPC"
        );

        let request = format!(
            "POST /openshell.v1.OpenShell/Health HTTP/1.1\r\nHost: 127.0.0.1:{}\r\nContent-Type: application/grpc\r\nTE: trailers\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
            addr.port()
        );

        let response = send_plain_http(addr, request).await;

        assert!(
            response.starts_with("HTTP/1.1 404 Not Found"),
            "plaintext service HTTP router should not serve gateway gRPC, got: {response:?}"
        );
        assert!(
            !response.contains("grpc-status: 0"),
            "plaintext service HTTP must not return a successful gRPC response: {response:?}"
        );
        stop_listener(shutdown, handle).await;
    }

    #[test]
    fn configured_compute_driver_triggers_auto_detection_when_empty() {
        fn available() -> bool {
            true
        }

        let mut registry = super::ComputeDriverRegistry::new();
        registry
            .install(
                super::ComputeDriverRegistration::new(
                    "detected",
                    100,
                    Some(available),
                    TestComputeDriverFactory,
                )
                .unwrap(),
            )
            .unwrap();
        let config = Config::new(None);
        let result =
            configured_compute_driver(&registry, &config, test_driver_startup(&config, None))
                .unwrap();

        let ConfiguredComputeDriver::Registered(registration) = result else {
            panic!("auto-detection must select a registered driver");
        };
        assert_eq!(registration.name, "detected");
    }

    #[test]
    fn registry_detection_reports_available_drivers_in_priority_order() {
        DETECTION_PROBE_ORDER.lock().unwrap().clear();
        let mut registry = super::ComputeDriverRegistry::new();
        registry
            .install(
                super::ComputeDriverRegistration::new(
                    "third",
                    300,
                    Some(available_third_probe),
                    TestComputeDriverFactory,
                )
                .unwrap(),
            )
            .unwrap();
        registry
            .install(
                super::ComputeDriverRegistration::new(
                    "first",
                    100,
                    Some(unavailable_first_probe),
                    TestComputeDriverFactory,
                )
                .unwrap(),
            )
            .unwrap();
        registry
            .install(
                super::ComputeDriverRegistration::new(
                    "second",
                    200,
                    Some(available_second_probe),
                    TestComputeDriverFactory,
                )
                .unwrap(),
            )
            .unwrap();

        let detection = registry.detect();
        assert_eq!(detection.selected(), Some("second"));
        assert_eq!(
            detection
                .available
                .iter()
                .map(String::as_str)
                .collect::<Vec<_>>(),
            vec!["second", "third"]
        );
        assert_eq!(
            DETECTION_PROBE_ORDER.lock().unwrap().as_slice(),
            ["first", "second", "third"]
        );
        assert_eq!(
            registry.installed_driver_names().collect::<Vec<_>>(),
            vec!["first", "second", "third"]
        );
    }

    #[test]
    fn configured_compute_driver_accepts_registered_name() {
        let config = Config::new(None).with_compute_driver("beta");
        let registry = test_compute_drivers();
        let driver =
            configured_compute_driver(&registry, &config, test_driver_startup(&config, None))
                .unwrap();
        assert_eq!(
            driver.telemetry_compute_driver(&registry).as_str(),
            "registered"
        );
        assert!(matches!(
            driver,
            ConfiguredComputeDriver::Registered(registration) if registration.name == "beta"
        ));
    }

    #[test]
    fn configured_compute_driver_resolves_named_remote() {
        let config = Config::new(None).with_compute_driver("kyma");
        let registry = test_compute_drivers();

        let driver =
            configured_compute_driver(&registry, &config, test_driver_startup(&config, None))
                .unwrap();
        assert_eq!(
            driver.telemetry_compute_driver(&registry).as_str(),
            "custom"
        );

        match driver {
            ConfiguredComputeDriver::Remote { name } => {
                assert_eq!(name, "kyma");
            }
            ConfiguredComputeDriver::Registered(other) => {
                panic!(
                    "expected remote driver, got registered driver {}",
                    other.name
                )
            }
        }
    }

    #[test]
    fn configured_compute_driver_uses_endpoint_override() {
        let config = Config::new(None)
            .with_compute_driver("alpha")
            .with_compute_driver_endpoint("alpha", "/run/openshell/alpha.sock");
        let registry = test_compute_drivers();

        let driver =
            configured_compute_driver(&registry, &config, test_driver_startup(&config, None))
                .unwrap();
        assert_eq!(
            driver.telemetry_compute_driver(&registry).as_str(),
            "registered"
        );
        assert!(matches!(
            driver,
            ConfiguredComputeDriver::Remote { name } if name == "alpha"
        ));
    }

    #[test]
    fn configured_compute_driver_uses_builtin_endpoint_override() {
        let config = Config::new(None)
            .with_compute_driver("beta")
            .with_compute_driver_endpoint("beta", "/run/openshell/beta.sock");

        let driver = configured_compute_driver(
            &test_compute_drivers(),
            &config,
            test_driver_startup(&config, None),
        )
        .unwrap();
        assert!(matches!(
            driver,
            ConfiguredComputeDriver::Remote { name } if name == "beta"
        ));
    }

    #[tokio::test]
    async fn failed_gateway_listener_bind_does_not_attempt_persisted_sandbox_start() {
        let occupied_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let occupied_address = occupied_listener.local_addr().unwrap();
        let start_attempted = AtomicBool::new(false);
        let result: openshell_core::Result<()> = async {
            let _listener = bind_gateway_listener(occupied_address).await?;
            start_attempted.store(true, Ordering::SeqCst);
            Ok(())
        }
        .await;

        assert!(
            result.is_err(),
            "binding the occupied gateway address should fail"
        );
        assert!(
            !start_attempted.load(Ordering::SeqCst),
            "persisted sandbox start must not run before the gateway listener is bound"
        );
    }
}