liminal_server/health/

endpoint.rs

use std::io::{Read, Write};
use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, PoisonError};
use std::thread::{self, JoinHandle};
use std::time::Duration;

use crate::ServerError;
use crate::server::listener::{
    loopback_interrupt_target, prepare_accepted_socket, shed_on_fd_exhaustion,
};

use super::checks::{SharedReadinessState, health_check, readiness_check};
use super::reissue::{
    OperatorCredentialReissueOutcome, OperatorCredentialReissueRefusal,
    OperatorCredentialReissueRequest, OperatorCredentialReissuer, SharedOperatorCredentialReissue,
};
use super::unloadable::{SharedUnloadableConversations, UnloadableConversationRecord};

use super::metrics_route;

const HEALTH_PATH: &str = "/health";
const READY_PATH: &str = "/ready";
const METRICS_PATH: &str = "/metrics";
const UNLOADABLE_PATH: &str = "/unloadable-conversations";
/// R18 amendment A7 (§0.18): the operator credential re-issue operation.
const REISSUE_PATH: &str = "/operator/credential-reissue";
const APPLICATION_JSON: &str = "application/json";
const READ_BUFFER_BYTES: usize = 2048;

/// Handle for a running health endpoint server.
///
/// W4 leg 2 (§4.2): the health accept worker BLOCKS in `accept` (kernel-parked,
/// zero idle wakes) rather than spinning a non-blocking poll with a backoff
/// sleep. Shutdown wakes the blocked `accept` with an explicit self-connect
/// interrupt to `interrupt_target` (the bound address, loopback-normalised),
/// mirroring the leg-1 listeners. Shutdown also interrupts an in-flight request
/// read directly via `active_stream`, so a silent client parked in
/// `handle_connection`'s read cannot defer shutdown by its admitted deadline.
#[derive(Debug)]
pub struct HealthServerHandle {
    local_addr: SocketAddr,
    /// Loopback-normalised self-connect target used to interrupt the blocking
    /// `accept` at shutdown.
    interrupt_target: SocketAddr,
    shutdown: Arc<AtomicBool>,
    /// Slot holding a `try_clone` of the request stream the worker is currently
    /// reading, if any. The worker registers it under the lock (with a shutdown
    /// recheck) before blocking on the request read and clears it on completion;
    /// `stop_worker` takes it and `shutdown(Both)`s it to interrupt an in-flight
    /// blocking read directly (TOLD) rather than waiting out the read's admitted
    /// deadline. At most one stream exists at a time (the worker is serial), so
    /// one slot suffices.
    active_stream: Arc<Mutex<Option<TcpStream>>>,
    /// The operator read surface for refused conversation loads, shared with
    /// the worker. The health server binds before any participant handler
    /// exists, so this starts empty and the participant's record is published
    /// into it by [`Self::install_unloadable_record`] once built. Pull-only:
    /// nothing reads it until a request arrives, so it cannot wake the worker.
    unloadable: SharedUnloadableConversations,
    /// The operator WRITE surface for A7 credential re-issue, shared with the
    /// worker. Empty until the participant is built, exactly like `unloadable`
    /// above and for the same reason. Call-driven only: nothing reads it until
    /// a request arrives, so it cannot wake the worker.
    reissue: SharedOperatorCredentialReissue,
    worker: Option<JoinHandle<Result<(), ServerError>>>,
    /// Count of `accept` calls issued by the worker (test observability for the
    /// zero-idle-wakes oracle: on a silent listener this stays at the single
    /// parked call). The worker always maintains the counter; only the host-side
    /// handle for reading it is test-scoped.
    #[cfg(test)]
    accept_attempts: Arc<AtomicU64>,
    /// Count of connections shed under fd exhaustion via the reserve descriptor
    /// (test observability for the shed helper reused from leg 1).
    #[cfg(test)]
    shed_count: Arc<AtomicU64>,
    /// Count of accepted requests the worker has entered (incremented just
    /// before it blocks reading the request). Test observability so a race can
    /// deterministically catch the worker mid-request rather than parked in
    /// `accept`.
    #[cfg(test)]
    requests_entered: Arc<AtomicU64>,
}

impl HealthServerHandle {
    /// Returns the bound address for the health endpoint server.
    #[must_use]
    pub const fn local_addr(&self) -> SocketAddr {
        self.local_addr
    }

    /// Publishes a participant's unloadable-conversation record onto
    /// `GET /unloadable-conversations`.
    ///
    /// Server startup binds the health endpoint FIRST — liveness has to be
    /// answerable while the rest of the server is still being built — so the
    /// record cannot be a constructor argument. Until this is called the route
    /// answers `participant_installed: false`, which is a different answer from
    /// "nothing is refused" and is reported as such.
    ///
    /// Installing shares the handler's live record rather than copying it: a
    /// refusal recorded after this call is reported by the next scrape. No
    /// thread, timer, or notification is created here.
    pub fn install_unloadable_record(&self, record: UnloadableConversationRecord) {
        self.unloadable.install(record);
    }

    /// Publishes the participant authority onto
    /// `POST /operator/credential-reissue` (R18 amendment A7, §0.18).
    ///
    /// Same timing constraint as [`Self::install_unloadable_record`]: the
    /// health endpoint binds before the participant exists, so the authority
    /// arrives here once built. Until it does the route answers 503 and says
    /// no participant is installed, which is a different answer from an
    /// unknown identity and is reported as such. No thread, timer, or
    /// notification is created here.
    pub fn install_credential_reissuer(&self, reissuer: Arc<dyn OperatorCredentialReissuer>) {
        self.reissue.install(reissuer);
    }

    /// Stops the health endpoint server and waits for its worker thread to exit.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::HealthEndpoint`] if the worker thread cannot be
    /// joined cleanly or if the server loop recorded a serving error.
    pub fn shutdown(mut self) -> Result<(), ServerError> {
        self.stop_worker()
    }

