mcpls_core/lsp/

client.rs

//! LSP client implementation with async request/response handling.

use std::collections::HashMap;
use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
use std::sync::{Arc, Mutex as StdMutex, PoisonError};

use lsp_types::{ErrorCodes, LspErrorCodes};
use serde::Serialize;
use serde::de::DeserializeOwned;
use serde_json::Value;
use tokio::sync::{Mutex, mpsc, oneshot};
use tokio::task::JoinHandle;
use tokio::time::{Duration, Instant, timeout, timeout_at};
use tracing::{debug, error, trace, warn};

use crate::config::{LanguageId, LspServerConfig, LspSettings, ServerId};
use crate::error::{BackgroundTask, Error, Result};
use crate::lsp::transport::{LspTransport, LspTransportReader};
use crate::lsp::types::{
    InboundMessage, JsonRpcError, JsonRpcNotification, JsonRpcOutcome, JsonRpcReply,
    JsonRpcRequest, JsonRpcResponse, JsonRpcVersion, LspNotification, RequestId,
};
use crate::lsp::{DropCounter, DropLog, Lane, NotificationSink, ServerMessage};
use crate::redaction::{RedactedText, Redactions};
use crate::util::MAX_ERROR_MESSAGE_CALLER_BYTES;

/// Maximum number of retry attempts for server-cancelled requests.
const SERVER_CANCELLED_MAX_RETRIES: u32 = 3;

/// Initial backoff delay for server-cancelled retries (milliseconds).
const SERVER_CANCELLED_INITIAL_DELAY_MS: u64 = 500;

/// Bounded capacity for the channel carrying fully-decoded inbound LSP
/// messages from [`spawn_reader_task`]'s background task to
/// [`LspClient::message_loop_inner`].
///
/// Backpressured (`send().await`, not `try_send`) unlike the best-effort
/// notification/lifecycle lanes: dropping a frame here would desync
/// request/response correlation or silently swallow a server-initiated
/// request. Matches the command channel's capacity -- both lanes carry
/// protocol-critical traffic at a similar cadence.
const READER_CHANNEL_CAPACITY: usize = 100;

/// How long [`LspClient::message_loop`] waits, after aborting the reader
/// task it no longer drains, for that task to actually finish dropping its
/// [`LspTransportReader`] (and the `ChildStdout` it owns).
///
/// `abort()` only requests cancellation -- the task's locals are dropped
/// once the runtime next polls it, not synchronously at the call site.
/// Mirrors `await_lsp_init_handle`'s reasoning in `crate::lib` for the same
/// pattern. Short: the task is either already parked in a cancel-safe
/// `.await` (aborts promptly) or has nothing left to do.
const READER_TASK_ABORT_GRACE: Duration = Duration::from_secs(1);

/// Deadline for a full LSP server shutdown (`shutdown` request, `exit`
/// notification, message-loop stop and child exit). Cleanup after an abort
/// may overrun it by a 50 ms settle.
pub const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);

/// How many undecodable inbound frames in a row the reader tolerates before it
/// treats the connection as broken.
///
/// A single frame that is not a valid message is dropped, so one bad message
/// does not take the server down. A server that emits only garbage never makes
/// progress, though: after this many consecutive undecodable frames the
/// connection is torn down and the server respawned like any other lost
/// connection.
pub const MAX_CONSECUTIVE_UNDECODABLE_FRAMES: u32 = 64;

const _: () = assert!(MAX_CONSECUTIVE_UNDECODABLE_FRAMES > 0);

/// How long [`LspClient::shutdown_until`] waits for an aborted message loop
/// to drop its command receiver before failing the pending requests.
const SHUTDOWN_ABORT_SETTLE: Duration = Duration::from_millis(50);

/// LSP request methods for which a `-32801` (`ContentModified`) error
/// response is safe to retry automatically -- also declared to servers via
/// `general.staleRequestSupport.retryOnContentModified` during initialize
/// (see [`crate::lsp::LspServer`]'s handshake).
///
/// Per the LSP spec, `ContentModified` means the server noticed the document
/// changed while it was computing a response; the (possibly stale) result may
/// still be useful, or the client may choose to cancel the request instead.
/// mcpls chooses to retry, which is safe for a read-only/idempotent request
/// (hover, references, diagnostics, ...): a stale response is simply
/// discarded and superseded by a fresh one.
///
/// Deliberately excludes every method in
/// `crate::bridge::translator::edits` (`textDocument/rename`,
/// `textDocument/formatting`, `textDocument/codeAction`): their result is an
/// edit the MCP caller applies, and `-32801` means the document changed since
/// the request was issued, so a retry at the original position could return
/// an edit for content the caller no longer expects (e.g. renaming a
/// different symbol than the one originally at that position).
///
/// This list only gates `-32801`. `-32802` (`ServerCancelled`) retry is
/// unaffected and keeps retrying unconditionally for every method, as before.
pub const CONTENT_MODIFIED_RETRY_METHODS: &[&str] = &[
    "textDocument/signatureHelp",
    "textDocument/inlayHint",
    "textDocument/completion",
    "textDocument/prepareCallHierarchy",
    "callHierarchy/incomingCalls",
    "callHierarchy/outgoingCalls",
    "textDocument/diagnostic",
    "textDocument/hover",
    "textDocument/definition",
    "textDocument/references",
    "textDocument/implementation",
    "textDocument/typeDefinition",
    "textDocument/documentSymbol",
    "workspace/symbol",
];

/// Upper bound on the effective timeout for completion requests, regardless
/// of `request_timeout_seconds`.
///
/// Completions are latency-sensitive: a completion list that takes longer
/// than this is no longer useful to the caller. This is a deliberate MVP
/// ceiling, not an oversight — completions cannot be configured above this
/// value today. See [`LspClient::completion_timeout`].
const COMPLETION_TIMEOUT_CAP: Duration = Duration::from_secs(10);

/// Upper bound on the effective timeout for a single `codeAction/resolve`
/// request, regardless of `request_timeout_seconds`.
///
/// `handle_code_actions` (`bridge::translator::edits`) resolves up to
/// `MAX_CODE_ACTION_RESOLVES` deferred actions concurrently after the
/// initial `textDocument/codeAction` response; an uncapped per-resolve
/// timeout would let a large `request_timeout_seconds` configuration make
/// one tool call wait far longer than a caller expects for what is meant to
/// be a best-effort follow-up. Mirrors [`COMPLETION_TIMEOUT_CAP`]'s
/// reasoning. See [`LspClient::code_action_resolve_timeout`].
const CODE_ACTION_RESOLVE_TIMEOUT_CAP: Duration = Duration::from_secs(10);

/// Type alias for pending request tracking map.
type PendingRequests = HashMap<RequestId, oneshot::Sender<Result<Value>>>;

/// Spawns the dedicated background task that owns `reader` exclusively and
/// decodes inbound LSP frames in a loop, handing each one to
/// [`LspClient::message_loop_inner`] over the returned channel.
///
/// This is the fix for #451: [`LspTransportReader::receive`] is not
/// cancel-safe, so it must never run as a branch of the `select!` in
/// `message_loop_inner`, which also waits on `command_rx`. Running it here
/// instead, on a task driven only by its own `.await`s, means it can never
/// be cancelled mid-frame.
///
/// The task exits after sending one `Err` (I/O failure or EOF), once
/// `message_loop_inner` drops its end of the channel (e.g. on shutdown), or
/// when the returned [`JoinHandle`] is aborted. Callers must abort it once
/// they stop draining the channel -- see [`LspClient::message_loop`], the
/// only caller -- otherwise it stays parked in a blocking read holding the
/// underlying `ChildStdout` open indefinitely: not itself a correctness bug
/// (`LspServer::is_dead`, lifecycle.rs, checks the child process directly via
/// `try_wait()` and doesn't care whether we still hold its stdout open), but
/// a leaked task and file descriptor for the lifetime of that connection.
fn spawn_reader_task(
    mut reader: LspTransportReader,
) -> (JoinHandle<()>, mpsc::Receiver<Result<InboundMessage>>) {
    let (tx, rx) = mpsc::channel(READER_CHANNEL_CAPACITY);
    let handle = tokio::spawn(async move {
        let mut run = UndecodableRun::default();
        loop {
            let message = reader
                .receive()
                .await
                .and_then(|message| run.observe(message));
            let is_err = message.is_err();
            if tx.send(message).await.is_err() || is_err {
                break;
            }
        }
    });
    (handle, rx)
}

/// The current run of consecutive undecodable inbound frames.
#[derive(Debug, Default)]
struct UndecodableRun {
    in_a_row: u32,
    drops: DropCounter,
}

impl UndecodableRun {
    /// Passes `message` on. A decodable message ends the run; an undecodable
    /// one extends it. Drops are logged at `warn` at most once a minute, with
    /// the number dropped since the last line, and at `debug` otherwise, so a
    /// server that spams garbage cannot flood the log.
    ///
    /// # Errors
    ///
    /// [`Error::LspProtocolError`] once the run reaches
    /// [`MAX_CONSECUTIVE_UNDECODABLE_FRAMES`], so the connection is torn down.
    fn observe(&mut self, message: InboundMessage) -> Result<InboundMessage> {
        self.observe_at(message, std::time::Instant::now())
    }

    fn observe_at(
        &mut self,
        message: InboundMessage,
        now: std::time::Instant,
    ) -> Result<InboundMessage> {
        if !message.is_undecodable() {
            self.in_a_row = 0;
            return Ok(message);
        }
        self.in_a_row = self.in_a_row.saturating_add(1);
        if self.in_a_row >= MAX_CONSECUTIVE_UNDECODABLE_FRAMES {
            return Err(Error::LspProtocolError(RedactedText::fixed(
                "too many consecutive undecodable messages",
            )));
        }
        if let Some(suppressed) = self.drops.record(now) {
            warn!(
                "Dropped an undecodable LSP message ({suppressed} others dropped since the last \
                 warning); further ones are logged at debug for a minute"
            );
        } else {
            debug!(
                "Dropped an undecodable LSP message ({} in a row)",
                self.in_a_row
            );
        }
        Ok(message)
    }
}

/// Why requests still pending on a client are being failed, and so which
/// error they resolve with.
#[derive(Debug, Clone, Default)]
pub enum PendingFailure {
    /// The connection is gone.
    #[default]
    Terminated,
    /// The server was deliberately restarted; the caller may retry.
    Restarted(ServerId),
}

impl PendingFailure {
    fn error(&self) -> Error {
        match self {
            Self::Terminated => Error::ServerTerminated,
            Self::Restarted(server_id) => Error::ServerRestarted {
                server_id: server_id.clone(),
            },
        }
    }
}

/// LSP client with async request/response handling.
///
/// This client manages communication with an LSP server, handling:
/// - Concurrent requests with unique ID tracking
/// - Background message loop for receiving responses
/// - Timeout support for all requests
/// - Graceful shutdown
#[derive(Debug)]
pub struct LspClient {
    /// Configuration for this LSP server.
    config: LspServerConfig,

    /// Current server state.
    state: Arc<Mutex<super::ServerState>>,

    /// Atomic counter for request IDs.
    request_counter: Arc<AtomicI64>,

    /// Command sender for outbound messages.
    command_tx: mpsc::Sender<ClientCommand>,

    /// Requests awaiting a response, shared with the background message loop.
    ///
    /// Exposed here (not just captured by the loop) so [`Self::request`] can
    /// remove its own entry on timeout instead of leaking it, and so a
    /// connection known to be dead can fail its stragglers immediately via
    /// [`Self::fail_pending_requests`] rather than leaving each to discover
    /// that only when its own timeout elapses.
    pending_requests: Arc<Mutex<PendingRequests>>,

    /// The error stragglers in `pending_requests` fail with, shared by every
    /// clone and by the message loop's exit-path drain.
    pending_failure: Arc<StdMutex<PendingFailure>>,

    /// Background receiver task handle.
    receiver_task: Option<JoinHandle<Result<()>>>,

    /// Secrets hidden from text derived from server output.
    redactions: Arc<Redactions>,

    /// Identity of the connection, shared by every clone.
    connection_id: ConnectionId,
}

/// Identity of one client connection, and so of one server process.
///
/// Clones of a client share it; every client built by this process gets a
/// distinct one, so a result tied to a replaced process is told apart from the
/// running one.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ConnectionId(u64);

impl ConnectionId {
    fn next() -> Self {
        static NEXT: AtomicU64 = AtomicU64::new(1);
        Self(NEXT.fetch_add(1, Ordering::Relaxed))
    }
}

impl Clone for LspClient {
    /// Creates a clone that shares the underlying connection.
    ///
    /// The clone does not own the receiver task and cannot perform shutdown.
    /// All clones share the same command channel for sending requests.
    fn clone(&self) -> Self {
        Self {
            config: self.config.clone(),
            state: Arc::clone(&self.state),
            request_counter: Arc::clone(&self.request_counter),
            command_tx: self.command_tx.clone(),
            pending_requests: Arc::clone(&self.pending_requests),
            pending_failure: Arc::clone(&self.pending_failure),
            receiver_task: None,
            redactions: Arc::clone(&self.redactions),
            connection_id: self.connection_id,
        }
    }
}

/// Commands for client control.
enum ClientCommand {
    /// Send a request and wait for response.
    SendRequest { request: JsonRpcRequest },
    /// Send a notification (no response expected).
    SendNotification {
        method: String,
        params: Option<Value>,
    },
    /// Shutdown the client.
    Shutdown,
}

impl LspClient {
    /// Create a new LSP client with the given configuration.
    ///
    /// The client starts in an uninitialized state. Call `initialize()` to
    /// start the server and complete the initialization handshake.
    #[must_use]
    pub fn new(config: LspServerConfig) -> Self {
        // Placeholder channel - the receiver is intentionally dropped since
        // the client starts uninitialized. A real channel is created when
        // `from_transport` or `from_transport_with_notifications` is called.
        let (command_tx, _command_rx) = mpsc::channel(1); // Minimal capacity for placeholder

        Self {
            config,
            state: Arc::new(Mutex::new(super::ServerState::Uninitialized)),
            request_counter: Arc::new(AtomicI64::new(1)),
            command_tx,
            pending_requests: Arc::new(Mutex::new(HashMap::new())),
            pending_failure: Arc::default(),
            receiver_task: None,
            redactions: Arc::default(),
            connection_id: ConnectionId::next(),
        }
    }

    /// Create client from transport (for testing or custom spawning).
    ///
    /// This method initializes the background message loop with the provided transport.
    #[cfg(test)]
    pub(crate) fn from_transport(
        config: LspServerConfig,
        transport: (LspTransport, LspTransportReader),
    ) -> Self {
        let state = Arc::new(Mutex::new(super::ServerState::Initializing));
        let request_counter = Arc::new(AtomicI64::new(1));
        let pending_requests = Arc::new(Mutex::new(HashMap::new()));
        let pending_failure = Arc::<StdMutex<PendingFailure>>::default();

        let (command_tx, command_rx) = mpsc::channel(100);

        let receiver_task = tokio::spawn(Self::message_loop(
            transport,
            command_rx,
            Arc::clone(&pending_requests),
            Arc::clone(&pending_failure),
            None,
            None,
            Arc::default(),
            Self::shared_settings(&config),
        ));

        Self {
            config,
            state,
            request_counter,
            command_tx,
            pending_requests,
            pending_failure,
            receiver_task: Some(receiver_task),
            redactions: Arc::default(),
            connection_id: ConnectionId::next(),
        }
    }

    /// Create client from transport with notification forwarding.
    ///
    /// Notifications are parsed and split across two lanes (P3): diagnostics/
    /// log/showMessage go through `notification_tx`; `$/progress` `begin`/
    /// `end` frames and unrecognized notifications (`LspNotification::Other`,
    /// which carries rust-analyzer's `experimental/serverStatus`) go through
    /// `lifecycle_tx` instead. A `$/progress` `report` frame is never
    /// enqueued on either lane -- see [`Self::message_loop_inner`].
    pub(crate) fn from_transport_with_notifications(
        config: LspServerConfig,
        transport: (LspTransport, LspTransportReader),
        notification_tx: NotificationSink,
        lifecycle_tx: mpsc::Sender<LspNotification>,
        redactions: Arc<Redactions>,
    ) -> Self {
        let state = Arc::new(Mutex::new(super::ServerState::Initializing));
        let request_counter = Arc::new(AtomicI64::new(1));
        let pending_requests = Arc::new(Mutex::new(HashMap::new()));
        let pending_failure = Arc::<StdMutex<PendingFailure>>::default();

        let (command_tx, command_rx) = mpsc::channel(100);

        let receiver_task = tokio::spawn(Self::message_loop(
            transport,
            command_rx,
            Arc::clone(&pending_requests),
            Arc::clone(&pending_failure),
            Some(notification_tx),
            Some(lifecycle_tx),
            Arc::clone(&redactions),
            Self::shared_settings(&config),
        ));

        Self {
            config,
            state,
            request_counter,
            command_tx,
            pending_requests,
            pending_failure,
            receiver_task: Some(receiver_task),
            redactions,
            connection_id: ConnectionId::next(),
        }
    }

