camel-component-grpc 0.33.0

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex, OnceLock};
use std::task::{Context, Poll};
use std::time::Duration;

use arc_swap::ArcSwap;
use camel_api::CamelError;
use camel_api::backoff::{BackoffConfig, BackoffState};
use camel_component_api::tls_source::ServerTlsSource;
use futures::StreamExt;
use hyper::server::conn::http2;
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use tokio::sync::{OnceCell, RwLock, mpsc};
use tokio_rustls::TlsAcceptor;
use tonic::body::Body as TonicBody;
use tonic::codec::Streaming;
use tonic::{Request, Response, Status};
use tower::Service;
use tracing::{debug, error};

use camel_api::security_policy::{AccessMode, AuthPrincipal, RouteSecurityPlan};
use camel_auth::CredentialSource;
use camel_auth::{AuthenticatedPrincipal, ProviderRegistry};
use camel_component_api::{RuntimeObservability, SecurityContext};

use crate::codec::RawBytesCodec;
use crate::config::{GrpcServerConfig, ServerTransport};
use crate::consumer::{GrpcReply, GrpcRequestEnvelope, GrpcStreamItem};
use crate::mode::GrpcMode;

pub(crate) type GrpcDispatchEntry = (
    mpsc::Sender<GrpcRequestEnvelope>,
    GrpcMode,
    Option<Arc<dyn camel_auth::TokenAuthenticator>>,
    Vec<CredentialSource>,
    Option<Arc<GrpcKernelAuth>>,
);

pub(crate) type GrpcDispatchTable = Arc<RwLock<HashMap<String, GrpcDispatchEntry>>>;

/// Kernel authentication state captured when a route's dispatch entry is
/// constructed (`unify-transport-auth`, Task 2.1).
///
/// Construction-order lifecycle: the compiled plan and the provider registry
/// arrive from the route's [`SecurityContext`] when the entry is inserted
/// into the dispatch table — before any request reaches the per-request
/// handlers. The interceptor is created with the plan already captured;
/// it never reads a post-hoc setter. A context lacking either piece keeps
/// the legacy authenticator path (which still demands credentials).
pub(crate) struct GrpcKernelAuth {
    pub(crate) plan: RouteSecurityPlan,
    pub(crate) providers: Arc<ProviderRegistry>,
}

impl GrpcKernelAuth {
    /// Capture the kernel state from a route's security context.
    ///
    /// `None` unless both the compiled plan and the provider registry are
    /// present: a plan without providers can never mint a principal, and a
    /// registry without a plan has nothing to enforce (fail-closed to the
    /// legacy path rather than a silently unauthenticated route).
    pub(crate) fn from_security_context(ctx: &SecurityContext) -> Option<Self> {
        Some(Self {
            plan: ctx.plan.clone()?,
            providers: ctx.providers.clone()?,
        })
    }
}

type ServerKey = (String, u16);

/// Shared, atomically-swappable TLS acceptor. `None` for plaintext.
/// Wrapping in ArcSwap lets the cert be hot-reloaded at runtime by calling
/// `.store()` on the inner ArcSwap (see Task 5 reload handler).
type SharedTlsAcceptor = Option<Arc<ArcSwap<TlsAcceptor>>>;

struct ServerHandle {
    dispatch: GrpcDispatchTable,
    task: tokio::task::JoinHandle<()>,
    transport: ServerTransport,
    tls_acceptor: SharedTlsAcceptor,
    tls_source: Option<ServerTlsSource>,
}

pub(crate) struct GrpcServerRegistry {
    inner: Mutex<HashMap<ServerKey, Arc<OnceCell<ServerHandle>>>>,
}

impl GrpcServerRegistry {
    pub(crate) fn global() -> &'static Self {
        static INSTANCE: OnceLock<GrpcServerRegistry> = OnceLock::new();
        INSTANCE.get_or_init(|| GrpcServerRegistry {
            inner: Mutex::new(HashMap::new()),
        })
    }

    pub(crate) async fn get_or_spawn(
        &'static self,
        host: &str,
        port: u16,
        config: GrpcServerConfig,
        runtime: Arc<dyn RuntimeObservability>,
    ) -> Result<GrpcDispatchTable, CamelError> {
        let host_owned = host.to_string();

        let cell = {
            let mut guard = self.inner.lock().map_err(|_| {
                CamelError::EndpointCreationFailed("GrpcServerRegistry lock poisoned".into())
            })?;
            let key = (host.to_string(), port);
            // Evict dead server so a fresh one can spawn (rc-4s65).
            if let Some(existing) = guard.get(&key)
                && let Some(handle) = existing.get()
                && handle.task.is_finished()
            {
                guard.remove(&key);
            }
            guard
                .entry(key)
                .or_insert_with(|| Arc::new(OnceCell::new()))
                .clone()
        };

        let handle = cell
            .get_or_try_init(|| async {
                let route_id = format!("grpc-server:{host_owned}:{port}");
                let (tls_acceptor, tls_source) = match build_tls_acceptor(&config.transport) {
                    Ok(v) => v,
                    Err(e) => {
                        runtime.health().force_unhealthy_for_route(
                            &route_id,
                            "g:grpc:tls-read",
                            &format!("{e}"),
                        );
                        // log-policy: outside-contract
                        error!(error = %e, "grpc TLS config build failed");
                        return Err(e);
                    }
                };
                let addr = format!("{host_owned}:{port}");
                let listener = tokio::net::TcpListener::bind(&addr).await.map_err(|e| {
                    CamelError::EndpointCreationFailed(format!(
                        "failed to bind gRPC server on {addr}: {e}"
                    ))
                })?;
                let dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
                let rt = Arc::clone(&runtime);
                let task = tokio::spawn(run_grpc_server(
                    listener,
                    Arc::clone(&dispatch),
                    config.clone(),
                    tls_acceptor.clone(),
                    rt,
                ));
                let handle = ServerHandle {
                    dispatch,
                    task,
                    transport: config.transport.clone(),
                    tls_acceptor,
                    tls_source,
                };
                // Register reload handler (exactly-once: inside OnceCell init closure).
                // Note: gRPC servers are process-lifetime (no release/eviction path),
                // so handlers are never unregistered. If eviction is added later,
                // add TlsReloadRegistry::global().unregister() there.
                if let (Some(acceptor), Some(source)) =
                    (handle.tls_acceptor.as_ref(), handle.tls_source.as_ref())
                {
                    let handler = Arc::new(crate::tls_reload::GrpcReloadHandler::new(
                        acceptor.clone(),
                        source.clone(),
                        host_owned.clone(),
                        port,
                    ));
                    camel_component_api::tls_source::TlsReloadRegistry::global().register(handler);
                }
                Ok::<ServerHandle, CamelError>(handle)
            })
            .await?;

        validate_server_handle(handle, &config.transport, &host_owned, port)?;
        Ok(Arc::clone(&handle.dispatch))
    }

    pub(crate) async fn get_or_spawn_with_listener(
        &'static self,
        listener: tokio::net::TcpListener,
        host: &str,
        port: u16,
        config: GrpcServerConfig,
        runtime: Arc<dyn RuntimeObservability>,
    ) -> Result<GrpcDispatchTable, CamelError> {
        let host_owned = host.to_string();

        let cell = {
            let mut guard = self.inner.lock().map_err(|_| {
                CamelError::EndpointCreationFailed("GrpcServerRegistry lock poisoned".into())
            })?;
            let key = (host.to_string(), port);
            // Evict dead server so a fresh one can spawn (rc-4s65).
            if let Some(existing) = guard.get(&key)
                && let Some(handle) = existing.get()
                && handle.task.is_finished()
            {
                guard.remove(&key);
            }
            guard
                .entry(key)
                .or_insert_with(|| Arc::new(OnceCell::new()))
                .clone()
        };

        let handle = cell
            .get_or_try_init(|| async {
                let route_id = format!("grpc-server:{host_owned}:{port}");
                let (tls_acceptor, tls_source) = match build_tls_acceptor(&config.transport) {
                    Ok(v) => v,
                    Err(e) => {
                        runtime.health().force_unhealthy_for_route(
                            &route_id,
                            "g:grpc:tls-read",
                            &format!("{e}"),
                        );
                        // log-policy: outside-contract
                        error!(error = %e, "grpc TLS config build failed");
                        return Err(e);
                    }
                };
                let dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
                let rt = Arc::clone(&runtime);
                let task = tokio::spawn(run_grpc_server(
                    listener,
                    Arc::clone(&dispatch),
                    config.clone(),
                    tls_acceptor.clone(),
                    rt,
                ));
                let handle = ServerHandle {
                    dispatch,
                    task,
                    transport: config.transport.clone(),
                    tls_acceptor,
                    tls_source,
                };
                // Register reload handler (exactly-once: inside OnceCell init closure).
                // Note: gRPC servers are process-lifetime (no release/eviction path),
                // so handlers are never unregistered. If eviction is added later,
                // add TlsReloadRegistry::global().unregister() there.
                if let (Some(acceptor), Some(source)) =
                    (handle.tls_acceptor.as_ref(), handle.tls_source.as_ref())
                {
                    let handler = Arc::new(crate::tls_reload::GrpcReloadHandler::new(
                        acceptor.clone(),
                        source.clone(),
                        host_owned.clone(),
                        port,
                    ));
                    camel_component_api::tls_source::TlsReloadRegistry::global().register(handler);
                }
                Ok::<ServerHandle, CamelError>(handle)
            })
            .await?;

        validate_server_handle(handle, &config.transport, &host_owned, port)?;
        Ok(Arc::clone(&handle.dispatch))
    }

    pub(crate) async fn unregister(&self, host: &str, port: u16, path: &str) {
        let key = (host.to_string(), port);
        let dispatch = {
            let guard = match self.inner.lock() {
                Ok(g) => g,
                Err(_) => return,
            };
            let Some(cell) = guard.get(&key) else {
                return;
            };
            let Some(handle) = cell.get() else {
                return;
            };
            Arc::clone(&handle.dispatch)
        };
        // Remove this route's path from the dispatch table. The shared
        // HTTP/2 server stays alive — gRPC servers are process-lifetime.
        let mut table = dispatch.write().await;
        table.remove(path);
    }
}