    fn stop_worker(&mut self) -> Result<(), ServerError> {
        self.shutdown.store(true, Ordering::SeqCst);
        let Some(worker) = self.worker.take() else {
            return Ok(());
        };
        // Interrupt an in-flight request read directly (TOLD): the flag store
        // above happens-before this lock acquire, so any stream the worker
        // registered before observing the flag is taken here and shut down,
        // waking its blocked read at once. `shutdown(Both)` on the clone reaches
        // the same underlying socket the worker is reading (a dup'd fd). If the
        // worker is instead parked in `accept`, the slot is empty and the
        // self-connect below wakes it. The guard is released (statement end)
        // before the socket call, so no lock is held across `shutdown`.
        let in_flight = self
            .active_stream
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .take();
        if let Some(stream) = in_flight {
            let _ = stream.shutdown(Shutdown::Both);
        }
        // Explicit cross-platform interrupt (mirrors the leg-1 listeners): a
        // single self-connect wakes a blocked `accept`. The worker sees the
        // shutdown flag it observed under the same ordering and sheds the woken
        // (spurious) socket rather than serving it — at most one spurious accept
        // per interrupt. If the worker already exited (a real accept then flag
        // check), the listener is gone and this connect fails fast; the join then
        // returns immediately.
        if let Ok(waker) = TcpStream::connect(self.interrupt_target) {
            drop(waker);
        }

        worker.join().map_err(|_| ServerError::HealthEndpoint {
            message: "health endpoint worker thread terminated unexpectedly".to_owned(),
        })?
    }

    /// `accept` calls issued by the worker (test observability, oracle 8).
    #[cfg(test)]
    fn accept_attempts(&self) -> u64 {
        self.accept_attempts.load(Ordering::SeqCst)
    }

    /// Connections shed under fd exhaustion (test observability).
    #[cfg(test)]
    fn shed_count(&self) -> u64 {
        self.shed_count.load(Ordering::SeqCst)
    }

    /// Accepted requests the worker has entered (test observability).
    #[cfg(test)]
    fn requests_entered(&self) -> u64 {
        self.requests_entered.load(Ordering::SeqCst)
    }
}

impl Drop for HealthServerHandle {
    fn drop(&mut self) {
        if let Err(error) = self.stop_worker() {
            tracing::debug!(%error, "health endpoint shutdown during drop failed");
        }
    }
}

/// Starts the health endpoint HTTP server on a distinct health bind address.
///
/// The returned server handle is independent from the main wire protocol
/// listener. Binding the health endpoint does not mark the main listener ready.
///
/// # Errors
///
/// Returns [`ServerError::HealthEndpoint`] when the health listener cannot bind
/// or cannot report its local address. The listener stays BLOCKING (W4 leg 2):
/// the accept worker kernel-parks in `accept` with zero idle wakes; shutdown
/// wakes it via the self-connect interrupt.
pub fn start_health_server(
    bind_address: SocketAddr,
    readiness: SharedReadinessState,
) -> Result<HealthServerHandle, ServerError> {
    let listener =
        TcpListener::bind(bind_address).map_err(|error| ServerError::HealthEndpoint {
            message: format!("failed to bind health endpoint at {bind_address}: {error}"),
        })?;
    // The listener stays BLOCKING (W4 leg 2): the accept worker kernel-parks in
    // `accept` with zero idle wakes; shutdown wakes it via the self-connect
    // interrupt below.
    let local_addr = listener
        .local_addr()
        .map_err(|error| ServerError::HealthEndpoint {
            message: format!("failed to inspect health endpoint listener address: {error}"),
        })?;
    let interrupt_target = loopback_interrupt_target(local_addr);
    let shutdown = Arc::new(AtomicBool::new(false));
    let active_stream = Arc::new(Mutex::new(None));
    let unloadable = SharedUnloadableConversations::default();
    let reissue = SharedOperatorCredentialReissue::default();
    let accept_attempts = Arc::new(AtomicU64::new(0));
    let shed_count = Arc::new(AtomicU64::new(0));
    let requests_entered = Arc::new(AtomicU64::new(0));
    let worker_shutdown = Arc::clone(&shutdown);
    let worker_stream = Arc::clone(&active_stream);
    let worker_attempts = Arc::clone(&accept_attempts);
    let worker_shed = Arc::clone(&shed_count);
    let worker_entered = Arc::clone(&requests_entered);
    let served = ServedState {
        readiness,
        unloadable: unloadable.clone(),
        reissue: reissue.clone(),
    };
    let worker = thread::spawn(move || {
        serve(
            &listener,
            &served,
            &worker_shutdown,
            &worker_stream,
            &worker_attempts,
            &worker_shed,
            &worker_entered,
        )
    });

    Ok(HealthServerHandle {
        local_addr,
        interrupt_target,
        shutdown,
        active_stream,
        unloadable,
        reissue,
        worker: Some(worker),
        #[cfg(test)]
        accept_attempts,
        #[cfg(test)]
        shed_count,
        #[cfg(test)]
        requests_entered,
    })
}

/// Everything a request is answered from, carried as one value so the worker's
/// parameter list does not grow a slot per operator surface.
#[derive(Debug, Clone)]
struct ServedState {
    readiness: SharedReadinessState,
    unloadable: SharedUnloadableConversations,
    reissue: SharedOperatorCredentialReissue,
}

fn serve(
    listener: &TcpListener,
    served: &ServedState,
    shutdown: &AtomicBool,
    active_stream: &Mutex<Option<TcpStream>>,
    accept_attempts: &AtomicU64,
    shed_count: &AtomicU64,
    requests_entered: &AtomicU64,
) -> Result<(), ServerError> {
    // One reserve descriptor held for the shed-with-spare-fd EMFILE policy,
    // reusing the leg-1 helper.
    let mut reserve = listener.try_clone().ok();
    while !shutdown.load(Ordering::SeqCst) {
        accept_attempts.fetch_add(1, Ordering::SeqCst);
        match listener.accept() {
            Ok((stream, ..)) => {
                // Register the in-flight stream and re-check shutdown atomically
                // under the slot lock. If shutdown already fired, shed without
                // serving (no request slips past the broadcast); otherwise the
                // registered clone lets `stop_worker` interrupt this request's
                // blocking read directly (TOLD). Pairing the flag load here with
                // `stop_worker`'s flag-store-then-lock means a registration can
                // never be missed: either the worker sees the flag and sheds, or
                // shutdown finds the clone in the slot.
                let admitted = {
                    let mut slot = active_stream.lock().unwrap_or_else(PoisonError::into_inner);
                    if shutdown.load(Ordering::SeqCst) {
                        false
                    } else {
                        *slot = stream.try_clone().ok();
                        true
                    }
                };
                if !admitted {
                    drop(stream);
                    continue;
                }
                // A `/metrics` body runs to many KB, so the response spans several
                // segments and Nagle would hold its trailing partial segment for the
                // scraper's delayed ACK. Same preparation as the leg-1 accept loops.
                prepare_accepted_socket(&stream, None);
                requests_entered.fetch_add(1, Ordering::SeqCst);
                // A per-connection error (e.g. a TCP probe that connects but sends no HTTP
                // data within the read timeout) must NOT terminate the serve loop — otherwise
                // a single port probe kills the health server for the process lifetime and
                // subsequent liveness/readiness probes get connection-refused. Only fatal
                // listener-level accept errors (below) terminate serving.
                let result = handle_connection(stream, served);
                // The request is done: clear the slot so a later shutdown finds
                // nothing to interrupt (and never shuts down an unrelated stream).
                // A no-op if `stop_worker` already took the clone.
                *active_stream.lock().unwrap_or_else(PoisonError::into_inner) = None;
                if let Err(error) = result {
                    tracing::debug!(%error, "health endpoint connection error");
                }
            }
            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
            Err(error) if is_transient_accept_error(&error) => {
                shed_on_fd_exhaustion(listener, &mut reserve, shed_count, &error);
            }
            Err(error) => {
                return Err(ServerError::HealthEndpoint {
                    message: format!("health endpoint accept failed: {error}"),
                });
            }
        }
    }

    Ok(())
}