    /// The configured settings, shared with the message loop.
    fn shared_settings(config: &LspServerConfig) -> Option<Arc<LspSettings>> {
        config.settings.clone().map(Arc::new)
    }

    /// Identity of the connection this client and its clones talk over.
    pub(crate) const fn connection_id(&self) -> ConnectionId {
        self.connection_id
    }

    /// Secrets this client hides from text derived from server output.
    pub(crate) const fn redactions(&self) -> &Arc<Redactions> {
        &self.redactions
    }

    /// Get the language ID for this client.
    #[must_use]
    pub const fn language_id(&self) -> &LanguageId {
        &self.config.language_id
    }

    /// Get the current server state.
    pub async fn state(&self) -> super::ServerState {
        *self.state.lock().await
    }

    /// The timeout applied to a single LSP request attempt, derived from
    /// [`LspServerConfig::request_timeout_seconds`].
    ///
    /// This bounds one attempt, not a whole tool call: [`Self::request`]
    /// retries up to `SERVER_CANCELLED_MAX_RETRIES` (3) additional times on a
    /// `-32802` (`ServerCancelled`) or `-32801` (`ContentModified`) response,
    /// sharing one attempt budget between the two codes, so the worst-case
    /// latency for a single tool call is `4 * request_timeout() + 3.5s` (the
    /// sum of the retry backoff delays).
    ///
    /// Exception: `get_code_actions` (`bridge::translator::edits`) can add a
    /// second, concurrent round of requests on top of this bound -- up to
    /// `MAX_CODE_ACTION_RESOLVES` `codeAction/resolve` calls, each retried
    /// under the same rules but bounded by [`Self::code_action_resolve_timeout`]
    /// rather than this timeout. Since those run concurrently with each
    /// other (not with the initial `textDocument/codeAction` request), the
    /// worst case for that one tool call is
    /// `(4 * request_timeout() + 3.5s) + (4 * code_action_resolve_timeout() + 3.5s)`,
    /// not a multiple scaling with the number of resolved actions.
    ///
    /// The value is a [`TimeoutSecs`], so it is already within 1 second to
    /// [`MAX_TIMEOUT_SECONDS`] and needs no clamping here.
    ///
    /// [`TimeoutSecs`]: crate::config::TimeoutSecs
    /// [`MAX_TIMEOUT_SECONDS`]: crate::config::MAX_TIMEOUT_SECONDS
    ///
    /// # Examples
    ///
    /// ```
    /// use std::time::Duration;
    /// use mcpls_core::config::{LspServerConfig, TimeoutSecs};
    /// use mcpls_core::lsp::LspClient;
    ///
    /// let mut config = LspServerConfig::rust_analyzer();
    /// config.request_timeout_seconds = TimeoutSecs::new(45).unwrap();
    /// let client = LspClient::new(config);
    ///
    /// assert_eq!(client.request_timeout(), Duration::from_secs(45));
    /// ```
    #[must_use]
    pub const fn request_timeout(&self) -> Duration {
        self.config.request_timeout_seconds.as_duration()
    }

    /// The timeout applied to completion (`textDocument/completion`) requests.
    ///
    /// Equal to [`Self::request_timeout`], capped at 10 seconds. Completions
    /// cannot be configured above this cap by any
    /// value of `request_timeout_seconds` — if that proves insufficient in
    /// practice, the fix is a dedicated `completion_timeout_seconds` field,
    /// not raising this cap.
    ///
    /// # Examples
    ///
    /// ```
    /// use std::time::Duration;
    /// use mcpls_core::config::{LspServerConfig, TimeoutSecs};
    /// use mcpls_core::lsp::LspClient;
    ///
    /// let mut config = LspServerConfig::rust_analyzer();
    /// config.request_timeout_seconds = TimeoutSecs::new(300).unwrap();
    /// let client = LspClient::new(config);
    ///
    /// // Capped at 10s even though request_timeout_seconds is 300.
    /// assert_eq!(client.completion_timeout(), Duration::from_secs(10));
    /// assert!(client.completion_timeout() <= client.request_timeout());
    /// ```
    #[must_use]
    pub fn completion_timeout(&self) -> Duration {
        self.request_timeout().min(COMPLETION_TIMEOUT_CAP)
    }

    /// The timeout applied to a single `codeAction/resolve` request.
    ///
    /// Equal to [`Self::request_timeout`], capped at 10 seconds.
    ///
    /// # Examples
    ///
    /// ```
    /// use std::time::Duration;
    /// use mcpls_core::config::{LspServerConfig, TimeoutSecs};
    /// use mcpls_core::lsp::LspClient;
    ///
    /// let mut config = LspServerConfig::rust_analyzer();
    /// config.request_timeout_seconds = TimeoutSecs::new(300).unwrap();
    /// let client = LspClient::new(config);
    ///
    /// // Capped at 10s even though request_timeout_seconds is 300.
    /// assert_eq!(client.code_action_resolve_timeout(), Duration::from_secs(10));
    /// assert!(client.code_action_resolve_timeout() <= client.request_timeout());
    /// ```
    #[must_use]
    pub fn code_action_resolve_timeout(&self) -> Duration {
        self.request_timeout().min(CODE_ACTION_RESOLVE_TIMEOUT_CAP)
    }

    /// Registers before enqueuing the send command (not before the message
    /// loop's own transport read), cleaning up the entry if the send fails.
    async fn register_and_send_request(
        &self,
        request: JsonRpcRequest,
        response_tx: oneshot::Sender<Result<Value>>,
    ) -> Result<()> {
        let id = request.id.clone();

        self.pending_requests
            .lock()
            .await
            .insert(id.clone(), response_tx);

        if self
            .command_tx
            .send(ClientCommand::SendRequest { request })
            .await
            .is_err()
        {
            self.pending_requests.lock().await.remove(&id);
            return Err(Error::ServerTerminated);
        }

        Ok(())
    }

    /// Send request and wait for response with timeout.
    ///
    /// Automatically retries up to 3 times when the server returns error code
    /// -32802 (`ServerCancelled`, any method) or -32801 (`ContentModified`,
    /// only for methods in `CONTENT_MODIFIED_RETRY_METHODS`) -- gated by
    /// `data.retriggerRequest` when present -- using exponential backoff
    /// starting at 500 ms. Both codes share the same attempt budget: a
    /// request that hits -32801 then -32802 does not get 8 attempts, only 4.
    ///
    /// This method owns the severity of LSP error-response logging (#392):
    /// a transient error about to be retried logs at `warn!`, while `error!`
    /// is reserved for an error actually surfaced to the caller -- retry
    /// exhaustion or a non-retryable code/method combination. This is
    /// deliberately not decided in `message_loop_inner`, which parses the
    /// response before knowing whether a retry will follow.
    ///
    /// # Type Parameters
    ///
    /// * `P` - The type of the request parameters (must be serializable)
    /// * `R` - The type of the response result (must be deserializable)
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Server has shut down
    /// - Request times out
    /// - Response cannot be deserialized
    /// - LSP server returns an error
    pub async fn request<P, R>(
        &self,
        method: &str,
        params: P,
        timeout_duration: Duration,
    ) -> Result<R>
    where
        P: Serialize,
        R: DeserializeOwned,
    {
        Box::pin(self.request_inner(method, params, timeout_duration))
            .await
            .map_err(UnclassifiedError::surface)
    }

    async fn request_inner<P, R>(
        &self,
        method: &str,
        params: P,
        timeout_duration: Duration,
    ) -> std::result::Result<R, UnclassifiedError>
    where
        P: Serialize,
        R: DeserializeOwned,
    {
        let params_value = Self::omit_null_params(
            serde_json::to_value(params).map_err(|e| UnclassifiedError::logged(e.into()))?,
        );
        let mut delay_ms = SERVER_CANCELLED_INITIAL_DELAY_MS;

        for attempt in 0..=SERVER_CANCELLED_MAX_RETRIES {
            if attempt > 0 {
                debug!(
                    "Retrying {} (attempt {}/{}), backoff={}ms",
                    method, attempt, SERVER_CANCELLED_MAX_RETRIES, delay_ms
                );
                tokio::time::sleep(Duration::from_millis(delay_ms)).await;
                delay_ms = delay_ms.saturating_mul(2);
            }

            let id = RequestId::Number(self.request_counter.fetch_add(1, Ordering::SeqCst));
            let (response_tx, response_rx) = oneshot::channel();
            let request = JsonRpcRequest {
                jsonrpc: JsonRpcVersion,
                id: id.clone(),
                method: method.to_string(),
                params: params_value.clone(),
            };

            debug!("Sending request: {} (id={:?})", method, id);

            self.register_and_send_request(request, response_tx)
                .await
                .map_err(UnclassifiedError::logged)?;

            let outcome = match timeout(timeout_duration, response_rx).await {
                Ok(received) => {
                    received.map_err(|_| UnclassifiedError::logged(Error::ServerTerminated))?
                }
                Err(_elapsed) => {
                    // The response may still arrive after this point (the
                    // server is just slow, not dead), but nothing will ever
                    // read it again -- drop the now-orphaned entry instead of
                    // leaking it in `pending_requests` forever.
                    self.pending_requests.lock().await.remove(&id);
                    return Err(UnclassifiedError::logged(Error::Timeout(timeout_duration)));
                }
            };

            match outcome {
                Ok(result_value) => {
                    return serde_json::from_value(result_value).map_err(|e| {
                        UnclassifiedError::logged(
                            self.redactions.protocol_error(format_args!(
                                "Failed to deserialize response: {e}"
                            )),
                        )
                    });
                }
                Err(Error::LspServerError {
                    code,
                    message,
                    data,
                }) if (LspErrorCodes::from(code) == LspErrorCodes::ServerCancelled
                    || (LspErrorCodes::from(code) == LspErrorCodes::ContentModified
                        && CONTENT_MODIFIED_RETRY_METHODS.contains(&method)))
                    && Self::should_retrigger(data.as_ref()) =>
                {
                    if attempt == SERVER_CANCELLED_MAX_RETRIES {
                        // Same "LSP error response: ..." prefix as the
                        // non-retryable branch below -- a log-grep alert on
                        // that prefix must catch every error actually
                        // surfaced to the caller, retry exhaustion included.
                        error!(
                            "LSP error response: {} (code {}) on '{}' (id={:?}), retries exhausted",
                            Self::truncate_error_message_for_log(&message),
                            code,
                            method,
                            id
                        );
                        return Err(UnclassifiedError::logged(Error::LspServerError {
                            code,
                            message,
                            data,
                        }));
                    }
                    warn!(
                        "LSP error response: {} (code {}) on '{}' (id={:?}), will retry",
                        Self::truncate_error_message_for_log(&message),
                        code,
                        method,
                        id
                    );
                    // continue loop for next attempt
                }
                Err(Error::LspServerError {
                    code,
                    message,
                    data,
                }) => {
                    return Err(UnclassifiedError::server_response(
                        ServerErrorResponse::new(code, message, data),
                        method,
                        id,
                    ));
                }
                Err(e) => return Err(UnclassifiedError::logged(e)),
            }
        }

        Err(UnclassifiedError::logged(Error::ServerTerminated))
    }

    /// Like [`Self::request_typed`], but a server error response is returned
    /// unlogged, wrapped in an [`UnclassifiedError`] so the caller decides
    /// whether it is an error surfaced to the user or an expected outcome.
    ///
    /// # Errors
    ///
    /// See [`Self::request`].
    pub(crate) async fn request_typed_classified<R>(
        &self,
        params: R::Params,
        timeout_duration: Duration,
    ) -> std::result::Result<R::Result, UnclassifiedError>
    where
        R: lsp_types::Request,
    {
        Box::pin(self.request_inner(R::METHOD.as_str(), params, timeout_duration)).await
    }

    /// Send a typed LSP request, deriving both the method string and the
    /// result type from `R`'s [`lsp_types::Request`] implementation so they
    /// cannot drift independently -- unlike [`Self::request`], which takes
    /// the method string and result type as two unchecked, hand-picked
    /// values.
    ///
    /// # Errors
    ///
    /// See [`Self::request`].
    pub async fn request_typed<R>(
        &self,
        params: R::Params,
        timeout_duration: Duration,
    ) -> Result<R::Result>
    where
        R: lsp_types::Request,
    {
        self.request(R::METHOD.as_str(), params, timeout_duration)
            .await
    }

    /// Returns true when the error data from a retryable "please retry" LSP
    /// response (`ServerCancelled` -32802 or `ContentModified` -32801)
    /// indicates a retry should happen.
    ///
    /// The LSP spec defines `data.retriggerRequest` only for diagnostic
    /// requests' `ServerCancelled` responses (`DiagnosticServerCancellationData`)
    /// -- it is not part of the general `ServerCancelled` or `ContentModified`
    /// contract for other methods. mcpls checks this field whenever it is
    /// present regardless of method (harmless for non-diagnostic methods,
    /// since a compliant server won't send it there) and defaults to
    /// retrying when the field is absent, since retrying is the more useful
    /// default for a request that would otherwise surface as a hard error to
    /// the MCP caller.
    fn should_retrigger(data: Option<&Value>) -> bool {
        data.is_none_or(|v| {
            v.get("retriggerRequest")
                .and_then(Value::as_bool)
                .unwrap_or(true)
        })
    }

    /// Whether the message loop this client owns has stopped (it returned
    /// or panicked), leaving the connection permanently dead.
    ///
    /// Always `false` for a clone or a placeholder client: only the owning
    /// client holds the loop handle.
    pub(crate) fn is_message_loop_finished(&self) -> bool {
        self.receiver_task
            .as_ref()
            .is_some_and(JoinHandle::is_finished)
    }

    /// Fail every request still parked in `pending_requests` with
    /// `Error::ServerTerminated`, instead of leaving each to discover a dead
    /// connection only when its own timeout elapses.
    ///
    /// Intended for a client that is about to be discarded -- e.g.
    /// superseded by a respawned replacement for the same server -- so
    /// callers still waiting on it unblock immediately.
    pub(crate) async fn fail_pending_requests(&self) {
        Self::drain_and_fail_pending(&self.pending_requests, &self.pending_failure).await;
    }

    /// Make every request still pending, or failed later by the loop's exit
    /// drain, resolve with [`Error::ServerRestarted`] instead of
    /// [`Error::ServerTerminated`].
    pub(crate) fn mark_restarted(&self, server_id: ServerId) {
        *self
            .pending_failure
            .lock()
            .unwrap_or_else(PoisonError::into_inner) = PendingFailure::Restarted(server_id);
    }

    /// Drains `pending`, resolving each remaining sender to
    /// `Err(Error::ServerTerminated)`. Shared by [`Self::fail_pending_requests`]
    /// and [`Self::message_loop`]'s exit-path cleanup (#458).
    async fn drain_and_fail_pending(
        pending: &Arc<Mutex<PendingRequests>>,
        failure: &StdMutex<PendingFailure>,
    ) {
        for (_, sender) in pending.lock().await.drain() {
            let error = failure
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .error();
            let _ = sender.send(Err(error));
        }
    }

    /// Send notification (fire-and-forget, no response expected).
    ///
    /// # Errors
    ///
    /// Returns an error if the server has shut down.
    pub async fn notify<P>(&self, method: &str, params: P) -> Result<()>
    where
        P: Serialize,
    {
        let params_value = Self::omit_null_params(serde_json::to_value(params)?);

        debug!("Sending notification: {}", method);

        self.command_tx
            .send(ClientCommand::SendNotification {
                method: method.to_string(),
                params: params_value,
            })
            .await
            .map_err(|_| Error::ServerTerminated)?;

        Ok(())
    }