/// Validate a server handle after registry lookup: reject transport mismatches
/// (Plaintext↔Tls OR Tls(A)↔Tls(B)) and reap finished tasks.
fn validate_server_handle(
    handle: &ServerHandle,
    requested: &ServerTransport,
    host: &str,
    port: u16,
) -> Result<(), CamelError> {
    if &handle.transport != requested {
        return Err(CamelError::EndpointCreationFailed(format!(
            "gRPC server {host}:{port} already bound with transport={:?}; \
             requested {:?} — refusing to mix incompatible transport configs on one listener",
            handle.transport, requested,
        )));
    }
    if handle.task.is_finished() {
        return Err(CamelError::EndpointCreationFailed(
            "gRPC server task has terminated unexpectedly".into(),
        ));
    }
    Ok(())
}

/// Build a TlsAcceptor wrapped in ArcSwap from ServerTransport.
/// Returns (None, None) for Plaintext. Hard-errors on missing/invalid cert.
///
/// The acceptor is wrapped in `Arc<ArcSwap<TlsAcceptor>>` so the cert can be
/// hot-swapped at runtime by calling `.store()` on the ArcSwap (Task 5).
/// ALPN `h2` is set for HTTP/2 negotiation.
fn build_tls_acceptor(
    transport: &ServerTransport,
) -> Result<(SharedTlsAcceptor, Option<ServerTlsSource>), CamelError> {
    match transport {
        ServerTransport::Plaintext => Ok((None, None)),
        ServerTransport::Tls(cfg) => {
            let source = ServerTlsSource {
                cert_path: std::path::PathBuf::from(&cfg.server_cert_path),
                key_path: std::path::PathBuf::from(&cfg.server_key_path),
                client_ca_path: cfg.client_ca_path.as_ref().map(std::path::PathBuf::from),
            };
            let mut server_cfg = source.build_server_config()?;
            server_cfg.alpn_protocols = vec![b"h2".to_vec()];
            let acceptor = TlsAcceptor::from(Arc::new(server_cfg));
            Ok((
                Some(Arc::new(ArcSwap::from_pointee(acceptor))),
                Some(source),
            ))
        }
    }
}

/// Serve h2 over a concrete IO type. Generic so both plaintext (TcpStream)
/// and TLS (TlsStream<TcpStream>) call sites type-check.
async fn serve_h2<I>(io: I, dispatch: GrpcDispatchTable, config: GrpcServerConfig)
where
    I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
{
    let io = TokioIo::new(io);
    let service = service_fn(move |req| {
        let dispatch = dispatch.clone();
        handle_grpc_request(req, dispatch)
    });
    let mut builder = http2::Builder::new(hyper_util::rt::TokioExecutor::new());
    if let Some(max_len) = config.max_receive_message_len {
        let frame_size = max_len.clamp(16_384, 16_777_215) as u32;
        builder.max_frame_size(frame_size);
    }
    if let Err(e) = builder.serve_connection(io, service).await {
        debug!(error = %e, "gRPC connection error");
    }
}

/// Capped exponential backoff config for the gRPC accept loop.
/// Starts at 10ms, doubles each failure, caps at 5s.
fn accept_backoff_config() -> BackoffConfig {
    BackoffConfig {
        initial_delay: Duration::from_millis(10),
        multiplier: 2.0,
        max_delay: Duration::from_secs(5),
    }
}

async fn run_grpc_server(
    listener: tokio::net::TcpListener,
    dispatch: GrpcDispatchTable,
    config: GrpcServerConfig,
    tls_acceptor: SharedTlsAcceptor,
    runtime: Arc<dyn RuntimeObservability>,
) {
    // Q-B1: route_id derived from listener local address. The accept loop runs
    // below route dispatch — multiple GrpcEndpoints register on one shared
    // listener, so no single per-route route_id is correct. The local address
    // is stable, attributable, and meaningful to operators.
    let route_id = listener
        .local_addr()
        .map(|addr| format!("grpc-server:{addr}"))
        .unwrap_or_else(|_| "grpc-server:unknown".to_string());

    let mut backoff = BackoffState::new(accept_backoff_config());

    loop {
        let (stream, _) = match listener.accept().await {
            Ok(s) => {
                backoff.reset();
                s
            }
            Err(e) => {
                runtime
                    .metrics()
                    .increment_errors(&route_id, "e:grpc:accept");
                // log-policy: outside-contract
                error!(error = %e, "gRPC server accept error");
                let delay = backoff.next_delay();
                tokio::time::sleep(delay).await;
                continue;
            }
        };

        let tls_acceptor = tls_acceptor.clone();
        let config = config.clone();
        let dispatch = dispatch.clone();
        let rt_metrics = runtime.clone();
        let route_id_clone = route_id.clone();
        tokio::spawn(async move {
            match tls_acceptor.as_ref() {
                None => serve_h2(stream, dispatch, config).await,
                Some(swap) => {
                    // load_full() returns Arc<TlsAcceptor> — owned snapshot, await-safe.
                    // Re-reading per connection enables atomic cert hot-swap via .store().
                    let acceptor = swap.load_full();
                    match acceptor.accept(stream).await {
                        Ok(tls_stream) => serve_h2(tls_stream, dispatch, config).await,
                        Err(e) => {
                            rt_metrics
                                .metrics()
                                .increment_errors(&route_id_clone, "e:grpc:tls-accept");
                            debug!(error = %e, "gRPC TLS handshake error");
                        }
                    }
                }
            }
        });
    }
}

// ── Response stream type ───────────────────────────────────────────────────

type ResponseStream = Pin<Box<dyn futures::Stream<Item = Result<Vec<u8>, Status>> + Send>>;

// ── Manual stream implementation (no async-stream dep) ─────────────────────

struct GrpcItemStream {
    rx: mpsc::Receiver<GrpcStreamItem>,
}

