mcpls_core/bridge/

indexing.rs

//! Workspace-indexing readiness tracking.
//!
//! Extracted from `bridge::notifications` (already ~2900 lines) so the
//! settle/latch transition logic below -- the part most likely to grow a
//! subtle off-by-one -- has its own directly unit-testable surface. See
//! [`IndexingTracker`] for the entry point; [`NotificationCache`](super::NotificationCache)
//! owns one and delegates every indexing-related call to it.
//!
//! Two independent signal sources feed the same per-server state:
//! rust-analyzer's custom `experimental/serverStatus` notification (#421,
//! [`IndexingTracker::observe_server_status`]), and the generic LSP
//! `$/progress` `begin`/`end` sequence every spec-compliant server may send
//! once mcpls advertises `window.workDoneProgress` (this issue,
//! [`IndexingTracker::observe_progress`]). [`IndexingSignalSource`] keeps
//! them from fighting over the same entry -- see that type's docs.

use std::collections::HashMap;
use std::time::Duration;

use lsp_types::{ProgressParams, ProgressToken};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use tokio::time::Instant;

use crate::config::ServerId;
use crate::lsp::types::ProgressKind;

/// How a server's tracked [`IndexingState`] is reset.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IndexingReset {
    /// Forget everything tracked: the state reverts to
    /// [`IndexingState::Unknown`] until the next signal.
    Forget,
    /// The server was replaced by a new process: a server that has reported a
    /// recognized signal before reads [`IndexingState::Loading`] for `within`
    /// or until its replacement reports one, whichever comes first. A server
    /// that never reported one behaves as after [`Self::Forget`], so it is not
    /// delayed.
    AwaitReplacement {
        /// Upper bound of the pre-signal [`IndexingState::Loading`] reading.
        within: Duration,
    },
}

/// Workspace-indexing readiness of a routed LSP server.
///
/// Tracked separately from the `initialize`/`initialized` handshake
/// completion (`ServerState::is_ready`). A server can finish the handshake
/// and still be mid-index for tens of seconds afterward, during which
/// whole-workspace queries (hover, definition, references, completions,
/// code actions) can silently return an empty/`null` result
/// indistinguishable from a genuine "nothing found".
///
/// `Unknown` and `Ready` are treated identically by
/// `Translator::wait_for_indexing_ready` (proceed without waiting): a server
/// that never emits a recognized readiness signal must never be penalized
/// with an artificial delay, and the only way to tell "no signal ever comes"
/// apart from "just hasn't reported yet" would require guessing at a
/// server's protocol support, so both stay unblocked. Only `Loading` -- a
/// positive signal that indexing is actively in progress -- triggers a
/// bounded wait.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum IndexingState {
    /// No recognized workspace-readiness signal has been observed for this
    /// server yet.
    #[default]
    Unknown,
    /// A recognized signal reported that indexing is still in progress.
    Loading,
    /// A recognized signal reported that the initial workspace load is
    /// complete.
    Ready,
}

/// Escape hatch for `IndexingTracker`'s readiness gate, configured per
/// server via `LspServerConfig::indexing`.
///
/// `Disabled` pins `IndexingTracker::state` at [`IndexingState::Unknown`]
/// unconditionally -- the same fail-open path a server that has simply never
/// sent a recognized signal already takes -- rather than adding a new branch
/// to the gate itself. Exists for a server whose `$/progress`/`serverStatus`
/// shape doesn't fit this tracker's assumptions (Accepted cost 1/3 in the
/// design): per-server granularity is enough, since `Error::WorkspaceIndexing`
/// already names the offending `server_id`.
///
/// # Examples
///
/// ```
/// use mcpls_core::bridge::IndexingPolicy;
///
/// assert_eq!(IndexingPolicy::default(), IndexingPolicy::Auto);
/// let json = serde_json::to_string(&IndexingPolicy::Disabled).unwrap();
/// assert_eq!(json, "\"disabled\"");
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum IndexingPolicy {
    /// Track readiness normally from whatever signals the server sends.
    #[default]
    Auto,
    /// Never gate on this server; `indexing_state` always reads `Unknown`.
    Disabled,
}

impl IndexingPolicy {
    /// Whether this is the default [`Self::Auto`] policy.
    ///
    /// Used as `LspServerConfig::indexing`'s `skip_serializing_if` (M4):
    /// without it, the generated default `mcpls.toml` would write `indexing
    /// = "auto"` into every one of its ~30 builtin server entries, unlike
    /// every other optional field in that struct, which is omitted at its
    /// default. Takes `&self`, not `self`, since `skip_serializing_if`
    /// requires `fn(&T) -> bool` (matching `Option::is_none`'s convention).
    #[expect(
        clippy::trivially_copy_pass_by_ref,
        reason = "serde's `skip_serializing_if` requires `fn(&T) -> bool`"
    )]
    #[must_use]
    pub(crate) const fn is_auto(&self) -> bool {
        matches!(self, Self::Auto)
    }
}

/// Which signal source a server has reported.
///
/// Both sources feed the same per-server state, so once one has spoken it must
/// not be silently overwritten by stale reasoning from the other (S2/#421): a
/// ready rust-analyzer `serverStatus` reading must not be knocked back to
/// `Loading` by a later `$/progress` sequence it never asked for, and
/// conversely a `ServerStatus` signal -- authoritative, since it is
/// rust-analyzer's own purpose-built readiness notification -- always
/// overrides whatever a generic `$/progress` sequence guessed, in any state.
/// See [`IndexingTracker::observe_server_status`] and
/// [`IndexingTracker::observe_progress`] for the exact precedence rules.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum IndexingSignalSource {
    /// rust-analyzer's `experimental/serverStatus`.
    ServerStatus,
    /// A generic `$/progress` `begin`/`end` sequence.
    Progress,
}

/// What is tracked for one server: the wait for a replacement's first signal,
/// or what a real signal produced.
#[derive(Debug, Clone)]
enum TrackedIndexing {
    /// A replacement process has not reported a signal yet; reads
    /// [`IndexingState::Loading`] until `since + within` has elapsed.
    AwaitingFirstSignal { since: Instant, within: Duration },
    /// A real signal has been observed.
    Signalled(Signalled),
}

/// What a `serverStatus` notification reported.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StatusReading {
    /// `quiescent: false`: indexing is in progress.
    Loading,
    /// `quiescent: true`: the initial workspace load is complete.
    Ready,
}

impl From<StatusReading> for IndexingState {
    fn from(reading: StatusReading) -> Self {
        match reading {
            StatusReading::Loading => Self::Loading,
            StatusReading::Ready => Self::Ready,
        }
    }
}