    /// Send a typed LSP notification, deriving the method string from `N`'s
    /// [`lsp_types::Notification`] implementation so it cannot drift from the
    /// params type -- the notification counterpart to [`Self::request_typed`].
    ///
    /// # Errors
    ///
    /// See [`Self::notify`].
    pub async fn notify_typed<N>(&self, params: N::Params) -> Result<()>
    where
        N: lsp_types::Notification,
    {
        self.notify(N::METHOD.as_str(), params).await
    }

    /// Shutdown client gracefully, with a deadline of [`SHUTDOWN_TIMEOUT`].
    ///
    /// This sends a shutdown command to the background task and waits for it to complete.
    /// Only the client that owns the message loop (the one held by
    /// [`LspServer`](crate::lsp::LspServer)) can abort a wedged loop; calling this on a
    /// clone still stops the shared loop for every clone but cannot abort it.
    ///
    /// # Errors
    ///
    /// Returns [`Error::ShutdownTimeout`] if the loop did not stop in time (it is then
    /// aborted and every pending request fails with [`Error::ServerTerminated`]), or
    /// another error if the background task failed.
    #[allow(
        clippy::arithmetic_side_effects,
        reason = "SHUTDOWN_TIMEOUT is a small constant"
    )]
    pub async fn shutdown(mut self) -> Result<()> {
        self.shutdown_until(Instant::now() + SHUTDOWN_TIMEOUT).await
    }

    /// As [`Self::shutdown`], with a caller-chosen absolute deadline.
    pub(crate) async fn shutdown_until(&mut self, deadline: Instant) -> Result<()> {
        debug!("Shutting down LSP client");

        let mut task = self.receiver_task.take();
        let command_tx = &self.command_tx;
        let outcome = timeout_at(deadline, async {
            let _ = command_tx.send(ClientCommand::Shutdown).await;
            match task.as_mut() {
                Some(task) => match task.await {
                    Ok(result) => result,
                    Err(source) => Err(Error::TaskFailed {
                        task: BackgroundTask::LspReceiver,
                        source,
                    }),
                },
                None => Ok(()),
            }
        })
        .await;

        let result = outcome.unwrap_or(Err(Error::ShutdownTimeout));
        if matches!(result, Err(Error::ShutdownTimeout)) {
            if let Some(task) = task {
                task.abort();
                // Lets the aborted loop drop `command_rx`, so a racing
                // request cannot enqueue after the drain below (#458).
                let _ = timeout(SHUTDOWN_ABORT_SETTLE, task).await;
            }
            Self::drain_and_fail_pending(&self.pending_requests, &self.pending_failure).await;
        }

        *self.state.lock().await = super::ServerState::Shutdown;

        result
    }

    /// Background task: handle message I/O.
    ///
    /// This task runs in the background, handling:
    /// - Outbound requests and notifications
    /// - Inbound responses and server notifications
    /// - Matching responses to pending requests
    #[allow(
        clippy::too_many_arguments,
        reason = "one-to-one with the spawned loop's channels and shared state"
    )]
    async fn message_loop(
        transport: (LspTransport, LspTransportReader),
        mut command_rx: mpsc::Receiver<ClientCommand>,
        pending_requests: Arc<Mutex<PendingRequests>>,
        pending_failure: Arc<StdMutex<PendingFailure>>,
        notification_tx: Option<NotificationSink>,
        lifecycle_tx: Option<mpsc::Sender<LspNotification>>,
        redactions: Arc<Redactions>,
        settings: Option<Arc<LspSettings>>,
    ) -> Result<()> {
        debug!("Message loop started");
        let (mut transport, reader) = transport;
        let (reader_handle, mut msg_rx) = spawn_reader_task(reader);
        let result = {
            // Aborts the reader task the instant `message_loop_inner` returns, even via `?` (#451).
            let _abort_reader_on_drop = crate::util::AbortOnDrop(&reader_handle);
            Self::message_loop_inner(
                &mut transport,
                &mut msg_rx,
                &mut command_rx,
                &pending_requests,
                notification_tx.as_ref(),
                lifecycle_tx.as_ref(),
                &redactions,
                settings.as_deref(),
            )
            .await
        };
        let _ = timeout(READER_TASK_ABORT_GRACE, reader_handle).await;
        // Dropped before the drain, not just at fn-exit: otherwise a
        // `register_and_send_request` racing the drain could still insert
        // into `pending_requests` and succeed its `command_tx.send(..)`,
        // leaving that entry unfailed until its own timeout (#458 S1). Once
        // this is gone, every later `send` fails and the caller cleans up
        // its own entry.
        drop(command_rx);
        // Runs after `message_loop_inner` returns, so a response the Shutdown
        // drain (#451) already resolved is gone from the map by now -- only
        // requests with no answer get failed here (#458).
        Self::drain_and_fail_pending(&pending_requests, &pending_failure).await;
        if let Err(ref e) = result {
            error!("Message loop exiting with error: {}", e);
        } else {
            debug!("Message loop exiting normally");
        }
        result
    }

    /// Maps a `null` params value to an omitted `params` field. LSP methods
    /// with `params: void` (`shutdown`, `exit`) must go out without the key:
    /// tsgo rejects `"params": null` with `-32602 expected empty, got: null`.
    fn omit_null_params(params: Value) -> Option<Value> {
        if params.is_null() { None } else { Some(params) }
    }

    /// Truncate an LSP server's error message for the `tracing::error!` log
    /// line, bounding it to at most [`crate::util::MAX_LOG_STRING_BYTES`]
    /// bytes (the full formatted string is slightly longer).
    ///
    /// Log-line use only -- the message forwarded to the MCP caller in
    /// [`Error::LspServerError`] is truncated separately, to the larger
    /// [`MAX_ERROR_MESSAGE_CALLER_BYTES`] (#313).
    fn truncate_error_message_for_log(message: &str) -> String {
        crate::util::truncate_str(message, crate::util::MAX_LOG_STRING_BYTES)
    }

    /// Which of the two notification lanes (P3) `notification` belongs on,
    /// or `None` if it must be dropped before reaching either.
    ///
    /// `notification_tx` (returned as `"notification"`) carries diagnostics/
    /// log/showMessage; `lifecycle_tx` (returned as `"lifecycle"`) carries
    /// `$/progress` `begin`/`end` frames and `Other` (which carries e.g.
    /// rust-analyzer's `experimental/serverStatus`) -- splitting them means
    /// a high-volume diagnostics publisher can never starve out a
    /// low-volume readiness signal, or vice versa. The lane name is
    /// returned alongside the sender purely for the drop-warning log at the
    /// call site (Fix 6) -- it plays no role in routing.
    ///
    /// A `$/progress` `report` frame -- the high-volume case a
    /// `report`-per-package emitter like gopls can produce -- is classified
    /// via [`crate::lsp::types::ProgressKind::from_value`] (shared with
    /// `bridge::indexing::IndexingTracker::observe_progress` so the two
    /// can't drift out of sync on which `kind`s are recognized -- Fix 9)
    /// and dropped before it ever reaches either channel (S3). An
    /// *unparseable* `$/progress` notification
    /// (e.g. missing the mandatory `token` field) falls back to
    /// `LspNotification::Other { method: "$/progress", .. }` at parse time
    /// (`LspNotification::parse`) rather than `Progress`, so it must be
    /// dropped here too by matching on `method` -- otherwise a server whose
    /// `report` payloads fail to deserialize could bypass the `kind`-based
    /// filter above entirely by sending malformed frames (security LOW /
    /// M1).
    /// Whether `notification` is handed to the log/lifecycle lanes, as opposed
    /// to the diagnostics mailbox or the floor.
    fn reaches_a_lane(
        notification: &LspNotification,
        has_sink: bool,
        lifecycle_tx: Option<&mpsc::Sender<LspNotification>>,
    ) -> bool {
        match notification {
            LspNotification::PublishDiagnostics(_) => false,
            LspNotification::LogMessage(_) | LspNotification::ShowMessage(_) => has_sink,
            other => Self::lifecycle_lane(other, lifecycle_tx).is_some(),
        }
    }

    fn lifecycle_lane<'a>(
        notification: &LspNotification,
        lifecycle_tx: Option<&'a mpsc::Sender<LspNotification>>,
    ) -> Option<&'a mpsc::Sender<LspNotification>> {
        match notification {
            LspNotification::PublishDiagnostics(_)
            | LspNotification::LogMessage(_)
            | LspNotification::ShowMessage(_) => None,
            LspNotification::Progress(params) => {
                crate::lsp::types::ProgressKind::from_value(&params.value).and(lifecycle_tx)
            }
            LspNotification::Other { method, .. } if method.as_ref() == "$/progress" => None,
            LspNotification::Other { .. } => lifecycle_tx,
        }
    }

    #[allow(
        clippy::too_many_lines,
        clippy::too_many_arguments,
        reason = "one select loop over the transport, channels and shared state"
    )]
    async fn message_loop_inner(
        transport: &mut LspTransport,
        msg_rx: &mut mpsc::Receiver<Result<InboundMessage>>,
        command_rx: &mut mpsc::Receiver<ClientCommand>,
        pending_requests: &Arc<Mutex<PendingRequests>>,
        notification_tx: Option<&NotificationSink>,
        lifecycle_tx: Option<&mpsc::Sender<LspNotification>>,
        redactions: &Redactions,
        settings: Option<&LspSettings>,
    ) -> Result<()> {
        loop {
            tokio::select! {
                Some(command) = command_rx.recv() => {
                    match command {
                        ClientCommand::SendRequest { request } => {
                            let value = serde_json::to_value(&request)?;
                            transport.send(&value).await?;
                        }
                        ClientCommand::SendNotification { method, params } => {
                            let notification = serde_json::to_value(JsonRpcNotification {
                                jsonrpc: JsonRpcVersion,
                                method,
                                params,
                            })?;
                            transport.send(&notification).await?;
                        }
                        ClientCommand::Shutdown => {
                            debug!("Client shutdown requested");
                            // The reader task can run ahead of us (#451): drain whatever it
                            // already queued instead of dropping it, or a response to a
                            // concurrent caller's in-flight request would be silently lost.
                            while let Ok(message) = msg_rx.try_recv() {
                                match message {
                                    Ok(m) => {
                                        Self::handle_inbound_message(
                                            transport,
                                            m,
                                            pending_requests,
                                            notification_tx,
                                            lifecycle_tx,
                                            redactions,
                                            settings,
                                        )
                                        .await?;
                                    }
                                    Err(e) => {
                                        error!("Transport receive error while draining on shutdown: {}", e);
                                        break;
                                    }
                                }
                            }
                            break;
                        }
                    }
                }

                // Cancel-safe, unlike the `transport.receive()` this replaces (#451): reading itself happens off this `select!`.
                message = msg_rx.recv() => {
                    let message = match message {
                        Some(Ok(m)) => m,
                        Some(Err(e)) => {
                            error!("Transport receive error: {}", e);
                            return Err(e);
                        }
                        None => {
                            // Reader task only exits after sending an `Err`, so this means it panicked.
                            error!("Transport receive error: reader task ended unexpectedly");
                            return Err(Error::ServerTerminated);
                        }
                    };
                    Self::handle_inbound_message(
                        transport,
                        message,
                        pending_requests,
                        notification_tx,
                        lifecycle_tx,
                        redactions,
                        settings,
                    )
                    .await?;
                }
            }
        }

        Ok(())
    }

    /// Hands `response` to the request waiting for it, if any.
    async fn complete_pending(
        pending_requests: &Arc<Mutex<PendingRequests>>,
        response: JsonRpcResponse,
        redactions: &Redactions,
    ) {
        trace!("Received response: id={:?}", response.id);

        let sender = pending_requests.lock().await.remove(&response.id);

        if let Some(sender) = sender {
            match response.outcome {
                JsonRpcOutcome::Error(error) => {
                    // Deliberately not logged at `error!` here: this fires
                    // for every attempt, before `LspClient::request`'s retry
                    // loop knows whether the error is transient and about to
                    // be retried (-32802, or -32801 for an allowlisted
                    // method). Logging unconditionally at this point would
                    // emit a spurious ERROR line for errors that are retried
                    // and succeed. `request` logs at `warn!` on retry and
                    // `error!` once the error is actually surfaced to the
                    // caller (retry exhaustion or a non-retryable error);
                    // the response id is already traced above.
                    let message = redactions.apply(&error.message);
                    trace!(
                        "LSP error response: {} (code {})",
                        Self::truncate_error_message_for_log(&message),
                        error.code
                    );
                    // Truncated separately from the log line, to the larger
                    // MAX_ERROR_MESSAGE_CALLER_BYTES -- the raw message is
                    // unbounded and attacker-influenceable (#313), but a
                    // log-line-sized cut would also clip legitimate long
                    // errors before the model ever sees them (S2).
                    let caller_message =
                        crate::util::truncate_str(&message, MAX_ERROR_MESSAGE_CALLER_BYTES);
                    let _ = sender.send(Err(Error::LspServerError {
                        code: error.code,
                        message: caller_message,
                        data: error.data,
                    }));
                }
                JsonRpcOutcome::Result(result) => {
                    let _ = sender.send(Ok(result));
                }
            }
        } else {
            warn!(
                "Received response for unknown request ID: {:?}",
                response.id
            );
        }
    }

    /// Processes one fully-decoded inbound LSP message: resolves a matching
    /// pending request, answers a server-initiated request, or forwards a
    /// notification to its lane. Shared by `message_loop_inner`'s normal
    /// `msg_rx.recv()` branch and its `Shutdown` drain, so a message handled
    /// during either path behaves identically.
    async fn handle_inbound_message(
        transport: &mut LspTransport,
        message: InboundMessage,
        pending_requests: &Arc<Mutex<PendingRequests>>,
        notification_tx: Option<&NotificationSink>,
        lifecycle_tx: Option<&mpsc::Sender<LspNotification>>,
        redactions: &Redactions,
        settings: Option<&LspSettings>,
    ) -> Result<()> {
        match message {
            InboundMessage::Response(response) => {
                Self::complete_pending(pending_requests, response, redactions).await;
            }
            InboundMessage::UndecodableResponse { id } => {
                warn!("Failed the request answered by an undecodable response: id={id:?}");
                let sender = pending_requests.lock().await.remove(&id);
                if let Some(sender) = sender {
                    let _ = sender.send(Err(Error::LspProtocolError(RedactedText::fixed(
                        "undecodable response",
                    ))));
                }
            }
            InboundMessage::UndecodableRequest { id } => {
                debug!("Answering an undecodable server request with an error: id={id:?}");
                let reply = JsonRpcReply::new(
                    id,
                    JsonRpcOutcome::Error(JsonRpcError::new(
                        ErrorCodes::InvalidRequest,
                        "Invalid Request: the request could not be decoded",
                    )),
                );
                transport.send(&serde_json::to_value(&reply)?).await?;
            }
            InboundMessage::UndecodableNotification | InboundMessage::UndecodableFrame => {}
            InboundMessage::Request(request) => {
                debug!(
                    "Received server request: {} (id={:?})",
                    request.method, request.id
                );
                let reply = Self::server_request_reply(request, settings);
                let value = serde_json::to_value(&reply)?;
                transport.send(&value).await?;
            }
            InboundMessage::Notification(notification) => {
                debug!("Received notification: {}", notification.method);

                // Parse notification into typed variant
                let mut typed = LspNotification::parse(&notification.method, notification.params);
                Self::redact_notification(&mut typed, redactions);

                if tracing::enabled!(tracing::Level::TRACE)
                    && Self::reaches_a_lane(&typed, notification_tx.is_some(), lifecycle_tx)
                {
                    trace!(
                        "Forwarding notification: {}",
                        redactions.apply(&format!("{typed:?}"))
                    );
                }

                // Diagnostics have their own delivery path: the mailbox coalesces
                // per file and records what it cannot hold, so a burst is never
                // dropped unrecorded.
                let delivered = match typed {
                    LspNotification::PublishDiagnostics(params) => {
                        if let Some(sink) = notification_tx {
                            debug!(
                                "Forwarding diagnostics for {}: {} items",
                                params.uri.as_ref(),
                                params.diagnostics.len()
                            );
                            sink.publishes().publish(params);
                        }
                        return Ok(());
                    }
                    LspNotification::LogMessage(params) => notification_tx.map(|sink| {
                        (
                            Lane::Notification,
                            sink.messages().try_send(ServerMessage::Log(params)).is_ok(),
                        )
                    }),
                    LspNotification::ShowMessage(params) => notification_tx.map(|sink| {
                        (
                            Lane::Notification,
                            sink.messages()
                                .try_send(ServerMessage::Show(params))
                                .is_ok(),
                        )
                    }),
                    other => Self::lifecycle_lane(&other, lifecycle_tx)
                        .map(|tx| (Lane::Lifecycle, tx.try_send(other).is_ok())),
                };

                if let Some((lane, false)) = delivered {
                    DropLog::of(lane).record(&notification.method);
                }
            }
        }

        Ok(())
    }

    /// Redacts the human-readable text a server controls before it reaches
    /// the notification cache, from where MCP tools and the diagnostics
    /// resource return it.
    pub(crate) fn redact_notification(notification: &mut LspNotification, redactions: &Redactions) {
        if redactions.is_empty() {
            return;
        }
        match notification {
            LspNotification::LogMessage(params) => {
                redactions.redact_in_place(&mut params.message);
            }
            LspNotification::ShowMessage(params) => {
                redactions.redact_in_place(&mut params.message);
            }
            LspNotification::PublishDiagnostics(params) => {
                for diagnostic in &mut params.diagnostics {
                    redactions.redact_diagnostic(diagnostic);
                }
            }
            LspNotification::Progress(params) => redactions.redact_progress(&mut params.value),
            LspNotification::Other { .. } => {}
        }
    }

    fn server_request_reply(
        request: JsonRpcRequest,
        settings: Option<&LspSettings>,
    ) -> JsonRpcReply {
        let outcome =
            Self::server_request_result(&request.method, request.params.as_ref(), settings);
        JsonRpcReply::new(request.id, outcome.into())
    }

    fn server_request_result(
        method: &str,
        params: Option<&Value>,
        settings: Option<&LspSettings>,
    ) -> std::result::Result<Value, JsonRpcError> {
        match method {
            "client/registerCapability"
            | "client/unregisterCapability"
            | "workspace/workspaceFolders"
            | "workspace/diagnostic/refresh"
            | "workspace/semanticTokens/refresh"
            | "workspace/inlayHint/refresh"
            | "workspace/codeLens/refresh"
            | "window/showMessageRequest"
            | "window/workDoneProgress/create" => Ok(Value::Null),
            "workspace/configuration" => Self::workspace_configuration_result(params, settings),
            "workspace/applyEdit" => Ok(serde_json::json!({ "applied": false })),
            _ => Err(JsonRpcError::new(
                ErrorCodes::MethodNotFound,
                format!("Unhandled server request: {method}"),
            )),
        }
    }

    /// Answers `workspace/configuration`: one entry per requested item.
    ///
    /// Without settings every item gets `Null`, whatever its shape and even
    /// when `items` is absent (the reply before settings existed). With
    /// settings an item gets its section (the whole object when `section` is
    /// absent, `Null` when it is unknown, not a string, or the item is not an
    /// object); only a missing
    /// `items` list is answered with `-32602`. `scopeUri` is never read.
    fn workspace_configuration_result(
        params: Option<&Value>,
        settings: Option<&LspSettings>,
    ) -> std::result::Result<Value, JsonRpcError> {
        let items = params
            .and_then(|params| params.get("items"))
            .and_then(Value::as_array);
        let Some(settings) = settings else {
            return Ok(Value::Array(vec![Value::Null; items.map_or(0, Vec::len)]));
        };
        let items = items.ok_or_else(|| {
            JsonRpcError::new(
                ErrorCodes::InvalidParams,
                "Invalid params for workspace/configuration",
            )
        })?;
        let sections = items
            .iter()
            .map(|item| match ConfigurationItem::parse(item) {
                ConfigurationItem::Whole => settings.section(None),
                ConfigurationItem::Section(section) => settings.section(Some(section)),
                ConfigurationItem::Uninterpretable => Value::Null,
            })
            .collect();
        Ok(Value::Array(sections))
    }
}