impl futures::Stream for GrpcItemStream {
    type Item = Result<Vec<u8>, Status>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match self.rx.poll_recv(cx) {
            Poll::Ready(Some(GrpcStreamItem::Message(bytes))) => Poll::Ready(Some(Ok(bytes))),
            Poll::Ready(Some(GrpcStreamItem::Error(status))) => Poll::Ready(Some(Err(status))),
            Poll::Ready(Some(GrpcStreamItem::Done)) => Poll::Ready(None),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

// ── Authentication helper ──────────────────────────────────────────────────

async fn extract_principal(
    authenticator: &dyn camel_auth::TokenAuthenticator,
    metadata: &tonic::metadata::MetadataMap,
    sources: &[CredentialSource],
) -> Result<camel_api::security_policy::Principal, tonic::Status> {
    let header_map = metadata_to_header_map(metadata);
    let uri = http::Uri::from_static("/");

    let token = match camel_auth::extract_token_multi(&header_map, &uri, sources) {
        Some(extracted) => extracted.token,
        None => {
            return Err(tonic::Status::unauthenticated(
                "missing or malformed credentials",
            ));
        }
    };

    match authenticator.authenticate_bearer(&token).await {
        Ok(p) => Ok(p),
        Err(e) => Err(auth_error_to_status(e)),
    }
}

/// Map an authentication error onto the transport denial idiom (`tonic::Status`).
fn auth_error_to_status(e: camel_api::CamelError) -> tonic::Status {
    match e {
        camel_api::CamelError::Unauthenticated(msg) => tonic::Status::unauthenticated(msg),
        camel_api::CamelError::ProcessorError(ref msg)
            if msg.contains("auth provider unavailable") =>
        {
            tonic::Status::unavailable(msg.clone())
        }
        other => {
            // log-policy: system-broken
            tracing::error!(error = %other, "gRPC authentication error");
            tonic::Status::internal(other.to_string())
        }
    }
}

/// Per-request authentication at the gRPC boundary.
///
/// Kernel path (plan captured at dispatch-entry construction): an
/// `AccessMode::Public` plan skips extraction entirely — pass-through,
/// `kernel_authenticate` is never called. Any other mode extracts per
/// `plan.credential_sources` (authorization metadata + named headers) and
/// mints the sealed principal through the kernel. No local JWT parsing.
///
/// Legacy path (route without a compiled plan): the route's configured
/// authenticator over the configured sources, unchanged.
///
/// Returns the plain principal (the policy-evaluation view) and, when the
/// kernel minted one, the sealed carrier principal for the request's
/// exchange.
async fn authenticate_request(
    legacy_authenticator: Option<&dyn camel_auth::TokenAuthenticator>,
    legacy_sources: &[CredentialSource],
    kernel: Option<&GrpcKernelAuth>,
    metadata: &tonic::metadata::MetadataMap,
) -> Result<
    (
        Option<camel_api::security_policy::Principal>,
        Option<AuthenticatedPrincipal>,
    ),
    tonic::Status,
> {
    if let Some(kernel) = kernel {
        if matches!(kernel.plan.access_mode, AccessMode::Public) {
            return Ok((None, None));
        }
        let header_map = metadata_to_header_map(metadata);
        let uri = http::Uri::from_static("/");
        let extracted =
            camel_auth::extract_token_multi(&header_map, &uri, &kernel.plan.credential_sources)
                .ok_or_else(|| {
                    tonic::Status::unauthenticated("missing or malformed credentials")
                })?;
        let principal =
            camel_auth::kernel_authenticate(&kernel.plan, &kernel.providers, &extracted)
                .await
                .map_err(auth_error_to_status)?;
        let view = principal.principal().clone();
        Ok((Some(view), Some(principal)))
    } else if let Some(authenticator) = legacy_authenticator {
        let principal = extract_principal(authenticator, metadata, legacy_sources).await?;
        Ok((Some(principal), None))
    } else {
        Ok((None, None))
    }
}

/// Convert tonic metadata into an `http::HeaderMap` for the shared extraction
/// helper. ASCII keys map to lowercase HTTP header names; binary keys
/// (`-bin` suffix) are skipped, as are values that fail header validation.
fn metadata_to_header_map(metadata: &tonic::metadata::MetadataMap) -> http::HeaderMap {
    use tonic::metadata::KeyAndValueRef;

    let mut headers = http::HeaderMap::new();
    for key_and_value in metadata.iter() {
        let KeyAndValueRef::Ascii(key, value) = key_and_value else {
            continue;
        };
        let Ok(header_name) = http::header::HeaderName::try_from(key.as_str()) else {
            continue;
        };
        let Ok(value_str) = value.to_str() else {
            continue;
        };
        let Ok(header_value) = http::header::HeaderValue::from_str(value_str) else {
            continue;
        };
        headers.append(header_name, header_value);
    }
    headers
}

// ── Mode-aware dispatch ────────────────────────────────────────────────────

async fn handle_grpc_request(
    req: hyper::Request<hyper::body::Incoming>,
    dispatch: GrpcDispatchTable,
) -> Result<hyper::Response<TonicBody>, std::convert::Infallible> {
    let path = req.uri().path().to_string();

    let entry = {
        let table = dispatch.read().await;
        table.get(&path).map(|(tx, mode, auth, sources, kernel)| {
            (
                tx.clone(),
                *mode,
                auth.clone(),
                sources.clone(),
                kernel.clone(),
            )
        })
    };

    let Some((sender, mode, authenticator_opt, credential_sources, kernel)) = entry else {
        let handler = UnimplementedHandler;
        let mut grpc = tonic::server::Grpc::new(RawBytesCodec);
        let response = grpc.unary(handler, req).await;
        return Ok(response);
    };

    let mut grpc = tonic::server::Grpc::new(RawBytesCodec);

    match mode {
        GrpcMode::Unary => {
            let handler = UnaryHandler {
                sender,
                authenticator_opt,
                credential_sources,
                kernel,
            };
            let response = grpc.unary(handler, req).await;
            Ok(response)
        }
        GrpcMode::ServerStreaming => {
            let handler = ServerStreamingHandler {
                sender,
                authenticator_opt,
                credential_sources,
                kernel,
            };
            let response = grpc.server_streaming(handler, req).await;
            Ok(response)
        }
        GrpcMode::ClientStreaming => {
            let handler = ClientStreamingHandler {
                sender,
                authenticator_opt,
                credential_sources,
                kernel,
            };
            let response = grpc.client_streaming(handler, req).await;
            Ok(response)
        }
        GrpcMode::Bidi => {
            let handler = BidiHandler {
                sender,
                authenticator_opt,
                credential_sources,
                kernel,
            };
            let response = grpc.streaming(handler, req).await;
            Ok(response)
        }
    }
}

// ── Unimplemented handler (fallback) ───────────────────────────────────────

struct UnimplementedHandler;

impl Service<Request<Vec<u8>>> for UnimplementedHandler {
    type Response = Response<Vec<u8>>;
    type Error = Status;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, _req: Request<Vec<u8>>) -> Self::Future {
        Box::pin(async { Err(Status::unimplemented("no handler for path")) })
    }
}

// ── Unary handler ──────────────────────────────────────────────────────────

struct UnaryHandler {
    sender: mpsc::Sender<GrpcRequestEnvelope>,
    authenticator_opt: Option<Arc<dyn camel_auth::TokenAuthenticator>>,
    credential_sources: Vec<CredentialSource>,
    kernel: Option<Arc<GrpcKernelAuth>>,
}

impl tonic::server::UnaryService<Vec<u8>> for UnaryHandler {
    type Response = Vec<u8>;
    type Future = Pin<Box<dyn Future<Output = Result<Response<Self::Response>, Status>> + Send>>;

    fn call(&mut self, req: Request<Vec<u8>>) -> Self::Future {
        let authenticator_opt = self.authenticator_opt.clone();
        let credential_sources = self.credential_sources.clone();
        let kernel = self.kernel.clone();
        let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
        let sender = self.sender.clone();

        Box::pin(async move {
            let (principal, kernel_principal) = authenticate_request(
                authenticator_opt.as_deref(),
                &credential_sources,
                kernel.as_deref(),
                req.metadata(),
            )
            .await?;

            let envelope = GrpcRequestEnvelope::Unary {
                metadata: req.metadata().clone(),
                body: req.into_inner(),
                reply_tx,
                principal,
                kernel_principal,
            };
            sender
                .send(envelope)
                .await
                .map_err(|_| Status::unavailable("consumer stopped"))?;
            match reply_rx
                .await
                .map_err(|_| Status::internal("reply channel dropped"))?
            {
                GrpcReply::Ok(bytes) => Ok(Response::new(bytes)),
                GrpcReply::Err(status) => Err(status),
            }
        })
    }
}

// ── Server-streaming handler ───────────────────────────────────────────────

struct ServerStreamingHandler {
    sender: mpsc::Sender<GrpcRequestEnvelope>,
    authenticator_opt: Option<Arc<dyn camel_auth::TokenAuthenticator>>,
    credential_sources: Vec<CredentialSource>,
    kernel: Option<Arc<GrpcKernelAuth>>,
}

impl tonic::server::ServerStreamingService<Vec<u8>> for ServerStreamingHandler {
    type Response = Vec<u8>;
    type ResponseStream = ResponseStream;
    type Future =
        Pin<Box<dyn Future<Output = Result<Response<Self::ResponseStream>, Status>> + Send>>;

    fn call(&mut self, req: Request<Vec<u8>>) -> Self::Future {
        let authenticator_opt = self.authenticator_opt.clone();
        let credential_sources = self.credential_sources.clone();
        let kernel = self.kernel.clone();
        let (reply_tx, reply_rx) = mpsc::channel::<GrpcStreamItem>(64);
        let sender = self.sender.clone();

        Box::pin(async move {
            let (principal, kernel_principal) = authenticate_request(
                authenticator_opt.as_deref(),
                &credential_sources,
                kernel.as_deref(),
                req.metadata(),
            )
            .await?;

            let envelope = GrpcRequestEnvelope::ServerStreaming {
                metadata: req.metadata().clone(),
                body: req.into_inner(),
                reply_tx,
                principal,
                kernel_principal,
            };
            sender
                .send(envelope)
                .await
                .map_err(|_| Status::unavailable("consumer stopped"))?;
            Ok(Response::new(
                Box::pin(GrpcItemStream { rx: reply_rx }) as ResponseStream
            ))
        })
    }
}

// ── Client-streaming handler ───────────────────────────────────────────────

struct ClientStreamingHandler {
    sender: mpsc::Sender<GrpcRequestEnvelope>,
    authenticator_opt: Option<Arc<dyn camel_auth::TokenAuthenticator>>,
    credential_sources: Vec<CredentialSource>,
    kernel: Option<Arc<GrpcKernelAuth>>,
}

impl tonic::server::ClientStreamingService<Vec<u8>> for ClientStreamingHandler {
    type Response = Vec<u8>;
    type Future = Pin<Box<dyn Future<Output = Result<Response<Self::Response>, Status>> + Send>>;