/// A tracked [`IndexingState`] plus the bookkeeping its source needs.
///
/// `last_updated` is the *only* timestamp of either variant (N5): both signal
/// sources refresh it on every accepted update, so [`IndexingTracker::state`]'s
/// staleness check (`INDEXING_STALENESS_BOUND`) works identically for either
/// source without a second, source-specific clock to keep in sync. For a
/// `Progress` source this also means a multi-phase load whose individual
/// phases each stay under the staleness bound remains gated for its whole
/// duration, even if the load as a whole runs past `INDEXING_STALENESS_BOUND`
/// -- only a single phase that itself never reports a boundary for that long
/// fails open.
#[derive(Debug, Clone)]
enum Signalled {
    /// Directly authoritative: the reading is the state.
    ServerStatus {
        reading: StatusReading,
        last_updated: Instant,
    },
    /// Recomputed on every read from the progress bookkeeping.
    Progress(ProgressEntry),
}

/// The bookkeeping of a `$/progress`-sourced readiness.
#[derive(Debug, Clone)]
struct ProgressEntry {
    last_updated: Instant,
    /// Progress tokens with an outstanding `begin` and no matching `end`
    /// yet, each paired with when its `begin` was accepted.
    ///
    /// A `HashMap`, not a `HashSet` (S1 / security HIGH): a single lost
    /// `end` -- realistic, since `try_send` drops frames once the
    /// lifecycle lane is full -- used to leave a token in the set forever,
    /// pinning [`IndexingTracker::state`] at `Loading` for the process
    /// lifetime even while unrelated later frames on the same server kept
    /// refreshing `last_updated` and masking the entry-wide staleness
    /// self-heal below. `observe_progress` evicts tokens older than
    /// `INDEXING_STALENESS_BOUND` on every write, and `state` additionally
    /// treats any individually stale token as absent even between writes,
    /// so the self-heal no longer depends on `last_updated` alone. Also
    /// capped at [`PROGRESS_OPEN_CAP`] so worst-case memory is bounded
    /// independent of eviction timing.
    open: HashMap<ProgressToken, Instant>,
    /// When `open` most recently became empty, if it ever has. `None` means
    /// "never been empty" -- critically different from "became empty a long
    /// time ago" (N2): both `state` and `latch` reads must treat them
    /// differently, or a fresh entry's very first `begin` reads/latches as
    /// though progress had already settled.
    empty_since: Option<Instant>,
    /// Once set, `observe_progress` ignores this server's `$/progress`
    /// stream forever and `state` reads `Ready` unconditionally. Set when a
    /// `begin` arrives after `open` has sat empty for at least
    /// `PROGRESS_LATCH_IDLE` -- the generic signal for "the workspace-load
    /// phase is over and everything after this is a per-request progress
    /// sequence (formatting, a single completion, ...), not more indexing".
    latched: bool,
}

impl ProgressEntry {
    /// A freshly observed entry, with no outstanding `begin` yet --
    /// `observe_progress` fills in `open`/`empty_since` right after this
    /// returns.
    fn new() -> Self {
        Self {
            last_updated: Instant::now(),
            open: HashMap::new(),
            empty_since: None,
            latched: false,
        }
    }
}

/// Custom notification method rust-analyzer uses to report workspace-load
/// progress; requires opting in at `initialize` (see `LspServer::initialize`).
const SERVER_STATUS_METHOD: &str = "experimental/serverStatus";

/// Boolean field on a [`SERVER_STATUS_METHOD`] payload: `true` once
/// rust-analyzer's initial workspace load is complete, `false` while it is
/// still in progress.
const QUIESCENT_FIELD: &str = "quiescent";

/// Once a `Loading` entry has gone this long without a fresh signal, reads
/// stop trusting it -- see [`IndexingTracker::state`].
///
/// Deliberately larger than `navigation::INDEXING_READY_TIMEOUT` (30s) and
/// anchored to the signal's own age, not to any individual caller's wait:
/// a caller that times out must never affect another concurrent or later
/// caller's deadline, only the age of the last real signal does. See
/// `navigation.rs`'s const-asserts for the cross-checked ordering between
/// this, `INDEXING_READY_TIMEOUT`, and the two constants below.
pub const INDEXING_STALENESS_BOUND: Duration = Duration::from_mins(1);

/// Default maximum time, in seconds, `Translator::wait_for_indexing_ready`
/// waits for a routed LSP server to report indexing readiness.
///
/// Only takes effect once a readiness signal has shown indexing is actually
/// in progress. A raw `u64` (not a `Duration`), so
/// `config::default_indexing_ready_timeout_seconds` can share this single
/// source of truth without a config -> translator module dependency;
/// `navigation::INDEXING_READY_TIMEOUT` derives its `Duration` from this
/// same value. Overridable via `workspace.indexing_ready_timeout_seconds` in
/// `mcpls.toml` (#424).
pub const DEFAULT_INDEXING_READY_TIMEOUT_SECS: u64 = 30;

/// How long a `Progress`-sourced entry's `open` set must stay empty before a
/// *read* trusts it as settled and reports `Ready` -- the read-path
/// counterpart to [`PROGRESS_LATCH_IDLE`] below.
///
/// Covers the ordinary inter-phase gap in a multi-phase load (e.g. gopls
/// ending "Setting up workspace" and taking a few seconds to run `go list`
/// before beginning "Loading packages"): without this, a read landing in
/// that gap would report `Ready` mid-load. 3s, not 1s: shortening it only
/// narrows this real gap-covering window while buying nothing, since the
/// cost is a few seconds of latency paid once per server lifetime and only
/// by servers that emit progress at all.
pub const PROGRESS_SETTLE: Duration = Duration::from_secs(3);

/// How long a `Progress`-sourced entry's `open` set must stay empty before a
/// `begin` write latches it `Ready` forever, instead of re-gating.
///
/// Deliberately a *separate*, larger threshold than [`PROGRESS_SETTLE`]
/// (N1): collapsing them into one 3s threshold reintroduced the exact
/// regression this two-threshold split fixes. gopls's own multi-phase load
/// routinely leaves a several-second gap between ending one phase and
/// beginning the next -- with a single 3s threshold, that ordinary gap
/// would permanently latch the tracker `Ready` after the *first* phase,
/// leaving the actual package-loading phase (the part that matters) to run
/// completely ungated. 30s is long enough that only a genuine end of the
/// workspace-load phase -- followed by a real per-request sequence
/// (formatting, one completion) -- crosses it.
pub const PROGRESS_LATCH_IDLE: Duration = Duration::from_secs(30);