/// Logs a server error response that reached the caller, under the prefix a
/// log-grep alert watches for.
fn log_surfaced_error(method: &str, id: &RequestId, code: i32, message: &str) {
    error!(
        "LSP error response: {} (code {}) on '{}' (id={:?})",
        LspClient::truncate_error_message_for_log(message),
        code,
        method,
        id
    );
}

/// A failed request whose severity is still undecided.
///
/// A server error response is logged only once the caller says whether it is
/// surfaced as an error ([`Self::surface`], ERROR) or turned into an expected
/// outcome ([`Self::handled`], DEBUG). There is deliberately no conversion
/// into [`Error`], so `?` does not compile and the choice cannot be skipped.
#[derive(Debug)]
#[must_use = "call `surface()` or `handled()` to log and unwrap the error"]
pub struct UnclassifiedError(Box<Unclassified>);

/// An error response from the server, kept as the [`Error::LspServerError`] it
/// becomes so a caller can classify it, and constructible from nothing else.
#[derive(Debug)]
struct ServerErrorResponse {
    code: i32,
    message: String,
    data: Option<Value>,
    /// The [`Error`] built from the fields above, once a caller asked to see it.
    error: std::sync::OnceLock<Error>,
}

impl ServerErrorResponse {
    const fn new(code: i32, message: String, data: Option<Value>) -> Self {
        Self {
            code,
            message,
            data,
            error: std::sync::OnceLock::new(),
        }
    }

    /// The code and message of the response.
    const fn code_and_message(&self) -> (i32, &str) {
        (self.code, self.message.as_str())
    }

    /// The response as the error it becomes, for classification.
    fn as_error(&self) -> &Error {
        self.error.get_or_init(|| Error::LspServerError {
            code: self.code,
            message: self.message.clone(),
            data: self.data.clone(),
        })
    }

    fn into_error(self) -> Error {
        self.error.into_inner().unwrap_or(Error::LspServerError {
            code: self.code,
            message: self.message,
            data: self.data,
        })
    }
}

/// The two states of an [`UnclassifiedError`]: a failure already logged where
/// it happened, and a server error response nobody has logged yet.
#[derive(Debug)]
enum Unclassified {
    Logged(Error),
    ServerResponse {
        response: ServerErrorResponse,
        method: String,
        id: RequestId,
    },
}

impl Unclassified {
    fn error(&self) -> &Error {
        match self {
            Self::Logged(error) => error,
            Self::ServerResponse { response, .. } => response.as_error(),
        }
    }

    /// Hands over the error, first passing an unlogged server error response
    /// to `report` as (method, id, code, message).
    fn finish(self, report: impl FnOnce(&str, &RequestId, i32, &str)) -> Error {
        match self {
            Self::Logged(error) => error,
            Self::ServerResponse {
                response,
                method,
                id,
            } => {
                let (code, message) = response.code_and_message();
                report(&method, &id, code, message);
                response.into_error()
            }
        }
    }
}

fn log_handled_error(method: &str, id: &RequestId, code: i32, message: &str) {
    debug!(
        "LSP error response handled by the caller: {} (code {}) on '{}' (id={:?})",
        LspClient::truncate_error_message_for_log(message),
        code,
        method,
        id
    );
}

impl UnclassifiedError {
    fn logged(error: Error) -> Self {
        Self(Box::new(Unclassified::Logged(error)))
    }

    fn server_response(response: ServerErrorResponse, method: &str, id: RequestId) -> Self {
        Self(Box::new(Unclassified::ServerResponse {
            response,
            method: method.to_owned(),
            id,
        }))
    }

    /// Hands over the state, leaving a logged placeholder behind so that the
    /// `Drop` below has nothing left to report.
    fn take(mut self) -> Unclassified {
        std::mem::replace(&mut *self.0, Unclassified::Logged(Error::ServerTerminated))
    }

    /// The wrapped error, for classification before logging.
    pub(crate) fn error(&self) -> &Error {
        self.0.error()
    }

    /// The error returned to the caller as a failure; a server error response
    /// is logged at ERROR.
    pub(crate) fn surface(self) -> Error {
        self.take().finish(log_surfaced_error)
    }

    /// The error of an expected outcome; a server error response is logged at
    /// DEBUG.
    pub(crate) fn handled(self) -> Error {
        self.take().finish(log_handled_error)
    }
}

/// A dropped, unclassified server error response is logged at ERROR, so
/// `.ok()`, `let _ =` or an ignoring match arm cannot hide it.
impl Drop for UnclassifiedError {
    fn drop(&mut self) {
        let inner = std::mem::replace(&mut *self.0, Unclassified::Logged(Error::ServerTerminated));
        drop(inner.finish(log_surfaced_error));
    }
}

/// How one `workspace/configuration` item selects part of the settings.
enum ConfigurationItem<'a> {
    Whole,
    Section(&'a str),
    Uninterpretable,
}

impl<'a> ConfigurationItem<'a> {
    fn parse(item: &'a Value) -> Self {
        let Value::Object(object) = item else {
            return Self::Uninterpretable;
        };
        match object.get("section") {
            None | Some(Value::Null) => Self::Whole,
            Some(Value::String(section)) => Self::Section(section),
            Some(_) => Self::Uninterpretable,
        }
    }
}

#[cfg(test)]
mod tests {
    use std::assert_matches;

    use super::*;
    use crate::config::TimeoutSecs;
    use crate::lsp::types::JsonRpcResponse;

    #[test]
    fn test_unclassified_server_response_always_reports_code_and_message() {
        let unclassified = UnclassifiedError::server_response(
            ServerErrorResponse::new(-32001, "boom".to_owned(), None),
            "textDocument/hover",
            RequestId::Number(3),
        );
        let mut reported = None;

        let error = unclassified.take().finish(|method, id, code, message| {
            reported = Some((method.to_owned(), id.clone(), code, message.to_owned()));
        });

        assert_eq!(
            reported,
            Some((
                "textDocument/hover".to_owned(),
                RequestId::Number(3),
                -32001,
                "boom".to_owned()
            ))
        );
        assert_matches!(error, Error::LspServerError { code: -32001, .. });
    }

    /// The `trace!` of a forwarded notification must not print a configured
    /// secret that the server echoed into an unrecognized notification.
    #[tokio::test]
    async fn test_forwarded_notification_trace_is_redacted() {
        use tracing_subscriber::prelude::*;

        use crate::test_lsp::CapturedLogs;

        let redactions =
            Redactions::new([("API_TOKEN".to_owned(), "SuperSecretValue123".to_owned())]);
        let logs = CapturedLogs::default();
        let _guard = tracing::subscriber::set_default(
            tracing_subscriber::registry()
                .with(tracing_subscriber::filter::LevelFilter::TRACE)
                .with(logs.clone()),
        );
        let (mut transport, _reader) = LspTransport::new(tokio::io::sink(), tokio::io::empty());
        let (lifecycle_tx, mut lifecycle_rx) = mpsc::channel(4);
        let pending = Arc::new(Mutex::new(HashMap::new()));
        let notification = InboundMessage::Notification(JsonRpcNotification {
            jsonrpc: JsonRpcVersion,
            method: "custom/echo".to_string(),
            params: Some(serde_json::json!({"env": "SuperSecretValue123"})),
        });

        LspClient::handle_inbound_message(
            &mut transport,
            notification,
            &pending,
            None,
            Some(&lifecycle_tx),
            &redactions,
            None,
        )
        .await
        .unwrap();

        assert!(lifecycle_rx.recv().await.is_some());
        let output = logs.messages().join("\n");
        assert!(output.contains("Forwarding notification"), "{output}");
        assert!(!output.contains("SuperSecretValue123"), "{output}");
    }

    fn secret_diagnostics_params() -> Value {
        let range = serde_json::json!({
            "start": {"line": 0, "character": 0},
            "end": {"line": 0, "character": 1}
        });
        serde_json::json!({
            "uri": "file:///w/main.rs",
            "diagnostics": [{
                "range": range,
                "message": "leak SuperSecretValue123",
                "data": {"hint": ["SuperSecretValue123"]}
            }]
        })
    }

    /// #583: the reader task redacts a pushed diagnostic before it reaches the
    /// notification lane, from where the cache and the MCP tools read it.
    #[tokio::test]
    async fn test_reader_task_forwards_pushed_diagnostics_redacted() {
        use crate::test_lsp::{fake_lsp_client_with_redactions, write_notification};

        let redactions =
            Redactions::new([("API_TOKEN".to_owned(), "SuperSecretValue123".to_owned())]);
        let (_client, mut server, mut lanes) = fake_lsp_client_with_redactions(redactions);

        write_notification(
            &mut server.read_half_stdin,
            "textDocument/publishDiagnostics",
            secret_diagnostics_params(),
        )
        .await;

        let forwarded = timeout(Duration::from_secs(2), lanes.publishes.recv())
            .await
            .unwrap()
            .unwrap();
        let shown = format!("{forwarded:?}");
        assert!(shown.contains("[redacted:API_TOKEN]"), "{shown}");
        assert!(!shown.contains("SuperSecretValue123"), "{shown}");
    }

    /// With nothing to hide the notification is forwarded as received.
    #[test]
    fn test_redact_notification_leaves_text_alone_without_secrets() {
        let mut notification = LspNotification::parse(
            "textDocument/publishDiagnostics",
            Some(secret_diagnostics_params()),
        );
        let before = format!("{notification:?}");

        LspClient::redact_notification(&mut notification, &Redactions::default());
        assert_eq!(format!("{notification:?}"), before);

        let unrelated = Redactions::new([("API_TOKEN".to_owned(), "NotInTheText99".to_owned())]);
        LspClient::redact_notification(&mut notification, &unrelated);
        assert_eq!(format!("{notification:?}"), before);

        let matching =
            Redactions::new([("API_TOKEN".to_owned(), "SuperSecretValue123".to_owned())]);
        LspClient::redact_notification(&mut notification, &matching);
        assert!(!format!("{notification:?}").contains("SuperSecretValue123"));
    }

    #[test]
    fn test_request_id_generation() {
        let counter = AtomicI64::new(1);

        let id1 = counter.fetch_add(1, Ordering::SeqCst);
        let id2 = counter.fetch_add(1, Ordering::SeqCst);
        let id3 = counter.fetch_add(1, Ordering::SeqCst);

        assert_eq!(id1, 1);
        assert_eq!(id2, 2);
        assert_eq!(id3, 3);
    }

    #[test]
    fn test_client_creation() {
        let config = LspServerConfig::rust_analyzer();

        let client = LspClient::new(config);
        assert_eq!(client.language_id(), "rust");
    }

    #[test]
    fn test_client_clone() {
        let config = LspServerConfig::rust_analyzer();
        let client = LspClient::new(config);

        #[allow(clippy::redundant_clone, reason = "the test exercises `Clone` itself")]
        let cloned = client.clone();
        assert_eq!(cloned.language_id(), "rust");

        assert!(
            cloned.receiver_task.is_none(),
            "Cloned client should not own receiver task"
        );
    }

    #[test]
    fn test_request_timeout_and_completion_timeout_at_default() {
        let config = LspServerConfig::rust_analyzer();
        let client = LspClient::new(config);

        assert_eq!(client.request_timeout(), Duration::from_secs(30));
        assert_eq!(client.completion_timeout(), Duration::from_secs(10));
    }

    #[test]
    fn test_completion_timeout_clamps_to_ten_seconds() {
        for secs in [1, 2, 3, 30, 300] {
            let mut config = LspServerConfig::rust_analyzer();
            config.request_timeout_seconds = TimeoutSecs::new(secs).unwrap();
            let client = LspClient::new(config);

            assert_eq!(
                client.completion_timeout(),
                Duration::from_secs(secs.min(10)),
                "request_timeout_seconds={secs}"
            );
            assert!(client.completion_timeout() <= client.request_timeout());
        }
    }

    #[test]
    fn test_code_action_resolve_timeout_clamps_to_ten_seconds() {
        for secs in [1, 2, 3, 30, 300] {
            let mut config = LspServerConfig::rust_analyzer();
            config.request_timeout_seconds = TimeoutSecs::new(secs).unwrap();
            let client = LspClient::new(config);

            assert_eq!(
                client.code_action_resolve_timeout(),
                Duration::from_secs(secs.min(10)),
                "request_timeout_seconds={secs}"
            );
            assert!(client.code_action_resolve_timeout() <= client.request_timeout());
        }
    }