    fn call(&mut self, req: Request<Streaming<Vec<u8>>>) -> Self::Future {
        let authenticator_opt = self.authenticator_opt.clone();
        let credential_sources = self.credential_sources.clone();
        let kernel = self.kernel.clone();
        let (body_tx, body_rx) = mpsc::channel::<Vec<u8>>(64);
        let (reply_tx, reply_rx) = tokio::sync::oneshot::channel::<GrpcReply>();
        let sender = self.sender.clone();

        Box::pin(async move {
            let (principal, kernel_principal) = authenticate_request(
                authenticator_opt.as_deref(),
                &credential_sources,
                kernel.as_deref(),
                req.metadata(),
            )
            .await?;

            let envelope = GrpcRequestEnvelope::ClientStreaming {
                metadata: req.metadata().clone(),
                body_rx,
                reply_tx,
                principal,
                kernel_principal,
            };

            let forward_handle = tokio::spawn(async move {
                let mut stream = req.into_inner();
                while let Some(result) = stream.next().await {
                    match result {
                        Ok(bytes) => {
                            if body_tx.send(bytes).await.is_err() {
                                break;
                            }
                        }
                        Err(status) => {
                            tracing::warn!(error = %status, "client streaming decode error");
                            return Some(status);
                        }
                    }
                }
                None
            });

            sender
                .send(envelope)
                .await
                .map_err(|_| Status::unavailable("consumer stopped"))?;

            let reply = reply_rx
                .await
                .map_err(|_| Status::internal("reply channel dropped"))?;

            // If the inbound stream had a decode error, propagate it instead of the consumer's reply.
            if let Ok(Some(status)) = forward_handle.await {
                return Err(status);
            }

            match reply {
                GrpcReply::Ok(bytes) => Ok(Response::new(bytes)),
                GrpcReply::Err(status) => Err(status),
            }
        })
    }
}

// ── Bidi handler ───────────────────────────────────────────────────────────

struct BidiHandler {
    sender: mpsc::Sender<GrpcRequestEnvelope>,
    authenticator_opt: Option<Arc<dyn camel_auth::TokenAuthenticator>>,
    credential_sources: Vec<CredentialSource>,
    kernel: Option<Arc<GrpcKernelAuth>>,
}

impl tonic::server::StreamingService<Vec<u8>> for BidiHandler {
    type Response = Vec<u8>;
    type ResponseStream = ResponseStream;
    type Future =
        Pin<Box<dyn Future<Output = Result<Response<Self::ResponseStream>, Status>> + Send>>;

