feat: preconfigure Python with Astral ty and Ruff

docs/architecture.md

@@ -29,7 +29,7 @@ Frontends inspect `count()` for the optional pending motion prefix; parsing and
 Frontends inspect `leader_pending()` for the cached display of a pending leader sequence; prefix recognition, cancellation, and dispatch remain editor-owned.
 Frontends inspect `buffer_id()`, `buffer_count()`, `has_buffer(id)`, and the ascending-ID `buffers()` iterator of borrowed path/dirty/active metadata. Existing text, cursor, selection, search, and mode methods describe only the active buffer.
 Frontends inspect `file_picker()` for query, presentation, view, explorer-local INSERT/filter subfocus, selection, and errors, and iterate borrowed `FileRow` metadata (actual path, display label, directory flag, visible depth, expansion). Dock side is a logical preference. They allocate rectangles, add folder suffixes/colors, and render; hierarchy, filtering, parent/child navigation, focus transitions, and selected-file opening remain backend-owned.
-Frontends inspect `diagnostics()` and `diagnostic_at_cursor()` for validated active-buffer annotations. Diagnostics are owned by core LSP or a plugin and buffer revision; text changes and owner/server retirement clear them. The TUI underlines annotated text, colors affected gutter numbers by severity, and displays the primary caret's diagnostic message without replacing search/selection backgrounds.
+Frontends inspect `diagnostics()` and `diagnostic_at_cursor()` for validated active-buffer annotations. Diagnostics are owned by core LSP or a plugin and buffer revision; text changes clear them. Native server retirement removes only that server's contribution, preserving sibling diagnostics. The TUI underlines annotated text, colors affected gutter numbers by severity, and displays the primary caret's diagnostic message without replacing search/selection backgrounds.
 Frontends inspect `hover()` for optional read-only content and logical-line/grapheme scroll offsets. Popup request eligibility, dismissal, and scrolling stay editor/session-owned; the TUI clips the panel inside the document workspace without changing the primary cursor or viewport.
 
 ## Terminal lifecycle and layout
@@ -52,6 +52,8 @@ macOS uses fixed JXA/AppKit `NSPasteboardTypeString` access through `osascript`,
 
 The native module initializes each server, sends `initialization_options` as `initializationOptions`, answers `workspace/configuration` from `settings` (including dotted sections; missing sections return `null`), honors open/change/save capabilities, synchronizes in-memory text including hidden dirty buffers, and sends close events. Incremental servers receive whole-document range replacements. Process/pipe I/O and requests stay on bounded background workers; input and idle diagnostic redraw do not wait for server responses.
 
+Language configurations can own an ordered server stack. The first matching language wins extension overlap; all its matching servers receive document events, while the first handles hover/definition. The Python preset uses ty then Ruff. LSP merges each worker's latest diagnostic snapshot in assignment order before emitting the unchanged core-owned update. Empty publications/retirement remove only their source; revision/path/root changes and close discard all contributions. Worker snapshots retain identity so queued pre-rename diagnostics cannot attach to a new path with the same text revision. Combined diagnostic message bytes remain bounded to 4 MiB; an overflowing publication retires its worker instead of discarding healthy sibling annotations. Workers are stored in config/root order, avoiding per-poll key cloning/sorting. Shutdown flushes the exit notification and waits for peer termination within the existing bounded grace period before process cleanup.
+
 ## Ruby action catalog
 
 Ruby builds `config.core` from the Rust action catalog before loading configuration or plugin scripts. Dotted identifiers create nested namespaces automatically; each namespace and Action value is frozen. Action names are not duplicated in Ruby, so Rust catalog additions/renames immediately change the Ruby interface without code generation or handwritten exports.

docs/configuration.md

@@ -138,7 +138,7 @@ LSP runs in Rust, independently of Ruby plugin hosts. **Every language server is
 Vamp.configure do |config|
   config.lsp["rust"] = true # rust-analyzer, extensions: ["rs"]
   config.lsp["ruby"] = true # solargraph stdio, extensions: ["rb"]