/// EMFILE/ENFILE resource exhaustion is transient, exactly as the leg-1 accept
/// loops treat it (mirrors the sibling WebSocket listener's local predicate).
fn is_transient_accept_error(error: &std::io::Error) -> bool {
    matches!(error.raw_os_error(), Some(code) if code == 24 || code == 23)
}

fn handle_connection(mut stream: TcpStream, served: &ServedState) -> Result<(), ServerError> {
    stream
        .set_nonblocking(false)
        .map_err(|error| ServerError::HealthEndpoint {
            message: format!("failed to configure health request stream: {error}"),
        })?;
    stream
        .set_read_timeout(Some(Duration::from_secs(2)))
        .map_err(|error| ServerError::HealthEndpoint {
            message: format!("failed to set health request read timeout: {error}"),
        })?;

    let mut buffer = [0_u8; READ_BUFFER_BYTES];
    let bytes_read = stream
        .read(&mut buffer)
        .map_err(|error| ServerError::HealthEndpoint {
            message: format!("failed to read health request: {error}"),
        })?;

    if bytes_read == 0 {
        return Ok(());
    }

    let response = response_for_request(&buffer[..bytes_read], served)?;
    stream
        .write_all(&response)
        .map_err(|error| ServerError::HealthEndpoint {
            message: format!("failed to write health response: {error}"),
        })?;
    stream.flush().map_err(|error| ServerError::HealthEndpoint {
        message: format!("failed to flush health response: {error}"),
    })
}

fn response_for_request(request: &[u8], served: &ServedState) -> Result<Vec<u8>, ServerError> {
    let Ok(request) = std::str::from_utf8(request) else {
        return Ok(empty_response(StatusCode::BadRequest));
    };
    let Some((method, path)) = parse_request_line(request) else {
        return Ok(empty_response(StatusCode::BadRequest));
    };

    match (method, path) {
        ("GET", HEALTH_PATH) => json_response(StatusCode::Ok, &health_check()),
        ("GET", READY_PATH) => {
            let status = readiness_check(&served.readiness.snapshot());
            let status_code = if status.ready {
                StatusCode::Ok
            } else {
                StatusCode::ServiceUnavailable
            };
            json_response(status_code, &status)
        }
        ("GET", METRICS_PATH) => Ok(response(
            StatusCode::Ok,
            Some(metrics_route::CONTENT_TYPE),
            metrics_route::render_body().as_bytes(),
        )),
        // The refused-load surface. Reading it is a snapshot of a record
        // another part of the server already maintains — this route computes
        // nothing, schedules nothing, and touches no conversation.
        ("GET", UNLOADABLE_PATH) => json_response(StatusCode::Ok, &served.unloadable.status()),
        (_, HEALTH_PATH | READY_PATH | METRICS_PATH | UNLOADABLE_PATH) => {
            Ok(empty_response(StatusCode::MethodNotAllowed))
        }
        // R18 amendment A7. Matched on the path with its query string SPLIT
        // OFF, and only here: the four routes above keep matching the whole
        // request target exactly as they always have, so a query string on
        // `/health` is still a 404 and no existing route's answer moves.
        _ if path.split('?').next() == Some(REISSUE_PATH) => {
            credential_reissue_response(method, path, served)
        }
        _ => Ok(empty_response(StatusCode::NotFound)),
    }
}

/// Answers one `OperatorCredentialReissue` call (R18 amendment A7, §0.18).
///
/// # Why the inputs ride the query string
///
/// The three inputs are fixed by §0.18 (`conversation_id`, `participant_id`,
/// `expected_current_generation`) and none of them is a secret — the SECRET
/// travels only in the response. The request shape is a build decision, and a
/// query string is chosen because this endpoint reads each request with ONE
/// bounded read and no message framing: a body would be present only when the
/// client happened to put it in the same segment, so parsing one would make
/// the route's answer depend on TCP segmentation. Teaching the shared request
/// reader to frame bodies would change the serving discipline of every route
/// here, which is not this lane's to do.
fn credential_reissue_response(
    method: &str,
    path: &str,
    served: &ServedState,
) -> Result<Vec<u8>, ServerError> {
    if method != "POST" {
        return Ok(empty_response(StatusCode::MethodNotAllowed));
    }
    let query = path.split_once('?').map_or("", |(_, query)| query);
    let Some(request) = parse_reissue_query(query) else {
        return Ok(empty_response(StatusCode::BadRequest));
    };
    match served.reissue.reissue(request) {
        // No participant is configured on this node. Distinct from every
        // identity answer, and reported as such rather than as a lookup miss.
        Ok(None) => Ok(empty_response(StatusCode::ServiceUnavailable)),
        Ok(Some(OperatorCredentialReissueOutcome::Issued(issued))) => {
            json_response(StatusCode::Ok, &issued)
        }
        Ok(Some(OperatorCredentialReissueOutcome::Refused(refusal))) => {
            json_response(reissue_refusal_status(&refusal), &refusal)
        }
        Err(error) => {
            // The operation could not be DECIDED. The operator is told, and
            // the text is the refusal's own — never a bare close.
            tracing::error!(%error, "operator credential re-issue could not be decided");
            json_response(
                StatusCode::ServiceUnavailable,
                &serde_json::json!({ "error": error.message }),
            )
        }
    }
}