    #[test]
    fn test_request_timeout_independent_per_server() {
        let mut config_a = LspServerConfig::rust_analyzer();
        config_a.request_timeout_seconds = TimeoutSecs::new(5).unwrap();
        let mut config_b = LspServerConfig::pyright();
        config_b.request_timeout_seconds = TimeoutSecs::new(15).unwrap();

        let client_a = LspClient::new(config_a);
        let client_b = LspClient::new(config_b);

        assert_eq!(client_a.request_timeout(), Duration::from_secs(5));
        assert_eq!(client_b.request_timeout(), Duration::from_secs(15));
    }

    #[test]
    fn test_register_capability_request_is_acknowledged() {
        let request = JsonRpcRequest {
            jsonrpc: JsonRpcVersion,
            id: RequestId::String("ts1".to_string()),
            method: "client/registerCapability".to_string(),
            params: Some(serde_json::json!({ "registrations": [] })),
        };

        let reply = reply_json(request);

        assert_eq!(reply["id"], serde_json::json!("ts1"));
        assert_eq!(reply["jsonrpc"], serde_json::json!("2.0"));
        assert_eq!(reply["result"], Value::Null);
        assert!(reply.get("error").is_none());
    }

    #[test]
    fn test_workspace_configuration_request_returns_null_per_item_without_settings() {
        let result = LspClient::workspace_configuration_result(
            Some(&serde_json::json!({
                "items": [{ "section": "typescript" }, { "section": "editor" }]
            })),
            None,
        );

        assert_eq!(result.unwrap(), serde_json::json!([null, null]));
    }

    fn sample_settings() -> LspSettings {
        serde_json::from_value(serde_json::json!({
            "python.analysis.typeCheckingMode": "strict",
            "gopls": { "ui.semanticTokens": true }
        }))
        .unwrap()
    }

    #[test]
    fn test_workspace_configuration_serves_settings_sections() {
        let settings = sample_settings();
        let result = LspClient::workspace_configuration_result(
            Some(&serde_json::json!({
                "items": [
                    { "section": "python.analysis" },
                    { "section": "gopls", "scopeUri": "file:///ignored" },
                    { "section": "missing" },
                    {}
                ]
            })),
            Some(&settings),
        )
        .unwrap();

        assert_eq!(
            result,
            serde_json::json!([
                { "typeCheckingMode": "strict" },
                { "ui.semanticTokens": true },
                null,
                settings.to_value()
            ])
        );
    }

    #[test]
    fn test_workspace_configuration_non_object_item_gets_null() {
        let settings = sample_settings();
        let result = LspClient::workspace_configuration_result(
            Some(&serde_json::json!({
                "items": [
                    { "section": "python.analysis" },
                    { "section": "missing.x" },
                    {},
                    { "section": 5 },
                    "x",
                    null,
                    7
                ]
            })),
            Some(&settings),
        )
        .unwrap();

        assert_eq!(
            result,
            serde_json::json!([
                { "typeCheckingMode": "strict" },
                null,
                settings.to_value(),
                null,
                null,
                null,
                null
            ])
        );
    }

    #[tokio::test]
    async fn test_message_loop_answers_workspace_configuration_from_settings() {
        use tokio::io::BufReader;

        use crate::test_lsp::{fake_lsp_client_with_config, read_framed_message, write_request};

        let mut config = LspServerConfig::rust_analyzer();
        config.settings = Some(sample_settings());
        let (_client, mut server) = fake_lsp_client_with_config(config);

        write_request(
            &mut server.read_half_stdin,
            &serde_json::json!(7),
            "workspace/configuration",
            serde_json::json!({ "items": [{ "section": "python.analysis", "scopeUri": 1 }] }),
        )
        .await;
        let mut reader = BufReader::new(&mut server.write_stdout);
        let response = read_framed_message(&mut reader).await;

        assert_eq!(response["id"], 7);
        assert_eq!(
            response["result"],
            serde_json::json!([{ "typeCheckingMode": "strict" }])
        );
    }

    #[test]
    fn test_workspace_configuration_is_lenient_about_item_shape() {
        let settings = sample_settings();
        let params = serde_json::json!({
            "items": [
                { "section": 5 },
                { "section": null },
                3,
                { "section": "gopls", "scopeUri": 42 }
            ]
        });

        let with = LspClient::workspace_configuration_result(Some(&params), Some(&settings));
        let without = LspClient::workspace_configuration_result(Some(&params), None);

        assert_eq!(
            with.unwrap(),
            serde_json::json!([
                null,
                settings.to_value(),
                null,
                { "ui.semanticTokens": true }
            ])
        );
        assert_eq!(
            without.unwrap(),
            serde_json::json!([null, null, null, null])
        );
    }

    #[test]
    fn test_workspace_configuration_without_settings_matches_the_old_reply_for_bad_params() {
        for params in [
            None,
            Some(serde_json::json!({})),
            Some(serde_json::json!({ "items": "x" })),
        ] {
            let result = LspClient::workspace_configuration_result(params.as_ref(), None);
            assert_eq!(result.unwrap(), serde_json::json!([]));
        }
    }

    #[test]
    fn test_workspace_configuration_with_settings_rejects_missing_items() {
        for params in [None, Some(serde_json::json!({ "items": "nope" }))] {
            let error = LspClient::workspace_configuration_result(
                params.as_ref(),
                Some(&sample_settings()),
            )
            .unwrap_err();
            assert_eq!(error.code, -32602);
        }
    }

    /// P1: without this, no spec-compliant LSP server may ever initiate
    /// `$/progress` at all (per LSP 3.17, a server needs a successful
    /// `window/workDoneProgress/create` response before it may report
    /// server-initiated progress for a token).
    #[test]
    fn test_work_done_progress_create_request_is_acknowledged() {
        let request = JsonRpcRequest {
            jsonrpc: JsonRpcVersion,
            id: RequestId::String("wdp1".to_string()),
            method: "window/workDoneProgress/create".to_string(),
            params: Some(serde_json::json!({ "token": "indexing" })),
        };

        let reply = reply_json(request);

        assert_eq!(reply["result"], Value::Null);
        assert!(reply.get("error").is_none());
    }

    /// S3 (Fix 7): a `report`-kind `$/progress` frame must never be
    /// enqueued on either notification lane -- the actual mechanism
    /// protecting against a `report`-per-package emitter like gopls
    /// overrunning the bounded lifecycle channel.
    #[test]
    fn test_report_progress_frame_reaches_neither_lane() {
        let (lifecycle_tx, _lifecycle_rx) = mpsc::channel(8);

        let report = LspNotification::Progress(lsp_types::ProgressParams {
            token: lsp_types::ProgressToken::Int(1),
            value: serde_json::json!({ "kind": "report", "percentage": 50 }),
        });

        let destination = LspClient::lifecycle_lane(&report, Some(&lifecycle_tx));

        assert!(
            destination.is_none(),
            "a report-kind frame must be dropped before reaching either lane"
        );
    }

    /// Fix 3 / M1: an unparseable `$/progress` notification (e.g. missing
    /// the mandatory `token` field) falls back to `LspNotification::Other {
    /// method: "$/progress", .. }` at parse time -- this must be dropped
    /// the same as a well-formed `report` frame, not routed to the
    /// lifecycle lane, or a server whose payloads fail to deserialize could
    /// bypass the kind-based filter entirely.
    #[test]
    fn test_malformed_progress_other_reaches_neither_lane() {
        let (lifecycle_tx, _lifecycle_rx) = mpsc::channel(8);

        let malformed = LspNotification::Other {
            method: std::borrow::Cow::Borrowed("$/progress"),
            params: None,
        };

        let destination = LspClient::lifecycle_lane(&malformed, Some(&lifecycle_tx));

        assert!(
            destination.is_none(),
            "a malformed $/progress frame must be dropped, not routed to the lifecycle lane"
        );
    }

    /// P3 sanity check alongside the two tests above: a `begin` frame and a
    /// genuine `Other` notification (e.g. rust-analyzer's
    /// `experimental/serverStatus`) must still reach the lifecycle lane --
    /// the report/malformed filters must not have overcorrected.
    #[test]
    fn test_begin_and_other_notifications_reach_lifecycle_lane() {
        let (lifecycle_tx, _lifecycle_rx) = mpsc::channel(8);

        let begin = LspNotification::Progress(lsp_types::ProgressParams {
            token: lsp_types::ProgressToken::Int(1),
            value: serde_json::json!({ "kind": "begin", "title": "Indexing" }),
        });
        assert!(LspClient::lifecycle_lane(&begin, Some(&lifecycle_tx)).is_some());

        let server_status = LspNotification::Other {
            method: std::borrow::Cow::Borrowed("experimental/serverStatus"),
            params: Some(serde_json::json!({ "quiescent": false })),
        };
        assert!(LspClient::lifecycle_lane(&server_status, Some(&lifecycle_tx)).is_some());
    }

    #[test]
    fn test_unknown_server_request_returns_method_not_found() {
        let request = JsonRpcRequest {
            jsonrpc: JsonRpcVersion,
            id: RequestId::String("unknown-1".to_string()),
            method: "custom/request".to_string(),
            params: None,
        };

        let reply = reply_json(request);

        assert!(reply.get("result").is_none());
        assert_eq!(reply["error"]["code"], serde_json::json!(-32601));
        assert_eq!(
            reply["error"]["message"],
            serde_json::json!("Unhandled server request: custom/request")
        );
    }

    fn reply_json(request: JsonRpcRequest) -> Value {
        serde_json::to_value(LspClient::server_request_reply(request, None)).unwrap()
    }

    #[tokio::test]
    async fn test_null_response_handling() {
        use crate::lsp::types::{JsonRpcResponse, RequestId};

        let pending_requests: Arc<Mutex<PendingRequests>> = Arc::new(Mutex::new(HashMap::new()));

        let (response_tx, response_rx) = oneshot::channel::<Result<Value>>();

        pending_requests
            .lock()
            .await
            .insert(RequestId::Number(1), response_tx);

        let null_response = JsonRpcResponse {
            jsonrpc: JsonRpcVersion,
            id: RequestId::Number(1),
            outcome: JsonRpcOutcome::Result(Value::Null),
        };

        let sender = pending_requests.lock().await.remove(&null_response.id);
        if let Some(sender) = sender {
            let _ = sender.send(Ok(Value::Null));
        }

        let timeout_result =
            tokio::time::timeout(tokio::time::Duration::from_millis(100), response_rx).await;

        assert!(timeout_result.is_ok(), "Should not timeout");

        let channel_result = timeout_result.unwrap();
        assert!(
            channel_result.is_ok(),
            "Channel should not be closed: {:?}",
            channel_result.err()
        );

        let response = channel_result.unwrap();
        assert!(
            response.is_ok(),
            "Should receive Ok(Value::Null), not Err: {:?}",
            response.err()
        );

        let value = response.unwrap();
        assert_eq!(value, Value::Null, "Should receive Value::Null");
    }

    #[tokio::test]
    async fn test_error_response_handling() {
        use crate::lsp::types::{JsonRpcError, JsonRpcResponse, RequestId};

        let pending_requests: Arc<Mutex<PendingRequests>> = Arc::new(Mutex::new(HashMap::new()));
        let (response_tx, response_rx) = oneshot::channel::<Result<Value>>();

        pending_requests
            .lock()
            .await
            .insert(RequestId::Number(1), response_tx);

        let error_response = JsonRpcResponse {
            jsonrpc: JsonRpcVersion,
            id: RequestId::Number(1),
            outcome: JsonRpcOutcome::Error(JsonRpcError {
                code: -32601,
                message: "Method not found".to_string(),
                data: None,
            }),
        };

        let sender = pending_requests.lock().await.remove(&error_response.id);
        if let Some(sender) = sender
            && let JsonRpcOutcome::Error(error) = error_response.outcome
        {
            let _ = sender.send(Err(Error::LspServerError {
                code: error.code,
                message: error.message,
                data: error.data,
            }));
        }

        let result = response_rx.await.unwrap();
        assert!(result.is_err(), "Should receive error");

        if let Err(Error::LspServerError { code, message, .. }) = result {
            assert_eq!(code, -32601);
            assert_eq!(message, "Method not found");
        } else {
            panic!("Expected LspServerError");
        }
    }

    #[tokio::test]
    async fn test_unknown_request_id() {
        use crate::lsp::types::{JsonRpcResponse, RequestId};

        let pending_requests: Arc<Mutex<PendingRequests>> = Arc::new(Mutex::new(HashMap::new()));

        let response = JsonRpcResponse {
            jsonrpc: JsonRpcVersion,
            id: RequestId::Number(999),
            outcome: JsonRpcOutcome::Result(Value::Null),
        };

        let sender = pending_requests.lock().await.remove(&response.id);
        assert!(sender.is_none(), "Should not find sender for unknown ID");
    }

    #[test]
    fn test_truncate_error_message_for_log_handles_multibyte_boundary() {
        // 199 ASCII bytes followed by a 3-byte UTF-8 char ('€') straddles the byte-200 cut.
        let message = format!("{}€{}", "x".repeat(199), "y".repeat(50));

        let truncated = LspClient::truncate_error_message_for_log(&message);

        // Cutting before the multi-byte char keeps the message valid UTF-8 (no panic) and
        // pins the payload to 199 bytes, not 200.
        assert_eq!(truncated, format!("{}... (truncated)", "x".repeat(199)));
    }

    #[test]
    fn test_truncate_error_message_for_log_no_truncation_at_or_below_limit() {
        let exact = "x".repeat(200);
        assert_eq!(LspClient::truncate_error_message_for_log(&exact), exact);
        assert_eq!(LspClient::truncate_error_message_for_log(""), "");
    }

    #[test]
    fn test_truncate_error_message_for_log_truncates_just_above_limit() {
        let message = "x".repeat(201);
        assert_eq!(
            LspClient::truncate_error_message_for_log(&message),
            format!("{}... (truncated)", "x".repeat(200))
        );
    }

    #[test]
    fn test_truncate_error_message_for_log_handles_wide_char_at_limit() {
        // A 4-byte emoji run straddling every possible alignment near the byte-200 boundary.
        let message = format!("{}{}", "x".repeat(197), "🦀".repeat(10));

        let truncated = LspClient::truncate_error_message_for_log(&message);

        assert_eq!(truncated, format!("{}... (truncated)", "x".repeat(197)));
    }

    #[tokio::test]
    async fn test_concurrent_request_ids() {
        let counter = Arc::new(AtomicI64::new(1));

        let counter1 = Arc::clone(&counter);
        let counter2 = Arc::clone(&counter);
        let counter3 = Arc::clone(&counter);

        let handles = vec![
            tokio::spawn(async move { counter1.fetch_add(1, Ordering::SeqCst) }),
            tokio::spawn(async move { counter2.fetch_add(1, Ordering::SeqCst) }),
            tokio::spawn(async move { counter3.fetch_add(1, Ordering::SeqCst) }),
        ];

        let mut ids = Vec::new();
        for handle in handles {
            ids.push(handle.await.unwrap());
        }

        ids.sort_unstable();
        assert_eq!(ids, vec![1, 2, 3], "IDs should be unique and sequential");
    }

    #[test]
    fn test_jsonrpc_version_constant() {
        assert_eq!(
            serde_json::to_value(JsonRpcVersion).unwrap(),
            serde_json::json!("2.0")
        );
    }

    /// #239 regression: a request that times out must remove its own entry
    /// from `pending_requests` instead of leaking it. `sleep` is used as the
    /// "server": it never writes anything to stdout, so no response can ever
    /// arrive and the request is guaranteed to time out rather than race a
    /// real answer.
    ///
    /// Unix-only: spawns a real `sleep` subprocess, which is unavailable on
    /// the Windows CI runner.
    #[cfg(unix)]
    #[tokio::test]
    async fn test_request_timeout_removes_pending_entry() {
        let mut child = tokio::process::Command::new("sleep")
            .arg("2")
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .kill_on_drop(true)
            .spawn()
            .unwrap();
        let stdin = child.stdin.take().unwrap();
        let stdout = child.stdout.take().unwrap();

        let transport = LspTransport::new(stdin, stdout);
        let client = LspClient::from_transport(LspServerConfig::rust_analyzer(), transport);

        let result: Result<Value> = client
            .request(
                "textDocument/hover",
                serde_json::json!({}),
                Duration::from_millis(50),
            )
            .await;

        assert_matches!(result, Err(Error::Timeout(_)), "got {result:?}");
        assert!(
            client.pending_requests.lock().await.is_empty(),
            "timed-out request must not remain in pending_requests"
        );
    }