/// Hard cap on the number of distinct tokens tracked in one server's
/// `open` set (security HIGH).
///
/// 64 is far above any real server's concurrent-operation count. Bounds
/// worst-case memory independent of eviction timing -- a server flooding
/// many distinct never-ending tokens within a single
/// `INDEXING_STALENESS_BOUND` window would otherwise grow `open` without
/// limit before the time-based eviction in `observe_progress` ever gets a
/// chance to shrink it. Once reached, `observe_progress` stops inserting
/// and `state` fails open (`Unknown`) rather than trusting an accumulator
/// this large.
const PROGRESS_OPEN_CAP: usize = 64;

/// Maximum accepted length, in bytes, of a `ProgressToken::String` before
/// [`IndexingTracker::observe_progress`] rejects the frame outright
/// (security HIGH).
///
/// Without this, [`PROGRESS_OPEN_CAP`] bounds token *count* but not size --
/// a server-chosen token string is otherwise bounded only by the
/// transport's 10 MiB `MAX_CONTENT_LENGTH`, so the count cap alone could
/// still admit up to `PROGRESS_OPEN_CAP * 10 MiB` of retained token bytes.
/// A few hundred bytes is far more than any real progress token needs.
const PROGRESS_TOKEN_MAX_LEN: usize = 256;

/// Everything known about one server's indexing readiness, so no per-server
/// fact can exist without the others.
#[derive(Debug, Default)]
struct ServerIndexing {
    policy: IndexingPolicy,
    /// The signal source the server last reported; survives
    /// [`IndexingTracker::reset`] like `policy` and decides whether a
    /// replacement is awaited ([`IndexingReset::AwaitReplacement`]).
    seen: Option<IndexingSignalSource>,
    tracked: Option<TrackedIndexing>,
}

impl ServerIndexing {
    fn is_disabled(&self) -> bool {
        self.policy == IndexingPolicy::Disabled
    }

    fn log_wait_ended(&self, server_id: &ServerId) {
        if matches!(
            self.tracked,
            Some(TrackedIndexing::AwaitingFirstSignal { .. })
        ) {
            tracing::debug!(%server_id, "indexing signal ends the pre-signal wait");
        }
    }
}

/// Per-server workspace-indexing readiness state plus the escape-hatch
/// policy, owned by [`NotificationCache`](super::NotificationCache) and
/// delegated to for every indexing-related call.
#[derive(Debug, Default)]
pub struct IndexingTracker {
    servers: HashMap<ServerId, ServerIndexing>,
}

impl IndexingTracker {
    /// Create an empty tracker: every server starts at [`IndexingState::Unknown`]
    /// under [`IndexingPolicy::Auto`].
    pub(crate) fn new() -> Self {
        Self::default()
    }

    /// Configure `server_id`'s [`IndexingPolicy`], set once at server
    /// registration from `LspServerConfig::indexing` before any signal for
    /// it can arrive.
    pub(crate) fn set_policy(&mut self, server_id: ServerId, policy: IndexingPolicy) {
        self.servers.entry(server_id).or_default().policy = policy;
    }

    fn is_disabled(&self, server_id: &ServerId) -> bool {
        self.servers
            .get(server_id)
            .is_some_and(ServerIndexing::is_disabled)
    }

    /// Record a workspace-readiness signal from an unrecognized/custom LSP
    /// notification (`LspNotification::Other`).
    ///
    /// Currently recognizes rust-analyzer's `experimental/serverStatus`
    /// notification: a `quiescent` boolean of `false` marks the server
    /// [`IndexingState::Loading`], `true` marks it [`IndexingState::Ready`].
    /// Any other method, or a `serverStatus` payload missing/malformed the
    /// field, leaves the current entry entirely untouched -- including its
    /// source -- rather than erroring: a malformed frame must not mark
    /// this server `ServerStatus`-sourced and thereby permanently disable
    /// its `$/progress` path via [`Self::observe_progress`]'s stickiness
    /// check (N6).
    ///
    /// Once a `ServerStatus`-sourced entry reaches [`IndexingState::Ready`]
    /// it never regresses on its own (sticky); a `ServerStatus` signal
    /// always overwrites a `Progress`-sourced entry regardless of that
    /// entry's current state, since it is the more authoritative source.
    pub(crate) fn observe_server_status(
        &mut self,
        server_id: &ServerId,
        method: &str,
        params: Option<&serde_json::Value>,
    ) {
        if self.is_disabled(server_id) {
            return;
        }
        if method != SERVER_STATUS_METHOD {
            return;
        }
        let Some(quiescent) = params
            .and_then(|p| p.get(QUIESCENT_FIELD))
            .and_then(serde_json::Value::as_bool)
        else {
            return;
        };

        let server = self.servers.entry(server_id.clone()).or_default();
        let sticky_ready = matches!(
            server.tracked,
            Some(TrackedIndexing::Signalled(Signalled::ServerStatus {
                reading: StatusReading::Ready,
                ..
            }))
        );
        if sticky_ready {
            return;
        }

        server.seen = Some(IndexingSignalSource::ServerStatus);
        server.log_wait_ended(server_id);
        server.tracked = Some(TrackedIndexing::Signalled(Signalled::ServerStatus {
            reading: if quiescent {
                StatusReading::Ready
            } else {
                StatusReading::Loading
            },
            last_updated: Instant::now(),
        }));
    }