/// The status each typed refusal is served with.
///
/// A lookup miss is a 404 and every guard refusal is a 409: the identity
/// resolved, and the operation was refused by a live fact about it. The
/// discriminator an operator branches on is the body's `refusal` field, never
/// the status — the status is the HTTP-shaped summary of it.
const fn reissue_refusal_status(refusal: &OperatorCredentialReissueRefusal) -> StatusCode {
    match refusal {
        OperatorCredentialReissueRefusal::ConversationUnknown { .. }
        | OperatorCredentialReissueRefusal::ParticipantUnknown { .. } => StatusCode::NotFound,
        OperatorCredentialReissueRefusal::Retired { .. }
        | OperatorCredentialReissueRefusal::LiveBinding { .. }
        | OperatorCredentialReissueRefusal::DetachReplayOpen { .. }
        | OperatorCredentialReissueRefusal::LiveReceipt { .. }
        | OperatorCredentialReissueRefusal::GenerationMismatch { .. } => StatusCode::Conflict,
    }
}

/// Parses the three §0.18 inputs, or answers `None` for anything else.
///
/// Every field is mandatory and every value is a plain decimal `u64`. An
/// unknown parameter, a repeat, or a missing one is refused rather than
/// defaulted: an operator who mistypes `expected_current_generation` must not
/// have a zero silently substituted for it.
fn parse_reissue_query(query: &str) -> Option<OperatorCredentialReissueRequest> {
    let mut conversation_id = None;
    let mut participant_id = None;
    let mut expected_current_generation = None;
    for pair in query.split('&') {
        let (name, value) = pair.split_once('=')?;
        let value = value.parse::<u64>().ok()?;
        let slot = match name {
            "conversation_id" => &mut conversation_id,
            "participant_id" => &mut participant_id,
            "expected_current_generation" => &mut expected_current_generation,
            _ => return None,
        };
        if slot.replace(value).is_some() {
            return None;
        }
    }
    Some(OperatorCredentialReissueRequest {
        conversation_id: conversation_id?,
        participant_id: participant_id?,
        expected_current_generation: expected_current_generation?,
    })
}

fn parse_request_line(request: &str) -> Option<(&str, &str)> {
    let request_line = request.lines().next()?;
    let mut parts = request_line.split_whitespace();
    let method = parts.next()?;
    let path = parts.next()?;
    parts.next()?;

    Some((method, path))
}

fn json_response<T>(status: StatusCode, value: &T) -> Result<Vec<u8>, ServerError>
where
    T: serde::Serialize,
{
    let body = serde_json::to_vec(value).map_err(|error| ServerError::HealthEndpoint {
        message: format!("failed to serialize health response: {error}"),
    })?;
    Ok(response(status, Some(APPLICATION_JSON), &body))
}

fn empty_response(status: StatusCode) -> Vec<u8> {
    response(status, None, &[])
}

fn response(status: StatusCode, content_type: Option<&str>, body: &[u8]) -> Vec<u8> {
    let mut response = Vec::new();
    let status_line = format!("HTTP/1.1 {} {}\r\n", status.code(), status.reason());
    response.extend_from_slice(status_line.as_bytes());
    response.extend_from_slice(format!("Content-Length: {}\r\n", body.len()).as_bytes());
    response.extend_from_slice(b"Connection: close\r\n");
    if let Some(content_type) = content_type {
        response.extend_from_slice(format!("Content-Type: {content_type}\r\n").as_bytes());
    }
    response.extend_from_slice(b"\r\n");
    response.extend_from_slice(body);
    response
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StatusCode {
    Ok,
    BadRequest,
    NotFound,
    MethodNotAllowed,
    Conflict,
    ServiceUnavailable,
}

impl StatusCode {
    const fn code(self) -> u16 {
        match self {
            Self::Ok => 200,
            Self::BadRequest => 400,
            Self::NotFound => 404,
            Self::MethodNotAllowed => 405,
            Self::Conflict => 409,
            Self::ServiceUnavailable => 503,
        }
    }

    const fn reason(self) -> &'static str {
        match self {
            Self::Ok => "OK",
            Self::BadRequest => "Bad Request",
            Self::NotFound => "Not Found",
            Self::MethodNotAllowed => "Method Not Allowed",
            Self::Conflict => "Conflict",
            Self::ServiceUnavailable => "Service Unavailable",
        }
    }
}

#[cfg(test)]
mod tests {
    use std::io::{Read, Write};
    use std::net::{SocketAddr, TcpStream};
    use std::sync::{Arc, Mutex, PoisonError};
    use std::thread;
    use std::time::{Duration, Instant};

    use serde_json::Value;

    use super::{
        OperatorCredentialReissueRefusal, OperatorCredentialReissueRequest,
        OperatorCredentialReissuer, ServedState, response_for_request, start_health_server,
    };
    use crate::health::checks::{
        ClusterReadiness, ReadinessCondition, ReadinessState, SharedReadinessState,
    };

    fn loopback_ephemeral() -> Result<SocketAddr, Box<dyn std::error::Error>> {
        Ok("127.0.0.1:0".parse()?)
    }

    /// The request-level tests answer from readiness alone; the refused-load
    /// surface stays uninstalled, which is the state a server is in before its
    /// participant handler exists.
    fn served(readiness: SharedReadinessState) -> ServedState {
        ServedState {
            readiness,
            unloadable: crate::health::unloadable::SharedUnloadableConversations::default(),
            reissue: crate::health::reissue::SharedOperatorCredentialReissue::default(),
        }
    }

    fn get(address: SocketAddr, path: &str) -> Result<String, Box<dyn std::error::Error>> {
        let mut stream = TcpStream::connect(address)?;
        stream.set_read_timeout(Some(Duration::from_secs(2)))?;
        let request = format!("GET {path} HTTP/1.1\r\nHost: localhost\r\n\r\n");
        stream.write_all(request.as_bytes())?;

        let mut response = String::new();
        stream.read_to_string(&mut response)?;
        Ok(response)
    }

    fn assert_status(response: &str, status: u16) {
        let expected = format!("HTTP/1.1 {status} ");
        assert!(
            response.starts_with(&expected),
            "response status did not start with {expected}: {response}"
        );
    }

    fn body(response: &str) -> Result<&str, Box<dyn std::error::Error>> {
        let Some((_headers, body)) = response.split_once("\r\n\r\n") else {
            return Err("response did not contain a header/body separator".into());
        };
        Ok(body)
    }

    fn json_body(response: &str) -> Result<Value, Box<dyn std::error::Error>> {
        Ok(serde_json::from_str(body(response)?)?)
    }