    /// #249 continuation: a client about to be discarded (e.g. superseded by
    /// a respawned replacement) must fail every still-pending request
    /// immediately rather than leaving callers to wait out their timeout.
    #[tokio::test]
    async fn test_fail_pending_requests_resolves_all_as_server_terminated() {
        let pending_requests: Arc<Mutex<PendingRequests>> = Arc::new(Mutex::new(HashMap::new()));
        let (command_tx, _command_rx) = mpsc::channel(1);

        let client = LspClient {
            config: LspServerConfig::rust_analyzer(),
            state: Arc::new(Mutex::new(super::super::ServerState::Ready)),
            request_counter: Arc::new(AtomicI64::new(1)),
            command_tx,
            pending_requests: Arc::clone(&pending_requests),
            pending_failure: Arc::default(),
            receiver_task: None,
            redactions: Arc::default(),
            connection_id: ConnectionId::next(),
        };

        let (tx1, rx1) = oneshot::channel::<Result<Value>>();
        let (tx2, rx2) = oneshot::channel::<Result<Value>>();
        pending_requests
            .lock()
            .await
            .insert(RequestId::Number(1), tx1);
        pending_requests
            .lock()
            .await
            .insert(RequestId::Number(2), tx2);

        client.fail_pending_requests().await;

        assert!(pending_requests.lock().await.is_empty());
        assert_matches!(rx1.await.unwrap(), Err(Error::ServerTerminated));
        assert_matches!(rx2.await.unwrap(), Err(Error::ServerTerminated));
    }

    /// Pins where a request is cancelled (#518): once the request is on the
    /// wire, a timeout or a dropped caller must leave the client able to
    /// serve the next request, with no stale reply crossing over to it. The
    /// paused clock makes the timeout deterministic.
    mod timeout_cancellation {
        use tokio::io::BufReader;
        use tokio::task::JoinHandle;

        use super::*;
        use crate::test_lsp::{fake_lsp_client, read_framed_message, write_response};

        const TIMEOUT: Duration = Duration::from_secs(5);

        fn spawn_hover(client: &LspClient) -> JoinHandle<Result<Value>> {
            let client = client.clone();
            tokio::spawn(async move {
                client
                    .request("textDocument/hover", serde_json::json!({}), TIMEOUT)
                    .await
            })
        }

        #[tokio::test(start_paused = true)]
        async fn test_timeout_while_awaiting_reply_clears_pending_and_discards_late_reply() {
            let (client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);

            let first = spawn_hover(&client);
            let first_request = read_framed_message(&mut reader).await;

            tokio::time::advance(TIMEOUT + Duration::from_secs(1)).await;
            assert_matches!(
                first.await.unwrap(),
                Err(Error::Timeout(limit)) if limit == TIMEOUT,
                "request awaiting its reply must time out"
            );
            assert!(
                client.pending_requests.lock().await.is_empty(),
                "timed-out request must not remain in pending_requests"
            );

            write_response(
                &mut server.read_half_stdin,
                &first_request["id"],
                serde_json::json!("stale"),
            )
            .await;

            let second = spawn_hover(&client);
            let second_request = read_framed_message(&mut reader).await;
            assert_ne!(second_request["id"], first_request["id"]);
            write_response(
                &mut server.read_half_stdin,
                &second_request["id"],
                serde_json::json!("fresh"),
            )
            .await;

            assert_eq!(second.await.unwrap().unwrap(), serde_json::json!("fresh"));
            assert!(client.pending_requests.lock().await.is_empty());
        }

        #[tokio::test]
        async fn test_undecodable_response_fails_only_its_request() {
            use crate::test_lsp::write_raw_frame;

            let (client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);

            let first = spawn_hover(&client);
            let first_request = read_framed_message(&mut reader).await;
            let deep = format!("{}1{}", "{\"parent\":".repeat(200), "}".repeat(200));
            write_raw_frame(
                &mut server.read_half_stdin,
                &format!(
                    r#"{{"jsonrpc":"2.0","id":{},"result":{deep}}}"#,
                    first_request["id"]
                ),
            )
            .await;

            assert_matches!(
                first.await.unwrap(),
                Err(Error::LspProtocolError(message)) if message.as_str() == "undecodable response"
            );

            let second = spawn_hover(&client);
            let second_request = read_framed_message(&mut reader).await;
            write_response(
                &mut server.read_half_stdin,
                &second_request["id"],
                serde_json::json!("alive"),
            )
            .await;
            assert_eq!(second.await.unwrap().unwrap(), serde_json::json!("alive"));
            assert!(client.pending_requests.lock().await.is_empty());
        }

        fn deeply_nested() -> String {
            format!("{}1{}", "{\"parent\":".repeat(200), "}".repeat(200))
        }

        /// #681: invalid UTF-8 in a response body fails its own request only.
        #[tokio::test]
        async fn test_invalid_utf8_response_fails_only_its_request() {
            use tokio::io::AsyncWriteExt as _;

            let (client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);

            let first = spawn_hover(&client);
            let first_request = read_framed_message(&mut reader).await;
            let mut body = format!(
                r#"{{"jsonrpc":"2.0","id":{},"result":"x"#,
                first_request["id"]
            )
            .into_bytes();
            body.extend_from_slice(&[0xff, 0xfe]);
            body.extend_from_slice(br#"y"}"#);
            let stdin = &mut server.read_half_stdin;
            stdin
                .write_all(format!("Content-Length: {}\r\n\r\n", body.len()).as_bytes())
                .await
                .unwrap();
            stdin.write_all(&body).await.unwrap();
            stdin.flush().await.unwrap();

            assert_matches!(
                first.await.unwrap(),
                Err(Error::LspProtocolError(message)) if message.as_str() == "undecodable response"
            );

            let second = spawn_hover(&client);
            let second_request = read_framed_message(&mut reader).await;
            write_response(
                &mut server.read_half_stdin,
                &second_request["id"],
                serde_json::json!("alive"),
            )
            .await;
            assert_eq!(second.await.unwrap().unwrap(), serde_json::json!("alive"));
        }

        /// #681: undecodable frames are warned about at most once a minute,
        /// with the number dropped silently since the last warning.
        #[test]
        fn test_undecodable_warning_is_time_limited_and_counts_the_suppressed() {
            use tracing_subscriber::layer::SubscriberExt as _;

            let captured = crate::test_lsp::CapturedLogs::default();
            let _guard = tracing::subscriber::set_default(
                tracing_subscriber::registry().with(captured.clone()),
            );
            let mut run = UndecodableRun::default();
            let start = std::time::Instant::now();
            let at = |secs| start + Duration::from_secs(secs);
            for secs in [0, 1, 2, 30] {
                run.observe_at(InboundMessage::UndecodableFrame, at(secs))
                    .unwrap();
            }
            run.observe_at(InboundMessage::UndecodableFrame, at(61))
                .unwrap();

            let warnings: Vec<_> = captured
                .entries()
                .into_iter()
                .filter(|(level, _)| *level == tracing::Level::WARN)
                .map(|(_, message)| message)
                .collect();
            assert_eq!(warnings.len(), 2, "{warnings:?}");
            assert!(warnings[0].contains("(0 others"), "{warnings:?}");
            assert!(warnings[1].contains("(3 others"), "{warnings:?}");
        }

        #[tokio::test]
        async fn test_undecodable_server_request_is_answered_with_an_error() {
            use crate::test_lsp::write_raw_frame;

            let (client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);

            write_raw_frame(
                &mut server.read_half_stdin,
                &format!(
                    r#"{{"jsonrpc":"2.0","id":"srv-1","method":"client/registerCapability","params":{}}}"#,
                    deeply_nested()
                ),
            )
            .await;
            let reply = read_framed_message(&mut reader).await;

            assert_eq!(reply["id"], serde_json::json!("srv-1"));
            assert_eq!(reply["error"]["code"], serde_json::json!(-32600));

            let hover = spawn_hover(&client);
            let request = read_framed_message(&mut reader).await;
            write_response(
                &mut server.read_half_stdin,
                &request["id"],
                serde_json::json!("alive"),
            )
            .await;
            assert_eq!(hover.await.unwrap().unwrap(), serde_json::json!("alive"));
        }

        #[tokio::test]
        async fn test_undecodable_notification_is_dropped_and_the_connection_stays_up() {
            use crate::test_lsp::write_raw_frame;

            let (client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);
            let hover = spawn_hover(&client);
            let request = read_framed_message(&mut reader).await;

            write_raw_frame(
                &mut server.read_half_stdin,
                &format!(
                    r#"{{"jsonrpc":"2.0","method":"window/logMessage","params":{}}}"#,
                    deeply_nested()
                ),
            )
            .await;
            write_response(
                &mut server.read_half_stdin,
                &request["id"],
                serde_json::json!("alive"),
            )
            .await;

            assert_eq!(hover.await.unwrap().unwrap(), serde_json::json!("alive"));
        }

        #[tokio::test]
        async fn test_a_run_of_undecodable_frames_ends_the_connection() {
            use crate::test_lsp::write_raw_frame;

            let (client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);
            let hover = spawn_hover(&client);
            let _request = read_framed_message(&mut reader).await;

            for _ in 0..MAX_CONSECUTIVE_UNDECODABLE_FRAMES {
                write_raw_frame(&mut server.read_half_stdin, "{not json").await;
            }

            assert_matches!(hover.await.unwrap(), Err(Error::ServerTerminated));
        }

        #[tokio::test]
        async fn test_a_decodable_message_ends_a_run_of_undecodable_frames() {
            use crate::test_lsp::write_raw_frame;

            let (client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);
            let hover = spawn_hover(&client);
            let request = read_framed_message(&mut reader).await;

            let almost = MAX_CONSECUTIVE_UNDECODABLE_FRAMES - 1;
            for _ in 0..2 {
                for _ in 0..almost {
                    write_raw_frame(&mut server.read_half_stdin, "{not json").await;
                }
                write_raw_frame(
                    &mut server.read_half_stdin,
                    r#"{"jsonrpc":"2.0","method":"$/progress"}"#,
                )
                .await;
            }
            write_response(
                &mut server.read_half_stdin,
                &request["id"],
                serde_json::json!("alive"),
            )
            .await;

            assert_eq!(hover.await.unwrap().unwrap(), serde_json::json!("alive"));
        }

        #[tokio::test(start_paused = true)]
        async fn test_dropped_caller_after_send_is_cleared_by_late_reply() {
            let (client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);

            let first = spawn_hover(&client);
            let first_request = read_framed_message(&mut reader).await;
            first.abort();
            assert!(first.await.unwrap_err().is_cancelled());

            write_response(
                &mut server.read_half_stdin,
                &first_request["id"],
                serde_json::json!("orphaned"),
            )
            .await;

            let second = spawn_hover(&client);
            let second_request = read_framed_message(&mut reader).await;
            write_response(
                &mut server.read_half_stdin,
                &second_request["id"],
                serde_json::json!("fresh"),
            )
            .await;

            assert_eq!(second.await.unwrap().unwrap(), serde_json::json!("fresh"));
            assert!(
                client.pending_requests.lock().await.is_empty(),
                "the orphaned reply must clear the abandoned entry"
            );
        }
    }

    #[test]
    fn test_should_retrigger_defaults_to_true_when_data_absent() {
        assert!(LspClient::should_retrigger(None));
    }

    #[test]
    fn test_should_retrigger_false_when_flag_false() {
        assert!(!LspClient::should_retrigger(Some(&serde_json::json!({
            "retriggerRequest": false
        }))));
    }

    #[test]
    fn test_should_retrigger_true_when_flag_true() {
        assert!(LspClient::should_retrigger(Some(&serde_json::json!({
            "retriggerRequest": true
        }))));
    }

    /// Wire-level checks that `params: void` LSP methods go out without a
    /// `params` key. tsgo rejects `"params": null` on `shutdown` with
    /// `-32602 expected empty, got: null` and then ignores the `exit` that
    /// follows, so mcpls fell through to the kill-on-timeout path.
    mod void_params_wire {
        use tokio::io::BufReader;

        use super::*;
        use crate::test_lsp::{fake_lsp_client, read_framed_message, write_response};

        #[tokio::test]
        async fn test_request_with_null_params_omits_params_key() {
            let (client, mut server) = fake_lsp_client();

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>("shutdown", Value::Null, Duration::from_secs(5))
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);
            let request = read_framed_message(&mut reader).await;

            assert_eq!(request["method"], "shutdown");
            assert!(
                request.get("params").is_none(),
                "null params must be omitted, got: {request}"
            );

            write_response(&mut server.read_half_stdin, &request["id"], Value::Null).await;
            request_task.await.unwrap().unwrap();
        }

        #[tokio::test]
        async fn test_notify_with_null_params_omits_params_key() {
            let (client, mut server) = fake_lsp_client();

            client.notify("exit", Value::Null).await.unwrap();

            let mut reader = BufReader::new(&mut server.write_stdout);
            let notification = read_framed_message(&mut reader).await;

            assert_eq!(notification["method"], "exit");
            assert!(
                notification.get("params").is_none(),
                "null params must be omitted, got: {notification}"
            );
        }

        #[tokio::test]
        async fn test_notify_with_empty_object_params_keeps_params_key() {
            let (client, mut server) = fake_lsp_client();

            client
                .notify("initialized", lsp_types::InitializedParams {})
                .await
                .unwrap();

            let mut reader = BufReader::new(&mut server.write_stdout);
            let notification = read_framed_message(&mut reader).await;

            assert_eq!(notification["method"], "initialized");
            assert_eq!(
                notification["params"],
                serde_json::json!({}),
                "non-null params must still be sent"
            );
        }
    }

    mod retry_behavior {
        use tokio::io::{AsyncWriteExt, BufReader, DuplexStream};

        use super::*;
        use crate::test_lsp::{
            CapturedLogs, fake_lsp_client, read_framed_message, write_error_response,
            write_response as write_success_response,
        };

        /// Writes a framed JSON-RPC retryable error response — either
        /// `ServerCancelled` (-32802) or `ContentModified` (-32801) — with a
        /// `data.retriggerRequest` flag.
        ///
        /// Kept local rather than promoted to the shared `test_lsp` harness:
        /// the `data.retriggerRequest` field is specific to this retry-logic
        /// test suite, unlike the generic success/error responses above.
        async fn write_retryable_error_response(
            stdin: &mut DuplexStream,
            id: &Value,
            code: i32,
            message: &str,
            retrigger: bool,
        ) {
            let response = serde_json::json!({
                "jsonrpc": "2.0",
                "id": id,
                "error": {
                    "code": code,
                    "message": message,
                    "data": { "retriggerRequest": retrigger },
                },
            });
            let content = serde_json::to_string(&response).unwrap();
            let header = format!("Content-Length: {}\r\n\r\n", content.len());
            stdin.write_all(header.as_bytes()).await.unwrap();
            stdin.write_all(content.as_bytes()).await.unwrap();
            stdin.flush().await.unwrap();
        }

        // Not `start_paused`: the retry loop's real backoff sleeps
        // interleave with real async I/O on the duplex pipes below, and
        // paused virtual time does not reliably auto-advance across both.
        //
        // Also captures tracing output (#392): retry exhaustion is the one
        // scenario where every attempt but the last logs `warn!` and only
        // the last logs `error!`, so this doubles as that regression test
        // rather than duplicating the same ~3.5s wire choreography in a
        // second test just to assert on log severity.
        #[tokio::test]
        async fn test_retry_exhaustion_returns_original_server_cancelled_error() {
            use tracing_subscriber::layer::SubscriberExt as _;

            let (client, mut server) = fake_lsp_client();
            let captured = CapturedLogs::default();
            let subscriber = tracing_subscriber::registry().with(captured.clone());
            let guard = tracing::subscriber::set_default(subscriber);

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);
            // Initial attempt plus SERVER_CANCELLED_MAX_RETRIES retries: every
            // attempt gets ServerCancelled, so retries must exhaust rather
            // than loop forever or swallow the error.
            for _ in 0..=SERVER_CANCELLED_MAX_RETRIES {
                let request = read_framed_message(&mut reader).await;
                let id = request["id"].clone();
                write_retryable_error_response(
                    &mut server.read_half_stdin,
                    &id,
                    i32::from(LspErrorCodes::ServerCancelled),
                    "server cancelled the request",
                    true,
                )
                .await;
            }

            let result = request_task.await.unwrap();