-  config.lsp["python"] = false # disabled (also the default when absent)
+  config.lsp["python"] = true # ty server + ruff server, extensions: ["py"]
 
   config.lsp["my-language"] = {
     command: ["my-language-server", "--stdio"],
@@ -152,13 +152,34 @@ Vamp.configure do |config|
 end
 ```
 
-Keys are arbitrary nonempty strings, used as the protocol's `languageId`; they need no Tree-sitter grammar or hardcoded language enum. Absent/`false` entries are disabled. `true` enables a built-in preset: `"rust"` uses `rust-analyzer`/`rs`, `"python"` uses `pylsp`/`py`, and `"ruby"` uses `solargraph stdio`/`rb`. Unknown keys need a configuration object with nonempty `command` argv and `extensions` (without leading dots). Objects also explicitly enable servers; known presets supply omitted command/extensions. For example, `config.lsp["rust"] = { command: ["/custom/path/rust-analyzer"] }` overrides the executable while retaining `rs`.
+Keys are arbitrary nonempty strings, used as the protocol's `languageId`; they need no Tree-sitter grammar or hardcoded language enum. Absent/`false` entries are disabled. `true` enables a built-in preset: `"rust"` uses `rust-analyzer`/`rs`, `"python"` uses `ty server` plus `ruff server`/`py`, and `"ruby"` uses `solargraph stdio`/`rb`. Unknown keys need a configuration object with nonempty `command` argv or a `servers` list, plus `extensions` (without leading dots). Objects also explicitly enable servers; known presets supply omitted command/extensions. An explicit `command` replaces the entire preset with one server: `config.lsp["rust"] = { command: ["/custom/path/rust-analyzer"] }` retains `rs`, while a Python command override does not also launch Ruff.
 
-Only `command`, `extensions`, `initialization_options`, and `settings` are accepted. Both option/settings values must be JSON-compatible objects and default to `{}`. Invalid declarations reject configuration with a notice rather than activating a guessed server. Server commands are executable/argument arrays, never shell commands; use an absolute path if the executable is not on PATH. No server installation or restart is automatic.
+Only `command`, `extensions`, `initialization_options`, `settings`, and `servers` are accepted. Options/settings must be JSON-compatible objects. Without an explicit command/list, options apply to every default server; explicit `initialization_options` replaces preset defaults. `servers` must be a nonempty ordered list of objects with explicit `command` and optional per-server `initialization_options`/`settings`; it shares language-level extensions and cannot be combined with top-level command/options/settings. Invalid declarations reject configuration with a notice rather than activating a guessed server. Commands are executable/argument arrays, never shell commands; use an absolute path if the executable is not on PATH. No server installation or restart is automatic.
 
 Install Solargraph yourself with `gem install solargraph` in the project's Ruby environment. The Ruby preset launches `["solargraph", "stdio"]`, following [Solargraph's stdio integration](https://solargraph.org/guides/language-server), and uses [YARD annotations](https://solargraph.org/guides/yard) for type information. Explicit `command` overrides replace the entire argv: for a bundled installation, use `config.lsp["ruby"] = { command: ["bundle", "exec", "solargraph", "stdio"] }`; Vamp retains the `rb` extension but does not append `stdio` to your override.
 
-Servers start lazily only when a matching buffer opens. Extension matching is case-sensitive; if enabled languages overlap, the first language in sorted key order wins. One worker runs per language configuration/project root, with at most 16 enabled languages/server-project keys. Start vamp from your project root: its startup working directory is the workspace; automatic root discovery is not implemented.
+**Python (Astral):** install [ty](https://docs.astral.sh/ty/editors/) and [Ruff](https://docs.astral.sh/ruff/editors/setup/) yourself, for example with `uv tool install ty` and `uv tool install ruff`. The Python preset sends the same documents to both servers, uses ty for hover/definition, and merges type errors with Ruff lint diagnostics. Clearing or losing one server preserves the other's diagnostics. Formatting and code actions remain unsupported by Vamp.
+
+Use `servers` to customize the stack independently; order selects the hover/definition server:
+
+```ruby
+config.lsp["python"] = {
+  servers: [
+    {
+      command: ["ty", "server"],
+      initialization_options: { experimental: { useUv: "off" } }
+    },
+    {
+      command: ["ruff", "server"],
+      initialization_options: { settings: { lint: { extendSelect: ["I"] } } }
+    }
+  ]
+}
+```
+
+**uv remains optional project tooling.** Run project setup/synchronization yourself; [ty discovers an existing `.venv`](https://docs.astral.sh/ty/reference/configuration/#python) without a uv launcher. Vamp neither runs `uv sync` nor changes dependencies/lockfiles. The default ty preset sets `experimental.useUv: "off"` so inherited `TY_UV` cannot enable ty's [experimental environment-management integration](https://docs.astral.sh/ty/reference/editor-settings/). Ruff's initialization options default to `{}`. When replacing ty's initialization options, retain that flag if you want the same policy.
+
+Servers start lazily only when a matching buffer opens. Extension matching is case-sensitive; if enabled languages overlap, the first language in sorted key order wins. All matching servers in that language run, and the first handles hover/definition and the diagnostic-status command without fallback. Each server/project root gets one worker, with at most 16 configured servers and 16 server/project keys; Python uses two slots. Start vamp from your project root: its startup working directory is the workspace; automatic root discovery is not implemented.
 
 See [native LSP architecture](architecture.md#native-lsp) for synchronization and worker behavior.
 

docs/development.md

@@ -35,7 +35,7 @@ Rust tests live alongside their modules; native LSP integration checks are in [`
 - Editor checks cover Unicode editing, multi-target state, undo/redo, saving failures, and modal paste acceptance and rejection atomicity.
 - Session and Ruby checks exercise real Ruby execution, command/capability ownership, effect validation, configuration fallback, plugin retirement, crash/timeout recovery, and inherited-pipe cleanup.
 - Unix terminal-hook checks use isolated child processes and real PTYs to keep process-global hooks out of the parallel runner. They cover cleanup on entry failure, normal return, I/O errors, and panics, including original panic-payload propagation.
-- Native LSP integration checks launch framed-protocol fixture subprocesses, covering initialization/settings, opt-in startup, synchronization, UTF-16/file-URI conversion, dirty targets, stale requests/diagnostics, and transport failures.
+- Native LSP integration checks launch framed-protocol fixture subprocesses, covering initialization/settings, opt-in startup, synchronization, UTF-16/file-URI conversion, dirty targets, stale requests/diagnostics, and transport failures. Stack checks cover independent initialization, diagnostic merging/clearing, sibling failure isolation in active/parked buffers, same-revision rename/close identity, aggregate limits, first-server position routing, and both shutdowns. Semantic ordering uses protocol/file gates; elapsed deadlines only bound stuck subprocesses.
 - Hover regression checks cover content forms, sanitization, cancelled/superseded replies, invalid results, scrolling/dismissal, Unicode clipping, and preserved editing state.
 
 Clipboard checks use isolated providers rather than reading or overwriting the desktop clipboard.

examples/config.rb

@@ -13,6 +13,7 @@ Vamp.configure do |config|
   # Native LSP is disabled by default; install a server and explicitly opt in.
   # config.lsp["rust"] = true # built-in rust-analyzer preset
   # config.lsp["ruby"] = true # built-in Solargraph stdio preset (gem install solargraph)
+  # config.lsp["python"] = true # built-in ty + Ruff stack (install both separately)
   # config.lsp["my-language"] = {
   #   command: ["my-language-server", "--stdio"], extensions: ["myl"],
   #   initialization_options: {}, settings: {}

src/lsp.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::{HashMap, VecDeque},
+    collections::{BTreeMap, HashMap, HashSet, VecDeque},
     io::{Read, Write},
     path::{Component, Path, PathBuf},
     process::{Child, ChildStderr, ChildStdin, ChildStdout, Command, Stdio},
@@ -39,7 +39,7 @@ pub struct ServerConfig {
     pub settings: Value,
 }
 
-#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 struct ServerKey {
     config: usize,
     root: PathBuf,
@@ -47,12 +47,26 @@ struct ServerKey {
 
 struct CachedDocument {
     snapshot: Arc<DocumentSnapshot>,
-    server: Option<ServerKey>,
+    servers: Vec<ServerKey>,
+    diagnostics: HashMap<usize, Contribution>,
+}
+
+struct Contribution {
+    items: Vec<Diagnostic>,
+    bytes: usize,
+}
+
+enum Output {
+    Update(PluginUpdate),
+    Diagnostics {
+        snapshot: Arc<DocumentSnapshot>,
+        items: Vec<Diagnostic>,
+    },
 }
 
 struct Worker {
     input: SyncSender<Input>,
-    output: Receiver<PluginUpdate>,
+    output: Receiver<Output>,
     failure: Receiver<String>,
     cancelled: Arc<AtomicBool>,
     retired: bool,
@@ -98,8 +112,9 @@ pub struct Lsp {
     configs: Vec<Arc<ServerConfig>>,
     cwd: PathBuf,
     documents: HashMap<usize, CachedDocument>,
-    servers: HashMap<ServerKey, Worker>,
+    servers: BTreeMap<ServerKey, Worker>,
     updates: Vec<PluginUpdate>,
+    changed_diagnostics: HashSet<usize>,
 }
 
 impl Lsp {
@@ -108,8 +123,9 @@ impl Lsp {
             configs: configs.into_iter().map(Arc::new).collect(),
             cwd: std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")),
             documents: HashMap::new(),
-            servers: HashMap::new(),
+            servers: BTreeMap::new(),
             updates: Vec::new(),
+            changed_diagnostics: HashSet::new(),
         }
     }
 
@@ -133,31 +149,65 @@ impl Lsp {
             .map(|cached| Arc::clone(&cached.snapshot))
             .unwrap_or_else(|| Arc::new(event.document.clone()));
         let mut startup_error = None;
-        let key = snapshot.path.as_ref().and_then(|path| {
+        if matches!(event.event, DocumentEventKind::Close) {
+            self.documents.remove(&snapshot.id);
+            self.changed_diagnostics.remove(&snapshot.id);
+        } else if self.documents.get(&snapshot.id).is_none_or(|cached| {
+            cached.snapshot.path != snapshot.path || cached.snapshot.root != snapshot.root
+        }) {
+            let keys = snapshot
+                .path
+                .as_ref()
+                .and_then(|path| {
                     let extension = Path::new(path).extension()?.to_str()?;
-            let config = self.configs.iter().position(|config| {
+                    let matches = |config: &ServerConfig| {
                         config
                             .extensions
                             .iter()
                             .any(|candidate| candidate == extension)
-            })?;
+                    };
-            Some(ServerKey {
+                    let language = &self
-                config,
+                        .configs
-                root: absolute(Path::new(&snapshot.root), &self.cwd),
+                        .iter()
+                        .find(|config| matches(config))?
+                        .language_id;
+                    let root = absolute(Path::new(&snapshot.root), &self.cwd);
+                    Some(
+                        self.configs
+                            .iter()
+                            .enumerate()
+                            .filter(|(_, config)| {
+                                config.language_id == *language && matches(config)
                             })
-        });
+                            .map(|(config, _)| ServerKey {
-        if matches!(event.event, DocumentEventKind::Close) {
+                                config,
-            self.documents.remove(&snapshot.id);
+                                root: root.clone(),
-        } else {
+                            })
+                            .collect::<Vec<_>>(),
+                    )
+                })
+                .unwrap_or_default();
             self.documents.insert(
                 snapshot.id,
                 CachedDocument {
                     snapshot: Arc::clone(&snapshot),
-                    server: key.clone(),
+                    servers: keys,
+                    diagnostics: HashMap::new(),
                 },
             );
-            if let Some(key) = &key {
+            self.changed_diagnostics.insert(snapshot.id);
-                startup_error = self.start(key).err();
+            for index in 0..self.documents[&snapshot.id].servers.len() {
+                let key = self.documents[&snapshot.id].servers[index].clone();
+                if let Err(error) = self.start(&key) {
+                    startup_error.get_or_insert(error);
+                }
+            }
+        } else {
+            let cached = self.documents.get_mut(&snapshot.id).unwrap();
+            if !Arc::ptr_eq(&cached.snapshot, &snapshot) {
+                cached.snapshot = Arc::clone(&snapshot);
+                cached.diagnostics.clear();
+                self.changed_diagnostics.insert(snapshot.id);
             }
         }
         let mut failures = Vec::new();
@@ -170,7 +220,10 @@ impl Lsp {
                 .try_send(Input::Document {
                     snapshot: Arc::clone(&snapshot),
                     event: event.event,
-                    served: key.as_ref() == Some(server_key),
+                    served: self
+                        .documents
+                        .get(&snapshot.id)
+                        .is_some_and(|doc| doc.servers.contains(server_key)),
                 })
                 .is_err()
             {
@@ -245,13 +298,85 @@ impl Lsp {
         for (key, reason) in failures {
             self.retire(&key, reason);
         }
-        let mut updates = std::mem::take(&mut self.updates);
+        // Ordered storage keeps attribution deterministic without per-poll sorting.
-        for worker in self.servers.values() {
+        for (key, worker) in &mut self.servers {
-            if !worker.retired {
+            if worker.retired {
-                updates.extend(worker.output.try_iter().take(QUEUE_SIZE));
+                continue;
+            }
+            let first_update = self.updates.len();
+            for _ in 0..QUEUE_SIZE {
+                let Ok(output) = worker.output.try_recv() else {
+                    break;
+                };
+                match output {
+                    Output::Update(update) => self.updates.push(update),
+                    Output::Diagnostics { snapshot, items } => {
+                        let Some(doc) = self.documents.get_mut(&snapshot.id).filter(|doc| {
+                            doc.servers.contains(key) && Arc::ptr_eq(&doc.snapshot, &snapshot)
+                        }) else {
+                            continue;
+                        };
+                        let bytes = items.iter().map(|item| item.message.len()).sum::<usize>();
+                        let siblings = doc
+                            .diagnostics
+                            .iter()
+                            .filter(|(config, _)| **config != key.config)
+                            .map(|(_, contribution)| contribution.bytes)
+                            .sum::<usize>();
+                        if bytes > MAX_BODY.saturating_sub(siblings) {
+                            self.updates.truncate(first_update);
+                            Self::retire_worker(
+                                key,
+                                worker,
+                                &mut self.documents,
+                                &mut self.changed_diagnostics,
+                                &mut self.updates,
+                                "LSP server stopped: combined diagnostic messages exceed 4 MiB"
+                                    .into(),
+                            );
+                            break;
+                        }
+                        if items.is_empty() {
+                            doc.diagnostics.remove(&key.config);
+                        } else {
+                            doc.diagnostics
+                                .insert(key.config, Contribution { items, bytes });
+                        }
+                        self.changed_diagnostics.insert(snapshot.id);
                     }
                 }
-        updates
+            }
+        }
+        let effects: Vec<_> = self
+            .changed_diagnostics
+            .drain()
+            .filter_map(|id| {
+                let doc = self.documents.get(&id)?;
+                let count = doc
+                    .diagnostics
+                    .values()
+                    .map(|value| value.items.len())
+                    .sum();
+                let mut items = Vec::with_capacity(count);
+                for key in &doc.servers {
+                    if let Some(contribution) = doc.diagnostics.get(&key.config) {
+                        items.extend(contribution.items.iter().cloned());
+                    }
+                }
+                Some(UpdateEffect::Diagnostics {
+                    buffer_id: id,
+                    revision: doc.snapshot.revision,
+                    items,
+                })
+            })
+            .collect();
+        if !effects.is_empty() {
+            self.updates.push(PluginUpdate {
+                request_id: None,
+                effects,
+            });
+        }
+        std::mem::take(&mut self.updates)
     }
 
     fn start(&mut self, key: &ServerKey) -> Result<(), String> {
@@ -271,7 +396,7 @@ impl Lsp {
         let initial: Vec<_> = self
             .documents
             .values()
-            .map(|doc| (Arc::clone(&doc.snapshot), doc.server.as_ref() == Some(key)))
+            .map(|doc| (Arc::clone(&doc.snapshot), doc.servers.contains(key)))
             .collect();
         let started = thread::Builder::new()
             .name(format!("lsp-{}", config.language_id))
@@ -328,7 +453,7 @@ impl Lsp {
         let key = self
             .documents
             .get(&buffer_id)
-            .and_then(|doc| doc.server.clone());
+            .and_then(|doc| doc.servers.first().cloned());
         if let Some(key) = key
             && let Some(worker) = self.servers.get(&key)
             && !worker.retired
@@ -353,23 +478,39 @@ impl Lsp {
         let Some(worker) = self.servers.get_mut(key) else {
             return;
         };
+        Self::retire_worker(
+            key,
+            worker,
+            &mut self.documents,
+            &mut self.changed_diagnostics,
+            &mut self.updates,
+            reason,
+        );
+    }
+
+    fn retire_worker(
+        key: &ServerKey,
+        worker: &mut Worker,
+        documents: &mut HashMap<usize, CachedDocument>,
+        changed: &mut HashSet<usize>,
+        updates: &mut Vec<PluginUpdate>,
+        reason: String,
+    ) {
         if worker.retired {
             return;
         }
         worker.retired = true;
         worker.cancelled.store(true, Ordering::Release);
-        let mut effects: Vec<_> = self
+        for (id, doc) in documents {
-            .documents
+            if doc.servers.contains(key) {
-            .values()
+                doc.diagnostics.remove(&key.config);
-            .filter(|doc| doc.server.as_ref() == Some(key))
+                changed.insert(*id);
-            .map(|doc| clear(&doc.snapshot))
+            }
-            .collect();
+        }
-        effects.push(UpdateEffect::Message { text: reason });
+        // Retirement cleanup remains independent of the command-message queue.
-        // At most one retirement update per bounded worker, independently of the
+        updates.push(PluginUpdate {
-        // command-message queue, so a full queue cannot hide diagnostic cleanup.
-        self.updates.push(PluginUpdate {
             request_id: None,
-            effects,
+            effects: vec![UpdateEffect::Message { text: reason }],
         });
     }
 }
@@ -415,7 +556,7 @@ struct Runtime {
     config: Arc<ServerConfig>,
     root: PathBuf,
     input: Receiver<Input>,
-    output: SyncSender<PluginUpdate>,
+    output: SyncSender<Output>,
     cancelled: Arc<AtomicBool>,
     transport: Transport,
     snapshots: HashMap<usize, Arc<DocumentSnapshot>>,
@@ -434,7 +575,7 @@ impl Runtime {
         config: Arc<ServerConfig>,
         root: PathBuf,
         input: Receiver<Input>,
-        output: SyncSender<PluginUpdate>,
+        output: SyncSender<Output>,
         cancelled: Arc<AtomicBool>,
     ) -> Result<Self, String> {
         let transport = Transport::spawn(&config.command, &root)?;
@@ -528,7 +669,22 @@ impl Runtime {
     }
 
     fn emit(&self, update: PluginUpdate) -> Result<(), String> {
-        self.output.try_send(update).map_err(|error| match error {
+        self.send(Output::Update(update))
+    }
+
+    fn emit_diagnostics(
+        &self,
+        snapshot: &Arc<DocumentSnapshot>,
+        items: Vec<Diagnostic>,
+    ) -> Result<(), String> {
+        self.send(Output::Diagnostics {
+            snapshot: Arc::clone(snapshot),
+            items,
+        })
+    }
+
+    fn send(&self, output: Output) -> Result<(), String> {
+        self.output.try_send(output).map_err(|error| match error {
             TrySendError::Full(_) => "LSP incoming update queue is full".into(),
             TrySendError::Disconnected(_) => "LSP client disconnected".into(),
         })
@@ -553,10 +709,7 @@ impl Runtime {
         if !served {
             if let Some(old) = old {
                 self.close(&old)?;
-                self.emit(PluginUpdate {
+                self.emit_diagnostics(&snapshot, Vec::new())?;
-                    request_id: None,
-                    effects: vec![clear(&snapshot)],
-                })?;
             }
             return Ok(());
         }
@@ -587,10 +740,7 @@ impl Runtime {
             saved: matches!(event, DocumentEventKind::Save),
         };
         if changed {
-            self.emit(PluginUpdate {
+            self.emit_diagnostics(&doc.snapshot, Vec::new())?;
-                request_id: None,
-                effects: vec![clear(&doc.snapshot)],
-            })?;
         }
         if self.ready {
             if !doc.opened {
@@ -937,21 +1087,14 @@ impl Runtime {
                 ),
             });
         }
-        self.emit(PluginUpdate {
+        self.emit_diagnostics(&doc.snapshot, diagnostics)
-            request_id: None,
-            effects: vec![UpdateEffect::Diagnostics {
-                buffer_id: doc.snapshot.id,
-                revision: doc.snapshot.revision,
-                items: diagnostics,
-            }],
-        })
     }
 
     fn definition_result(&self, source: Source, result: &Value) -> Result<(), String> {
-        if !self
+        if self
             .documents
             .get(&source.buffer_id)
-            .is_some_and(|doc| doc.snapshot.revision == source.revision)
+            .is_none_or(|doc| doc.snapshot.revision != source.revision)
         {
             return Ok(());
         }
@@ -1084,7 +1227,11 @@ impl Runtime {
             if self.transport.flush().is_err() || self.transport.read().is_err() {
                 break;
             }
-            if exiting && self.transport.flushed() {
+            // A flushed exit notification is not proof the peer has consumed it.
+            if exiting
+                && self.transport.flushed()
+                && self.transport.child.try_wait().ok().flatten().is_some()
+            {
                 break;
             }
             match self.transport.frame() {
@@ -1651,14 +1798,6 @@ unsafe extern "C" {
     fn kill(pid: i32, signal: i32) -> i32;
 }
 
-fn clear(snapshot: &DocumentSnapshot) -> UpdateEffect {
-    UpdateEffect::Diagnostics {
-        buffer_id: snapshot.id,
-        revision: snapshot.revision,
-        items: Vec::new(),
-    }
-}
-
 fn message(text: &str, request_id: Option<u64>) -> PluginUpdate {
     PluginUpdate {
         request_id,

src/lsp_config.rs

@@ -10,10 +10,10 @@ pub fn resolve(value: Value) -> Result<Vec<ServerConfig>, String> {
         if language_id.is_empty() || language_id.chars().any(char::is_control) {
             return Err("LSP language keys must not be empty or contain control characters".into());
         }
-        let preset: Option<(&[&str], &str)> = match language_id.as_str() {
+        let preset: Option<(&[&[&str]], &str)> = match language_id.as_str() {
-            "rust" => Some((&["rust-analyzer"], "rs")),
+            "rust" => Some((&[&["rust-analyzer"]], "rs")),
-            "python" => Some((&["pylsp"], "py")),
+            "python" => Some((&[&["ty", "server"], &["ruff", "server"]], "py")),
-            "ruby" => Some((&["solargraph", "stdio"], "rb")),
+            "ruby" => Some((&[&["solargraph", "stdio"]], "rb")),
             _ => None,
         };
         let mut options = match value {
@@ -36,55 +36,134 @@ pub fn resolve(value: Value) -> Result<Vec<ServerConfig>, String> {
         for key in options.keys() {
             if !matches!(
                 key.as_str(),
-                "command" | "extensions" | "initialization_options" | "settings"
+                "command" | "extensions" | "initialization_options" | "settings" | "servers"
             ) {
                 return Err(format!(
                     "LSP language {language_id:?} has unknown option {key:?}"
                 ));
             }
         }
-        let command = match options.remove("command") {
+        let mut extensions = match options.remove("extensions") {
-            Some(value) => command(value, &language_id)?,
-            None => match preset {
-                Some((command, _)) => command.iter().map(|argument| (*argument).into()).collect(),
-                None => {
-                    return Err(format!(
-                        "LSP language {language_id:?} requires command argv"
-                    ));
-                }
-            },
-        };
-        let extensions = match options.remove("extensions") {
             Some(value) => extensions(value, &language_id)?,
             None => match preset {
                 Some((_, extension)) => vec![extension.into()],
                 None => return Err(format!("LSP language {language_id:?} requires extensions")),
             },
         };
+        let explicit_servers = options.contains_key("servers");
+        let configurations = match options.remove("servers") {
+            Some(value) => {
+                if ["command", "initialization_options", "settings"]
+                    .iter()
+                    .any(|key| options.contains_key(*key))
+                {
+                    return Err(format!(
+                        "LSP language {language_id:?} servers cannot be combined with command, initialization_options, or settings"
+                    ));
+                }
+                let Value::Array(configurations) = value else {
+                    return Err(format!(
+                        "LSP language {language_id:?} servers must be a nonempty array of configuration objects"
+                    ));
+                };
+                if configurations.is_empty() {
+                    return Err(format!(
+                        "LSP language {language_id:?} servers must be a nonempty array of configuration objects"
+                    ));
+                }
+                configurations
+            }
+            None if options.contains_key("command") => vec![Value::Object(options)],
+            None => {
+                let Some((commands, _)) = preset else {
+                    return Err(format!(
+                        "LSP language {language_id:?} requires command argv or servers"
+                    ));
+                };
+                let mut configurations = Vec::with_capacity(commands.len());
+                for (index, command) in commands.iter().enumerate() {
+                    let mut configuration = if index + 1 == commands.len() {
+                        std::mem::take(&mut options)
+                    } else {
+                        options.clone()
+                    };
+                    configuration.insert(
+                        "command".into(),
+                        Value::Array(
+                            command
+                                .iter()
+                                .map(|argument| Value::from(*argument))
+                                .collect(),
+                        ),
+                    );
+                    if language_id == "python" && index == 0 {
+                        configuration
+                            .entry("initialization_options")
+                            .or_insert_with(
+                                || serde_json::json!({"experimental": {"useUv": "off"}}),
+                            );
+                    }
+                    configurations.push(Value::Object(configuration));
+                }
+                configurations
+            }
+        };
+        if servers.len() + configurations.len() > 16 {
+            return Err("LSP configuration supports at most 16 configured servers".into());
+        }
+        let configuration_count = configurations.len();
+        let mut language_id = language_id;
+        for (index, configuration) in configurations.into_iter().enumerate() {
+            let context = if explicit_servers {
+                format!("LSP language {language_id:?} servers[{index}]")
+            } else {
+                format!("LSP language {language_id:?}")
+            };
+            let Value::Object(mut configuration) = configuration else {
+                return Err(format!("{context} must be a configuration object"));
+            };
+            for key in configuration.keys() {
+                if !matches!(
+                    key.as_str(),
+                    "command" | "initialization_options" | "settings"
+                ) {
+                    return Err(format!("{context} has unknown option {key:?}"));
+                }
+            }
+            let command = match configuration.remove("command") {
+                Some(value) => command(value, &context)?,
+                None => return Err(format!("{context} requires command argv")),
+            };
             let initialization_options = object(
-            options.remove("initialization_options"),
+                configuration.remove("initialization_options"),
-            &language_id,
+                &context,
                 "initialization_options",
             )?;
-        let settings = object(options.remove("settings"), &language_id, "settings")?;
+            let settings = object(configuration.remove("settings"), &context, "settings")?;
             servers.push(ServerConfig {
-            language_id,
+                language_id: if index + 1 == configuration_count {
+                    std::mem::take(&mut language_id)
+                } else {
+                    language_id.clone()
+                },
                 command,
-            extensions,
+                extensions: if index + 1 == configuration_count {
+                    std::mem::take(&mut extensions)
+                } else {
+                    extensions.clone()
+                },
                 initialization_options,
                 settings,
             });
-        if servers.len() > 16 {
-            return Err("LSP configuration supports at most 16 enabled languages".into());
         }
     }
     Ok(servers)
 }
 
-fn command(value: Value, language: &str) -> Result<Vec<String>, String> {
+fn command(value: Value, context: &str) -> Result<Vec<String>, String> {
     let error = || {
         format!(
-            "LSP language {language:?} command must be nonempty argv with a nonempty executable and no NUL bytes"
+            "{context} command must be nonempty argv with a nonempty executable and no NUL bytes"
         )
     };
     let Value::Array(values) = value else {
@@ -137,12 +216,10 @@ fn extensions(value: Value, language: &str) -> Result<Vec<String>, String> {
     Ok(extensions)
 }
 
-fn object(value: Option<Value>, language: &str, option: &str) -> Result<Value, String> {
+fn object(value: Option<Value>, context: &str, option: &str) -> Result<Value, String> {
     match value {
         None => Ok(Value::Object(serde_json::Map::new())),
         Some(value) if value.is_object() => Ok(value),
-        Some(_) => Err(format!(
+        Some(_) => Err(format!("{context} {option} must be a JSON object")),
-            "LSP language {language:?} {option} must be a JSON object"
-        )),
     }
 }

src/main.rs

@@ -37,8 +37,9 @@ Ruby config: $XDG_CONFIG_HOME/vamp/config.rb or ~/.config/vamp/config.rb
 Set VAMP_RUBY to choose the Ruby executable.
 
 Native LSP is disabled by default: config.lsp[\"rust\"] = true enables a preset
-Presets: rust -> rust-analyzer, python -> pylsp, ruby -> solargraph stdio
+Presets: rust -> rust-analyzer, python -> ty server + ruff server, ruby -> solargraph stdio
 Custom languages: config.lsp[\"my-language\"] = { command: [...], extensions: [...] }
+Server stacks: config.lsp[\"python\"] = { servers: [{ command: [...] }, { command: [...] }] }
 LSP commands: :lsp.definition, :lsp.diagnostics, :lsp.hover
 Hover binding: config.bind \"K\", config.core.lsp.hover (no default key)";
 

src/ruby.rs

@@ -981,9 +981,9 @@ end"#,
 end"#,
         );
         let config = load_config(&path).unwrap();
+        assert_eq!(config.lsp.len(), 4);
         for (language, command, extension) in [
             ("rust", &["rust-analyzer"][..], "rs"),
-            ("python", &["pylsp"][..], "py"),
             ("ruby", &["solargraph", "stdio"][..], "rb"),
         ] {
             let server = config
@@ -996,6 +996,25 @@ end"#,
             assert_eq!(server.initialization_options, serde_json::json!({}));
             assert_eq!(server.settings, serde_json::json!({}));
         }
+        let python: Vec<_> = config
+            .lsp
+            .iter()
+            .filter(|server| server.language_id == "python")
+            .collect();
+        assert_eq!(python.len(), 2);
+        for (server, command, initialization_options) in [
+            (
+                python[0],
+                ["ty", "server"],
+                serde_json::json!({"experimental": {"useUv": "off"}}),
+            ),
+            (python[1], ["ruff", "server"], serde_json::json!({})),
+        ] {
+            assert_eq!(server.command, command);
+            assert_eq!(server.extensions, ["py"]);
+            assert_eq!(server.initialization_options, initialization_options);
+            assert_eq!(server.settings, serde_json::json!({}));
+        }
 
         let path = fixture.file(
             "config.rb",
@@ -1005,7 +1024,7 @@ end"#,
     initialization_options: {cargo: {features: ["α"]}},
     settings: {"rust-analyzer" => {check: {command: "clippy"}}}
   }
-  config.lsp["python"] = {command: ["custom pylsp", "--stdio"]}
+  config.lsp["python"] = {command: ["custom-python-server", "--stdio"]}
   config.lsp["ruby"] = {command: ["bundle", "exec", "solargraph", "stdio"]}
   config.lsp["my-language.c++"] = {
     command: ["custom server", "--flag", ""],
@@ -1035,10 +1054,18 @@ end"#,
             .iter()
             .find(|server| server.language_id == "python")
             .unwrap();
-        assert_eq!(python.command, ["custom pylsp", "--stdio"]);
+        assert_eq!(python.command, ["custom-python-server", "--stdio"]);
         assert_eq!(python.extensions, ["py"]);
         assert_eq!(python.initialization_options, serde_json::json!({}));
         assert_eq!(python.settings, serde_json::json!({}));
+        assert_eq!(
+            config
+                .lsp
+                .iter()
+                .filter(|server| server.language_id == "python")
+                .count(),
+            1
+        );
         let ruby = config
             .lsp
             .iter()
@@ -1060,6 +1087,182 @@ end"#,
         );
     }
 
+    #[test]
+    fn lsp_python_common_options_and_ordered_server_overrides() {
+        let fixture = Fixture::new();
+        let path = fixture.file(
+            "config.rb",
+            r#"Vamp.configure do |config|
+  config.lsp["python"] = {
+    extensions: ["py", "pyi"],
+    initialization_options: {diagnosticMode: "workspace"},
+    settings: {shared: true}
+  }
+end"#,
+        );
+        let config = load_config(&path).unwrap();
+        assert_eq!(config.lsp.len(), 2);
+        for (server, command) in config
+            .lsp
+            .iter()
+            .zip([["ty", "server"], ["ruff", "server"]])
+        {
+            assert_eq!(server.language_id, "python");
+            assert_eq!(server.command, command);
+            assert_eq!(server.extensions, ["py", "pyi"]);
+            assert_eq!(
+                server.initialization_options,
+                serde_json::json!({"diagnosticMode": "workspace"})
+            );
+            assert_eq!(server.settings, serde_json::json!({"shared": true}));
+        }
+        let path = fixture.file(
+            "config.rb",
+            r#"Vamp.configure do |config|
+  config.lsp["python"] = {servers: [
+    {command: ["type-server", "--stdio"], initialization_options: {experimental: {useUv: "off"}}, settings: {first: true}},
+    {command: ["lint-server"], initialization_options: {settings: {lint: {enable: true}}}}
+  ]}
+  config.lsp["custom"] = {extensions: ["custom"], servers: [
+    {command: ["custom-first"]}, {command: ["custom-second"], settings: {second: true}}
+  ]}
+end"#,
+        );
+        let config = load_config(&path).unwrap();
+        for (language, commands, extensions) in [
+            (
+                "python",
+                [&["type-server", "--stdio"][..], &["lint-server"][..]],
+                ["py"],
+            ),
+            (
+                "custom",
+                [&["custom-first"][..], &["custom-second"][..]],
+                ["custom"],
+            ),
+        ] {
+            let servers: Vec<_> = config
+                .lsp
+                .iter()
+                .filter(|server| server.language_id == language)
+                .collect();
+            assert_eq!(servers.len(), 2);
+            for (server, command) in servers.iter().zip(commands) {
+                assert_eq!(server.command, command);
+                assert_eq!(server.extensions, extensions);
+            }
+        }
+        let python: Vec<_> = config
+            .lsp
+            .iter()
+            .filter(|server| server.language_id == "python")
+            .collect();
+        assert_eq!(
+            python[0].initialization_options,
+            serde_json::json!({"experimental": {"useUv": "off"}})
+        );
+        assert_eq!(python[0].settings, serde_json::json!({"first": true}));
+        assert_eq!(
+            python[1].initialization_options,
+            serde_json::json!({"settings": {"lint": {"enable": true}}})
+        );
+        assert_eq!(python[1].settings, serde_json::json!({}));
+        let custom: Vec<_> = config
+            .lsp
+            .iter()
+            .filter(|server| server.language_id == "custom")
+            .collect();
+        assert_eq!(custom[0].initialization_options, serde_json::json!({}));
+        assert_eq!(custom[1].settings, serde_json::json!({"second": true}));
+        let path = fixture.file(
+            "config.rb",
+            r#"Vamp.configure { |config| config.lsp["python"] = {settings: {shared: true}} }"#,
+        );
+        let config = load_config(&path).unwrap();
+        assert_eq!(
+            config.lsp[0].initialization_options,
+            serde_json::json!({"experimental": {"useUv": "off"}})
+        );
+        assert_eq!(config.lsp[1].initialization_options, serde_json::json!({}));
+    }
+
+    #[test]
+    fn lsp_server_lists_reject_invalid_shapes_and_ambiguous_options_locally() {
+        let fixture = Fixture::new();
+        for (options, error) in [
+            ("{servers: nil}", "nonempty array"),
+            ("{servers: {command: ['server']}}", "nonempty array"),
+            ("{servers: []}", "nonempty array"),
+            (
+                "{servers: [true]}",
+                "servers[0] must be a configuration object",
+            ),
+            ("{servers: [{}]}", "servers[0] requires command argv"),
+            ("{servers: [{command: []}]}", "servers[0] command must be"),
+            (
+                "{servers: [{command: ['server'], extensions: ['py']}]}",
+                "servers[0] has unknown option",
+            ),
+            (
+                "{servers: [{command: ['server'], servers: []}]}",
+                "servers[0] has unknown option",
+            ),
+            (
+                "{servers: [{command: ['server'], initialization_options: []}]}",
+                "servers[0] initialization_options must be",
+            ),
+            (
+                "{servers: [{command: ['server']}, {command: ['other'], settings: nil}]}",
+                "servers[1] settings must be",
+            ),
+            (
+                "{servers: [{command: ['server']}], command: ['other']}",
+                "cannot be combined",
+            ),
+            (
+                "{servers: [{command: ['server']}], initialization_options: {}}",
+                "cannot be combined",
+            ),
+            (
+                "{servers: [{command: ['server']}], settings: {}}",
+                "cannot be combined",
+            ),
+        ] {
+            let path = fixture.file(
+                "config.rb",
+                &format!("Vamp.configure {{ |config| config.lsp[\"python\"] = {options} }}"),
+            );
+            let actual = load_config(&path).unwrap_err();
+            assert!(
+                actual.starts_with("Ruby config: LSP language \"python\""),
+                "{actual}"
+            );
+            assert!(actual.contains(error), "{options}: {actual}");
+        }
+    }
+
+    #[test]
+    fn lsp_server_limit_is_enforced_at_configuration_startup() {
+        let fixture = Fixture::new();
+        for count in [16, 17] {
+            let path = fixture.file(
+                "config.rb",
+                &format!(
+                    "Vamp.configure {{ |config| config.lsp[\"python\"] = {{servers: Array.new({count}) {{ {{command: [\"server\"]}} }} }} }}"
+                ),
+            );
+            let result = load_config(&path);
+            if count == 16 {
+                assert_eq!(result.unwrap().lsp.len(), 16);
+            } else {
+                assert_eq!(
+                    result.unwrap_err(),
+                    "Ruby config: LSP configuration supports at most 16 configured servers"
+                );
+            }
+        }
+    }
+
     #[test]
     fn lsp_configuration_rejects_invalid_keys_values_and_json_options() {
         let fixture = Fixture::new();
@@ -1074,6 +1277,7 @@ end"#,
             r#"config.lsp["rust"] = "yes""#,
             r#"config.lsp["custom"] = {}"#,
             r#"config.lsp["custom"] = {command: ["server"]}"#,
+            r#"config.lsp["custom"] = {servers: [{command: ["server"]}]}"#,
             r#"config.lsp["custom"] = {extensions: ["custom"]}"#,
             r#"config.lsp["rust"] = {command: "server"}"#,
             r#"config.lsp["rust"] = {command: []}"#,
@@ -1146,6 +1350,7 @@ end"#,
             serde_json::json!({"": false}),
             serde_json::json!({"bad\u{0085}key": false}),
             serde_json::json!({"rust": null}),
+            serde_json::json!({"python": []}),
             serde_json::json!({"unknown": true}),
             serde_json::json!({"custom": {}}),
             serde_json::json!({"rust": {"command": ["tool", "bad\0argument"]}}),
@@ -1153,6 +1358,11 @@ end"#,
             serde_json::json!({"rust": {"initialization_options": []}}),
             serde_json::json!({"rust": {"settings": "not-an-object"}}),
             serde_json::json!({"rust": {"unexpected": true}}),
+            serde_json::json!({"python": {"servers": []}}),
+            serde_json::json!({"python": {"servers": [null]}}),
+            serde_json::json!({"python": {"servers": [{}]}}),
+            serde_json::json!({"python": {"servers": [{"command": ["server"], "extensions": ["py"]}]}}),
+            serde_json::json!({"python": {"servers": [{"command": ["server"]}], "settings": {}}}),
         ] {
             frame["lsp"] = value;
             let Response::Config { lsp, .. } =
@@ -1162,7 +1372,7 @@ end"#,
             };
             assert!(crate::lsp_config::resolve(lsp).is_err(), "{frame}");
         }
-        let languages: serde_json::Map<String, serde_json::Value> = (0..17)
+        let mut languages: serde_json::Map<String, serde_json::Value> = (0..14)
             .map(|index| {
                 (
                     format!("custom-{index}"),
@@ -1170,13 +1380,24 @@ end"#,
                 )
             })
             .collect();
+        languages.insert("python".into(), serde_json::json!(true));
         frame["lsp"] = serde_json::Value::Object(languages);
         let Response::Config { lsp, .. } =
             serde_json::from_value::<Response>(frame.clone()).unwrap()
         else {
             panic!("expected config frame");
         };
-        assert!(crate::lsp_config::resolve(lsp).is_err());
+        assert_eq!(crate::lsp_config::resolve(lsp).unwrap().len(), 16);
+        frame["lsp"]["ruby"] = serde_json::json!(true);
+        let Response::Config { lsp, .. } =
+            serde_json::from_value::<Response>(frame.clone()).unwrap()
+        else {
+            panic!("expected config frame");
+        };
+        assert_eq!(
+            crate::lsp_config::resolve(lsp).unwrap_err(),
+            "LSP configuration supports at most 16 configured servers"
+        );
         frame.as_object_mut().unwrap().remove("lsp");
         assert!(serde_json::from_value::<Response>(frame).is_err());
     }

tests/lsp.rs

@@ -69,8 +69,10 @@ initialized_sent = false
 held = []
 definition_count = 0
 source_uri = uri(source)
+documents = {}
+control_id = nil
 loop do
-  if initialize_id && !initialized_sent && (!['gated', 'hover-gated'].include?(mode) || File.exist?(release))
+  if initialize_id && !initialized_sent && (!['gated', 'hover-gated', 'controlled-gated'].include?(mode) || File.exist?(release))
     sync = case mode
            when 'numeric' then 1
            when 'none' then {'openClose' => false, 'change' => 0, 'save' => false}
@@ -78,8 +80,8 @@ loop do
            else {'openClose' => true, 'change' => 1, 'save' => {'includeText' => true}}
            end
     result(initialize_id, {'capabilities' => {'positionEncoding' => 'utf-16',
-      'textDocumentSync' => sync, 'definitionProvider' => mode == 'unsupported' ? false : {'workDoneProgress' => false},
+      'textDocumentSync' => sync, 'definitionProvider' => ['unsupported', 'controlled-unsupported'].include?(mode) ? false : {'workDoneProgress' => false},
-      'hoverProvider' => mode == 'hover-unsupported' ? false : {'workDoneProgress' => false}}})
+      'hoverProvider' => ['hover-unsupported', 'controlled-unsupported'].include?(mode) ? false : {'workDoneProgress' => false}}})
     initialized_sent = true
   end
   if !held.empty? && File.exist?(release) && ['stale', 'hover-stale'].include?(mode)
@@ -89,6 +91,30 @@ loop do
     held.clear
     emit({'method' => 'window/showMessage', 'params' => {'type' => 3, 'message' => 'stale response complete'}})
   end
+  control = release + '.control'
+  if File.exist?(control)
+    command = JSON.parse(File.read(control))
+    if command['id'] != control_id
+      control_id = command.fetch('id')
+      case command.fetch('action')
+      when 'publish', 'clear'
+        document = documents.fetch(command.fetch('uri', source_uri))
+        params = {'uri' => document['uri'], 'version' => command.fetch('version', document['version']),
+          'diagnostics' => command['action'] == 'clear' ? [] :
+            [{'range' => range(0, 2, 4), 'severity' => 2, 'message' => command.fetch('message')}]}
+        emit({'method' => 'textDocument/publishDiagnostics', 'params' => params})
+      when 'die'
+        exit! 7
+      when 'malformed'
+        STDOUT.write("Content-Length: nope\r\n\r\n")
+      else
+        raise 'unknown fixture action'
+      end
+      # The response proves the worker consumed every preceding publication.
+      emit({'id' => "control-#{control_id}", 'method' => 'workspace/configuration',
+            'params' => {'items' => []}})
+    end
+  end
   next unless IO.select([STDIN], nil, nil, 0.01)
   headers = {}
   loop do
@@ -114,7 +140,8 @@ loop do
     emit({'id' => 0, 'method' => '$/unsupported', 'params' => {}})
   when 'textDocument/didOpen'
     document = frame.fetch('params').fetch('textDocument')
-    if document['uri'] == source_uri
+    documents[document['uri']] = document
+    if document['uri'] == source_uri && !mode.start_with?('controlled')
       if mode == 'malformed'
         STDOUT.write("Content-Length: nope\r\n\r\n")
       elsif mode == 'dead'
@@ -125,7 +152,8 @@ loop do
     end
   when 'textDocument/didChange'
     document = frame.fetch('params').fetch('textDocument')
-    if document['uri'] == source_uri && !['malformed', 'dead'].include?(mode)
+    documents[document['uri']] = document
+    if document['uri'] == source_uri && !['malformed', 'dead'].include?(mode) && !mode.start_with?('controlled')
       diagnostics(document, 'obsolete must drop', old: true)
       diagnostics(document, "revision #{document['version']}")
     end
@@ -258,6 +286,103 @@ impl Fixture {
         .unwrap()
     }
 
+    fn worker_log(&self, worker: usize) -> PathBuf {
+        if worker == 0 {
+            self.log.clone()
+        } else {
+            self.root.join(format!("protocol-{worker}.jsonl"))
+        }
+    }
+
+    fn worker_release(&self, worker: usize) -> PathBuf {
+        if worker == 0 {
+            self.release.clone()
+        } else {
+            self.root.join(format!("release-{worker}"))
+        }
+    }
+
+    fn worker_frames(&self, worker: usize) -> Vec<Value> {
+        fs::read_to_string(self.worker_log(worker))
+            .unwrap_or_default()
+            .lines()
+            .filter_map(|line| serde_json::from_str(line).ok())
+            .collect()
+    }
+
+    fn worker_count(&self, worker: usize, method: &str) -> usize {
+        self.worker_frames(worker)
+            .iter()
+            .filter(|frame| frame["method"] == method)
+            .count()
+    }
+
+    fn open_stack(&self, modes: [&str; 2], overlap: bool) -> EditorSession {
+        self.open_stack_at(modes, overlap, &self.source)
+    }
+
+    fn open_stack_at(
+        &self,
+        modes: [&str; 2],
+        overlap: bool,
+        path: &std::path::Path,
+    ) -> EditorSession {
+        let server = self.root.join("server.rb");
+        fs::write(&server, SERVER).unwrap();
+        let command = |worker: usize, mode: &str| {
+            if mode == "spawn" {
+                json!([self.root.join("unavailable-language-server")])
+            } else {
+                json!([
+                    "ruby",
+                    server,
+                    mode,
+                    self.worker_log(worker),
+                    self.worker_release(worker),
+                    self.source,
+                    self.target,
+                    self.alternate
+                ])
+            }
+        };
+        let servers: Vec<_> = modes
+            .into_iter()
+            .enumerate()
+            .map(|(worker, mode)| json!({"command": command(worker, mode)}))
+            .collect();
+        let mut languages = json!({"python": {"extensions": ["rs"], "servers": servers}});
+        if overlap {
+            languages["zzz-other"] =
+                json!({"extensions": ["rs"], "command": command(2, "controlled")});
+        }
+        let config = format!(
+            "Vamp.configure do |config|\n config.clipboard_mode = 'separate'\n JSON.parse({}).each {{ |language, options| config.lsp[language] = options }}\n config.bind 'K', config.core.lsp.hover\nend\n",
+            serde_json::to_string(&languages.to_string()).unwrap(),
+        );
+        let config_path = self.root.join("config.rb");
+        fs::write(&config_path, config).unwrap();
+        EditorSession::open(
+            Some(path.to_path_buf()),
+            vec!["text".into(), "diagnostics".into()],
+            Some(&config_path),
+        )
+        .unwrap()
+    }
+
+    fn control(&self, worker: usize, id: u64, mut command: Value) {
+        command["id"] = json!(id);
+        let path = self.worker_release(worker).with_extension("control");
+        let temporary = path.with_extension("pending");
+        fs::write(&temporary, command.to_string()).unwrap();
+        fs::rename(temporary, path).unwrap();
+    }
+
+    fn acknowledged(&self, worker: usize, id: u64) -> bool {
+        self.worker_frames(worker)
+            .iter()
+            .any(|frame| frame["id"] == format!("control-{id}") && frame.get("result").is_some())
+    }
+
     fn frames(&self) -> Vec<Value> {
         fs::read_to_string(&self.log)
             .unwrap_or_default()
@@ -959,3 +1084,540 @@ fn lsp_hover_absent_unsupported_and_invalid_results_preserve_native_editing() {
         assert_eq!(fixture.method_count("textDocument/definition"), 0);
     }
 }
+
+fn diagnostic_messages(session: &EditorSession) -> Vec<&str> {
+    session
+        .editor()
+        .diagnostics()
+        .map(|diagnostic| diagnostic.message.as_str())
+        .collect()
+}
+
+fn wait_protocol(fixture: &Fixture, predicate: impl Fn() -> bool) {
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while !predicate() {
+        assert!(
+            Instant::now() < deadline,
+            "fixture protocol gate did not complete: {:?}",
+            fixture.frames()
+        );
+        thread::sleep(Duration::from_millis(2));
+    }
+    // Deliberately no session.poll(): outputs may be queued across a rename/close.
+}
+
+fn publish(
+    session: &mut EditorSession,
+    fixture: &Fixture,
+    worker: usize,
+    id: u64,
+    command: Value,
+    expected: &[&str],
+) {
+    fixture.control(worker, id, command);
+    until(session, fixture, |session| {
+        fixture.acknowledged(worker, id) && diagnostic_messages(session) == expected
+    });
+}
+
+#[test]
+fn lsp_python_stack_merges_in_assignment_order_and_routes_positions_only_to_ty() {
+    let fixture = Fixture::new();
+    let mut session = fixture.open_stack(["controlled-gated", "controlled"], true);
+    until(&mut session, &fixture, |_| {
+        fixture.worker_count(0, "initialize") == 1
+            && fixture.worker_count(1, "textDocument/didOpen") == 1
+    });
+    assert_eq!(fixture.worker_count(0, "textDocument/didOpen"), 0);
+    session.handle(Input::Char('A'));
+    session.handle(Input::Char('!'));
+    session.handle(Input::Escape);
+    assert_eq!(session.editor().lines(), &["A😀e\u{301}Z!", ""]);
+    session.execute("lsp.hover");
+    fixture.release();
+    until(&mut session, &fixture, |session| {
+        session.editor().hover().is_some()
+    });
+    assert_eq!(fixture.worker_count(0, "textDocument/hover"), 1);
+    assert_eq!(fixture.worker_count(1, "textDocument/hover"), 0);
+    for worker in 0..2 {
+        let frames = fixture.worker_frames(worker);
+        let initialize = frames
+            .iter()
+            .find(|frame| frame["method"] == "initialize")
+            .unwrap();
+        assert!(
+            initialize["params"]["capabilities"]["textDocument"]
+                .get("diagnostic")
+                .is_none()
+        );
+        let open = frames
+            .iter()
+            .find(|frame| frame["method"] == "textDocument/didOpen")
+            .unwrap();
+        assert_eq!(open["params"]["textDocument"]["languageId"], "python");
+    }
+    publish(
+        &mut session,
+        &fixture,
+        1,
+        1,
+        json!({"action": "publish", "message": "ruff"}),
+        &["ruff"],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        0,
+        1,
+        json!({"action": "publish", "message": "ty"}),
+        &["ty", "ruff"],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        0,
+        2,
+        json!({"action": "publish", "message": "ty replaced"}),
+        &["ty replaced", "ruff"],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        1,
+        2,
+        json!({"action": "clear"}),
+        &["ty replaced"],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        1,
+        3,
+        json!({"action": "publish", "message": "duplicate"}),
+        &["ty replaced", "duplicate"],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        0,
+        3,
+        json!({"action": "publish", "message": "duplicate"}),
+        &["duplicate", "duplicate"],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        0,
+        4,
+        json!({"action": "clear"}),
+        &["duplicate"],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        1,
+        4,
+        json!({"action": "clear"}),
+        &[],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        0,
+        5,
+        json!({"action": "publish", "message": "ty first"}),
+        &["ty first"],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        1,
+        5,
+        json!({"action": "publish", "message": "ruff second"}),
+        &["ty first", "ruff second"],
+    );
+
+    let target_id = edit_target(&mut session, &fixture);
+    session.execute("lsp.definition");
+    until(&mut session, &fixture, |session| {
+        session.editor().buffer_id() == target_id
+    });
+    assert_eq!(session.editor().lines(), &["😀e\u{301}disk", ""]);
+    assert_eq!(session.editor().cursor(), Cursor { row: 0, column: 1 });
+    assert_eq!(fixture.worker_count(0, "textDocument/definition"), 1);
+    assert_eq!(fixture.worker_count(1, "textDocument/definition"), 0);
+    session.execute("buffer 1");
+    assert_eq!(diagnostic_messages(&session), ["ty first", "ruff second"]);
+    session.handle(Input::Char('A'));
+    session.handle(Input::Char('?'));
+    session.handle(Input::Escape);
+    assert!(diagnostic_messages(&session).is_empty());
+    until(&mut session, &fixture, |_| {
+        (0..2).all(|worker| {
+            fixture.worker_frames(worker).iter().any(|frame| {
+                frame["method"] == "textDocument/didChange"
+                    && frame["params"]["contentChanges"][0]["text"] == "A😀e\u{301}Z!?\n"
+            })
+        })
+    });
+    publish(
+        &mut session,
+        &fixture,
+        1,
+        6,
+        json!({"action": "publish", "version": 1, "message": "obsolete"}),
+        &[],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        1,
+        7,
+        json!({"action": "publish", "message": "new lint"}),
+        &["new lint"],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        0,
+        6,
+        json!({"action": "clear"}),
+        &["new lint"],
+    );
+    session.execute("w");
+    until(&mut session, &fixture, |_| {
+        (0..2).all(|worker| fixture.worker_count(worker, "textDocument/didSave") == 1)
+    });
+    session.execute("bdelete 1");
+    until(&mut session, &fixture, |_| {
+        (0..2).all(|worker| fixture.worker_count(worker, "textDocument/didClose") == 1)
+    });
+    assert_eq!(session.editor().buffer_id(), target_id);
+    assert!(
+        !fixture.worker_log(2).exists(),
+        "overlapping losing language must never start"
+    );
+    drop(session);
+    for worker in 0..2 {
+        assert_eq!(fixture.worker_count(worker, "shutdown"), 1);
+        assert_eq!(fixture.worker_count(worker, "exit"), 1);
+    }
+}
+
+#[test]
+fn lsp_python_stack_failures_retire_only_the_offender_in_active_and_parked_buffers() {
+    for failing in 0..2 {
+        for failure in ["spawn", "malformed", "die"] {
+            let fixture = Fixture::new();
+            fs::write(&fixture.target, "A😀e\u{301}Z\n").unwrap();
+            let mut modes = ["controlled"; 2];
+            if failure == "spawn" {
+                modes[failing] = "spawn";
+            }
+            let mut session = fixture.open_stack(modes, true);
+            let survivor = 1 - failing;
+            until(&mut session, &fixture, |session| {
+                fixture.worker_count(survivor, "textDocument/didOpen") == 1
+                    && if failure == "spawn" {
+                        session.editor().message().contains("failed to start")
+                    } else {
+                        fixture.worker_count(failing, "textDocument/didOpen") == 1
+                    }
+            });
+            publish(
+                &mut session,
+                &fixture,
+                survivor,
+                1,
+                json!({"action": "publish", "message": "healthy source"}),
+                &["healthy source"],
+            );
+            if failure != "spawn" {
+                let expected = if failing == 0 {
+                    ["failed source", "healthy source"]
+                } else {
+                    ["healthy source", "failed source"]
+                };
+                publish(
+                    &mut session,
+                    &fixture,
+                    failing,
+                    1,
+                    json!({"action": "publish", "message": "failed source"}),
+                    &expected,
+                );
+            }
+            session.execute(&format!("edit {}", fixture.target.display()));
+            until(&mut session, &fixture, |_| {
+                fixture.worker_count(survivor, "textDocument/didOpen") == 2
+                    && (failure == "spawn"
+                        || fixture.worker_count(failing, "textDocument/didOpen") == 2)
+            });
+            let target_uri = fixture
+                .worker_frames(survivor)
+                .into_iter()
+                .filter(|frame| frame["method"] == "textDocument/didOpen")
+                .nth(1)
+                .unwrap()["params"]["textDocument"]["uri"]
+                .clone();
+            publish(
+                &mut session,
+                &fixture,
+                survivor,
+                2,
+                json!({"action": "publish", "uri": target_uri, "message": "healthy target"}),
+                &["healthy target"],
+            );
+            if failure != "spawn" {
+                let expected = if failing == 0 {
+                    ["failed target", "healthy target"]
+                } else {
+                    ["healthy target", "failed target"]
+                };
+                publish(
+                    &mut session,
+                    &fixture,
+                    failing,
+                    2,
+                    json!({"action": "publish", "uri": target_uri, "message": "failed target"}),
+                    &expected,
+                );
+                fixture.control(failing, 3, json!({"action": failure}));
+                until(&mut session, &fixture, |session| {
+                    session.editor().message().contains("server stopped")
+                        && diagnostic_messages(session) == ["healthy target"]
+                });
+            }
+            assert_eq!(diagnostic_messages(&session), ["healthy target"]);
+            session.execute("buffer 1");
+            assert_eq!(diagnostic_messages(&session), ["healthy source"]);
+            session.handle(Input::Char('A'));
+            session.handle(Input::Char('!'));
+            session.handle(Input::Escape);
+            assert_eq!(session.editor().lines(), &["A😀e\u{301}Z!", ""]);
+            assert!(diagnostic_messages(&session).is_empty());
+            until(&mut session, &fixture, |_| {
+                fixture.worker_count(survivor, "textDocument/didChange") == 1
+            });
+            publish(
+                &mut session,
+                &fixture,
+                survivor,
+                3,
+                json!({"action": "publish", "version": 1, "message": "stale healthy"}),
+                &[],
+            );
+            publish(
+                &mut session,
+                &fixture,
+                survivor,
+                4,
+                json!({"action": "publish", "message": "healthy edit"}),
+                &["healthy edit"],
+            );
+            if failing == 0 {
+                session.execute("lsp.definition");
+                until(&mut session, &fixture, |session| {
+                    session.editor().message().contains("no running server")
+                });
+                assert_eq!(fixture.worker_count(1, "textDocument/definition"), 0);
+            } else {
+                session.execute("lsp.hover");
+                until(&mut session, &fixture, |session| {
+                    session.editor().hover().is_some()
+                });
+            }
+            assert!(!fixture.worker_log(2).exists());
+        }
+    }
+}
+
+#[test]
+fn lsp_python_stack_discards_queued_same_revision_diagnostics_after_rename_and_close() {
+    let fixture = Fixture::new();
+    let mut session = fixture.open_stack(["controlled", "controlled"], false);
+    until(&mut session, &fixture, |_| {
+        (0..2).all(|worker| fixture.worker_count(worker, "textDocument/didOpen") == 1)
+    });
+    publish(
+        &mut session,
+        &fixture,
+        0,
+        1,
+        json!({"action": "publish", "message": "before rename"}),
+        &["before rename"],
+    );
+    let revision = session.editor().context().revision;
+    for worker in 0..2 {
+        fixture.control(
+            worker,
+            2,
+            json!({"action": "publish", "message": "queued old path"}),
+        );
+    }
+    wait_protocol(&fixture, || {
+        (0..2).all(|worker| fixture.acknowledged(worker, 2))
+    });
+    let renamed = fixture.root.join("renamed.rs");
+    session.execute(&format!("w {}", renamed.display()));
+    assert_eq!(session.editor().context().revision, revision);
+    until(&mut session, &fixture, |_| {
+        (0..2).all(|worker| {
+            fixture.worker_count(worker, "textDocument/didClose") == 1
+                && fixture.worker_count(worker, "textDocument/didOpen") == 2
+        })
+    });
+    assert!(diagnostic_messages(&session).is_empty());
+    let old_uri = fixture
+        .worker_frames(0)
+        .into_iter()
+        .find(|frame| frame["method"] == "textDocument/didOpen")
+        .unwrap()["params"]["textDocument"]["uri"]
+        .clone();
+    let new_uri = fixture
+        .worker_frames(0)
+        .into_iter()
+        .filter(|frame| frame["method"] == "textDocument/didOpen")
+        .nth(1)
+        .unwrap()["params"]["textDocument"]["uri"]
+        .clone();
+    publish(
+        &mut session,
+        &fixture,
+        0,
+        3,
+        json!({"action": "publish", "uri": old_uri, "message": "closed old path"}),
+        &[],
+    );
+    publish(
+        &mut session,
+        &fixture,
+        1,
+        3,
+        json!({"action": "publish", "uri": new_uri, "message": "new path"}),
+        &["new path"],
+    );
+    fixture.control(
+        1,
+        4,
+        json!({"action": "publish", "uri": new_uri, "message": "queued closed buffer"}),
+    );
+    wait_protocol(&fixture, || fixture.acknowledged(1, 4));
+    session.execute("bdelete 1");
+    until(&mut session, &fixture, |_| {
+        (0..2).all(|worker| fixture.worker_count(worker, "textDocument/didClose") == 2)
+    });
+    assert!(!session.editor().has_buffer(1));
+    assert!(diagnostic_messages(&session).is_empty());
+}
+
+#[test]
+fn lsp_python_stack_aggregate_message_limit_preserves_the_healthy_contribution() {
+    let fixture = Fixture::new();
+    let mut session = fixture.open_stack(["controlled", "controlled"], false);
+    until(&mut session, &fixture, |_| {
+        (0..2).all(|worker| fixture.worker_count(worker, "textDocument/didOpen") == 1)
+    });
+    let large = "x".repeat(2 * 1024 * 1024 + 1);
+    fixture.control(0, 1, json!({"action": "publish", "message": large}));
+    until(&mut session, &fixture, |session| {
+        fixture.acknowledged(0, 1) && session.editor().diagnostics().count() == 1
+    });
+    assert_eq!(
+        session.editor().diagnostics().next().unwrap().message.len(),
+        large.len()
+    );
+    fixture.control(1, 1, json!({"action": "publish", "message": large}));
+    until(&mut session, &fixture, |session| {
+        session
+            .editor()
+            .message()
+            .contains("combined diagnostic messages exceed 4 MiB")
+    });
+    assert_eq!(session.editor().diagnostics().count(), 1);
+    assert_eq!(
+        session.editor().diagnostics().next().unwrap().message.len(),
+        large.len()
+    );
+    publish(
+        &mut session,
+        &fixture,
+        0,
+        2,
+        json!({"action": "publish", "message": "healthy replacement"}),
+        &["healthy replacement"],
+    );
+    session.handle(Input::Char('A'));
+    session.handle(Input::Char('!'));
+    session.handle(Input::Escape);
+    assert_eq!(session.editor().lines(), &["A😀e\u{301}Z!", ""]);
+}
+
+#[test]
+fn lsp_python_stack_seeds_dirty_unserved_documents_before_lazy_startup() {
+    let mut fixture = Fixture::new();
+    fixture.target = fixture.root.join("target.txt");
+    fs::write(&fixture.target, "disk\n").unwrap();
+    let mut session = fixture.open_stack_at(["controlled", "controlled"], false, &fixture.target);
+    session.execute("lsp.diagnostics");
+    until(&mut session, &fixture, |session| {
+        session.editor().message().contains("no running server")
+    });
+    assert!(!fixture.worker_log(0).exists());
+    assert!(!fixture.worker_log(1).exists());
+    session.handle(Input::Char('i'));
+    session.handle(Input::Paste("😀e\u{301}".into()));
+    session.handle(Input::Escape);
+    session.execute(&format!("edit {}", fixture.source.display()));
+    until(&mut session, &fixture, |_| {
+        (0..2).all(|worker| fixture.worker_count(worker, "textDocument/didOpen") == 1)
+    });
+    session.execute("lsp.definition");
+    until(&mut session, &fixture, |session| {
+        session.editor().buffer_id() == 1
+    });
+    assert_eq!(session.editor().lines(), &["😀e\u{301}disk", ""]);
+    assert_eq!(session.editor().cursor(), Cursor { row: 0, column: 1 });
+    assert!(session.editor().dirty());
+    assert_eq!(fs::read_to_string(&fixture.target).unwrap(), "disk\n");
+    assert_eq!(fixture.worker_count(0, "textDocument/definition"), 1);
+    assert_eq!(fixture.worker_count(1, "textDocument/definition"), 0);
+    for worker in 0..2 {
+        assert_eq!(
+            fixture.worker_count(worker, "textDocument/didOpen"),
+            1,
+            "the unserved target is cached, not opened on the wire"
+        );
+    }
+}
+
+#[test]
+fn lsp_python_stack_does_not_fall_back_to_ruff_when_ty_lacks_position_capabilities() {
+    let fixture = Fixture::new();
+    let mut session = fixture.open_stack(["controlled-unsupported", "controlled"], false);
+    until(&mut session, &fixture, |_| {
+        (0..2).all(|worker| fixture.worker_count(worker, "textDocument/didOpen") == 1)
+    });
+    publish(
+        &mut session,
+        &fixture,
+        1,
+        1,
+        json!({"action": "publish", "message": "lint remains available"}),
+        &["lint remains available"],
+    );
+    for command in ["lsp.definition", "lsp.hover"] {
+        let previous_message = session.editor().message().to_owned();
+        session.execute(command);
+        until(&mut session, &fixture, |session| {
+            session.editor().message() != previous_message
+        });
+        assert_eq!(session.editor().buffer_id(), 1);
+        assert_eq!(diagnostic_messages(&session), ["lint remains available"]);
+    }
+    for worker in 0..2 {
+        assert_eq!(fixture.worker_count(worker, "textDocument/definition"), 0);
+        assert_eq!(fixture.worker_count(worker, "textDocument/hover"), 0);
+    }
+}