    /// Record a `$/progress` notification toward `server_id`'s readiness.
    ///
    /// A no-op once a well-formed `experimental/serverStatus` signal has
    /// been seen for this server (P2b/S2): that source is authoritative and
    /// sticky, so a `$/progress` sequence arriving after it (e.g.
    /// rust-analyzer's own pre-`serverStatus` startup progress, or a later
    /// per-request sequence) must never resurrect or override it.
    ///
    /// Otherwise implements the settle/latch transition (see
    /// [`PROGRESS_SETTLE`]/[`PROGRESS_LATCH_IDLE`] docs for the two
    /// thresholds this balances):
    /// - an unparseable frame (`report`, or missing/unrecognized `kind`) is
    ///   ignored outright, as is a `ProgressToken::String` longer than
    ///   [`PROGRESS_TOKEN_MAX_LEN`] (security HIGH);
    /// - once `latched`, every later frame is ignored;
    /// - every accepted frame first evicts tokens whose `begin` is older
    ///   than `INDEXING_STALENESS_BOUND` from `open` (S1) -- otherwise a
    ///   single lost `end` leaves a token in `open` forever, and unrelated
    ///   later frames on the same server keep refreshing `last_updated`
    ///   and masking the entry-wide staleness self-heal that would
    ///   otherwise catch it;
    /// - `begin`: if `open` is empty and has been for at least
    ///   [`PROGRESS_LATCH_IDLE`], latches instead of reopening -- **the
    ///   `PROGRESS_LATCH_IDLE` check only fires when `empty_since` is
    ///   `Some`**; a fresh entry's very first `begin` has `empty_since ==
    ///   None` ("never been empty", not "empty forever ago") and must
    ///   always open normally, never latch (N2) -- otherwise the entire
    ///   feature silently no-ops on the very first workspace load. Anything
    ///   short of the latch threshold inserts the token (unless
    ///   [`PROGRESS_OPEN_CAP`] is already reached, security HIGH: the frame
    ///   is then dropped rather than grown further) and clears
    ///   `empty_since` normally, including the ordinary inter-phase gap a
    ///   multi-phase load leaves between `PROGRESS_SETTLE` and
    ///   `PROGRESS_LATCH_IDLE` (N1) -- that gap must re-gate, not latch;
    /// - `end`: sets `empty_since` **only if this token was actually open**
    ///   (`open.remove` returned `Some`) and `open` is now empty (N3) -- an
    ///   unmatched `end` (a dropped `begin`, or one arriving after latching)
    ///   must not fabricate a settle window on an already-empty entry.
    pub(crate) fn observe_progress(&mut self, server_id: &ServerId, params: &ProgressParams) {
        if self.is_disabled(server_id) {
            return;
        }
        let Some(kind) = ProgressKind::from_value(&params.value) else {
            return;
        };
        if let Some(server) = self.servers.get(server_id) {
            match &server.tracked {
                Some(TrackedIndexing::Signalled(Signalled::ServerStatus { .. })) => return,
                // Progress frames of a server that reports `serverStatus` are not the signal awaited.
                Some(TrackedIndexing::AwaitingFirstSignal { .. })
                    if server.seen != Some(IndexingSignalSource::Progress) =>
                {
                    return;
                }
                _ => {}
            }
        }
        if let ProgressToken::String(token) = &params.token
            && token.len() > PROGRESS_TOKEN_MAX_LEN
        {
            // Reject outright, before PROGRESS_OPEN_CAP could be multiplied by an oversized token (security HIGH).
            return;
        }

        let server = self.servers.entry(server_id.clone()).or_default();
        server.seen.get_or_insert(IndexingSignalSource::Progress);
        server.log_wait_ended(server_id);
        let tracked = server.tracked.get_or_insert_with(|| {
            TrackedIndexing::Signalled(Signalled::Progress(ProgressEntry::new()))
        });
        if matches!(tracked, TrackedIndexing::AwaitingFirstSignal { .. }) {
            *tracked = TrackedIndexing::Signalled(Signalled::Progress(ProgressEntry::new()));
        }
        let TrackedIndexing::Signalled(Signalled::Progress(entry)) = tracked else {
            return;
        };
        if entry.latched {
            return;
        }

        let now = Instant::now();
        // Age out tokens whose `begin` never got a matching `end` (S1) -- see `ProgressEntry::open`'s doc.
        entry
            .open
            .retain(|_, began| began.elapsed() < INDEXING_STALENESS_BOUND);

        match kind {
            ProgressKind::Begin => {
                if entry.open.is_empty()
                    && entry
                        .empty_since
                        .is_some_and(|since| since.elapsed() >= PROGRESS_LATCH_IDLE)
                {
                    entry.latched = true;
                    entry.open.clear();
                    return;
                }
                if entry.open.len() < PROGRESS_OPEN_CAP {
                    entry.open.insert(params.token.clone(), now);
                }
                // Else at PROGRESS_OPEN_CAP already -- deliberately not inserted; `state` fails open instead.
                entry.empty_since = None;
            }
            ProgressKind::End => {
                if entry.open.remove(&params.token).is_some() && entry.open.is_empty() {
                    entry.empty_since = Some(now);
                }
            }
        }
        entry.last_updated = now;
    }

    /// Current tracked workspace-indexing readiness for `server_id`.
    ///
    /// Returns [`IndexingState::Unknown`] if no signal has ever been
    /// observed for `server_id`, or if it is configured
    /// [`IndexingPolicy::Disabled`].
    ///
    /// A `ServerStatus`-sourced `Loading` entry older than
    /// `INDEXING_STALENESS_BOUND` reads back as `Unknown` rather than
    /// `Loading`: the self-heal for a `quiescent: true` notification dropped
    /// by a full channel, or a server that stalled mid-index. A
    /// `Progress`-sourced entry applies the same staleness bound to
    /// `last_updated` regardless of `open`/`latched` as a coarse entry-wide
    /// gate, then: `latched` reads `Ready` unconditionally; `open.len() >=
    /// PROGRESS_OPEN_CAP` reads `Unknown` (security HIGH -- too many
    /// concurrently open tokens to track reliably, matching
    /// `observe_progress`'s refusal to grow `open` further); otherwise any
    /// token in `open` individually younger than `INDEXING_STALENESS_BOUND`
    /// reads `Loading` (S1: computed here, not just via write-time
    /// eviction, so a token that went stale purely with the passage of time
    /// between writes still stops pinning `Loading` on the very next read).
    /// If `open` is truly empty, reads `Loading` for [`PROGRESS_SETTLE`]
    /// after `empty_since`, `Ready` once past that -- but only when
    /// `empty_since` is `Some`, i.e. a real `begin`-then-`end` transition
    /// was actually observed to complete. `empty_since == None` reads
    /// `Unknown` instead of `Ready`: the only way to reach `open` empty
    /// with `empty_since` still `None` is an entry whose first-ever
    /// `$/progress` frame was an unmatched `end` (e.g. its `begin` was
    /// dropped) -- there is no evidence indexing ever finished, so this
    /// must fail open rather than falsely report readiness. If `open`
    /// is non-empty but every token in it is individually stale (a lost
    /// `end` that unrelated later frames kept `last_updated` fresh enough to
    /// survive the entry-wide gate above), also reads `Unknown` rather than
    /// running the settle/latch logic meant for a genuinely-observed `end`.
    pub(crate) fn state(&self, server_id: &ServerId) -> IndexingState {
        let Some(server) = self.servers.get(server_id) else {
            return IndexingState::Unknown;
        };
        if server.is_disabled() {
            return IndexingState::Unknown;
        }
        match &server.tracked {
            None => IndexingState::Unknown,
            Some(TrackedIndexing::AwaitingFirstSignal { since, within }) => {
                if since.elapsed() < *within {
                    IndexingState::Loading
                } else {
                    IndexingState::Unknown
                }
            }
            Some(TrackedIndexing::Signalled(signalled)) => Self::signalled_state(signalled),
        }
    }