    #[test]
    fn health_endpoint_returns_json_200_regardless_of_readiness()
    -> Result<(), Box<dyn std::error::Error>> {
        let readiness = SharedReadinessState::new(ReadinessState::default());
        let server = start_health_server(loopback_ephemeral()?, readiness)?;

        let response = get(server.local_addr(), "/health")?;
        server.shutdown()?;

        assert_status(&response, 200);
        assert!(response.contains("Content-Type: application/json\r\n"));
        let body = json_body(&response)?;
        assert_eq!(body["status"], "healthy");

        Ok(())
    }

    #[test]
    fn ready_endpoint_returns_503_before_main_listener_binds()
    -> Result<(), Box<dyn std::error::Error>> {
        let readiness = SharedReadinessState::new(ReadinessState::new(
            true,
            false,
            ClusterReadiness::NotConfigured,
        ));
        let server = start_health_server(loopback_ephemeral()?, readiness)?;

        let response = get(server.local_addr(), "/ready")?;
        server.shutdown()?;

        assert_status(&response, 503);
        assert!(response.contains("Content-Type: application/json\r\n"));
        let body = json_body(&response)?;
        assert_eq!(body["ready"], false);
        assert_eq!(body["unmet_conditions"][0], "listener_bound");

        Ok(())
    }

    #[test]
    fn ready_endpoint_returns_200_after_all_startup_gates() -> Result<(), Box<dyn std::error::Error>>
    {
        let readiness = SharedReadinessState::new(ReadinessState::ready_without_cluster());
        let server = start_health_server(loopback_ephemeral()?, readiness)?;

        let response = get(server.local_addr(), "/ready")?;
        server.shutdown()?;

        assert_status(&response, 200);
        let body = json_body(&response)?;
        assert_eq!(body["ready"], true);
        let Some(unmet_conditions) = body["unmet_conditions"].as_array() else {
            return Err("unmet_conditions should be an array".into());
        };
        assert!(unmet_conditions.is_empty());

        Ok(())
    }

    #[test]
    fn ready_endpoint_updates_from_shared_readiness_state() -> Result<(), Box<dyn std::error::Error>>
    {
        let readiness = SharedReadinessState::new(ReadinessState::default());
        let server = start_health_server(loopback_ephemeral()?, readiness.clone())?;

        let response = get(server.local_addr(), "/ready")?;
        assert_status(&response, 503);

        readiness.set_config_loaded(true);
        readiness.set_listener_bound(true);
        let response = get(server.local_addr(), "/ready")?;
        server.shutdown()?;

        assert_status(&response, 200);

        Ok(())
    }

    #[test]
    fn clustered_ready_transitions_503_to_200_when_membership_established()
    -> Result<(), Box<dyn std::error::Error>> {
        // A clustered server starts with the cluster gate unmet: config loaded and
        // listener bound, but membership not yet established (G2). /ready is 503.
        let readiness = SharedReadinessState::new(ReadinessState::new(
            true,
            true,
            ClusterReadiness::Configured {
                membership_established: false,
            },
        ));
        let server = start_health_server(loopback_ephemeral()?, readiness.clone())?;

        let response = get(server.local_addr(), "/ready")?;
        assert_status(&response, 503);
        let body = json_body(&response)?;
        assert_eq!(body["ready"], false);
        assert_eq!(
            body["unmet_conditions"][0],
            serde_json::to_value(ReadinessCondition::ClusterMembershipEstablished)?
        );

        // The cluster start's on_established hook flips exactly this flag; once set,
        // /ready must transition to 200 with no unmet conditions.
        readiness.set_cluster_membership_established(true);
        let response = get(server.local_addr(), "/ready")?;
        server.shutdown()?;

        assert_status(&response, 200);
        let body = json_body(&response)?;
        assert_eq!(body["ready"], true);
        let Some(unmet_conditions) = body["unmet_conditions"].as_array() else {
            return Err("unmet_conditions should be an array".into());
        };
        assert!(unmet_conditions.is_empty());

        Ok(())
    }

    #[test]
    fn cluster_readiness_is_listed_when_configured_but_not_joined()
    -> Result<(), Box<dyn std::error::Error>> {
        let readiness = SharedReadinessState::new(ReadinessState::new(
            true,
            true,
            ClusterReadiness::Configured {
                membership_established: false,
            },
        ));
        let response = response_for_request(b"GET /ready HTTP/1.1\r\n\r\n", &served(readiness))?;
        let response = String::from_utf8(response)?;

        assert_status(&response, 503);
        let body = json_body(&response)?;
        assert_eq!(
            body["unmet_conditions"][0],
            serde_json::to_value(ReadinessCondition::ClusterMembershipEstablished)?
        );

        Ok(())
    }

    /// THE UNLOADABLE-CONVERSATIONS READ SURFACE, RED UNTIL THE ROUTE EXISTS.
    ///
    /// The node already records every conversation it refused to load
    /// (`server/participant/production/handler.rs`, `record_unloadable`) and
    /// then offers an operator no way to read it back: the accessor has no
    /// caller. Containment without a read surface is a node that serves nothing
    /// on one conversation forever and answers no question about it.
    ///
    /// This pin fixes the shape of the answer and nothing else — that the route
    /// EXISTS, answers JSON 200, and carries the three fields an operator reads:
    /// how many conversations are refused, which ones, and whether a participant
    /// record is even attached (so a `count` of zero is never confused with a
    /// node that has no participant installed at all).
    ///
    /// The unknown-path arm in the same test is the discriminator: without it a
    /// 200 here would also be satisfied by a server that answers 200 to
    /// everything.
    #[test]
    fn unloadable_conversations_route_answers_the_operator_a_json_shape()
    -> Result<(), Box<dyn std::error::Error>> {
        let readiness = SharedReadinessState::new(ReadinessState::default());
        let server = start_health_server(loopback_ephemeral()?, readiness)?;

        let response = get(server.local_addr(), "/unloadable-conversations")?;
        let unknown = get(server.local_addr(), "/unloadable-conversations-typo")?;
        server.shutdown()?;

        // The discriminator first: a neighbouring path is still not served, so
        // the 200 below is this route's own answer.
        assert_status(&unknown, 404);

        assert_status(&response, 200);
        assert!(
            response.contains("Content-Type: application/json\r\n"),
            "the unloadable-conversations route must answer JSON: {response}"
        );
        let body = json_body(&response)?;
        assert_eq!(
            body["count"], 0,
            "a server with no participant record attached refuses nothing: {body}"
        );
        assert_eq!(
            body["participant_installed"], false,
            "no participant record is attached to this server, and the surface must say so \
             rather than let a zero count read as a clean node: {body}"
        );
        let Some(conversations) = body["conversations"].as_array() else {
            return Err("conversations should be an array".into());
        };
        assert!(
            conversations.is_empty(),
            "no conversation was refused: {conversations:?}"
        );

        Ok(())
    }