            match result {
                Err(Error::LspServerError {
                    code,
                    message,
                    data,
                }) => {
                    // Assert the exact original error surfaces, not merely
                    // "some error with this code" -- a freshly constructed
                    // placeholder error would satisfy a code-only check.
                    assert_eq!(code, i32::from(LspErrorCodes::ServerCancelled));
                    assert_eq!(message, "server cancelled the request");
                    assert_eq!(data, Some(serde_json::json!({ "retriggerRequest": true })));
                }
                other => panic!("expected exhausted ServerCancelled error, got {other:?}"),
            }

            drop(guard);
            let logs = captured.entries();
            assert_eq!(
                logs.iter()
                    .filter(|(level, _)| *level == tracing::Level::ERROR)
                    .count(),
                1,
                "exactly the final exhausted attempt must log at ERROR, got: {logs:?}"
            );
            assert!(
                logs.iter()
                    .any(|(level, msg)| *level == tracing::Level::ERROR
                        && msg.contains("LSP error response")
                        && msg.contains("retries exhausted")),
                "expected an ERROR log sharing the 'LSP error response' prefix and naming \
                 retry exhaustion, got: {logs:?}"
            );
            assert_eq!(
                logs.iter()
                    .filter(
                        |(level, msg)| *level == tracing::Level::WARN && msg.contains("will retry")
                    )
                    .count(),
                usize::try_from(SERVER_CANCELLED_MAX_RETRIES).unwrap(),
                "every attempt before the last must log a WARN 'will retry' line, got: {logs:?}"
            );
        }

        #[tokio::test]
        async fn test_retrigger_false_returns_immediately_without_retry() {
            let (client, mut server) = fake_lsp_client();

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);
            let request = read_framed_message(&mut reader).await;
            let id = request["id"].clone();
            write_retryable_error_response(
                &mut server.read_half_stdin,
                &id,
                i32::from(LspErrorCodes::ServerCancelled),
                "server cancelled the request",
                false,
            )
            .await;

            // With `retriggerRequest: false`, `should_retrigger`'s gate on
            // the retry branch must short-circuit the loop: the error
            // returns well under the first 500ms backoff, and no second
            // request is ever sent. If the `&& Self::should_retrigger(..)`
            // guard were ever dropped from the retry match arm, this would
            // instead retry and both assertions below would fail.
            let result = tokio::time::timeout(Duration::from_millis(200), request_task)
                .await
                .unwrap()
                .unwrap();

            match result {
                Err(Error::LspServerError { code, .. }) => {
                    assert_eq!(code, i32::from(LspErrorCodes::ServerCancelled));
                }
                other => panic!("expected immediate ServerCancelled error, got {other:?}"),
            }