    fn signalled_state(signalled: &Signalled) -> IndexingState {
        match signalled {
            Signalled::ServerStatus {
                reading,
                last_updated,
            } => {
                if *reading == StatusReading::Loading
                    && last_updated.elapsed() >= INDEXING_STALENESS_BOUND
                {
                    IndexingState::Unknown
                } else {
                    (*reading).into()
                }
            }
            Signalled::Progress(entry) => Self::progress_state(entry),
        }
    }

    fn progress_state(entry: &ProgressEntry) -> IndexingState {
        if entry.latched {
            return IndexingState::Ready;
        }
        if entry.last_updated.elapsed() >= INDEXING_STALENESS_BOUND {
            return IndexingState::Unknown;
        }
        if entry.open.len() >= PROGRESS_OPEN_CAP {
            return IndexingState::Unknown;
        }
        let has_live_open = entry
            .open
            .values()
            .any(|began| began.elapsed() < INDEXING_STALENESS_BOUND);
        if has_live_open {
            return IndexingState::Loading;
        }
        if entry.open.is_empty() {
            match entry.empty_since {
                Some(since) if since.elapsed() < PROGRESS_SETTLE => IndexingState::Loading,
                Some(_) => IndexingState::Ready,
                // No transition has ever been observed to settle.
                None => IndexingState::Unknown,
            }
        } else {
            IndexingState::Unknown
        }
    }