    #[test]
    fn unsupported_paths_are_not_served() -> Result<(), Box<dyn std::error::Error>> {
        let readiness = SharedReadinessState::default();
        let response = response_for_request(b"GET /unknown HTTP/1.1\r\n\r\n", &served(readiness))?;
        let response = String::from_utf8(response)?;

        assert_status(&response, 404);

        Ok(())
    }

    #[test]
    fn unsupported_methods_on_health_paths_are_rejected() -> Result<(), Box<dyn std::error::Error>>
    {
        let readiness = SharedReadinessState::default();
        let response = response_for_request(b"POST /health HTTP/1.1\r\n\r\n", &served(readiness))?;
        let response = String::from_utf8(response)?;

        assert_status(&response, 405);

        Ok(())
    }

    // -----------------------------------------------------------------------
    // R18 amendment A7 (§0.18) — the operator credential re-issue route
    // -----------------------------------------------------------------------

    /// A reissuer that answers whatever it was built with, and records the
    /// request it was handed so the route's parsing can be measured rather than
    /// assumed.
    #[derive(Debug)]
    struct RecordingReissuer {
        outcome: crate::health::reissue::OperatorCredentialReissueOutcome,
        seen: Mutex<Vec<OperatorCredentialReissueRequest>>,
    }

    impl OperatorCredentialReissuer for RecordingReissuer {
        fn reissue(
            &self,
            request: OperatorCredentialReissueRequest,
        ) -> Result<
            crate::health::reissue::OperatorCredentialReissueOutcome,
            crate::health::reissue::OperatorCredentialReissueError,
        > {
            self.seen
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .push(request);
            Ok(self.outcome.clone())
        }
    }

    fn served_with_reissuer(
        outcome: crate::health::reissue::OperatorCredentialReissueOutcome,
    ) -> (ServedState, Arc<RecordingReissuer>) {
        let reissuer = Arc::new(RecordingReissuer {
            outcome,
            seen: Mutex::new(Vec::new()),
        });
        let state = served(SharedReadinessState::default());
        state
            .reissue
            .install(Arc::clone(&reissuer) as Arc<dyn OperatorCredentialReissuer>);
        (state, reissuer)
    }

    const REISSUE_TARGET: &str = "/operator/credential-reissue?conversation_id=7&participant_id=3&\
         expected_current_generation=14";

    /// A node with no participant configured says so, rather than answering
    /// like a node whose identity is unknown.
    #[test]
    fn the_reissue_route_reports_an_uninstalled_participant()
    -> Result<(), Box<dyn std::error::Error>> {
        let request = format!("POST {REISSUE_TARGET} HTTP/1.1\r\n\r\n");
        let response =
            response_for_request(request.as_bytes(), &served(SharedReadinessState::default()))?;
        let response = String::from_utf8(response)?;

        assert_status(&response, 503);
        Ok(())
    }

    /// A committed re-issue serves the secret ONCE, and the three §0.18 inputs
    /// arrive at the authority exactly as the operator wrote them.
    #[test]
    fn the_reissue_route_carries_the_three_inputs_and_returns_the_secret_once()
    -> Result<(), Box<dyn std::error::Error>> {
        let issued = crate::health::reissue::OperatorCredentialReissued {
            conversation_id: 7,
            participant_id: 3,
            presented_generation: 14,
            issued_generation: 15,
            attach_secret: crate::health::reissue::encode_hex(&[0x5A; 32]),
        };
        let (state, reissuer) = served_with_reissuer(
            crate::health::reissue::OperatorCredentialReissueOutcome::Issued(issued),
        );
        let request = format!("POST {REISSUE_TARGET} HTTP/1.1\r\n\r\n");

        let response = String::from_utf8(response_for_request(request.as_bytes(), &state)?)?;

        assert_status(&response, 200);
        let body = json_body(&response)?;
        assert_eq!(body["issued_generation"], 15);
        assert_eq!(body["presented_generation"], 14);
        assert_eq!(body["attach_secret"], "5a".repeat(32));
        let seen = reissuer
            .seen
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone();
        assert_eq!(
            seen.as_slice(),
            [OperatorCredentialReissueRequest {
                conversation_id: 7,
                participant_id: 3,
                expected_current_generation: 14,
            }]
        );
        Ok(())
    }

    /// The NORMATIVE compare-and-set payload survives the route (§0.18 item 4),
    /// and a guard refusal is a 409 whose discriminator is a field.
    #[test]
    fn the_reissue_route_serves_the_normative_generation_pair()
    -> Result<(), Box<dyn std::error::Error>> {
        let (state, _) = served_with_reissuer(
            crate::health::reissue::OperatorCredentialReissueOutcome::Refused(
                OperatorCredentialReissueRefusal::GenerationMismatch {
                    conversation_id: 7,
                    participant_id: 3,
                    presented_generation: 14,
                    current_generation: 15,
                },
            ),
        );
        let request = format!("POST {REISSUE_TARGET} HTTP/1.1\r\n\r\n");

        let response = String::from_utf8(response_for_request(request.as_bytes(), &state)?)?;

        assert_status(&response, 409);
        let body = json_body(&response)?;
        assert_eq!(body["refusal"], "generation_mismatch");
        assert_eq!(body["presented_generation"], 14);
        assert_eq!(body["current_generation"], 15);
        Ok(())
    }

    /// A pre-guard lookup miss is a 404, and discloses nothing beyond the
    /// identifiers the operator presented.
    #[test]
    fn the_reissue_route_answers_a_lookup_miss_with_not_found()
    -> Result<(), Box<dyn std::error::Error>> {
        let (state, _) = served_with_reissuer(
            crate::health::reissue::OperatorCredentialReissueOutcome::Refused(
                OperatorCredentialReissueRefusal::ConversationUnknown { conversation_id: 7 },
            ),
        );
        let request = format!("POST {REISSUE_TARGET} HTTP/1.1\r\n\r\n");

        let response = String::from_utf8(response_for_request(request.as_bytes(), &state)?)?;

        assert_status(&response, 404);
        let body = json_body(&response)?;
        assert_eq!(body["refusal"], "conversation_unknown");
        assert_eq!(body["conversation_id"], 7);
        assert!(
            body.get("participant_id").is_none(),
            "an unknown-conversation refusal must disclose nothing beyond the presented \
             conversation id: {body}"
        );
        Ok(())
    }