            let second_request =
                tokio::time::timeout(Duration::from_millis(200), read_framed_message(&mut reader))
                    .await;
            assert!(
                second_request.is_err(),
                "no retry should have been sent after retriggerRequest: false"
            );
        }

        #[tokio::test]
        async fn test_retry_succeeds_after_one_server_cancelled_response() {
            let (client, mut server) = fake_lsp_client();

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);

            // First attempt is cancelled and must retrigger.
            let first = read_framed_message(&mut reader).await;
            write_retryable_error_response(
                &mut server.read_half_stdin,
                &first["id"].clone(),
                i32::from(LspErrorCodes::ServerCancelled),
                "server cancelled the request",
                true,
            )
            .await;

            // Second attempt (after backoff) succeeds -- proves the loop
            // genuinely re-sends the request rather than just counting down.
            let second = read_framed_message(&mut reader).await;
            assert_ne!(
                first["id"], second["id"],
                "retry must use a fresh request id"
            );
            let expected_result = serde_json::json!({ "contents": "resolved on retry" });
            write_success_response(
                &mut server.read_half_stdin,
                &second["id"].clone(),
                expected_result.clone(),
            )
            .await;

            let result = request_task.await.unwrap();
            assert_eq!(result.unwrap(), expected_result);
        }

        #[tokio::test]
        async fn test_retry_exhaustion_returns_original_content_modified_error() {
            let (client, mut server) = fake_lsp_client();

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);
            // Initial attempt plus SERVER_CANCELLED_MAX_RETRIES retries: every
            // attempt gets ContentModified, so retries must exhaust rather
            // than loop forever or swallow the error. -32801 shares the same
            // attempt budget as -32802 (FR-002), not an independent one.
            for _ in 0..=SERVER_CANCELLED_MAX_RETRIES {
                let request = read_framed_message(&mut reader).await;
                let id = request["id"].clone();
                write_retryable_error_response(
                    &mut server.read_half_stdin,
                    &id,
                    i32::from(LspErrorCodes::ContentModified),
                    "content modified",
                    true,
                )
                .await;
            }

            let result = request_task.await.unwrap();

            match result {
                Err(Error::LspServerError {
                    code,
                    message,
                    data,
                }) => {
                    // Assert the exact original error surfaces, not merely
                    // "some error with this code" -- a freshly constructed
                    // placeholder error would satisfy a code-only check.
                    assert_eq!(code, i32::from(LspErrorCodes::ContentModified));
                    assert_eq!(message, "content modified");
                    assert_eq!(data, Some(serde_json::json!({ "retriggerRequest": true })));
                }
                other => panic!("expected exhausted ContentModified error, got {other:?}"),
            }
        }

        #[tokio::test]
        async fn test_retrigger_false_returns_immediately_without_retry_for_content_modified() {
            let (client, mut server) = fake_lsp_client();

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);
            let request = read_framed_message(&mut reader).await;
            let id = request["id"].clone();
            write_retryable_error_response(
                &mut server.read_half_stdin,
                &id,
                i32::from(LspErrorCodes::ContentModified),
                "content modified",
                false,
            )
            .await;

            // Same `should_retrigger` gate as -32802: a non-spec-compliant
            // server sending `retriggerRequest: false` on -32801 must still
            // be honored (FR-006 resolution), short-circuiting the loop well
            // under the first 500ms backoff.
            let result = tokio::time::timeout(Duration::from_millis(200), request_task)
                .await
                .unwrap()
                .unwrap();

            match result {
                Err(Error::LspServerError { code, .. }) => {
                    assert_eq!(code, i32::from(LspErrorCodes::ContentModified));
                }
                other => panic!("expected immediate ContentModified error, got {other:?}"),
            }

            let second_request =
                tokio::time::timeout(Duration::from_millis(200), read_framed_message(&mut reader))
                    .await;
            assert!(
                second_request.is_err(),
                "no retry should have been sent after retriggerRequest: false"
            );
        }

        #[tokio::test]
        async fn test_retry_succeeds_after_one_content_modified_response() {
            let (client, mut server) = fake_lsp_client();

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);

            // First attempt gets ContentModified and must retrigger.
            let first = read_framed_message(&mut reader).await;
            write_retryable_error_response(
                &mut server.read_half_stdin,
                &first["id"].clone(),
                i32::from(LspErrorCodes::ContentModified),
                "content modified",
                true,
            )
            .await;

            // Second attempt (after backoff) succeeds -- proves the loop
            // genuinely re-sends the request rather than just counting down.
            let second = read_framed_message(&mut reader).await;
            assert_ne!(
                first["id"], second["id"],
                "retry must use a fresh request id"
            );
            let expected_result = serde_json::json!({ "contents": "resolved on retry" });
            write_success_response(
                &mut server.read_half_stdin,
                &second["id"].clone(),
                expected_result.clone(),
            )
            .await;

            let result = request_task.await.unwrap();
            assert_eq!(result.unwrap(), expected_result);
        }

        #[tokio::test]
        async fn test_content_modified_on_non_allowlisted_method_does_not_retry() {
            let (client, mut server) = fake_lsp_client();

            // `textDocument/rename` is deliberately excluded from
            // `CONTENT_MODIFIED_RETRY_METHODS` (its result is an edit the
            // caller applies at a position that may no longer be valid once
            // the document changed) -- a -32801 response for it must return
            // immediately even though `retriggerRequest: true` would pass
            // `should_retrigger`'s gate on its own.
            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/rename",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);
            let request = read_framed_message(&mut reader).await;
            let id = request["id"].clone();
            write_retryable_error_response(
                &mut server.read_half_stdin,
                &id,
                i32::from(LspErrorCodes::ContentModified),
                "content modified",
                true,
            )
            .await;

            let result = tokio::time::timeout(Duration::from_millis(200), request_task)
                .await
                .unwrap()
                .unwrap();

            match result {
                Err(Error::LspServerError { code, .. }) => {
                    assert_eq!(code, i32::from(LspErrorCodes::ContentModified));
                }
                other => panic!("expected immediate ContentModified error, got {other:?}"),
            }

            let second_request =
                tokio::time::timeout(Duration::from_millis(200), read_framed_message(&mut reader))
                    .await;
            assert!(
                second_request.is_err(),
                "no retry should have been sent for a non-allowlisted method"
            );
        }

        /// #313: an oversized, server-controlled error message must be
        /// truncated before it reaches the MCP caller in
        /// `Error::LspServerError`, not just before it is logged. Routes
        /// through the real `message_loop_inner` (via `fake_lsp_client`)
        /// rather than constructing the error by hand, so it actually
        /// exercises the fix.
        #[tokio::test]
        async fn test_oversized_error_message_truncated_for_caller() {
            let (client, mut server) = fake_lsp_client();

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);
            let request = read_framed_message(&mut reader).await;
            let id = request["id"].clone();
            let oversized_message = "x".repeat(MAX_ERROR_MESSAGE_CALLER_BYTES + 500);
            write_error_response(&mut server.read_half_stdin, &id, -32603, &oversized_message)
                .await;

            let result = request_task.await.unwrap();

            match result {
                Err(Error::LspServerError { code, message, .. }) => {
                    assert_eq!(code, -32603);
                    assert!(
                        message.len() < oversized_message.len(),
                        "caller-facing message must be truncated, got {} bytes",
                        message.len()
                    );
                    assert!(message.ends_with("... (truncated)"));
                }
                other => panic!("expected truncated LspServerError, got {other:?}"),
            }
        }

        /// #313 S2: a legitimate error message longer than the log-line cap
        /// (`crate::util::MAX_LOG_STRING_BYTES`, 200 bytes) but shorter than
        /// the caller-facing cap must reach the MCP caller intact -- the
        /// caller-facing budget must not silently collapse to the log
        /// budget.
        #[tokio::test]
        async fn test_error_message_between_log_and_caller_caps_reaches_caller_intact() {
            let (client, mut server) = fake_lsp_client();

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);
            let request = read_framed_message(&mut reader).await;
            let id = request["id"].clone();
            let message = "x".repeat(crate::util::MAX_LOG_STRING_BYTES + 50);
            write_error_response(&mut server.read_half_stdin, &id, -32603, &message).await;

            let result = request_task.await.unwrap();

            match result {
                Err(Error::LspServerError {
                    message: returned, ..
                }) => {
                    assert_eq!(
                        returned, message,
                        "message under the caller cap must not be truncated"
                    );
                }
                other => panic!("expected untruncated LspServerError, got {other:?}"),
            }
        }

        /// #392: a `-32802`/`-32801` error that gets retried and then
        /// succeeds must not log at `error!` -- only a `warn!` "will retry"
        /// line -- so log-based monitoring does not false-positive on a
        /// transient error the retry loop silently recovers from.
        #[tokio::test]
        async fn test_retried_error_that_recovers_does_not_log_error_level() {
            use tracing_subscriber::layer::SubscriberExt as _;

            let (client, mut server) = fake_lsp_client();
            let captured = CapturedLogs::default();
            let subscriber = tracing_subscriber::registry().with(captured.clone());
            let guard = tracing::subscriber::set_default(subscriber);

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);

            let first = read_framed_message(&mut reader).await;
            write_retryable_error_response(
                &mut server.read_half_stdin,
                &first["id"].clone(),
                i32::from(LspErrorCodes::ServerCancelled),
                "server cancelled the request",
                true,
            )
            .await;

            let second = read_framed_message(&mut reader).await;
            write_success_response(
                &mut server.read_half_stdin,
                &second["id"].clone(),
                serde_json::json!({ "contents": "resolved on retry" }),
            )
            .await;

            let result = request_task.await.unwrap();
            assert!(result.is_ok(), "expected retry to recover, got {result:?}");

            drop(guard);
            let logs = captured.entries();
            assert!(
                !logs
                    .iter()
                    .any(|(level, _)| *level == tracing::Level::ERROR),
                "a retried-and-recovered error must not log at ERROR, got: {logs:?}"
            );
            assert!(
                logs.iter().any(
                    |(level, msg)| *level == tracing::Level::WARN && msg.contains("will retry")
                ),
                "expected a WARN 'will retry' log line, got: {logs:?}"
            );
        }

        /// #392: a `-32801` (`ContentModified`) error for a method outside
        /// `CONTENT_MODIFIED_RETRY_METHODS` is never retried, so it must
        /// still surface at `error!` on the very first attempt.
        #[tokio::test]
        async fn test_non_retryable_error_logs_error_level() {
            use tracing_subscriber::layer::SubscriberExt as _;

            let (client, mut server) = fake_lsp_client();
            let captured = CapturedLogs::default();
            let subscriber = tracing_subscriber::registry().with(captured.clone());
            let guard = tracing::subscriber::set_default(subscriber);

            let request_task = tokio::spawn(async move {
                client
                    .request::<_, Value>(
                        "textDocument/rename",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let mut reader = BufReader::new(&mut server.write_stdout);
            let request = read_framed_message(&mut reader).await;
            let id = request["id"].clone();
            write_retryable_error_response(
                &mut server.read_half_stdin,
                &id,
                i32::from(LspErrorCodes::ContentModified),
                "content modified",
                true,
            )
            .await;

            let result = request_task.await.unwrap();
            assert!(result.is_err(), "expected a non-retryable error");

            drop(guard);
            let logs = captured.entries();
            assert!(
                logs.iter()
                    .any(|(level, msg)| *level == tracing::Level::ERROR
                        && msg.contains("LSP error response")
                        && msg.contains("content modified")),
                "a non-retryable error must still surface an ERROR log sharing the \
                 'LSP error response' prefix, got: {logs:?}"
            );
        }
    }

    /// Regression coverage for #451 (`LspTransportReader::receive` moved off
    /// the `select!` and onto a dedicated reader task, see
    /// [`super::spawn_reader_task`]).
    mod reader_task_regression {
        use tokio::io::BufReader;

        use super::*;
        use crate::test_lsp::{
            fake_lsp_client, inert_transport, read_framed_message, write_response,
        };

        /// The reader task only ever sends `Err` through the channel before
        /// exiting (see `spawn_reader_task`) -- a `None` from `msg_rx.recv()`
        /// means the task disappeared some other way (e.g. panicked). This is
        /// the one branch in `message_loop_inner` that has no equivalent in
        /// the pre-#451 code, so it needs its own direct test rather than
        /// relying on the full `fake_lsp_client` harness to provoke it.
        #[tokio::test]
        async fn test_message_loop_inner_treats_reader_channel_close_as_server_terminated() {
            let (mut transport, _reader) = inert_transport();
            let (_command_tx, mut command_rx) = mpsc::channel::<ClientCommand>(1);
            let (msg_tx, mut msg_rx) = mpsc::channel::<Result<InboundMessage>>(1);
            let pending_requests: Arc<Mutex<PendingRequests>> =
                Arc::new(Mutex::new(HashMap::new()));

            drop(msg_tx);

            let result = LspClient::message_loop_inner(
                &mut transport,
                &mut msg_rx,
                &mut command_rx,
                &pending_requests,
                None,
                None,
                &Redactions::default(),
                None,
            )
            .await;

            assert_matches!(result, Err(Error::ServerTerminated), "got {result:?}");
        }

        /// As above, with a real pending request parked at the time the
        /// reader task disappears -- `message_loop_inner` itself never
        /// touches `pending_requests` on this branch, so a caller blocked on
        /// it only unblocks via `message_loop`'s post-loop
        /// `drain_and_fail_pending` (#458), invoked here the same way
        /// `message_loop` does after `message_loop_inner` returns. Driving
        /// this exact branch through the real `message_loop` would require
        /// the reader task to panic (the only other way `msg_rx.recv()`
        /// returns `None`), which isn't reproducible without a production
        /// seam -- see `test_message_loop_fails_pending_requests_on_transport_error_exit`
        /// for the sibling exit path exercised through the real function.
        #[tokio::test]
        async fn test_message_loop_inner_reader_gone_then_drain_fails_pending_request() {
            let (mut transport, _reader) = inert_transport();
            let (_command_tx, mut command_rx) = mpsc::channel::<ClientCommand>(1);
            let (msg_tx, mut msg_rx) = mpsc::channel::<Result<InboundMessage>>(1);
            let pending_requests: Arc<Mutex<PendingRequests>> =
                Arc::new(Mutex::new(HashMap::new()));

            let (response_tx, response_rx) = oneshot::channel::<Result<Value>>();
            pending_requests
                .lock()
                .await
                .insert(RequestId::Number(1), response_tx);

            drop(msg_tx);

            let result = LspClient::message_loop_inner(
                &mut transport,
                &mut msg_rx,
                &mut command_rx,
                &pending_requests,
                None,
                None,
                &Redactions::default(),
                None,
            )
            .await;
            assert_matches!(result, Err(Error::ServerTerminated), "got {result:?}");

            LspClient::drain_and_fail_pending(&pending_requests, &StdMutex::default()).await;

            let received = response_rx.await.unwrap();
            assert_matches!(received, Err(Error::ServerTerminated), "got {received:?}");
            assert!(pending_requests.lock().await.is_empty());
        }

        /// The reader task can run ahead of `select!` and have already queued
        /// a fully-decoded response in `msg_rx` by the time a concurrent
        /// `Shutdown` command is what `select!` happens to pick. Before the
        /// drain fix, that queued response was silently dropped instead of
        /// resolving its caller's pending request -- the caller would then
        /// block until its own `request_timeout_seconds` elapsed instead of
        /// failing fast or succeeding.
        #[tokio::test]
        async fn test_shutdown_drains_buffered_responses_instead_of_dropping_them() {
            let (mut transport, _reader) = inert_transport();
            let (command_tx, mut command_rx) = mpsc::channel::<ClientCommand>(1);
            let (msg_tx, mut msg_rx) = mpsc::channel::<Result<InboundMessage>>(1);
            let pending_requests: Arc<Mutex<PendingRequests>> =
                Arc::new(Mutex::new(HashMap::new()));

            let id = RequestId::Number(1);
            let (response_tx, response_rx) = oneshot::channel::<Result<Value>>();
            pending_requests
                .lock()
                .await
                .insert(id.clone(), response_tx);

            msg_tx
                .send(Ok(InboundMessage::Response(JsonRpcResponse {
                    jsonrpc: JsonRpcVersion,
                    id: id.clone(),
                    outcome: JsonRpcOutcome::Result(serde_json::json!({ "ok": true })),
                })))
                .await
                .unwrap();
            command_tx.send(ClientCommand::Shutdown).await.unwrap();
            drop(command_tx);

            let result = LspClient::message_loop_inner(
                &mut transport,
                &mut msg_rx,
                &mut command_rx,
                &pending_requests,
                None,
                None,
                &Redactions::default(),
                None,
            )
            .await;

            assert!(result.is_ok(), "got {result:?}");
            // `Err` here means the sender was dropped without a reply -- i.e. the
            // buffered response was lost instead of resolving this request.
            let received = response_rx.await.unwrap();
            assert_eq!(received.unwrap(), serde_json::json!({ "ok": true }));
            assert!(
                pending_requests.lock().await.is_empty(),
                "the drained response must resolve its pending request entry"
            );
        }

        /// #458: before the fix, none of `message_loop`'s exit paths failed
        /// requests still parked in `pending_requests` -- each caller stayed
        /// blocked in its own `timeout(..)` until `request_timeout` elapsed.
        /// Drives the real `message_loop` (not `_inner`) over
        /// [`inert_transport`], whose reader side is dropped, so the reader
        /// task hits a transport receive error immediately and
        /// `message_loop` exits through that path.
        #[tokio::test]
        async fn test_message_loop_fails_pending_requests_on_transport_error_exit() {
            let (transport, reader) = inert_transport();
            let (_command_tx, command_rx) = mpsc::channel::<ClientCommand>(1);
            let pending_requests: Arc<Mutex<PendingRequests>> =
                Arc::new(Mutex::new(HashMap::new()));

            let (response_tx, response_rx) = oneshot::channel::<Result<Value>>();
            pending_requests
                .lock()
                .await
                .insert(RequestId::Number(1), response_tx);

            let result = LspClient::message_loop(
                (transport, reader),
                command_rx,
                Arc::clone(&pending_requests),
                Arc::default(),
                None,
                None,
                Arc::default(),
                None,
            )
            .await;

            assert!(result.is_err(), "got {result:?}");
            let received = response_rx.await.unwrap();
            assert_matches!(received, Err(Error::ServerTerminated), "got {received:?}");
            assert!(pending_requests.lock().await.is_empty());
        }

        /// #458: unit-level check that composing `message_loop_inner`'s
        /// Shutdown `try_recv` drain (#451) with a subsequent
        /// `drain_and_fail_pending` call -- the same two pieces
        /// `message_loop` itself calls in that order -- does not re-fail a
        /// request already resolved by a buffered response, while still
        /// failing a truly unanswered one. This does not exercise
        /// `message_loop`'s own statement ordering (see
        /// `test_message_loop_end_to_end_resolves_answered_then_fails_unanswered_on_shutdown`
        /// for that, driven through the real function).
        #[tokio::test]
        async fn test_message_loop_shutdown_resolves_answered_and_fails_unanswered_pending() {
            let (mut transport, _reader) = inert_transport();
            let (command_tx, mut command_rx) = mpsc::channel::<ClientCommand>(1);
            let (msg_tx, mut msg_rx) = mpsc::channel::<Result<InboundMessage>>(1);
            let pending_requests: Arc<Mutex<PendingRequests>> =
                Arc::new(Mutex::new(HashMap::new()));

            let answered_id = RequestId::Number(1);
            let (answered_tx, answered_rx) = oneshot::channel::<Result<Value>>();
            let unanswered_id = RequestId::Number(2);
            let (unanswered_tx, unanswered_rx) = oneshot::channel::<Result<Value>>();
            pending_requests
                .lock()
                .await
                .insert(answered_id.clone(), answered_tx);
            pending_requests
                .lock()
                .await
                .insert(unanswered_id.clone(), unanswered_tx);

            msg_tx
                .send(Ok(InboundMessage::Response(JsonRpcResponse {
                    jsonrpc: JsonRpcVersion,
                    id: answered_id,
                    outcome: JsonRpcOutcome::Result(serde_json::json!({ "ok": true })),
                })))
                .await
                .unwrap();
            command_tx.send(ClientCommand::Shutdown).await.unwrap();
            drop(command_tx);

            let inner_result = LspClient::message_loop_inner(
                &mut transport,
                &mut msg_rx,
                &mut command_rx,
                &pending_requests,
                None,
                None,
                &Redactions::default(),
                None,
            )
            .await;
            assert!(inner_result.is_ok(), "got {inner_result:?}");

            // Mirrors the drain `message_loop` runs after `message_loop_inner`
            // returns.
            LspClient::drain_and_fail_pending(&pending_requests, &StdMutex::default()).await;

            let answered = answered_rx.await.unwrap();
            assert_eq!(answered.unwrap(), serde_json::json!({ "ok": true }));

            let unanswered = unanswered_rx.await.unwrap();
            assert_matches!(
                unanswered,
                Err(Error::ServerTerminated),
                "got {unanswered:?}"
            );
            assert!(pending_requests.lock().await.is_empty());
        }

        /// #458 S2: end-to-end proof that `message_loop`'s post-loop drain
        /// runs *after* `message_loop_inner` returns, driven through the
        /// real `message_loop` (via `fake_lsp_client`), not a hand-recreated
        /// sequence. Deterministic without depending on `select!`
        /// nondeterminism: the answered request is awaited to completion via
        /// the loop's ordinary response path *before* `shutdown` is ever
        /// sent, so nothing here relies on which branch `select!` happens to
        /// pick when both are ready -- that race is unit-tested separately in
        /// `test_message_loop_shutdown_resolves_answered_and_fails_unanswered_pending`.
        #[tokio::test]
        async fn test_message_loop_end_to_end_resolves_answered_then_fails_unanswered_on_shutdown()
        {
            let (client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);

            let answered_client = client.clone();
            let answered_task = tokio::spawn(async move {
                answered_client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            let request = read_framed_message(&mut reader).await;
            let id = request["id"].clone();
            write_response(
                &mut server.read_half_stdin,
                &id,
                serde_json::json!({ "ok": true }),
            )
            .await;

            let answered = answered_task.await.unwrap().unwrap();
            assert_eq!(answered, serde_json::json!({ "ok": true }));

            let unanswered_client = client.clone();
            let unanswered_task = tokio::spawn(async move {
                unanswered_client
                    .request::<_, Value>(
                        "textDocument/definition",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });

            // Reading the framed bytes back confirms the request was fully
            // sent, which only happens after it is already registered in
            // `pending_requests` -- so shutdown below cannot race its insert.
            let _unanswered_request = read_framed_message(&mut reader).await;

            client.shutdown().await.unwrap();

            let unanswered = tokio::time::timeout(Duration::from_millis(200), unanswered_task)
                .await
                .unwrap_or_else(|_| {
                    panic!(
                        "unanswered request must fail immediately on shutdown, not hang until its own timeout"
                    )
                })
                .unwrap();
            assert_matches!(
                unanswered,
                Err(Error::ServerTerminated),
                "got {unanswered:?}"
            );
        }

        /// Drives dense, concurrent request/response traffic through the real
        /// `message_loop`/`spawn_reader_task` pair (via `fake_lsp_client`) and
        /// answers deliberately out of arrival order, so a bug that matched
        /// responses positionally instead of by id -- the failure mode a
        /// mid-frame desync would eventually cause -- would surface as a
        /// mismatched payload rather than a hang.
        #[tokio::test]
        async fn test_dense_concurrent_requests_all_resolve_to_matching_responses() {
            const REQUEST_COUNT: usize = 20;
            let (client, mut server) = fake_lsp_client();

            // A manual push loop, not `.map(..).collect()`: `tokio::spawn`
            // must run eagerly for every `i` right here, before
            // `server_task` starts answering below -- a lazily-iterated
            // combinator would spawn (and thus send) each request only as
            // its `JoinHandle` is later awaited, one at a time, defeating
            // the "dense concurrent" setup this test needs.
            let mut request_tasks = Vec::with_capacity(REQUEST_COUNT);
            for i in 0..REQUEST_COUNT {
                let client = client.clone();
                request_tasks.push(tokio::spawn(async move {
                    client
                        .request::<_, Value>(
                            "textDocument/hover",
                            serde_json::json!({ "n": i }),
                            Duration::from_secs(30),
                        )
                        .await
                }));
            }

            let server_task = tokio::spawn(async move {
                let mut reader = BufReader::new(&mut server.write_stdout);
                let mut requests = Vec::with_capacity(REQUEST_COUNT);
                for _ in 0..REQUEST_COUNT {
                    requests.push(read_framed_message(&mut reader).await);
                }
                for request in requests.into_iter().rev() {
                    let id = request["id"].clone();
                    let n = request["params"]["n"].clone();
                    write_response(
                        &mut server.read_half_stdin,
                        &id,
                        serde_json::json!({ "echo": n }),
                    )
                    .await;
                }
            });

            server_task.await.unwrap();

            for (i, task) in request_tasks.into_iter().enumerate() {
                let value = task
                    .await
                    .unwrap()
                    .unwrap_or_else(|e| panic!("request {i} failed: {e:?}"));
                assert_eq!(
                    value["echo"],
                    serde_json::json!(i),
                    "response for request {i} carried the wrong payload -- id/response mismatch"
                );
            }
        }

        /// Shutting down a clone stops the loop for every clone but cannot
        /// abort it, and the owner can still shut down afterwards.
        #[tokio::test]
        async fn test_shutdown_on_clone_stops_shared_loop_then_owner_shutdown_succeeds() {
            let (client, _server) = fake_lsp_client();

            client.clone().shutdown().await.unwrap();

            let request = timeout(
                Duration::from_secs(2),
                client.request::<_, Value>("x/y", serde_json::json!({}), Duration::from_secs(30)),
            )
            .await
            .unwrap_or_else(|_| panic!("request on a stopped loop must fail fast"));
            assert_matches!(request, Err(Error::ServerTerminated), "got {request:?}");

            client.shutdown().await.unwrap();
        }

        /// A deadline that has already passed still aborts the loop, reports
        /// `ShutdownTimeout`, and leaves the client in the `Shutdown` state.
        #[tokio::test]
        async fn test_shutdown_until_expired_deadline_reports_timeout() {
            let (mut client, _server) = fake_lsp_client();
            let probe = client.clone();

            let result = client
                .shutdown_until(Instant::now() - Duration::from_millis(50))
                .await;

            assert_matches!(result, Err(Error::ShutdownTimeout), "got {result:?}");
            assert_matches!(probe.state().await, crate::lsp::ServerState::Shutdown);
        }

        /// A message loop wedged writing to a server that never reads must not
        /// hang `shutdown_until`: it returns `ShutdownTimeout` at the deadline
        /// and fails the requests still in flight.
        #[tokio::test]
        async fn test_shutdown_until_aborts_wedged_loop_and_fails_pending() {
            let (mut client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);

            let in_flight_client = client.clone();
            let in_flight = tokio::spawn(async move {
                in_flight_client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });
            let _ = read_framed_message(&mut reader).await;

            // Larger than the mock pipe's capacity, so the loop blocks in `send`.
            client
                .notify("x/wedge", serde_json::json!({ "p": "a".repeat(1 << 20) }))
                .await
                .unwrap();

            let deadline = Instant::now() + Duration::from_millis(200);
            let result = timeout(Duration::from_secs(5), client.shutdown_until(deadline))
                .await
                .unwrap_or_else(|_| panic!("shutdown_until must honor its deadline"));
            assert_matches!(result, Err(Error::ShutdownTimeout), "got {result:?}");

            let in_flight = timeout(Duration::from_secs(1), in_flight)
                .await
                .unwrap_or_else(|_| panic!("in-flight request must be failed by the drain"))
                .unwrap();
            assert_matches!(in_flight, Err(Error::ServerTerminated), "got {in_flight:?}");
        }

        /// A client marked restarted fails its stragglers with the retryable
        /// `ServerRestarted`, both from the explicit drain and from the
        /// message loop's exit drain.
        #[tokio::test]
        async fn test_marked_restarted_client_fails_pending_with_server_restarted() {
            let (mut client, mut server) = fake_lsp_client();
            let mut reader = BufReader::new(&mut server.write_stdout);

            let in_flight_client = client.clone();
            let in_flight = tokio::spawn(async move {
                in_flight_client
                    .request::<_, Value>(
                        "textDocument/hover",
                        serde_json::json!({}),
                        Duration::from_secs(30),
                    )
                    .await
            });
            let _ = read_framed_message(&mut reader).await;

            client.mark_restarted(ServerId::from_static("rust"));
            client
                .shutdown_until(Instant::now() + Duration::from_millis(200))
                .await
                .ok();

            let in_flight = timeout(Duration::from_secs(1), in_flight)
                .await
                .unwrap_or_else(|_| panic!("in-flight request must be failed by the drain"))
                .unwrap();
            assert!(
                matches!(&in_flight, Err(Error::ServerRestarted { server_id }) if server_id.as_str() == "rust"),
                "got {in_flight:?}"
            );
        }

        /// #458: a request issued after the client has already shut down
        /// must fail immediately via `register_and_send_request`'s
        /// `command_tx.send(..)` check, not hang until its own
        /// `request_timeout`. Wrapped in a short outer `timeout` so a
        /// regression back to the slow path fails this test instead of
        /// merely making it slow.
        #[tokio::test]
        async fn test_request_after_shutdown_fails_fast_instead_of_hanging() {
            let (client, _server) = fake_lsp_client();
            let post_shutdown_client = client.clone();

            client.shutdown().await.unwrap();

            let result = tokio::time::timeout(
                Duration::from_millis(200),
                post_shutdown_client.request::<_, Value>(
                    "textDocument/hover",
                    serde_json::json!({}),
                    Duration::from_secs(30),
                ),
            )
            .await
            .unwrap_or_else(|_| {
                panic!(
                    "request after shutdown must fail immediately, not hang until its own timeout"
                )
            });

            assert_matches!(result, Err(Error::ServerTerminated), "got {result:?}");
        }
    }
}