    /// Reset `server_id`'s tracked entry as `reset` says. Touches neither its
    /// [`IndexingPolicy`] (see [`Self::set_policy`]) -- a respawned process
    /// still honors whatever the static config said -- nor which signal
    /// source it has reported before.
    pub(crate) fn reset(&mut self, server_id: &ServerId, reset: IndexingReset) {
        let Some(server) = self.servers.get_mut(server_id) else {
            return;
        };
        server.tracked = None;
        let IndexingReset::AwaitReplacement { within } = reset else {
            return;
        };
        if server.is_disabled() || server.seen.is_none() {
            return;
        }
        tracing::debug!(%server_id, ?within, "replacement awaits its first indexing signal");
        server.tracked = Some(TrackedIndexing::AwaitingFirstSignal {
            since: Instant::now(),
            within,
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn test_server() -> ServerId {
        ServerId::from_static("test-server")
    }

    fn progress(kind: &str, token: i32) -> ProgressParams {
        ProgressParams {
            token: ProgressToken::Int(token),
            value: serde_json::json!({ "kind": kind }),
        }
    }

    #[test]
    fn test_state_defaults_unknown() {
        let tracker = IndexingTracker::new();
        assert_eq!(tracker.state(&test_server()), IndexingState::Unknown);
    }

    #[test]
    fn test_observe_server_status_quiescent_false_marks_loading() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": false})),
        );
        assert_eq!(tracker.state(&server), IndexingState::Loading);
    }

    #[test]
    fn test_observe_server_status_quiescent_true_marks_ready() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": true})),
        );
        assert_eq!(tracker.state(&server), IndexingState::Ready);
    }

    #[test]
    fn test_observe_server_status_ignores_unrecognized_method() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_server_status(
            &server,
            "window/logMessage",
            Some(&serde_json::json!({"quiescent": false})),
        );
        assert_eq!(tracker.state(&server), IndexingState::Unknown);
    }

    /// N6: a malformed `serverStatus` payload must not mark the entry's
    /// source, or it would permanently disable that server's `$/progress`
    /// path via `observe_progress`'s stickiness check even though no
    /// well-formed `ServerStatus` signal was ever actually seen.
    #[test]
    fn test_malformed_server_status_does_not_mark_source_or_disable_progress() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_server_status(&server, "experimental/serverStatus", None);
        assert_eq!(tracker.state(&server), IndexingState::Unknown);

        tracker.observe_progress(&server, &progress("begin", 1));
        assert_eq!(
            tracker.state(&server),
            IndexingState::Loading,
            "the progress path must still be live after a malformed serverStatus payload"
        );
    }

    #[test]
    fn test_observe_server_status_ready_is_sticky() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": true})),
        );
        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": false})),
        );
        assert_eq!(tracker.state(&server), IndexingState::Ready);
    }

    #[test]
    fn test_observe_server_status_tracks_servers_independently() {
        let mut tracker = IndexingTracker::new();
        let rust = ServerId::from_static("rust");
        let python = ServerId::from_static("python");
        tracker.observe_server_status(
            &rust,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": false})),
        );
        assert_eq!(tracker.state(&rust), IndexingState::Loading);
        assert_eq!(tracker.state(&python), IndexingState::Unknown);
    }

    #[tokio::test(start_paused = true)]
    async fn test_state_treats_stale_server_status_loading_as_unknown() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": false})),
        );
        assert_eq!(tracker.state(&server), IndexingState::Loading);

        tokio::time::advance(INDEXING_STALENESS_BOUND.saturating_sub(Duration::from_secs(1))).await;
        assert_eq!(tracker.state(&server), IndexingState::Loading);

        tokio::time::advance(Duration::from_secs(2)).await;
        assert_eq!(tracker.state(&server), IndexingState::Unknown);
    }

    #[tokio::test(start_paused = true)]
    async fn test_observe_server_status_refreshes_staleness_clock() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": false})),
        );

        tokio::time::advance(INDEXING_STALENESS_BOUND.saturating_sub(Duration::from_secs(1))).await;
        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": false})),
        );

        tokio::time::advance(INDEXING_STALENESS_BOUND.saturating_sub(Duration::from_secs(1))).await;
        assert_eq!(tracker.state(&server), IndexingState::Loading);
    }

    /// N2: a fresh entry's very first `begin` must open (`Loading`), never
    /// latch straight to `Ready` -- the `empty_since == None` ("never been
    /// empty") case must not satisfy the latch-idle check.
    #[test]
    fn test_first_begin_on_fresh_entry_yields_loading_never_ready() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1));
        assert_eq!(tracker.state(&server), IndexingState::Loading);
    }

    /// N3: an `end` for a token that was never open must not create (or
    /// refresh) `empty_since` on an already-empty entry.
    #[tokio::test(start_paused = true)]
    async fn test_unmatched_end_does_not_create_empty_since() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        // Open and close one legitimate operation first, establishing a real (old) empty_since.
        tracker.observe_progress(&server, &progress("begin", 1));
        tracker.observe_progress(&server, &progress("end", 1));

        tokio::time::advance(PROGRESS_SETTLE + Duration::from_secs(1)).await;
        assert_eq!(
            tracker.state(&server),
            IndexingState::Ready,
            "settled after the real end"
        );

        // An end for a token that was never open must not refresh empty_since back to "just now".
        tracker.observe_progress(&server, &progress("end", 99));
        assert_eq!(
            tracker.state(&server),
            IndexingState::Ready,
            "an unmatched end must not manufacture a fresh settle window"
        );
    }

    /// Fix 8 (code-review round 2): a fresh entry whose first-ever
    /// `$/progress` frame is an `end` (e.g. its matching `begin` was
    /// dropped by a full lifecycle lane) must read `Unknown`, not `Ready`
    /// -- `empty_since == None` means "no load-to-quiescent transition has
    /// ever been observed", not "already settled".
    #[tokio::test(start_paused = true)]
    async fn test_end_as_first_ever_frame_reads_unknown_not_ready() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("end", 1)); // no prior begin

        assert_eq!(
            tracker.state(&server),
            IndexingState::Unknown,
            "an end with no observed prior begin must not read Ready immediately"
        );

        tokio::time::advance(PROGRESS_SETTLE + Duration::from_secs(1)).await;
        assert_eq!(
            tracker.state(&server),
            IndexingState::Unknown,
            "must still read Unknown once the settle window would have expired -- there was \
             never a real empty_since to settle from"
        );
    }

    /// `begin` after the `open` set has sat empty for at least
    /// `PROGRESS_LATCH_IDLE` latches the entry `Ready` forever.
    #[tokio::test(start_paused = true)]
    async fn test_begin_after_latch_idle_gap_latches_permanently() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1));
        tracker.observe_progress(&server, &progress("end", 1));

        tokio::time::advance(PROGRESS_LATCH_IDLE + Duration::from_secs(1)).await;
        tracker.observe_progress(&server, &progress("begin", 2));
        assert_eq!(
            tracker.state(&server),
            IndexingState::Ready,
            "a begin after PROGRESS_LATCH_IDLE of quiet must latch, not re-gate"
        );

        // Latched forever: even ending the "reopened" op, or a later begin, changes nothing.
        tracker.observe_progress(&server, &progress("end", 2));
        tracker.observe_progress(&server, &progress("begin", 3));
        assert_eq!(tracker.state(&server), IndexingState::Ready);
    }

    /// N1 (gopls regression guard): a `begin` after a gap strictly between
    /// `PROGRESS_SETTLE` and `PROGRESS_LATCH_IDLE` must re-gate as `Loading`,
    /// not latch -- this is the ordinary inter-phase gap shape gopls uses.
    #[tokio::test(start_paused = true)]
    async fn test_begin_after_mid_gap_regates_and_does_not_latch() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1));
        tracker.observe_progress(&server, &progress("end", 1));

        tokio::time::advance(Duration::from_secs(10)).await; // between 3s and 30s
        tracker.observe_progress(&server, &progress("begin", 2));
        assert_eq!(
            tracker.state(&server),
            IndexingState::Loading,
            "a mid-gap begin must re-gate the next phase, not latch it away"
        );
    }

    /// S1 (core): a phase gap shorter than `PROGRESS_SETTLE` must not read
    /// `Ready` mid-gap.
    #[tokio::test(start_paused = true)]
    async fn test_phase_gap_shorter_than_settle_does_not_read_ready() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1));
        tracker.observe_progress(&server, &progress("end", 1));

        tokio::time::advance(PROGRESS_SETTLE.checked_sub(Duration::from_secs(1)).unwrap()).await;
        assert_eq!(tracker.state(&server), IndexingState::Loading);
    }

    #[test]
    fn test_open_non_empty_reads_loading() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1));
        tracker.observe_progress(&server, &progress("begin", 2));
        tracker.observe_progress(&server, &progress("end", 1));
        assert_eq!(
            tracker.state(&server),
            IndexingState::Loading,
            "token 2 is still open"
        );
    }

    /// Multi-server isolation for the `Progress` source, mirroring
    /// `test_observe_server_status_tracks_servers_independently` for
    /// `ServerStatus` -- the same `HashMap<ServerId, _>` isolation, proven
    /// for the other source too.
    #[test]
    fn test_observe_progress_tracks_servers_independently() {
        let mut tracker = IndexingTracker::new();
        let rust = ServerId::from_static("rust");
        let go = ServerId::from_static("go");

        tracker.observe_progress(&rust, &progress("begin", 1));

        assert_eq!(
            tracker.state(&rust),
            IndexingState::Loading,
            "rust's token is still open"
        );
        assert_eq!(
            tracker.state(&go),
            IndexingState::Unknown,
            "go has received no progress signal of its own"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn test_progress_last_updated_past_staleness_bound_reads_unknown() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1));

        tokio::time::advance(INDEXING_STALENESS_BOUND + Duration::from_secs(1)).await;
        assert_eq!(
            tracker.state(&server),
            IndexingState::Unknown,
            "a single op with no phase boundary for over a minute must fail open"
        );
    }

    /// N5: a multi-phase load whose *individual* phases each refresh
    /// `last_updated` stays gated past the raw 60s bound, as long as no
    /// single phase itself runs that long uninterrupted.
    #[tokio::test(start_paused = true)]
    async fn test_multiphase_load_past_staleness_bound_with_boundaries_stays_loading() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1));

        tokio::time::advance(Duration::from_secs(40)).await;
        tracker.observe_progress(&server, &progress("end", 1));
        tracker.observe_progress(&server, &progress("begin", 2));

        tokio::time::advance(Duration::from_secs(40)).await;
        // 80s total elapsed, but every gap between refreshes stayed under INDEXING_STALENESS_BOUND (60s).
        assert_eq!(tracker.state(&server), IndexingState::Loading);
    }

    /// S1 (security HIGH regression guard): a lost `end` must not pin
    /// `state` at `Loading` for the process lifetime, even while unrelated
    /// later frames on the *same* server keep refreshing `last_updated` --
    /// the exact mechanism that let this bug survive the pre-existing
    /// entry-wide-only staleness self-heal.
    #[tokio::test(start_paused = true)]
    async fn test_lost_end_self_heals_after_staleness_bound() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1)); // never gets an `end`

        // Unrelated periodic activity on a different token keeps last_updated fresh.
        tokio::time::advance(Duration::from_secs(30)).await;
        tracker.observe_progress(&server, &progress("begin", 2));
        tracker.observe_progress(&server, &progress("end", 2));

        tokio::time::advance(Duration::from_secs(25)).await; // t=55s
        tracker.observe_progress(&server, &progress("begin", 3));
        tracker.observe_progress(&server, &progress("end", 3));

        assert_eq!(
            tracker.state(&server),
            IndexingState::Loading,
            "token 1 is still within its own staleness window at t=55s"
        );

        // Token 1's own begin (t=0) is now stale (t=61s), though last_updated (t=55s) is not.
        tokio::time::advance(Duration::from_secs(6)).await; // t=61s
        assert_eq!(
            tracker.state(&server),
            IndexingState::Unknown,
            "a token whose own begin exceeded INDEXING_STALENESS_BOUND must stop pinning \
             Loading even while unrelated later frames keep the entry-wide last_updated clock \
             fresh"
        );
    }

    /// Security HIGH: reaching `PROGRESS_OPEN_CAP` distinct open tokens
    /// must fail open (`Unknown`) instead of growing `open` further or
    /// staying `Loading` forever.
    #[test]
    fn test_open_token_cap_forces_unknown() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        for i in 0..(PROGRESS_OPEN_CAP - 1) {
            tracker.observe_progress(&server, &progress("begin", i32::try_from(i).unwrap()));
        }
        assert_eq!(
            tracker.state(&server),
            IndexingState::Loading,
            "PROGRESS_OPEN_CAP - 1 distinct open tokens must still read Loading"
        );

        tracker.observe_progress(
            &server,
            &progress("begin", i32::try_from(PROGRESS_OPEN_CAP - 1).unwrap()),
        );
        assert_eq!(
            tracker.state(&server),
            IndexingState::Unknown,
            "reaching PROGRESS_OPEN_CAP must fail open instead of staying Loading forever"
        );
    }

    /// Security HIGH: an oversized `ProgressToken::String` must be rejected
    /// outright, before it can multiply `PROGRESS_OPEN_CAP` by up to
    /// `MAX_CONTENT_LENGTH` (10 MiB) per token.
    #[test]
    fn test_oversized_string_token_is_rejected() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        let oversized = ProgressParams {
            token: ProgressToken::String("x".repeat(PROGRESS_TOKEN_MAX_LEN + 1)),
            value: serde_json::json!({ "kind": "begin" }),
        };
        tracker.observe_progress(&server, &oversized);
        assert_eq!(
            tracker.state(&server),
            IndexingState::Unknown,
            "an oversized token must be dropped outright, not tracked"
        );
    }

    #[test]
    fn test_malformed_progress_kind_ignored() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("report", 1));
        assert_eq!(tracker.state(&server), IndexingState::Unknown);

        let missing_kind = ProgressParams {
            token: ProgressToken::Int(1),
            value: serde_json::json!({}),
        };
        tracker.observe_progress(&server, &missing_kind);
        assert_eq!(tracker.state(&server), IndexingState::Unknown);
    }

    #[test]
    fn test_disabled_policy_pins_unknown() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.set_policy(server.clone(), IndexingPolicy::Disabled);

        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": false})),
        );
        tracker.observe_progress(&server, &progress("begin", 1));
        assert_eq!(tracker.state(&server), IndexingState::Unknown);
    }

    /// S2 (#421 regression guard): a settled `$/progress` sequence followed
    /// by `quiescent: false` must still report `Loading` -- the
    /// `ServerStatus` source must be able to override a `Progress` entry in
    /// any state, not just while that entry itself reads `Loading`.
    #[tokio::test(start_paused = true)]
    async fn test_progress_settle_then_quiescent_false_yields_loading() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1));
        tracker.observe_progress(&server, &progress("end", 1));
        tokio::time::advance(PROGRESS_SETTLE + Duration::from_secs(1)).await;
        assert_eq!(tracker.state(&server), IndexingState::Ready);

        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": false})),
        );
        assert_eq!(tracker.state(&server), IndexingState::Loading);
    }

    /// P2b: once a `ServerStatus` signal has been seen, later `$/progress`
    /// frames are ignored entirely -- rust-analyzer's own pre-`serverStatus`
    /// `$/progress` chatter (or a later per-request sequence) must never
    /// resurrect or override the authoritative source.
    #[test]
    fn test_server_status_entry_ignores_later_progress_frames() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_server_status(
            &server,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": true})),
        );
        tracker.observe_progress(&server, &progress("begin", 1));
        assert_eq!(tracker.state(&server), IndexingState::Ready);
    }

    #[test]
    fn test_reset_clears_entry_but_not_policy() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.set_policy(server.clone(), IndexingPolicy::Disabled);
        tracker.observe_progress(&server, &progress("begin", 1));
        tracker.reset(&server, IndexingReset::Forget);
        assert_eq!(
            tracker.state(&server),
            IndexingState::Unknown,
            "policy stays Disabled, so this reads Unknown regardless of the reset entry"
        );

        let mut auto_tracker = IndexingTracker::new();
        auto_tracker.observe_progress(&server, &progress("begin", 1));
        auto_tracker.reset(&server, IndexingReset::Forget);
        assert_eq!(auto_tracker.state(&server), IndexingState::Unknown);
    }

    const WITHIN: Duration = Duration::from_secs(30);

    fn server_status(tracker: &mut IndexingTracker, server: &ServerId, quiescent: bool) {
        tracker.observe_server_status(
            server,
            "experimental/serverStatus",
            Some(&serde_json::json!({ "quiescent": quiescent })),
        );
    }

    fn await_replacement(tracker: &mut IndexingTracker, server: &ServerId) {
        tracker.reset(server, IndexingReset::AwaitReplacement { within: WITHIN });
    }

    #[test]
    fn test_await_replacement_without_history_stays_unknown() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        await_replacement(&mut tracker, &server);
        assert_eq!(tracker.state(&server), IndexingState::Unknown);
    }

    #[test]
    fn test_await_replacement_after_ready_reads_loading() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        server_status(&mut tracker, &server, true);
        await_replacement(&mut tracker, &server);
        assert_eq!(tracker.state(&server), IndexingState::Loading);
    }

    #[test]
    fn test_await_replacement_after_loading_reads_loading() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        server_status(&mut tracker, &server, false);
        await_replacement(&mut tracker, &server);
        assert_eq!(tracker.state(&server), IndexingState::Loading);
    }

    #[tokio::test(start_paused = true)]
    async fn test_await_replacement_ends_after_within() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        server_status(&mut tracker, &server, true);
        await_replacement(&mut tracker, &server);

        tokio::time::advance(WITHIN.saturating_sub(Duration::from_secs(1))).await;
        assert_eq!(tracker.state(&server), IndexingState::Loading);
        tokio::time::advance(Duration::from_secs(2)).await;
        assert_eq!(tracker.state(&server), IndexingState::Unknown);
    }

    #[test]
    fn test_await_replacement_ready_signal_ends_wait_at_once() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        server_status(&mut tracker, &server, true);
        await_replacement(&mut tracker, &server);
        server_status(&mut tracker, &server, true);
        assert_eq!(tracker.state(&server), IndexingState::Ready);
    }

    #[test]
    fn test_await_replacement_loading_signal_is_a_real_loading_entry() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        server_status(&mut tracker, &server, true);
        await_replacement(&mut tracker, &server);
        server_status(&mut tracker, &server, false);
        assert_eq!(tracker.state(&server), IndexingState::Loading);
        server_status(&mut tracker, &server, true);
        assert_eq!(tracker.state(&server), IndexingState::Ready);
    }

    #[test]
    fn test_await_replacement_ignores_progress_of_a_server_status_server() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        server_status(&mut tracker, &server, true);
        await_replacement(&mut tracker, &server);
        tracker.observe_progress(&server, &progress("begin", 1));
        tracker.observe_progress(&server, &progress("end", 1));
        assert_eq!(
            tracker.state(&server),
            IndexingState::Loading,
            "startup progress must not end the wait for the serverStatus signal"
        );
    }

    #[test]
    fn test_await_replacement_progress_server_is_replaced_by_progress_entry() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.observe_progress(&server, &progress("begin", 1));
        tracker.observe_progress(&server, &progress("end", 1));
        await_replacement(&mut tracker, &server);
        assert_eq!(tracker.state(&server), IndexingState::Loading);

        tracker.observe_progress(&server, &progress("begin", 2));
        assert_eq!(tracker.state(&server), IndexingState::Loading);
        tracker.observe_progress(&server, &progress("end", 2));
        assert_eq!(
            tracker.state(&server),
            IndexingState::Loading,
            "settle window of the fresh progress entry"
        );
    }

    #[test]
    fn test_await_replacement_twice_restarts_the_wait() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        server_status(&mut tracker, &server, true);
        await_replacement(&mut tracker, &server);
        await_replacement(&mut tracker, &server);
        assert_eq!(tracker.state(&server), IndexingState::Loading);
        tracker.reset(&server, IndexingReset::Forget);
        assert_eq!(tracker.state(&server), IndexingState::Unknown);
    }

    #[test]
    fn test_await_replacement_respects_disabled_policy() {
        let mut tracker = IndexingTracker::new();
        let server = test_server();
        tracker.set_policy(server.clone(), IndexingPolicy::Disabled);
        await_replacement(&mut tracker, &server);
        assert_eq!(tracker.state(&server), IndexingState::Unknown);
    }

    #[test]
    fn test_await_replacement_tracks_servers_independently() {
        let mut tracker = IndexingTracker::new();
        let rust = ServerId::from_static("rust");
        let python = ServerId::from_static("python");
        server_status(&mut tracker, &rust, true);
        server_status(&mut tracker, &python, true);
        await_replacement(&mut tracker, &rust);
        assert_eq!(tracker.state(&rust), IndexingState::Loading);
        assert_eq!(tracker.state(&python), IndexingState::Ready);
    }

    /// End-to-end `$/progress` handling over a real [`LspClient`] on a mock
    /// transport: the server-initiated `create` request, the lane routing of
    /// `begin`/`report`/`end`, and the tracked [`IndexingState`] through the
    /// real notification pump.
    mod mock_server {
        use std::time::Duration;

        use serde_json::json;
        use tokio::io::BufReader;

        use super::*;
        use crate::lsp::LspNotification;
        use crate::test_lsp::{
            FakeServer, fake_lsp_client_with_lanes, read_framed_message,
            spawn_test_pump_over_lanes, write_notification, write_request,
        };

        async fn assert_create_request_is_acknowledged(server: &mut FakeServer) {
            write_request(
                &mut server.read_half_stdin,
                &json!(7),
                "window/workDoneProgress/create",
                json!({ "token": "idx" }),
            )
            .await;

            let reply = read_framed_message(&mut BufReader::new(&mut server.write_stdout)).await;
            assert_eq!(reply["id"], 7);
            assert_eq!(reply.get("result"), Some(&serde_json::Value::Null));
            assert!(reply.get("error").is_none(), "unexpected error: {reply}");
        }

        async fn send_progress(server: &mut FakeServer, kind: &str) {
            write_notification(
                &mut server.read_half_stdin,
                "$/progress",
                json!({ "token": "idx", "value": { "kind": kind, "title": "Indexing" } }),
            )
            .await;
        }

        fn kind_of(notification: &LspNotification) -> &str {
            let LspNotification::Progress(params) = notification else {
                panic!("expected a progress notification, got {notification:?}");
            };
            params.value["kind"].as_str().unwrap()
        }

        async fn wait_for(
            cache: &tokio::sync::Mutex<crate::bridge::NotificationCache>,
            id: &ServerId,
            expected: IndexingState,
        ) {
            for _ in 0..200 {
                if cache.lock().await.indexing_state(id) == expected {
                    return;
                }
                tokio::time::sleep(Duration::from_millis(1)).await;
            }
            panic!("indexing state never became {expected:?}");
        }

        #[tokio::test]
        async fn test_report_frame_never_reaches_the_lifecycle_lane() {
            let (_client, mut server, mut lanes) = fake_lsp_client_with_lanes();
            assert_create_request_is_acknowledged(&mut server).await;

            for kind in ["begin", "report", "end"] {
                send_progress(&mut server, kind).await;
            }

            let first = tokio::time::timeout(Duration::from_secs(5), lanes.lifecycle_rx.recv())
                .await
                .unwrap()
                .unwrap();
            let second = tokio::time::timeout(Duration::from_secs(5), lanes.lifecycle_rx.recv())
                .await
                .unwrap()
                .unwrap();
            assert_eq!((kind_of(&first), kind_of(&second)), ("begin", "end"));
            assert!(lanes.lifecycle_rx.try_recv().is_err());
        }

        #[tokio::test]
        async fn test_progress_sequence_moves_tracker_from_loading_to_ready() {
            let (_client, mut server, lanes) = fake_lsp_client_with_lanes();
            assert_create_request_is_acknowledged(&mut server).await;
            let (cache, _cancel) = spawn_test_pump_over_lanes(lanes);
            let id = ServerId::from_static("rust");
            tokio::time::pause();

            send_progress(&mut server, "begin").await;
            wait_for(&cache, &id, IndexingState::Loading).await;

            send_progress(&mut server, "report").await;
            send_progress(&mut server, "end").await;
            tokio::time::sleep(Duration::from_millis(10)).await;
            assert_eq!(
                cache.lock().await.indexing_state(&id),
                IndexingState::Loading,
                "still inside the settle window after `end`"
            );

            tokio::time::advance(PROGRESS_SETTLE + Duration::from_secs(1)).await;
            assert_eq!(cache.lock().await.indexing_state(&id), IndexingState::Ready);
        }
    }
}