    /// A malformed call is REFUSED, never defaulted. An operator who mistypes a
    /// generation must not have a zero silently substituted for it and a
    /// credential rotated on the strength of it.
    #[test]
    fn a_malformed_reissue_call_is_refused_and_never_reaches_the_authority()
    -> Result<(), Box<dyn std::error::Error>> {
        let (state, reissuer) = served_with_reissuer(
            crate::health::reissue::OperatorCredentialReissueOutcome::Refused(
                OperatorCredentialReissueRefusal::ConversationUnknown { conversation_id: 7 },
            ),
        );
        let malformed = [
            // A missing input.
            "/operator/credential-reissue?conversation_id=7&participant_id=3",
            // No inputs at all.
            "/operator/credential-reissue",
            // A non-numeric generation.
            "/operator/credential-reissue?conversation_id=7&participant_id=3&\
             expected_current_generation=fourteen",
            // An unknown parameter riding along.
            "/operator/credential-reissue?conversation_id=7&participant_id=3&\
             expected_current_generation=14&force=1",
            // A repeated parameter, where the last one silently winning would be
            // an operator's typo deciding which identity rotates.
            "/operator/credential-reissue?conversation_id=7&conversation_id=8&\
             participant_id=3&expected_current_generation=14",
        ];

        for target in malformed {
            let request = format!("POST {target} HTTP/1.1\r\n\r\n");
            let response = String::from_utf8(response_for_request(request.as_bytes(), &state)?)?;
            assert_status(&response, 400);
        }

        assert!(
            reissuer
                .seen
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .is_empty(),
            "a malformed call must never reach the serialized participant-state point"
        );
        Ok(())
    }

    /// The operation is a POST. A GET of the same target is refused rather than
    /// rotating a credential from a link someone clicked.
    #[test]
    fn the_reissue_route_refuses_every_other_method() -> Result<(), Box<dyn std::error::Error>> {
        let (state, reissuer) = served_with_reissuer(
            crate::health::reissue::OperatorCredentialReissueOutcome::Refused(
                OperatorCredentialReissueRefusal::ConversationUnknown { conversation_id: 7 },
            ),
        );

        for method in ["GET", "PUT", "DELETE"] {
            let request = format!("{method} {REISSUE_TARGET} HTTP/1.1\r\n\r\n");
            let response = String::from_utf8(response_for_request(request.as_bytes(), &state)?)?;
            assert_status(&response, 405);
        }

        assert!(
            reissuer
                .seen
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .is_empty()
        );
        Ok(())
    }

    /// ⛔ The query-string split must not have moved any EXISTING route's
    /// answer. `GET /health?x=1` was a 404 before A7 and stays one: the four
    /// original routes still match the whole request target exactly.
    #[test]
    fn the_reissue_route_did_not_move_any_existing_routes_answer()
    -> Result<(), Box<dyn std::error::Error>> {
        let state = served(SharedReadinessState::default());
        for target in [
            "/health?x=1",
            "/ready?x=1",
            "/metrics?x=1",
            "/unloadable-conversations?x=1",
        ] {
            let request = format!("GET {target} HTTP/1.1\r\n\r\n");
            let response = String::from_utf8(response_for_request(request.as_bytes(), &state)?)?;
            assert_status(&response, 404);
        }
        // POSITIVE CONTROL: the same routes without a query string still serve,
        // so the assertions above measure the query handling and not a broken
        // request line.
        let response = String::from_utf8(response_for_request(
            b"GET /health HTTP/1.1\r\n\r\n",
            &state,
        )?)?;
        assert_status(&response, 200);
        Ok(())
    }

    /// Oracle 8 (W4 leg 2) — on a quiet health listener the blocking accept is
    /// issued exactly once (the parked call) and never again: zero repeated
    /// accepts, zero application wakes after arming, with route behaviour
    /// unchanged (a real request is still served afterwards).
    #[test]
    fn silent_health_listener_has_zero_application_wakes() -> Result<(), Box<dyn std::error::Error>>
    {
        let server = start_health_server(loopback_ephemeral()?, SharedReadinessState::default())?;

        let deadline = Instant::now() + Duration::from_secs(2);
        while server.accept_attempts() < 1 && Instant::now() < deadline {
            thread::sleep(Duration::from_millis(5));
        }
        let armed = server.accept_attempts();
        assert_eq!(
            armed, 1,
            "the blocking accept is issued exactly once when parked"
        );

        thread::sleep(Duration::from_millis(200));
        assert_eq!(
            server.accept_attempts(),
            armed,
            "a silent health listener must not wake or re-accept"
        );
        assert_eq!(
            server.shed_count(),
            0,
            "a silent health listener sheds nothing"
        );

        // Route behaviour unchanged: a real request is still served after silence.
        let response = get(server.local_addr(), "/health")?;
        assert_status(&response, 200);
        let body = json_body(&response)?;
        assert_eq!(body["status"], "healthy");

        server.shutdown()?;
        Ok(())
    }

    /// Oracle 9 (W4 leg 2) — absence proof over this module's production source:
    /// the retired non-blocking flip and its `WouldBlock` + sleep poll must not
    /// appear in the health accept path.
    #[test]
    fn health_accept_source_has_no_wouldblock_sleep_poll() {
        const SOURCE: &str = include_str!("endpoint.rs");
        let production = SOURCE.split("mod tests").next().unwrap_or(SOURCE);
        for forbidden in [
            "set_nonblocking(true)",
            "ErrorKind::WouldBlock",
            "thread::sleep",
        ] {
            assert!(
                !production.contains(forbidden),
                "retired health accept-path source `{forbidden}` reappeared"
            );
        }
    }