    fn call(&mut self, req: Request<Streaming<Vec<u8>>>) -> Self::Future {
        let authenticator_opt = self.authenticator_opt.clone();
        let credential_sources = self.credential_sources.clone();
        let kernel = self.kernel.clone();
        let (body_tx, body_rx) = mpsc::channel::<Vec<u8>>(64);
        let (reply_tx, reply_rx) = mpsc::channel::<GrpcStreamItem>(64);
        let reply_tx_forward = reply_tx.clone();
        let sender = self.sender.clone();

        Box::pin(async move {
            let (principal, kernel_principal) = authenticate_request(
                authenticator_opt.as_deref(),
                &credential_sources,
                kernel.as_deref(),
                req.metadata(),
            )
            .await?;

            let envelope = GrpcRequestEnvelope::Bidi {
                metadata: req.metadata().clone(),
                body_rx,
                reply_tx,
                principal,
                kernel_principal,
            };

            tokio::spawn(async move {
                let mut stream = req.into_inner();
                while let Some(result) = stream.next().await {
                    match result {
                        Ok(bytes) => {
                            if body_tx.send(bytes).await.is_err() {
                                break;
                            }
                        }
                        Err(status) => {
                            tracing::warn!(error = %status, "bidi streaming decode error");
                            let _ = reply_tx_forward.send(GrpcStreamItem::Error(status)).await;
                            break;
                        }
                    }
                }
            });

            sender
                .send(envelope)
                .await
                .map_err(|_| Status::unavailable("consumer stopped"))?;

            Ok(Response::new(
                Box::pin(GrpcItemStream { rx: reply_rx }) as ResponseStream
            ))
        })
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::sync::Mutex;
    use std::task::Poll;
    use std::time::Duration;

    use camel_api::MetricsCollector;
    use camel_component_api::HealthCheckRegistry;
    use futures::{Stream, StreamExt};
    use tokio::sync::mpsc;
    use tonic::Status;
    use tonic::server::{ServerStreamingService, UnaryService};
    use tower::Service;

    use super::*;
    use crate::consumer::{GrpcReply, GrpcStreamItem};

    // -----------------------------------------------------------------------
    // Recording metrics collector for testing increment_errors calls
    // -----------------------------------------------------------------------

    struct RecordingMetrics {
        errors: Arc<Mutex<Vec<(String, String)>>>,
    }

    impl MetricsCollector for RecordingMetrics {
        fn record_exchange_duration(&self, _: &str, _: Duration) {}
        fn increment_errors(&self, route_id: &str, error_type: &str) {
            self.errors
                .lock()
                .unwrap()
                .push((route_id.to_string(), error_type.to_string()));
        }
        fn increment_exchanges(&self, _: &str) {}
        fn set_queue_depth(&self, _: &str, _: usize) {}
        fn record_circuit_breaker_change(&self, _: &str, _: &str, _: &str) {}
    }

    struct RecordingRuntime {
        metrics_collector: Arc<RecordingMetrics>,
    }

    impl RecordingRuntime {
        fn new(errors: Arc<Mutex<Vec<(String, String)>>>) -> Self {
            Self {
                metrics_collector: Arc::new(RecordingMetrics { errors }),
            }
        }
    }

    impl RuntimeObservability for RecordingRuntime {
        fn metrics(&self) -> Arc<dyn MetricsCollector> {
            self.metrics_collector.clone() as Arc<dyn MetricsCollector>
        }
        fn health(&self) -> Arc<dyn HealthCheckRegistry> {
            panic!("RecordingRuntime::health not used in this test")
        }
    }

    #[test]
    fn test_global_registry_returns_singleton() {
        let first = GrpcServerRegistry::global();
        let second = GrpcServerRegistry::global();
        assert!(std::ptr::eq(first, second));
    }

    #[tokio::test]
    async fn test_grpc_item_stream_yields_message() {
        let (tx, rx) = mpsc::channel::<GrpcStreamItem>(4);
        let mut stream = GrpcItemStream { rx };
        tx.send(GrpcStreamItem::Message(vec![1, 2, 3]))
            .await
            .unwrap();
        drop(tx);
        let item = stream.next().await.unwrap().unwrap();
        assert_eq!(item, vec![1, 2, 3]);
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn test_grpc_item_stream_yields_error() {
        let (tx, rx) = mpsc::channel::<GrpcStreamItem>(4);
        let mut stream = GrpcItemStream { rx };
        let status = Status::internal("test error");
        tx.send(GrpcStreamItem::Error(status.clone()))
            .await
            .unwrap();
        drop(tx);
        let item = stream.next().await.unwrap();
        assert!(item.is_err());
        assert_eq!(item.unwrap_err().code(), status.code());
    }

    #[tokio::test]
    async fn test_grpc_item_stream_yields_done_as_none() {
        let (tx, rx) = mpsc::channel::<GrpcStreamItem>(4);
        let mut stream = GrpcItemStream { rx };
        tx.send(GrpcStreamItem::Done).await.unwrap();
        drop(tx);
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn test_grpc_item_stream_closed_channel() {
        let (tx, rx) = mpsc::channel::<GrpcStreamItem>(4);
        let mut stream = GrpcItemStream { rx };
        drop(tx);
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn test_grpc_item_stream_multiple_messages() {
        let (tx, rx) = mpsc::channel::<GrpcStreamItem>(4);
        let stream = GrpcItemStream { rx };
        tx.send(GrpcStreamItem::Message(vec![1])).await.unwrap();
        tx.send(GrpcStreamItem::Message(vec![2])).await.unwrap();
        tx.send(GrpcStreamItem::Message(vec![3])).await.unwrap();
        drop(tx);
        let results: Vec<_> = stream.collect().await;
        assert_eq!(results.len(), 3);
        assert_eq!(results[0].as_ref().unwrap(), &vec![1]);
        assert_eq!(results[1].as_ref().unwrap(), &vec![2]);
        assert_eq!(results[2].as_ref().unwrap(), &vec![3]);
    }

    #[tokio::test]
    async fn test_grpc_item_stream_poll_pending() {
        let (_tx, rx) = mpsc::channel::<GrpcStreamItem>(4);
        let stream = GrpcItemStream { rx };
        let waker = futures::task::noop_waker();
        let mut cx = Context::from_waker(&waker);
        let mut stream_pinned = std::pin::Pin::new(Box::new(stream));
        assert!(matches!(
            Stream::poll_next(stream_pinned.as_mut(), &mut cx),
            Poll::Pending
        ));
    }

    #[test]
    fn test_unimplemented_handler_poll_ready() {
        let mut handler = UnimplementedHandler;
        let waker = futures::task::noop_waker();
        let mut cx = Context::from_waker(&waker);
        assert!(matches!(handler.poll_ready(&mut cx), Poll::Ready(Ok(()))));
    }

    #[tokio::test]
    async fn test_unimplemented_handler_returns_unimplemented_status() {
        let mut handler = UnimplementedHandler;
        let req = Request::new(vec![1, 2, 3]);
        let result = Service::call(&mut handler, req).await;
        assert!(result.is_err());
        let status = result.unwrap_err();
        assert_eq!(status.code(), tonic::Code::Unimplemented);
        assert_eq!(status.message(), "no handler for path");
    }

    #[tokio::test]
    async fn test_unregister_removes_path_from_dispatch() {
        let dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
        let (tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
        {
            let mut table = dispatch.write().await;
            table.insert(
                "/test.Service/Method".to_string(),
                (tx, GrpcMode::Unary, None, Vec::new(), None),
            );
        }
        assert!(dispatch.read().await.contains_key("/test.Service/Method"));
        {
            let mut table = dispatch.write().await;
            table.remove("/test.Service/Method");
        }
        assert!(!dispatch.read().await.contains_key("/test.Service/Method"));
    }

    #[tokio::test]
    async fn test_unregister_nonexistent_path_is_noop() {
        let registry = GrpcServerRegistry::global();
        let dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
        registry
            .unregister("localhost", 50051, "/nonexistent.Path/Method")
            .await;
        assert!(dispatch.read().await.is_empty());
    }

    #[test]
    fn test_server_key_equality() {
        let key1: ServerKey = ("localhost".to_string(), 50051);
        let key2: ServerKey = ("localhost".to_string(), 50051);
        let key3: ServerKey = ("localhost".to_string(), 50052);
        let key4: ServerKey = ("remotehost".to_string(), 50051);
        assert_eq!(key1, key2);
        assert_ne!(key1, key3);
        assert_ne!(key1, key4);
    }

    #[tokio::test]
    async fn test_dispatch_table_insert_and_retrieve() {
        let dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
        let (tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
        let path = "/pkg.Service/Method".to_string();
        {
            let mut table = dispatch.write().await;
            table.insert(
                path.clone(),
                (tx, GrpcMode::ServerStreaming, None, Vec::new(), None),
            );
        }
        let table = dispatch.read().await;
        let (_, mode, _, _, _) = table.get(&path).unwrap();
        assert_eq!(*mode, GrpcMode::ServerStreaming);
    }

    #[tokio::test]
    async fn test_dispatch_table_remove_returns_entry() {
        let dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
        let (tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
        let path = "/pkg.Service/Method".to_string();
        {
            let mut table = dispatch.write().await;
            table.insert(path.clone(), (tx, GrpcMode::Bidi, None, Vec::new(), None));
        }
        {
            let mut table = dispatch.write().await;
            let removed = table.remove(&path);
            assert!(removed.is_some());
            let (_, mode, _, _, _) = removed.unwrap();
            assert_eq!(mode, GrpcMode::Bidi);
        }
        assert!(dispatch.read().await.is_empty());
    }

    #[tokio::test]
    async fn test_dispatch_table_all_grpc_modes() {
        let dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
        let modes = [
            GrpcMode::Unary,
            GrpcMode::ServerStreaming,
            GrpcMode::ClientStreaming,
            GrpcMode::Bidi,
        ];
        {
            let mut table = dispatch.write().await;
            for (i, mode) in modes.iter().enumerate() {
                let (tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
                table.insert(format!("/svc/M{i}"), (tx, *mode, None, Vec::new(), None));
            }
        }
        let table = dispatch.read().await;
        assert_eq!(table.len(), 4);
        for (i, expected_mode) in modes.iter().enumerate() {
            let (_, mode, _, _, _) = table.get(&format!("/svc/M{i}")).unwrap();
            assert_eq!(*mode, *expected_mode);
        }
    }

    #[test]
    fn test_grpc_reply_variants() {
        let ok_reply = GrpcReply::Ok(vec![4, 5, 6]);
        match ok_reply {
            GrpcReply::Ok(bytes) => assert_eq!(bytes, vec![4, 5, 6]),
            GrpcReply::Err(_) => panic!("expected Ok"),
        }
        let err_reply = GrpcReply::Err(Status::not_found("missing"));
        match err_reply {
            GrpcReply::Ok(_) => panic!("expected Err"),
            GrpcReply::Err(s) => assert_eq!(s.code(), tonic::Code::NotFound),
        }
    }

    #[tokio::test]
    async fn test_grpc_request_envelope_unary() {
        let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
        let mut metadata = tonic::metadata::MetadataMap::new();
        metadata.insert("x-test", "value".parse().unwrap());
        let body = vec![10, 20, 30];
        let envelope = GrpcRequestEnvelope::Unary {
            metadata: metadata.clone(),
            body: body.clone(),
            reply_tx,
            kernel_principal: None,
            principal: None,
        };
        match envelope {
            GrpcRequestEnvelope::Unary {
                metadata: m,
                body: b,
                reply_tx: tx,
                ..
            } => {
                assert!(m.get("x-test").is_some());
                assert_eq!(b, body);
                let _ = tx.send(GrpcReply::Ok(vec![99]));
            }
            _ => panic!("expected Unary"),
        }
        let reply = reply_rx.await.unwrap();
        assert!(matches!(reply, GrpcReply::Ok(v) if v == vec![99]));
    }

    #[tokio::test]
    async fn test_grpc_request_envelope_server_streaming() {
        let (reply_tx, mut reply_rx) = mpsc::channel::<GrpcStreamItem>(4);
        let envelope = GrpcRequestEnvelope::ServerStreaming {
            metadata: tonic::metadata::MetadataMap::new(),
            body: vec![1],
            reply_tx,
            kernel_principal: None,
            principal: None,
        };
        match envelope {
            GrpcRequestEnvelope::ServerStreaming { reply_tx: tx, .. } => {
                tx.send(GrpcStreamItem::Message(vec![42])).await.unwrap();
                tx.send(GrpcStreamItem::Done).await.unwrap();
            }
            _ => panic!("expected ServerStreaming"),
        }
        match reply_rx.recv().await {
            Some(GrpcStreamItem::Message(b)) => assert_eq!(b, vec![42]),
            _ => panic!("expected Message(42)"),
        }
        assert!(matches!(reply_rx.recv().await, Some(GrpcStreamItem::Done)));
    }

    #[tokio::test]
    async fn test_grpc_request_envelope_client_streaming() {
        let (body_tx, body_rx) = mpsc::channel::<Vec<u8>>(4);
        let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
        let envelope = GrpcRequestEnvelope::ClientStreaming {
            metadata: tonic::metadata::MetadataMap::new(),
            body_rx,
            reply_tx,
            kernel_principal: None,
            principal: None,
        };
        let handle = tokio::spawn(async move {
            match envelope {
                GrpcRequestEnvelope::ClientStreaming {
                    body_rx: mut rx,
                    reply_tx: tx,
                    ..
                } => {
                    assert_eq!(rx.recv().await, Some(vec![1]));
                    assert_eq!(rx.recv().await, Some(vec![2]));
                    let _ = tx.send(GrpcReply::Ok(vec![99]));
                }
                _ => panic!("expected ClientStreaming"),
            }
        });
        body_tx.send(vec![1]).await.unwrap();
        body_tx.send(vec![2]).await.unwrap();
        drop(body_tx);
        handle.await.unwrap();
        let reply = reply_rx.await.unwrap();
        assert!(matches!(reply, GrpcReply::Ok(v) if v == vec![99]));
    }

    #[tokio::test]
    async fn test_grpc_request_envelope_bidi() {
        let (body_tx, body_rx) = mpsc::channel::<Vec<u8>>(4);
        let (reply_tx, mut reply_rx) = mpsc::channel::<GrpcStreamItem>(4);
        let envelope = GrpcRequestEnvelope::Bidi {
            metadata: tonic::metadata::MetadataMap::new(),
            body_rx,
            reply_tx,
            kernel_principal: None,
            principal: None,
        };
        let handle = tokio::spawn(async move {
            match envelope {
                GrpcRequestEnvelope::Bidi {
                    body_rx: mut rx,
                    reply_tx: tx,
                    ..
                } => {
                    assert_eq!(rx.recv().await, Some(vec![10]));
                    tx.send(GrpcStreamItem::Message(vec![20])).await.unwrap();
                }
                _ => panic!("expected Bidi"),
            }
        });
        body_tx.send(vec![10]).await.unwrap();
        match reply_rx.recv().await {
            Some(GrpcStreamItem::Message(b)) => assert_eq!(b, vec![20]),
            _ => panic!("expected Message(20)"),
        }
        handle.await.unwrap();
    }

    #[tokio::test]
    async fn test_get_or_spawn_with_listener_success() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let errors = Arc::new(Mutex::new(Vec::<(String, String)>::new()));
        let rt: Arc<dyn RuntimeObservability> = Arc::new(RecordingRuntime::new(errors));

        let dispatch = GrpcServerRegistry::global()
            .get_or_spawn_with_listener(
                listener,
                "127.0.0.1",
                port,
                GrpcServerConfig::default(),
                rt,
            )
            .await;
        assert!(dispatch.is_ok());
    }

    // Test-only: the std MutexGuard is held across yield/sleep awaits to
    // sequence the dead-task setup atomically. No real contention is possible
    // (fake handle), so the await-holding-lock lint is suppressed here.
    #[tokio::test]
    #[allow(clippy::await_holding_lock)]
    async fn test_dead_server_evicted_on_reuse() {
        let registry = GrpcServerRegistry::global();
        let port = 17899u16;

        // Insert a dead server handle into the registry.
        {
            let mut guard = registry.inner.lock().unwrap();
            let key = ("127.0.0.1".to_string(), port);
            let dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
            let cell = Arc::new(OnceCell::new());
            let dead_task = tokio::spawn(async {});
            // Yield + sleep so the spawned task completes without consuming the handle.
            tokio::task::yield_now().await;
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            assert!(dead_task.is_finished(), "task should be finished");
            cell.set(ServerHandle {
                dispatch,
                task: dead_task,
                transport: ServerTransport::Plaintext,
                tls_acceptor: None,
                tls_source: None,
            })
            .ok();
            guard.insert(key, cell);
        }

        // Now spawn a real server on the same port — the dead entry
        // must be evicted so a fresh server is created.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:17899")
            .await
            .unwrap();
        let errors = Arc::new(Mutex::new(Vec::<(String, String)>::new()));
        let rt: Arc<dyn RuntimeObservability> = Arc::new(RecordingRuntime::new(errors));

        let result = registry
            .get_or_spawn_with_listener(
                listener,
                "127.0.0.1",
                port,
                GrpcServerConfig::default(),
                rt,
            )
            .await;

        assert!(result.is_ok(), "dead server should be evicted");
    }

    #[tokio::test]
    async fn test_unregister_from_global_registry() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let errors = Arc::new(Mutex::new(Vec::<(String, String)>::new()));
        let rt: Arc<dyn RuntimeObservability> = Arc::new(RecordingRuntime::new(errors));

        let dispatch = GrpcServerRegistry::global()
            .get_or_spawn_with_listener(
                listener,
                "127.0.0.1",
                port,
                GrpcServerConfig::default(),
                rt,
            )
            .await
            .unwrap();

        let (tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
        let path = "/test.Unregister/Method".to_string();
        {
            let mut table = dispatch.write().await;
            table.insert(path.clone(), (tx, GrpcMode::Unary, None, Vec::new(), None));
        }
        assert!(dispatch.read().await.contains_key(&path));

        GrpcServerRegistry::global()
            .unregister("127.0.0.1", port, &path)
            .await;

        assert!(!dispatch.read().await.contains_key(&path));
    }

    #[test]
    fn test_unary_handler_is_constructable() {
        let (_tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
        let _handler = UnaryHandler {
            sender: _tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
    }

    #[test]
    fn test_server_streaming_handler_is_constructable() {
        let (_tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
        let _handler = ServerStreamingHandler {
            sender: _tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
    }

    #[test]
    fn test_client_streaming_handler_is_constructable() {
        let (_tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
        let _handler = ClientStreamingHandler {
            sender: _tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
    }

    #[test]
    fn test_bidi_handler_is_constructable() {
        let (_tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
        let _handler = BidiHandler {
            sender: _tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
    }

    #[tokio::test]
    async fn test_unary_handler_send_fails_when_consumer_stopped() {
        let (tx, rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        drop(rx);

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        let req = Request::new(vec![1, 2, 3]);
        let fut = handler.call(req);
        let handle = tokio::spawn(fut);

        let result = handle.await.unwrap();
        let err = result.unwrap_err();
        assert_eq!(err.code(), tonic::Code::Unavailable);
        assert!(err.message().contains("consumer stopped"));
    }

    #[tokio::test]
    async fn test_server_streaming_handler_send_fails_when_consumer_stopped() {
        let (tx, rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        drop(rx);

        let mut handler = ServerStreamingHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        let req = Request::new(vec![1, 2, 3]);
        let fut = handler.call(req);
        let handle = tokio::spawn(fut);

        match handle.await.unwrap() {
            Err(err) => {
                assert_eq!(err.code(), tonic::Code::Unavailable);
            }
            Ok(_) => panic!("expected error when consumer stopped"),
        }
    }

    #[tokio::test]
    async fn test_client_streaming_handler_send_fails_when_consumer_stopped() {
        let (tx, rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        drop(rx);

        let handler = ClientStreamingHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        assert!(handler.sender.is_closed());
    }

    #[tokio::test]
    async fn test_bidi_handler_send_fails_when_consumer_stopped() {
        let (tx, rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        drop(rx);

        let handler = BidiHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        assert!(handler.sender.is_closed());
    }

    #[tokio::test]
    async fn test_unary_handler_reply_channel_dropped() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(4);

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        let req = Request::new(vec![1, 2, 3]);
        let fut = handler.call(req);

        let handle = tokio::spawn(fut);

        let envelope = rx.recv().await.unwrap();
        match envelope {
            GrpcRequestEnvelope::Unary { reply_tx, .. } => {
                drop(reply_tx);
            }
            _ => panic!("expected Unary"),
        }

        let result = handle.await.unwrap();
        let err = result.unwrap_err();
        assert_eq!(err.code(), tonic::Code::Internal);
        assert!(err.message().contains("reply channel dropped"));
    }

    #[tokio::test]
    async fn test_unary_handler_returns_ok_response() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(4);

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        let req = Request::new(vec![10, 20, 30]);
        let fut = handler.call(req);
        let handle = tokio::spawn(fut);

        let envelope = rx.recv().await.unwrap();
        match envelope {
            GrpcRequestEnvelope::Unary { reply_tx, body, .. } => {
                assert_eq!(body, vec![10, 20, 30]);
                let _ = reply_tx.send(GrpcReply::Ok(vec![40, 50]));
            }
            _ => panic!("expected Unary"),
        }

        let result = handle.await.unwrap().unwrap();
        assert_eq!(result.into_inner(), vec![40, 50]);
    }

    #[tokio::test]
    async fn test_unary_handler_returns_error_response() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(4);

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        let req = Request::new(vec![1]);
        let fut = handler.call(req);
        let handle = tokio::spawn(fut);

        let envelope = rx.recv().await.unwrap();
        match envelope {
            GrpcRequestEnvelope::Unary { reply_tx, .. } => {
                let _ = reply_tx.send(GrpcReply::Err(Status::not_found("not found")));
            }
            _ => panic!("expected Unary"),
        }

        let result = handle.await.unwrap();
        let err = result.unwrap_err();
        assert_eq!(err.code(), tonic::Code::NotFound);
    }

    #[tokio::test]
    async fn test_server_streaming_handler_success() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(4);

        let mut handler = ServerStreamingHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        let req = Request::new(vec![1, 2]);
        let fut = handler.call(req);
        let handle = tokio::spawn(fut);

        let envelope = rx.recv().await.unwrap();
        match envelope {
            GrpcRequestEnvelope::ServerStreaming { reply_tx, body, .. } => {
                assert_eq!(body, vec![1, 2]);
                reply_tx
                    .send(GrpcStreamItem::Message(vec![100]))
                    .await
                    .unwrap();
                reply_tx.send(GrpcStreamItem::Done).await.unwrap();
            }
            _ => panic!("expected ServerStreaming"),
        }

        let result = handle.await.unwrap().unwrap();
        let mut stream = result.into_inner();
        assert_eq!(stream.next().await.unwrap().unwrap(), vec![100]);
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn test_server_streaming_handler_error_in_stream() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(4);

        let mut handler = ServerStreamingHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        let req = Request::new(vec![1]);
        let fut = handler.call(req);
        let handle = tokio::spawn(fut);

        let envelope = rx.recv().await.unwrap();
        match envelope {
            GrpcRequestEnvelope::ServerStreaming { reply_tx, .. } => {
                reply_tx
                    .send(GrpcStreamItem::Error(Status::internal("stream error")))
                    .await
                    .unwrap();
            }
            _ => panic!("expected ServerStreaming"),
        }

        let result = handle.await.unwrap().unwrap();
        let mut stream = result.into_inner();
        let item = stream.next().await.unwrap();
        let err = item.unwrap_err();
        assert_eq!(err.code(), tonic::Code::Internal);
    }

    #[tokio::test]
    async fn test_bidi_handler_forwards_items() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(4);
        let (reply_tx, _reply_rx) = mpsc::channel::<GrpcStreamItem>(4);

        let handler = BidiHandler {
            sender: tx,
            authenticator_opt: None,
            kernel: None,
            credential_sources: Vec::new(),
        };
        let (body_tx, body_rx) = mpsc::channel::<Vec<u8>>(4);
        let envelope_for_test = GrpcRequestEnvelope::Bidi {
            metadata: tonic::metadata::MetadataMap::new(),
            body_rx,
            reply_tx,
            kernel_principal: None,
            principal: None,
        };

        let send_result = handler.sender.send(envelope_for_test).await;
        assert!(send_result.is_ok());

        let received = rx.recv().await;
        assert!(received.is_some());

        body_tx.send(vec![10]).await.unwrap();
        body_tx.send(vec![20]).await.unwrap();
        drop(body_tx);
    }

    #[tokio::test]
    async fn test_grpc_stream_item_variants() {
        let msg = GrpcStreamItem::Message(vec![1, 2]);
        match msg {
            GrpcStreamItem::Message(b) => assert_eq!(b, vec![1, 2]),
            _ => panic!(),
        }

        let err = GrpcStreamItem::Error(Status::internal("err"));
        match err {
            GrpcStreamItem::Error(s) => assert_eq!(s.code(), tonic::Code::Internal),
            _ => panic!(),
        }

        let done = GrpcStreamItem::Done;
        match done {
            GrpcStreamItem::Done => {}
            _ => panic!(),
        }
    }

    #[derive(Debug)]
    struct MockAuthenticator {
        should_fail_unauthenticated: bool,
        should_fail_unavailable: bool,
    }

    #[async_trait::async_trait]
    impl camel_auth::TokenAuthenticator for MockAuthenticator {
        async fn authenticate_bearer(
            &self,
            _token: &str,
        ) -> Result<camel_api::security_policy::Principal, camel_api::CamelError> {
            if self.should_fail_unavailable {
                return Err(camel_api::CamelError::ProcessorError(
                    "auth provider unavailable".into(),
                ));
            }
            if self.should_fail_unauthenticated {
                return Err(camel_api::CamelError::Unauthenticated(
                    "invalid token".into(),
                ));
            }
            Ok(camel_api::security_policy::Principal {
                subject: "test-user".into(),
                issuer: "test-issuer".into(),
                audience: vec![],
                scopes: vec![],
                roles: vec![],
                claims: serde_json::json!({}),
            })
        }
    }

    #[test]
    fn grpc_plan_present_at_interceptor_construction() {
        // Task 2.1 lifecycle fix: the interceptor's kernel state must be
        // captured from the security context at construction time — the
        // plan and providers ride the dispatch entry, never a setter
        // patched on after the interceptor exists.
        let authenticator: Arc<MockAuthenticator> = Arc::new(MockAuthenticator {
            should_fail_unauthenticated: false,
            should_fail_unavailable: false,
        });
        let registry = Arc::new(ProviderRegistry::new());
        let plan = RouteSecurityPlan {
            access_mode: AccessMode::Authenticated,
            provider_ref: Some("idp-a".to_string()),
            transport: camel_api::security_policy::TransportId::Grpc,
            credential_sources: vec![CredentialSource::AuthorizationHeader],
            audience_binding: None,
        };

        let mut ctx = SecurityContext::new(GrantAllPolicy, authenticator)
            .with_credential_sources(plan.credential_sources.clone())
            .with_plan(plan.clone())
            .with_providers(Arc::clone(&registry));

        let kernel = GrpcKernelAuth::from_security_context(&ctx).expect("kernel captured"); // allow-unwrap
        assert!(matches!(kernel.plan.access_mode, AccessMode::Authenticated));
        assert_eq!(kernel.plan.provider_ref.as_deref(), Some("idp-a"));
        assert!(matches!(
            kernel.plan.transport,
            camel_api::security_policy::TransportId::Grpc
        ));
        assert_eq!(kernel.plan.credential_sources, plan.credential_sources);
        // Providers captured by Arc identity — the same registry the
        // SecurityContext carries, not a copy.
        assert!(Arc::ptr_eq(&kernel.providers, &registry));

        // Without a plan the interceptor is not kernel-driven (legacy path).
        ctx.plan = None;
        assert!(GrpcKernelAuth::from_security_context(&ctx).is_none());

        // Without providers a plan can never authenticate: no kernel either.
        ctx.plan = Some(plan);
        ctx.providers = None;
        assert!(GrpcKernelAuth::from_security_context(&ctx).is_none());
    }

    struct GrantAllPolicy;

    #[tonic::async_trait]
    impl camel_api::security_policy::SecurityPolicy for GrantAllPolicy {
        async fn evaluate(
            &self,
            _exchange: &mut camel_api::Exchange,
            _auth: &camel_api::security_policy::AuthContext<'_>,
        ) -> Result<camel_api::AuthorizationDecision, camel_api::CamelError> {
            Ok(camel_api::AuthorizationDecision::Granted {
                principal: camel_api::security_policy::Principal {
                    subject: "grant-all".into(),
                    issuer: "test".into(),
                    audience: vec![],
                    scopes: vec![],
                    roles: vec![],
                    claims: serde_json::Value::Null,
                },
            })
        }
    }

    #[tokio::test]
    async fn test_grpc_auth_valid_token() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        let authenticator: Option<Arc<dyn camel_auth::TokenAuthenticator>> =
            Some(Arc::new(MockAuthenticator {
                should_fail_unauthenticated: false,
                should_fail_unavailable: false,
            }));

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: authenticator,
            kernel: None,
            credential_sources: vec![CredentialSource::AuthorizationHeader],
        };

        let mut request = Request::new(vec![]);
        request
            .metadata_mut()
            .insert("authorization", "Bearer test-token".parse().unwrap());

        let handle = tokio::spawn(async move {
            let envelope = rx.recv().await.unwrap();
            match envelope {
                GrpcRequestEnvelope::Unary {
                    principal,
                    reply_tx,
                    ..
                } => {
                    assert!(principal.is_some());
                    let p = principal.unwrap();
                    assert_eq!(p.subject, "test-user");
                    assert_eq!(p.issuer, "test-issuer");
                    let _ = reply_tx.send(GrpcReply::Ok(vec![]));
                }
                _ => panic!("expected Unary"),
            }
        });

        let result = handler.call(request).await;
        assert!(result.is_ok());
        handle.await.unwrap();
    }

    #[tokio::test]
    async fn grpc_credential_sources_custom_header_authenticates() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        let authenticator: Option<Arc<dyn camel_auth::TokenAuthenticator>> =
            Some(Arc::new(MockAuthenticator {
                should_fail_unauthenticated: false,
                should_fail_unavailable: false,
            }));

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: authenticator,
            kernel: None,
            credential_sources: vec![CredentialSource::Header {
                name: "x-api-key".to_string(),
            }],
        };

        let mut request = Request::new(vec![]);
        request
            .metadata_mut()
            .insert("x-api-key", "secret-key".parse().unwrap());

        let handle = tokio::spawn(async move {
            let envelope = rx.recv().await.unwrap();
            match envelope {
                GrpcRequestEnvelope::Unary {
                    principal,
                    reply_tx,
                    ..
                } => {
                    assert!(principal.is_some());
                    let p = principal.unwrap();
                    assert_eq!(p.subject, "test-user");
                    let _ = reply_tx.send(GrpcReply::Ok(vec![]));
                }
                _ => panic!("expected Unary"),
            }
        });

        let result = handler.call(request).await;
        assert!(result.is_ok());
        handle.await.unwrap();
    }

    #[tokio::test]
    async fn grpc_credential_sources_default_bearer_unchanged() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        let authenticator: Option<Arc<dyn camel_auth::TokenAuthenticator>> =
            Some(Arc::new(MockAuthenticator {
                should_fail_unauthenticated: false,
                should_fail_unavailable: false,
            }));

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: authenticator,
            kernel: None,
            credential_sources: vec![CredentialSource::AuthorizationHeader],
        };

        let mut request = Request::new(vec![]);
        request
            .metadata_mut()
            .insert("authorization", "Bearer test-token".parse().unwrap());

        let handle = tokio::spawn(async move {
            let envelope = rx.recv().await.unwrap();
            match envelope {
                GrpcRequestEnvelope::Unary {
                    principal,
                    reply_tx,
                    ..
                } => {
                    assert!(principal.is_some());
                    let p = principal.unwrap();
                    assert_eq!(p.subject, "test-user");
                    let _ = reply_tx.send(GrpcReply::Ok(vec![]));
                }
                _ => panic!("expected Unary"),
            }
        });

        let result = handler.call(request).await;
        assert!(result.is_ok());
        handle.await.unwrap();
    }

    #[tokio::test]
    async fn test_grpc_auth_missing_token() {
        let (tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        let authenticator: Option<Arc<dyn camel_auth::TokenAuthenticator>> =
            Some(Arc::new(MockAuthenticator {
                should_fail_unauthenticated: false,
                should_fail_unavailable: false,
            }));

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: authenticator,
            kernel: None,
            credential_sources: vec![CredentialSource::AuthorizationHeader],
        };

        let request = Request::new(vec![]);

        let result = handler.call(request).await;
        assert!(result.is_err());
        let status = result.unwrap_err();
        assert_eq!(status.code(), tonic::Code::Unauthenticated);
    }

    #[tokio::test]
    async fn test_grpc_auth_invalid_token() {
        let (tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        let authenticator: Option<Arc<dyn camel_auth::TokenAuthenticator>> =
            Some(Arc::new(MockAuthenticator {
                should_fail_unauthenticated: true,
                should_fail_unavailable: false,
            }));

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: authenticator,
            kernel: None,
            credential_sources: vec![CredentialSource::AuthorizationHeader],
        };

        let mut request = Request::new(vec![]);
        request
            .metadata_mut()
            .insert("authorization", "Bearer bad-token".parse().unwrap());

        let result = handler.call(request).await;
        assert!(result.is_err());
        let status = result.unwrap_err();
        assert_eq!(status.code(), tonic::Code::Unauthenticated);
    }

    #[tokio::test]
    async fn test_grpc_auth_provider_unavailable() {
        let (tx, _rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        let authenticator: Option<Arc<dyn camel_auth::TokenAuthenticator>> =
            Some(Arc::new(MockAuthenticator {
                should_fail_unauthenticated: false,
                should_fail_unavailable: true,
            }));

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: authenticator,
            kernel: None,
            credential_sources: vec![CredentialSource::AuthorizationHeader],
        };

        let mut request = Request::new(vec![]);
        request
            .metadata_mut()
            .insert("authorization", "Bearer valid-token".parse().unwrap());

        let result = handler.call(request).await;
        assert!(result.is_err());
        let status = result.unwrap_err();
        assert_eq!(status.code(), tonic::Code::Unavailable);
    }

    #[tokio::test]
    async fn test_grpc_no_auth_configured() {
        let (tx, mut rx) = mpsc::channel::<GrpcRequestEnvelope>(1);
        let authenticator: Option<Arc<dyn camel_auth::TokenAuthenticator>> = None;

        let mut handler = UnaryHandler {
            sender: tx,
            authenticator_opt: authenticator,
            kernel: None,
            credential_sources: vec![CredentialSource::AuthorizationHeader],
        };

        let request = Request::new(vec![]);

        let handle = tokio::spawn(async move {
            let envelope = rx.recv().await.unwrap();
            match envelope {
                GrpcRequestEnvelope::Unary {
                    principal,
                    reply_tx,
                    ..
                } => {
                    assert!(principal.is_none());
                    let _ = reply_tx.send(GrpcReply::Ok(vec![]));
                }
                _ => panic!("expected Unary"),
            }
        });

        let result = handler.call(request).await;
        assert!(result.is_ok());
        handle.await.unwrap();
    }

    #[tokio::test]
    async fn test_server_handle_struct() {
        let dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
        let task = tokio::spawn(async {});
        let handle = ServerHandle {
            dispatch,
            task,
            transport: ServerTransport::Plaintext,
            tls_acceptor: None,
            tls_source: None,
        };
        let _ = handle;
    }

    // ── ADR-0012 (e) site regression test ──────────────────────────────────

    #[tokio::test]
    async fn test_run_grpc_server_route_id_derivation() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let route_id = listener
            .local_addr()
            .map(|addr| format!("grpc-server:{addr}"))
            .unwrap_or_else(|_| "grpc-server:unknown".to_string());
        assert!(!route_id.is_empty(), "route_id must not be empty");
        assert!(
            route_id.starts_with("grpc-server:"),
            "route_id should start with 'grpc-server:': got {route_id}"
        );
    }

    #[tokio::test]
    async fn test_increment_errors_recording_works() {
        // This test validates the metrics recording machinery for the accept
        // error branch without driving the real accept loop into an error.
        //
        // Driving the real loop requires platform-specific fd manipulation
        // (dup+shutdown) that causes the accept to return EINVAL on every
        // iteration — the loop spins indefinitely recording spurious errors,
        // which cannot be precisely asserted. The error condition is not
        // single-shot; it persists after shutdown.
        //
        // The accept error BRANCH (server.rs:189-196) is exercised indirectly:
        //   - test_run_grpc_server_happy_path confirms the accept loop runs
        //     and records zero errors on success.
        //   - This test confirms the recording subsystem correctly captures
        //     the call signature (route_id + label) that the error branch
        //     would emit.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let expected_route_id = listener
            .local_addr()
            .map(|addr| format!("grpc-server:{addr}"))
            .unwrap();
        let _dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
        let errors = Arc::new(Mutex::new(Vec::<(String, String)>::new()));
        let rt: Arc<dyn RuntimeObservability> = Arc::new(RecordingRuntime::new(errors.clone()));

        // Simulate the accept-error branch: verify the metric call signature
        rt.metrics()
            .increment_errors(&expected_route_id, "e:grpc:accept");

        let recorded = errors.lock().unwrap();
        assert_eq!(recorded.len(), 1, "expected one error record");
        assert_eq!(recorded[0].0, expected_route_id, "route_id mismatch");
        assert_eq!(recorded[0].1, "e:grpc:accept", "error label mismatch");
    }

    #[tokio::test]
    async fn test_run_grpc_server_happy_path() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let _dispatch: GrpcDispatchTable = Arc::new(RwLock::new(HashMap::new()));
        let errors = Arc::new(Mutex::new(Vec::<(String, String)>::new()));
        let rt: Arc<dyn RuntimeObservability> = Arc::new(RecordingRuntime::new(errors.clone()));

        let handle = tokio::spawn(run_grpc_server(
            listener,
            _dispatch,
            GrpcServerConfig::default(),
            None,
            rt,
        ));

        // Connect to verify the accept loop handles clients
        let conn =
            tokio::time::timeout(Duration::from_secs(2), tokio::net::TcpStream::connect(addr))
                .await;
        assert!(conn.is_ok(), "server should accept connections");

        // Verify no accept errors recorded on happy path
        {
            let recorded = errors.lock().unwrap();
            assert!(
                recorded.is_empty(),
                "no accept errors expected on happy path"
            );
        }

        handle.abort();
        let _ = handle.await;
    }

    /// Regression: Registry transport-mismatch — cannot mix TLS/plaintext on same listener.
    /// This test verifies the behavioral fail-closed path: first bind plaintext, then
    /// attempt TLS on the same port → must error with "transport" in the message.
    #[tokio::test]
    async fn test_registry_transport_mismatch_errors() {
        use crate::config::ServerTlsConfig;

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let errors = Arc::new(Mutex::new(Vec::<(String, String)>::new()));
        let rt: Arc<dyn RuntimeObservability> = Arc::new(RecordingRuntime::new(errors));

        // First, bind plaintext on this port
        let plaintext_config = GrpcServerConfig {
            max_receive_message_len: None,
            transport: ServerTransport::Plaintext,
        };
        let dispatch = GrpcServerRegistry::global()
            .get_or_spawn_with_listener(listener, "127.0.0.1", port, plaintext_config, rt.clone())
            .await;
        assert!(dispatch.is_ok(), "plaintext bind should succeed");

        // Now attempt TLS on the SAME port — MUST FAIL with transport mismatch
        let tls_config = GrpcServerConfig {
            max_receive_message_len: None,
            transport: ServerTransport::Tls(ServerTlsConfig {
                server_cert_path: "/nonexistent/cert.pem".to_string(),
                server_key_path: "/nonexistent/key.pem".to_string(),
                client_ca_path: None,
            }),
        };

        // Bind a new listener for the TLS attempt (will fail at validation, not bind)
        let listener2 = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

        // Use the SAME port as the first bind to trigger mismatch
        let result = GrpcServerRegistry::global()
            .get_or_spawn_with_listener(listener2, "127.0.0.1", port, tls_config, rt)
            .await;

        assert!(
            result.is_err(),
            "TLS on port already serving plaintext must fail (transport-mismatch)"
        );
        match result {
            Err(e) => {
                let err = e.to_string();
                assert!(
                    err.contains("transport"),
                    "error must mention transport mismatch: {err}"
                );
            }
            Ok(_) => panic!("expected error, got Ok"),
        }
    }

    #[tokio::test]
    async fn test_unregister_last_route_keeps_server_alive() {
        let registry = GrpcServerRegistry::global();
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let errors = Arc::new(Mutex::new(Vec::<(String, String)>::new()));
        let rt: Arc<dyn RuntimeObservability> = Arc::new(RecordingRuntime::new(errors));

        // Register 2 routes on same (host, port)
        let _dispatch1 = registry
            .get_or_spawn_with_listener(
                listener,
                "127.0.0.1",
                port,
                GrpcServerConfig::default(),
                rt.clone(),
            )
            .await
            .unwrap();

        // Second route on same port (listener not used, server already spawned)
        let dummy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let _dispatch2 = registry
            .get_or_spawn_with_listener(
                dummy_listener,
                "127.0.0.1",
                port,
                GrpcServerConfig::default(),
                rt,
            )
            .await
            .unwrap();

        let key = ("127.0.0.1".to_string(), port);

        // Keep a reference to the cell to check task state after unregister
        let cell = {
            let guard = registry.inner.lock().unwrap();
            guard.get(&key).unwrap().clone()
        };

        // Unregister route 1 -> task still alive
        registry.unregister("127.0.0.1", port, "/route1").await;
        {
            let handle = cell.get().expect("handle should exist");
            assert!(
                !handle.task.is_finished(),
                "task should still be alive after first unregister"
            );
        }

        // Unregister route 2 -> server stays alive (process-lifetime)
        registry.unregister("127.0.0.1", port, "/route2").await;
        tokio::time::sleep(Duration::from_millis(10)).await;
        {
            let handle = cell.get().expect("handle should exist");
            assert!(
                !handle.task.is_finished(),
                "task should still be alive — server is process-lifetime"
            );
        }

        // Entry stays in registry for potential restart.
        {
            let guard = registry.inner.lock().unwrap();
            assert!(
                guard.get(&key).is_some(),
                "entry should remain in registry — server kept alive for restart"
            );
        }
    }

    #[test]
    fn test_accept_backoff_config_values() {
        let cfg = accept_backoff_config();
        assert_eq!(cfg.initial_delay, Duration::from_millis(10));
        assert_eq!(cfg.multiplier, 2.0);
        assert_eq!(cfg.max_delay, Duration::from_secs(5));
    }
}