    /// Oracle 10 (W4 leg 2) — shutdown interrupts the blocking accept wait at
    /// every race point: before the worker arms, after it parks, concurrent with
    /// a pending connection, and after an accept returns. Each shutdown returns
    /// promptly (no sleep-poll) and joins cleanly (no worker leak); the released
    /// listener refuses further connects (no descriptor leak).
    #[test]
    fn health_shutdown_interrupts_accept_wait() -> Result<(), Box<dyn std::error::Error>> {
        // (a) shutdown immediately after start — possibly before the worker arms.
        let server = start_health_server(loopback_ephemeral()?, SharedReadinessState::default())?;
        let start = Instant::now();
        server.shutdown()?;
        assert!(
            start.elapsed() < Duration::from_secs(2),
            "shutdown before arming must interrupt promptly, not sleep-poll"
        );

        // (b) shutdown after the worker has parked the blocking accept.
        let server = start_health_server(loopback_ephemeral()?, SharedReadinessState::default())?;
        let deadline = Instant::now() + Duration::from_secs(2);
        while server.accept_attempts() < 1 && Instant::now() < deadline {
            thread::sleep(Duration::from_millis(5));
        }
        assert_eq!(
            server.accept_attempts(),
            1,
            "the worker parked before shutdown"
        );
        let parked_addr = server.local_addr();
        let start = Instant::now();
        server.shutdown()?;
        assert!(
            start.elapsed() < Duration::from_secs(2),
            "shutdown of a parked accept must interrupt promptly"
        );
        // No descriptor leak: the released listener refuses further connects.
        assert!(
            TcpStream::connect(parked_addr).is_err(),
            "the listener descriptor was released; further connects are refused"
        );

        // (c) shutdown concurrent with accept readiness: a client is pending.
        // Whether the worker is still parked in `accept` or has already accepted
        // and is blocked reading the silent client, shutdown interrupts promptly
        // (self-connect for the parked case, in-flight stream shutdown for the
        // reading case) — deterministically under the 2s read deadline. The bound
        // is tightened to 500 ms to reflect the TOLD interrupt: it must not
        // approach the read window that the pre-fix bytes deferred to (see
        // `shutdown_interrupts_in_flight_silent_request_read` for the dedicated
        // mid-read regression).
        let server = start_health_server(loopback_ephemeral()?, SharedReadinessState::default())?;
        let deadline = Instant::now() + Duration::from_secs(2);
        while server.accept_attempts() < 1 && Instant::now() < deadline {
            thread::sleep(Duration::from_millis(5));
        }
        let _pending = TcpStream::connect(server.local_addr())?;
        let start = Instant::now();
        server.shutdown()?;
        assert!(
            start.elapsed() < Duration::from_millis(500),
            "shutdown concurrent with a pending accept must interrupt promptly (TOLD), \
             not defer by a request read deadline: elapsed {:?}",
            start.elapsed()
        );

        // (d) shutdown after an accept returns and a request is served.
        let server = start_health_server(loopback_ephemeral()?, SharedReadinessState::default())?;
        let response = get(server.local_addr(), "/health")?;
        assert_status(&response, 200);
        let start = Instant::now();
        server.shutdown()?;
        assert!(
            start.elapsed() < Duration::from_secs(2),
            "shutdown after a served request must interrupt the next parked accept promptly"
        );

        Ok(())
    }

    /// Regression for oracle 10 case (c) (W4 leg 2 — tear-seat BOUNCE): an
    /// in-flight SILENT request must not defer shutdown by its read window. A
    /// client that connects and sends nothing parks the worker inside
    /// `handle_connection`'s blocking read (the admitted 2s slow-client
    /// deadline). Shutdown must interrupt that read directly (TOLD stream
    /// interrupt), NOT wait for the deadline to expire.
    ///
    /// The promptness bound is inherently a timing assertion: it is set to
    /// 500 ms — comfortably under the 2s read deadline (so the pre-fix bytes,
    /// where the self-connect interrupt merely queues behind the blocked read,
    /// red deterministically) and comfortably over scheduler-wakeup noise even
    /// under full-workspace parallel load (so the post-fix stream interrupt
    /// greens deterministically). The worker's entry into the read is observed
    /// via the `requests_entered` counter, not a sleep, so the race is caught
    /// deterministically mid-request.
    #[test]
    fn shutdown_interrupts_in_flight_silent_request_read() -> Result<(), Box<dyn std::error::Error>>
    {
        let server = start_health_server(loopback_ephemeral()?, SharedReadinessState::default())?;

        // Connect but send NOTHING and keep the socket open: the worker accepts
        // and blocks in handle_connection's read on the silent stream. The named
        // binding keeps the client alive (a bare `_` would drop it and end the
        // read early with EOF).
        let _silent = TcpStream::connect(server.local_addr())?;

        // Observe the worker has entered the in-flight request read (counter, not
        // a sleep) so shutdown is fired while it is genuinely blocked on the read.
        let deadline = Instant::now() + Duration::from_secs(2);
        while server.requests_entered() < 1 && Instant::now() < deadline {
            thread::sleep(Duration::from_millis(5));
        }
        assert_eq!(
            server.requests_entered(),
            1,
            "the worker entered the in-flight silent request read"
        );

        let start = Instant::now();
        server.shutdown()?;
        assert!(
            start.elapsed() < Duration::from_millis(500),
            "shutdown must interrupt an in-flight silent request read promptly (TOLD), \
             not defer by the read's admitted 2s deadline: elapsed {:?}",
            start.elapsed()
        );

        Ok(())
    }

    /// Oracle 11 (W4 leg 2, idle-honesty both-sides) — an unrelated served
    /// request grows the BUSY listener's accept-attempt counter while the silent
    /// listener's accept-attempt counter stays FLAT during the workload. The
    /// growing side proves the fixture cannot pass by hiding the workload (a
    /// frozen harness would leave the busy counter flat and fail); the flat side
    /// proves genuine silence rather than a global freeze.
    #[test]
    fn health_idle_grows_unrelated_counters_while_accept_stays_flat()
    -> Result<(), Box<dyn std::error::Error>> {
        let idle = start_health_server(loopback_ephemeral()?, SharedReadinessState::default())?;
        let busy = start_health_server(loopback_ephemeral()?, SharedReadinessState::default())?;

        let deadline = Instant::now() + Duration::from_secs(2);
        while (idle.accept_attempts() < 1 || busy.accept_attempts() < 1)
            && Instant::now() < deadline
        {
            thread::sleep(Duration::from_millis(5));
        }
        let idle_armed = idle.accept_attempts();
        assert_eq!(idle_armed, 1, "the idle listener parks exactly one accept");
        let busy_before = busy.accept_attempts();

        // Unrelated served workload on the BUSY listener: each served request
        // returns the parked accept and re-parks a fresh one, growing its counter.
        for _ in 0..5 {
            let response = get(busy.local_addr(), "/health")?;
            assert_status(&response, 200);
        }
        let deadline = Instant::now() + Duration::from_secs(2);
        while busy.accept_attempts() <= busy_before && Instant::now() < deadline {
            thread::sleep(Duration::from_millis(5));
        }

        assert!(
            busy.accept_attempts() > busy_before,
            "an unrelated served request grows the busy listener's accept counter"
        );
        assert_eq!(
            idle.accept_attempts(),
            idle_armed,
            "the silent listener's accept counter stays flat during the workload"
        );
        assert_eq!(idle.shed_count(), 0, "the silent listener sheds nothing");

        idle.shutdown()?;
        busy.shutdown()?;
        Ok(())
    }
}