diff --git a/builtin/runtime/lsp.lua b/builtin/runtime/lsp.lua index d3d4799..1879485 100644 --- a/builtin/runtime/lsp.lua +++ b/builtin/runtime/lsp.lua @@ -153,21 +153,51 @@ pmacs.hook.add("buffer.after-edit", function() pcall(pmacs.lsp.did_change, rec.server, rec.uri, rec.version, active_buffer_text()) end) --- Synchronous poll: tick the supervisor + manager in tight loops until --- `predicate()` returns a truthy value or the deadline elapses. Uses --- wall-clock `pmacs.now_ms` because `os.clock` counts CPU time, and --- the inner ticks block on subprocess I/O instead of burning CPU. --- The v0.2 LSP UX pass swaps this for async-await coroutines. -local function poll_until(predicate, timeout_ms) - timeout_ms = timeout_ms or 250 - local deadline = pmacs.now_ms() + timeout_ms - while pmacs.now_ms() < deadline do - pmacs.process._tick() - pmacs.lsp._tick() - local ok, value = pcall(predicate) - if ok and value then return value end +-- Async request surface (T M4.5 async bridge). The Rust manager +-- registers each `textDocument/*` request with the async runtime and +-- returns a job id; the JSON-RPC response (or a server-teardown +-- drain) settles it. `_request_*_raw` is the raw job-id-returning +-- binding (mirrors `pmacs.mcp._send_request_raw`); the wrappers below +-- hand back a `pmacs.workers` Handle whose `:await()` resumes the +-- caller when the response lands. The pre-v1.0 `poll_until` tick-loop +-- this replaced blocked the editor for the whole request; awaiting +-- yields the coroutine instead. +local workers_mod = pmacs.workers +assert(workers_mod and workers_mod._new_handle, + "pmacs.workers._new_handle missing; did async.lua load before lsp.lua?") +assert(pmacs.lsp._request_completion_raw, + "pmacs.lsp._request_completion_raw missing; lua_bindings::install_lsp not run?") +local new_handle = workers_mod._new_handle + +local function wrap_request(raw) + -- Raises on dispatch failure (e.g. "server not ready"), matching + -- `mcp.lua`'s wrapper; callers pcall the `request():await()` chain. + return function(...) + return new_handle(raw(...)) end - return nil +end + +pmacs.lsp.request_completion = wrap_request(pmacs.lsp._request_completion_raw) +pmacs.lsp.request_hover = wrap_request(pmacs.lsp._request_hover_raw) +pmacs.lsp.request_signature_help = wrap_request(pmacs.lsp._request_signature_help_raw) +pmacs.lsp.request_definition = wrap_request(pmacs.lsp._request_definition_raw) +pmacs.lsp.request_formatting = wrap_request(pmacs.lsp._request_formatting_raw) + +-- Render an `:await()` failure into a modeline-friendly reason. +-- `Handle:await()` raises `{ tag = "cancelled", ... }` when the +-- server went away mid-request (drain in `lsp.rs`) and +-- `{ tag = "failed", message = ... }` for a JSON-RPC error response; +-- a raw dispatch failure surfaces as a plain string. +local function lsp_await_error(err) + if type(err) == "table" then + if err.tag == "cancelled" then + return "server unavailable (request cancelled)" + elseif err.tag == "failed" then + return err.message or "server error" + end + return tostring(err.tag or "error") + end + return tostring(err) end -- Cursor positioning ------------------------------------------------------ @@ -235,6 +265,16 @@ end -- Commands ---------------------------------------------------------------- +-- Each command captures the cursor/target at invocation time, then +-- spawns a coroutine that awaits the request and reacts. The editor +-- never blocks: the command function returns immediately and the +-- modeline updates when the response lands (or the await fails). +-- `:await()` sequences the work and surfaces server-gone / server- +-- error as structured errors; the normalized typed store (hybrid +-- model) is still the read path, so LSP result-shape variance +-- (Location | Location[] | LocationLink[], MarkupContent, …) stays +-- parsed in one place in Rust rather than re-derived here. + function pmacs.lsp.go_to_definition() local rec = attached_for_active() if not rec then @@ -244,28 +284,28 @@ function pmacs.lsp.go_to_definition() local line = pmacs.editor.cursor_line() local col = pmacs.editor.cursor_col() pmacs.definition.clear(rec.server, rec.uri) - local ok = pcall(pmacs.lsp.request_definition, rec.server, rec.uri, line, col) - if not ok then - pmacs.editor.set_status("LSP: server not ready") - return - end - local locs = poll_until(function() - local ls = pmacs.definition.locations(rec.server, rec.uri) - if #ls > 0 then return ls end - return nil - end, 500) - if not locs then - pmacs.editor.set_status("LSP: no definition found") - return - end - local first = locs[1] - if first.uri == rec.uri then - move_active_cursor_to(first.line, first.col) - pmacs.editor.set_status(string.format( - "LSP: definition at %d:%d", first.line + 1, first.col + 1)) - else - pmacs.editor.set_status("LSP: definition lives in " .. first.uri) - end + pmacs.async(function() + local ok, err = pcall(function() + pmacs.lsp.request_definition(rec.server, rec.uri, line, col):await() + end) + if not ok then + pmacs.editor.set_status("LSP: " .. lsp_await_error(err)) + return + end + local locs = pmacs.definition.locations(rec.server, rec.uri) + if not locs or #locs == 0 then + pmacs.editor.set_status("LSP: no definition found") + return + end + local first = locs[1] + if first.uri == rec.uri then + move_active_cursor_to(first.line, first.col) + pmacs.editor.set_status(string.format( + "LSP: definition at %d:%d", first.line + 1, first.col + 1)) + else + pmacs.editor.set_status("LSP: definition lives in " .. first.uri) + end + end) end function pmacs.lsp.format_buffer() @@ -275,22 +315,22 @@ function pmacs.lsp.format_buffer() return end pmacs.formatting.clear(rec.server, rec.uri) - local ok = pcall(pmacs.lsp.request_formatting, rec.server, rec.uri, 4, true) - if not ok then - pmacs.editor.set_status("LSP: server not ready") - return - end - local edits = poll_until(function() - local es = pmacs.formatting.edits(rec.server, rec.uri) - if #es > 0 then return es end - return nil - end, 1000) - if not edits then - pmacs.editor.set_status("LSP: no formatting edits") - return - end - local n = apply_text_edits(edits) - pmacs.editor.set_status(string.format("LSP: applied %d edits", n)) + pmacs.async(function() + local ok, err = pcall(function() + pmacs.lsp.request_formatting(rec.server, rec.uri, 4, true):await() + end) + if not ok then + pmacs.editor.set_status("LSP: " .. lsp_await_error(err)) + return + end + local edits = pmacs.formatting.edits(rec.server, rec.uri) + if not edits or #edits == 0 then + pmacs.editor.set_status("LSP: no formatting edits") + return + end + local n = apply_text_edits(edits) + pmacs.editor.set_status(string.format("LSP: applied %d edits", n)) + end) end function pmacs.lsp.hover_at_cursor() @@ -299,25 +339,28 @@ function pmacs.lsp.hover_at_cursor() pmacs.editor.set_status("LSP: no server for active buffer") return end + local line = pmacs.editor.cursor_line() + local col = pmacs.editor.cursor_col() pmacs.hover.clear(rec.server, rec.uri) - local ok = pcall(pmacs.lsp.request_hover, rec.server, rec.uri, - pmacs.editor.cursor_line(), pmacs.editor.cursor_col()) - if not ok then - pmacs.editor.set_status("LSP: server not ready") - return - end - local hover = poll_until(function() - return pmacs.hover.current(rec.server, rec.uri) - end, 500) - if not hover then - pmacs.editor.set_status("LSP: no hover info") - return - end - -- v0.1 surfaces the first line of the hover body in the modeline. The - -- popup view subscribes to the same store; future work can wire one - -- in here when the keybinding is meant to surface a panel. - local first = (hover.contents or ""):match("^[^\n]*") or "" - pmacs.editor.set_status(first ~= "" and ("LSP: " .. first) or "LSP: hover empty") + pmacs.async(function() + local ok, err = pcall(function() + pmacs.lsp.request_hover(rec.server, rec.uri, line, col):await() + end) + if not ok then + pmacs.editor.set_status("LSP: " .. lsp_await_error(err)) + return + end + local hover = pmacs.hover.current(rec.server, rec.uri) + if not hover then + pmacs.editor.set_status("LSP: no hover info") + return + end + -- Surface the first line of the hover body in the modeline. The + -- popup view subscribes to the same store; a panel can wire in + -- here when the keybinding is meant to surface one. + local first = (hover.contents or ""):match("^[^\n]*") or "" + pmacs.editor.set_status(first ~= "" and ("LSP: " .. first) or "LSP: hover empty") + end) end function pmacs.lsp.signature_help_at_cursor() @@ -326,22 +369,25 @@ function pmacs.lsp.signature_help_at_cursor() pmacs.editor.set_status("LSP: no server for active buffer") return end + local line = pmacs.editor.cursor_line() + local col = pmacs.editor.cursor_col() pmacs.signature.clear(rec.server, rec.uri) - local ok = pcall(pmacs.lsp.request_signature_help, rec.server, rec.uri, - pmacs.editor.cursor_line(), pmacs.editor.cursor_col()) - if not ok then - pmacs.editor.set_status("LSP: server not ready") - return - end - local help = poll_until(function() - return pmacs.signature.current(rec.server, rec.uri) - end, 500) - if not help or not help.signatures or #help.signatures == 0 then - pmacs.editor.set_status("LSP: no signature help") - return - end - local active = help.signatures[(help.active_signature or 0) + 1] - pmacs.editor.set_status(active and ("LSP: " .. active.label) or "LSP: signature unknown") + pmacs.async(function() + local ok, err = pcall(function() + pmacs.lsp.request_signature_help(rec.server, rec.uri, line, col):await() + end) + if not ok then + pmacs.editor.set_status("LSP: " .. lsp_await_error(err)) + return + end + local help = pmacs.signature.current(rec.server, rec.uri) + if not help or not help.signatures or #help.signatures == 0 then + pmacs.editor.set_status("LSP: no signature help") + return + end + local active = help.signatures[(help.active_signature or 0) + 1] + pmacs.editor.set_status(active and ("LSP: " .. active.label) or "LSP: signature unknown") + end) end -- Default commands + keymap entries -------------------------------------- diff --git a/src/async_runtime.rs b/src/async_runtime.rs index 8aeca78..cad4235 100644 --- a/src/async_runtime.rs +++ b/src/async_runtime.rs @@ -330,6 +330,14 @@ pub enum JobKind { /// supervisor pipe. No worker thread is occupied for the /// round-trip. McpRequest, + /// External request/reply for an LSP server, settled the same way + /// as [`JobKind::McpRequest`]: the [`crate::lsp::LspManager`] + /// registers a pending entry via [`AsyncRuntime::register_external`] + /// when a `textDocument/*` request goes out and settles it via + /// `complete_external_ok` / `_failed` / `_cancelled` when the + /// JSON-RPC response is routed in `LspManager::tick`. No worker + /// thread is occupied for the round-trip ([T M4.5] async bridge). + LspRequest, } impl JobKind { @@ -348,6 +356,7 @@ impl JobKind { JobKind::FsChmod => "fs_chmod", JobKind::FsRemove => "fs_remove", JobKind::McpRequest => "mcp_request", + JobKind::LspRequest => "lsp_request", } } } diff --git a/src/bin/pmacs_fake_lsp.rs b/src/bin/pmacs_fake_lsp.rs index f894010..e1c70db 100644 --- a/src/bin/pmacs_fake_lsp.rs +++ b/src/bin/pmacs_fake_lsp.rs @@ -70,6 +70,27 @@ fn main() { .get("params") .cloned() .unwrap_or(serde_json::Value::Null); + // T M4.5 async-bridge failure-path test modes: + // * `error` — answer every `textDocument/*` request with a + // JSON-RPC error object (drives `Handle:await()` -> failed). + // * `silent` — accept the request but never answer it while + // staying alive (drives the request-timeout sweep, and + // makes supersede deterministic: a superseded handle can + // only settle via cancellation, never racing a response). + if method.starts_with("textDocument/") { + if mode == "error" && id.is_some() { + let resp = serde_json::json!({ + "jsonrpc": "2.0", + "id": id.clone(), + "error": { "code": -32603, "message": "synthetic error" } + }); + write_frame(&mut stdout, &resp); + continue; + } + if mode == "silent" { + continue; + } + } match (method.as_str(), id) { ("initialize", Some(idv)) => { let resp = serde_json::json!({ diff --git a/src/editor.rs b/src/editor.rs index 240cbc4..1c02deb 100644 --- a/src/editor.rs +++ b/src/editor.rs @@ -183,9 +183,12 @@ impl EditorState { // T M4.5 LSP manager. Wires onto the same supervisor so its // spawn/restart/I/O machinery is shared with `pmacs.process.*`. // The manager itself is reachable from Lua as `pmacs.lsp.*`. - let lsp_manager = - crate::lua_bindings::make_lsp_manager(lua_host.lua(), process_supervisor.clone()) - .expect("install pmacs.lsp"); + let lsp_manager = crate::lua_bindings::make_lsp_manager( + lua_host.lua(), + process_supervisor.clone(), + async_runtime.clone(), + ) + .expect("install pmacs.lsp"); // T M9.1 MCP manager. Wires onto the same supervisor that LSP // and `pmacs.process.*` use; the protocol-uniformity claim is // that this share is sufficient (no parallel dispatch path). diff --git a/src/lsp.rs b/src/lsp.rs index 283dbd0..0af86de 100644 --- a/src/lsp.rs +++ b/src/lsp.rs @@ -53,7 +53,7 @@ //! it matters. use std::cell::RefCell; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::rc::Rc; use std::sync::atomic::{AtomicU64, Ordering}; @@ -61,10 +61,12 @@ use std::time::{Duration, Instant}; use serde_json::{Map, Value, json}; +use crate::async_runtime::{JobId, JobKind, SharedAsyncRuntime}; use crate::process::{ ProcessEvent, ProcessEventKind, ProcessId, ProcessMode, ProcessSpec, ProcessState, RestartPolicy, Termination, }; +use crate::worker::CancellationToken; // --------------------------------------------------------------------------- // Identity @@ -494,6 +496,13 @@ pub struct LspClient { attempt: u32, /// When the manager should attempt the next restart, if any. next_restart_at: Option, + /// T M4.5: request ids we sent `$/cancelRequest` for after every + /// awaiter abandoned them. A server may still answer a cancelled + /// request (the cancel/response race); a late response whose id + /// is in this set is dropped silently instead of surfacing a + /// "response for unknown request id" `ProtocolError`. Mirrors + /// `mcp.rs`'s `cancelled_rids`. + cancelled_rids: HashSet, } impl LspClient { @@ -507,6 +516,7 @@ impl LspClient { pending: HashMap::new(), attempt: 0, next_restart_at: None, + cancelled_rids: HashSet::new(), } } @@ -574,6 +584,22 @@ pub struct LspManager { /// `request_completion` etc. record an entry here; `handle_response` /// consumes it to absorb the response into the correct store. pending_routes: HashMap<(LspServerId, u64), ResponseRoute>, + /// T M4.5 async bridge: `(server, request_id)` → the awaiters + /// parked on this request's async-runtime job(s). Settled in + /// [`Self::handle_response`] when the response routes, and + /// drained-cancelled wherever [`Self::pending_routes`] is purged + /// for a server (restart / exit / forget) so a coroutine never + /// parks forever on a server that went away. + pending_external: HashMap<(LspServerId, u64), PendingExternal>, + /// Async runtime handle. The bridge between the supervisor + /// reader-thread response delivery and Lua-side `Handle:await()` + /// resumption; mirrors [`crate::mcp::McpManager`]'s `runtime`. + runtime: SharedAsyncRuntime, + /// T M4.5 task #8: hard ceiling on how long an in-flight request + /// may park its awaiters before the sweep fails them with a + /// timeout. Generous by default (real servers can take seconds on + /// a cold cache); tunable from Lua via `pmacs.lsp.set_request_timeout_ms`. + request_timeout: Duration, /// T M4.8 status tracker. Folds every emitted [`LspEvent`] into a /// per-server [`crate::lsp_status::LspStatus`] for the modeline / /// `*lsp*` buffer. @@ -607,15 +633,64 @@ enum ResponseRoute { Formatting { uri: String }, } +/// One Lua-visible awaiter bound to an in-flight LSP request. Mirrors +/// [`crate::mcp`]'s `Awaiter`: each carries its own +/// [`CancellationToken`] (minted by [`SharedAsyncRuntime::register_external`]) +/// so a single handle can be cancelled without disturbing the +/// in-flight wire request or sibling awaiters. +#[derive(Clone, Debug)] +struct Awaiter { + job_id: JobId, + /// Per-awaiter cancellation token minted by `register_external`. + /// Flipped by `Handle:cancel()` or by supersede; the per-tick + /// [`LspManager::drain_cancelled_externals`] sweep observes it. + /// Mirrors `mcp.rs`'s `Awaiter::token`. + token: CancellationToken, +} + +/// In-flight `textDocument/*` request bound to one or more +/// async-runtime jobs. When the JSON-RPC response is routed in +/// [`LspManager::handle_response`] the response is absorbed into the +/// typed store (the hybrid model — popup/gutter consumers are +/// untouched) *and* every non-cancelled awaiter is settled via +/// [`SharedAsyncRuntime::complete_external_ok`] (or `_failed` on a +/// server error). Keyed `(server, request_id)` alongside +/// [`LspManager::pending_routes`]. T M4.5 async bridge. +#[derive(Clone, Debug)] +struct PendingExternal { + /// JSON-RPC method, kept for the `*workers*`/`*lsp*` observability + /// surface and protocol-error messages. + method: String, + /// Live awaiters. Never empty while this entry exists. The first + /// is the original caller; later entries are siblings. + awaiters: Vec, + /// When the request went on the wire. T M4.5 task #8: the + /// per-tick sweep fails awaiters whose request has outlived + /// [`LspManager::request_timeout`], so a wedged-but-alive server + /// can't park a coroutine forever (the pre-v1.0 `poll_until` had + /// 500/1000 ms ceilings; this is the async-era equivalent, but a + /// generous hard ceiling rather than a UX poll budget). + dispatched_at: Instant, +} + impl LspManager { - /// Construct a fresh manager wired to `supervisor`. + /// Construct a fresh manager wired to `supervisor` and the + /// editor's `runtime`. The runtime bridges the supervisor's + /// reader-thread response delivery to Lua-side `Handle:await()` + /// resumption (mirrors [`crate::mcp::McpManager::new`]). #[must_use] - pub fn new(supervisor: crate::lua_bindings::SharedProcessSupervisor) -> Self { + pub fn new( + supervisor: crate::lua_bindings::SharedProcessSupervisor, + runtime: SharedAsyncRuntime, + ) -> Self { Self { supervisor, + runtime, clients: HashMap::new(), process_to_server: HashMap::new(), pending: HashMap::new(), + pending_external: HashMap::new(), + request_timeout: Duration::from_secs(10), restart_backoff: Duration::from_millis(500), diag_store: crate::diag::make_shared_store(), completion_store: crate::completion::make_shared_store(), @@ -796,10 +871,18 @@ impl LspManager { client.next_restart_at = None; client.stdout = FrameParser::new(); client.pending.clear(); + // T M4.5 task #7: the new generation has a fresh request-id + // space; stale cancelled ids can never collide, so reset to + // keep the set from growing across restarts. + client.cancelled_rids.clear(); // T M4.7: drop any pending response routes for this server. // Their request ids belong to the previous generation; the // new server starts request id numbering fresh. self.pending_routes.retain(|(sid, _), _| *sid != id); + // T M4.5: the previous generation's in-flight awaiters will + // never get a response (new process, fresh id space) — wake + // them cancelled before the restart. + self.drain_external_cancelled(id); client.state = LspClientState::Starting; let proc_spec = client.spec.to_process_spec(); let pid = self.supervisor.borrow_mut().spawn(proc_spec)?; @@ -899,97 +982,262 @@ impl LspManager { Ok(()) } + /// Register an async-runtime awaiter for the just-sent request + /// `req_id` and return its [`JobId`] — the value the Lua surface + /// wraps in a `pmacs.workers` Handle. Called only after + /// [`Self::send_request`] succeeded, so the wire request is live + /// and a response (or a teardown drain) will settle it; no + /// rollback path is needed here (unlike `mcp.rs`, where the + /// register precedes the wire send). T M4.5 async bridge. + /// + /// T M4.5 task #7: the awaiter is registered with a supersede key + /// `lsp:{method}:{sid}:{uri}` — stable across calls for the same + /// (server, method, document). A newer request for the same thing + /// flips the prior job's [`CancellationToken`] through the async + /// runtime's supersede map; the next [`Self::drain_cancelled_externals`] + /// sweep settles that awaiter cancelled and `$/cancelRequest`s the + /// in-flight wire request. This is what keeps keystroke-driven + /// completion from piling up N in-flight requests on the server. + fn register_awaiter( + &mut self, + sid: LspServerId, + req_id: u64, + method: &str, + uri: &str, + ) -> JobId { + let supersede = format!("lsp:{method}:{}:{uri}", sid.raw()); + let (job_id, token) = self + .runtime + .register_external(JobKind::LspRequest, Some(&supersede)); + self.pending_external.insert( + (sid, req_id), + PendingExternal { + method: method.to_owned(), + awaiters: vec![Awaiter { job_id, token }], + dispatched_at: Instant::now(), + }, + ); + job_id + } + + /// T M4.5 task #8: override the per-request timeout (default 10s). + /// Exposed to Lua as `pmacs.lsp.set_request_timeout_ms` and used + /// by the await-path tests to force fast timeouts. + pub fn set_request_timeout(&mut self, timeout: Duration) { + self.request_timeout = timeout; + } + + /// T M4.5 async bridge: settle every awaiter for `sid` as + /// cancelled and drop its `pending_external` entries. Called + /// wherever [`Self::pending_routes`] is purged for a server + /// (restart generation flip / terminal exit / forget) so a + /// coroutine parked on a request whose server went away wakes + /// with `{ tag = "cancelled" }` instead of hanging forever. + /// Mirrors `mcp.rs`'s drain-on-exit. Idempotent: a second call + /// finds no entries (and `complete_external_cancelled` is itself + /// idempotent against double-completion). + fn drain_external_cancelled(&mut self, sid: LspServerId) { + let keys: Vec<(LspServerId, u64)> = self + .pending_external + .keys() + .filter(|(s, _)| *s == sid) + .copied() + .collect(); + for k in keys { + if let Some(p) = self.pending_external.remove(&k) { + for a in &p.awaiters { + self.runtime.complete_external_cancelled(a.job_id); + } + } + } + } + + /// T M4.5 task #7: per-tick per-awaiter cancellation sweep for + /// `sid`. An awaiter whose [`CancellationToken`] was flipped — + /// either by `Handle:cancel()` or by being superseded by a newer + /// same-key request (see [`Self::register_awaiter`]) — is removed + /// and settled cancelled; the in-flight wire request continues if + /// sibling awaiters still want it. When the *last* awaiter of a + /// request abandons it, the entry and its route are dropped, the + /// rid is recorded in `cancelled_rids` so a late response is + /// dropped silently, and `$/cancelRequest` is sent best-effort so + /// the server can stop working. Mirrors `mcp.rs`'s + /// `drain_cancelled_externals` (LSP has no resource cache, so the + /// cache-state plumbing is omitted). + /// + /// T M4.5 task #8: the same sweep also fails any awaiter whose + /// request has outlived [`Self::request_timeout`] — a server that + /// stays alive but never answers a particular id would otherwise + /// park its coroutine forever. Timed-out entries take the same + /// abandon path as fully-cancelled ones (`$/cancelRequest` + + /// `cancelled_rids`), but settle `failed` rather than `cancelled`. + fn drain_cancelled_externals(&mut self, sid: LspServerId) { + let mut cancelled_jobs: Vec = Vec::new(); + let mut timed_out_jobs: Vec<(JobId, String)> = Vec::new(); + let mut abandoned_rids: Vec = Vec::new(); + // Captured into locals so the `retain` closure doesn't have to + // borrow `self` (it already borrows `self.pending_external`). + let now = Instant::now(); + let timeout = self.request_timeout; + let timeout_ms = timeout.as_millis(); + self.pending_external.retain(|(s, rid), p| { + if *s != sid { + return true; + } + let mut still: Vec = Vec::with_capacity(p.awaiters.len()); + for a in p.awaiters.drain(..) { + if a.token.is_cancelled() { + cancelled_jobs.push(a.job_id); + } else { + still.push(a); + } + } + // T M4.5 task #8: any awaiter that survived the cancel + // pass but whose request has outlived the timeout fails + // now — the server is alive but not answering this id. + if !still.is_empty() && now.duration_since(p.dispatched_at) >= timeout { + for a in still.drain(..) { + timed_out_jobs.push((a.job_id, p.method.clone())); + } + } + p.awaiters = still; + if p.awaiters.is_empty() { + abandoned_rids.push(*rid); + false + } else { + true + } + }); + for job_id in cancelled_jobs { + self.runtime.complete_external_cancelled(job_id); + } + for (job_id, method) in timed_out_jobs { + self.runtime.complete_external_failed( + job_id, + format!("LSP {method}: request timed out after {timeout_ms}ms"), + ); + } + for rid in abandoned_rids { + self.pending_routes.remove(&(sid, rid)); + if let Some(client) = self.clients.get_mut(&sid) { + client.pending.remove(&rid); + client.cancelled_rids.insert(rid); + } + self.send_cancel_request(sid, rid); + } + } + + /// Send `$/cancelRequest { id }` to `sid`, best-effort. A server + /// that is not accepting writes (stopped / crashed) is skipped by + /// [`Self::send_notification`]'s state guard; the `Err` is + /// intentionally ignored. T M4.5 task #7. + fn send_cancel_request(&mut self, sid: LspServerId, rid: u64) { + let _ = self.send_notification(sid, "$/cancelRequest", json!({ "id": rid })); + } + /// Send `textDocument/completion` for `uri` at `(line, col)`. /// The response is absorbed into the completion store at - /// `(sid, uri)`; the same `LspEventKind::Response` event is also - /// emitted so direct observers can see the raw payload. - /// Returns the JSON-RPC request id. + /// `(sid, uri)` (popup consumers untouched) and also settles the + /// returned awaiter. The same `LspEventKind::Response` event is + /// still emitted for raw observers. Returns the async-runtime + /// [`JobId`] the response will settle (`type JobId = u64`, so the + /// signature is unchanged for existing Rust callers — only the + /// value's meaning moved from JSON-RPC id to job id, and no + /// caller consumes it). T M4.5 async bridge. pub fn request_completion( &mut self, sid: LspServerId, uri: impl Into, line: u32, col: u32, - ) -> Result { + ) -> Result { let uri = uri.into(); let params = json!({ "textDocument": { "uri": uri.clone() }, "position": { "line": line, "character": col } }); let req_id = self.send_request(sid, "textDocument/completion", params)?; + let job_id = self.register_awaiter(sid, req_id, "textDocument/completion", &uri); self.pending_routes .insert((sid, req_id), ResponseRoute::Completion { uri }); - Ok(req_id) + Ok(job_id) } - /// Send `textDocument/hover` for `uri` at `(line, col)`. + /// Send `textDocument/hover` for `uri` at `(line, col)`. Returns + /// the async-runtime [`JobId`] the response will settle. pub fn request_hover( &mut self, sid: LspServerId, uri: impl Into, line: u32, col: u32, - ) -> Result { + ) -> Result { let uri = uri.into(); let params = json!({ "textDocument": { "uri": uri.clone() }, "position": { "line": line, "character": col } }); let req_id = self.send_request(sid, "textDocument/hover", params)?; + let job_id = self.register_awaiter(sid, req_id, "textDocument/hover", &uri); self.pending_routes .insert((sid, req_id), ResponseRoute::Hover { uri }); - Ok(req_id) + Ok(job_id) } /// Send `textDocument/signatureHelp` for `uri` at `(line, col)`. + /// Returns the async-runtime [`JobId`] the response will settle. pub fn request_signature_help( &mut self, sid: LspServerId, uri: impl Into, line: u32, col: u32, - ) -> Result { + ) -> Result { let uri = uri.into(); let params = json!({ "textDocument": { "uri": uri.clone() }, "position": { "line": line, "character": col } }); let req_id = self.send_request(sid, "textDocument/signatureHelp", params)?; + let job_id = self.register_awaiter(sid, req_id, "textDocument/signatureHelp", &uri); self.pending_routes .insert((sid, req_id), ResponseRoute::Signature { uri }); - Ok(req_id) + Ok(job_id) } /// Send `textDocument/definition` for `uri` at `(line, col)`. The /// response is absorbed into the definition store at `(sid, uri)`. + /// Returns the async-runtime [`JobId`] the response will settle. pub fn request_definition( &mut self, sid: LspServerId, uri: impl Into, line: u32, col: u32, - ) -> Result { + ) -> Result { let uri = uri.into(); let params = json!({ "textDocument": { "uri": uri.clone() }, "position": { "line": line, "character": col } }); let req_id = self.send_request(sid, "textDocument/definition", params)?; + let job_id = self.register_awaiter(sid, req_id, "textDocument/definition", &uri); self.pending_routes .insert((sid, req_id), ResponseRoute::Definition { uri }); - Ok(req_id) + Ok(job_id) } /// Send `textDocument/formatting` for `uri` with `tab_size` / /// `insert_spaces` formatting options. The response is absorbed - /// into the formatting store at `(sid, uri)`. + /// into the formatting store at `(sid, uri)`. Returns the + /// async-runtime [`JobId`] the response will settle. pub fn request_formatting( &mut self, sid: LspServerId, uri: impl Into, tab_size: u32, insert_spaces: bool, - ) -> Result { + ) -> Result { let uri = uri.into(); let params = json!({ "textDocument": { "uri": uri.clone() }, @@ -999,9 +1247,10 @@ impl LspManager { } }); let req_id = self.send_request(sid, "textDocument/formatting", params)?; + let job_id = self.register_awaiter(sid, req_id, "textDocument/formatting", &uri); self.pending_routes .insert((sid, req_id), ResponseRoute::Formatting { uri }); - Ok(req_id) + Ok(job_id) } /// Reply to a server-initiated request. @@ -1036,6 +1285,10 @@ impl LspManager { let server_ids: Vec = self.clients.keys().copied().collect(); for sid in server_ids { self.drain_process_events(sid); + // T M4.5 task #7: after this tick's responses are absorbed + // (above), reap any awaiter cancelled or superseded since + // the last tick. + self.drain_cancelled_externals(sid); self.maybe_restart(sid); } // T M4.8: housekeeping for the status tracker (releases stale @@ -1228,6 +1481,11 @@ impl LspManager { !was_shutdown && should_restart(client.spec.restart), ) }; + // T M4.5: the server is gone (clean stop or crash) and will + // never answer its in-flight requests. Wake every awaiter + // cancelled now — not at the eventual restart/forget, which + // may never come under `LspRestartPolicy::Never`. + self.drain_external_cancelled(sid); if was_shutdown { self.push_event(sid, at, LspEventKind::Stopped); } else { @@ -1363,6 +1621,14 @@ impl LspManager { let Some(client) = self.clients.get_mut(&sid) else { return; }; + // T M4.5 task #7: the cancel/response race. We sent + // `$/cancelRequest` for this id after every awaiter + // abandoned it; a server may answer anyway. By definition + // no awaiter is listening, so drop it silently rather + // than emitting a `ProtocolError` for an expected outcome. + if client.cancelled_rids.remove(&rid) { + return; + } client.pending.remove(&rid).unwrap_or_default() }; if method.is_empty() { @@ -1423,6 +1689,28 @@ impl LspManager { { self.absorb_routed_response(sid, &route, value); } + // T M4.5 async bridge: settle every awaiter parked on this + // request. Deliberately independent of the store-absorb guard + // above — a null result (e.g. "no hover here") is still a + // successful response and must wake `:await()` with `nil` + // rather than leave the coroutine parked forever. A server + // error settles the awaiter as failed so `:await()` raises the + // structured `{ tag = "failed" }`. The typed store was already + // populated above when present (hybrid model); popup/gutter + // consumers are unaffected by this block. + if let Some(p) = self.pending_external.remove(&(sid, rid)) { + if let Some(err) = error.as_ref() { + let msg = format!("LSP {} error {}: {}", p.method, err.code, err.message); + for a in &p.awaiters { + self.runtime.complete_external_failed(a.job_id, msg.clone()); + } + } else { + let value = result.clone().unwrap_or(Value::Null); + for a in &p.awaiters { + self.runtime.complete_external_ok(a.job_id, value.clone()); + } + } + } // Generic response. self.push_event( sid, @@ -1659,6 +1947,10 @@ impl LspManager { self.clients.remove(&sid); self.pending.remove(&sid); self.pending_routes.retain(|(s, _), _| *s != sid); + // T M4.5: belt-and-braces — on_exit already drained on the + // terminal transition; this catches any awaiter registered + // between exit and forget. Idempotent. + self.drain_external_cancelled(sid); self.status_tracker.forget(sid); // T M4.9: drop the project scoping so the next // ensure_server_for_project call spawns a fresh server. diff --git a/src/lua_bindings.rs b/src/lua_bindings.rs index 5390349..fd10477 100644 --- a/src/lua_bindings.rs +++ b/src/lua_bindings.rs @@ -7117,16 +7117,31 @@ pub fn install_lsp(lua: &Lua, manager: &SharedLspManager) -> mlua::Result<()> { } { + // T M4.5 task #8: hard per-request timeout, tunable from + // init.lua (e.g. raise it for a slow language server on a + // cold cache, or lower it in tests). let m = manager.clone(); lsp_mod.set( - "request_completion", + "set_request_timeout_ms", + lua.create_function(move |_, ms: u64| { + m.borrow_mut() + .set_request_timeout(std::time::Duration::from_millis(ms)); + Ok(()) + })?, + )?; + } + + { + let m = manager.clone(); + lsp_mod.set( + "_request_completion_raw", lua.create_function( move |_, (id, uri, line, col): (LspServerIdLua, String, u32, u32)| { - let req_id = m + let job_id = m .borrow_mut() .request_completion(id.0, uri, line, col) .map_err(mlua::Error::external)?; - Ok(req_id) + Ok(job_id) }, )?, )?; @@ -7135,14 +7150,14 @@ pub fn install_lsp(lua: &Lua, manager: &SharedLspManager) -> mlua::Result<()> { { let m = manager.clone(); lsp_mod.set( - "request_hover", + "_request_hover_raw", lua.create_function( move |_, (id, uri, line, col): (LspServerIdLua, String, u32, u32)| { - let req_id = m + let job_id = m .borrow_mut() .request_hover(id.0, uri, line, col) .map_err(mlua::Error::external)?; - Ok(req_id) + Ok(job_id) }, )?, )?; @@ -7151,14 +7166,14 @@ pub fn install_lsp(lua: &Lua, manager: &SharedLspManager) -> mlua::Result<()> { { let m = manager.clone(); lsp_mod.set( - "request_signature_help", + "_request_signature_help_raw", lua.create_function( move |_, (id, uri, line, col): (LspServerIdLua, String, u32, u32)| { - let req_id = m + let job_id = m .borrow_mut() .request_signature_help(id.0, uri, line, col) .map_err(mlua::Error::external)?; - Ok(req_id) + Ok(job_id) }, )?, )?; @@ -7167,14 +7182,14 @@ pub fn install_lsp(lua: &Lua, manager: &SharedLspManager) -> mlua::Result<()> { { let m = manager.clone(); lsp_mod.set( - "request_definition", + "_request_definition_raw", lua.create_function( move |_, (id, uri, line, col): (LspServerIdLua, String, u32, u32)| { - let req_id = m + let job_id = m .borrow_mut() .request_definition(id.0, uri, line, col) .map_err(mlua::Error::external)?; - Ok(req_id) + Ok(job_id) }, )?, )?; @@ -7183,7 +7198,7 @@ pub fn install_lsp(lua: &Lua, manager: &SharedLspManager) -> mlua::Result<()> { { let m = manager.clone(); lsp_mod.set( - "request_formatting", + "_request_formatting_raw", lua.create_function( move |_, (id, uri, tab_size, insert_spaces): ( @@ -7192,11 +7207,11 @@ pub fn install_lsp(lua: &Lua, manager: &SharedLspManager) -> mlua::Result<()> { u32, Option, )| { - let req_id = m + let job_id = m .borrow_mut() .request_formatting(id.0, uri, tab_size, insert_spaces.unwrap_or(true)) .map_err(mlua::Error::external)?; - Ok(req_id) + Ok(job_id) }, )?, )?; @@ -7424,8 +7439,9 @@ pub fn install_lsp(lua: &Lua, manager: &SharedLspManager) -> mlua::Result<()> { pub fn make_lsp_manager( lua: &Lua, supervisor: SharedProcessSupervisor, + runtime: crate::async_runtime::SharedAsyncRuntime, ) -> mlua::Result { - let manager = Rc::new(RefCell::new(LspManager::new(supervisor))); + let manager = Rc::new(RefCell::new(LspManager::new(supervisor, runtime))); install_lsp(lua, &manager)?; install_diag(lua, &manager)?; install_completion(lua, &manager)?; diff --git a/tests/m4_acceptance.rs b/tests/m4_acceptance.rs index e28d615..b2e26c9 100644 --- a/tests/m4_acceptance.rs +++ b/tests/m4_acceptance.rs @@ -1069,7 +1069,12 @@ fn make_lsp_test_manager() -> ( use std::rc::Rc; let sup = Rc::new(RefCell::new(pmacs::process::ProcessSupervisor::new())); - let mgr = Rc::new(RefCell::new(LspManager::new(sup.clone()))); + // The manager owns the runtime Rc; these store-assertion tests + // never tick it, so the registered external entries are simply + // never drained (harmless). The await-path tests (T M4.5 task #9) + // use a separate helper that also returns the runtime. + let runtime = Rc::new(AsyncRuntime::with_pool_size(1)); + let mgr = Rc::new(RefCell::new(LspManager::new(sup.clone(), runtime))); (sup, mgr) } @@ -3368,3 +3373,276 @@ fn project_search_boundary_round_trips_via_lua() { "set_search_boundary(nil) must clear back to nil" ); } + +// --------------------------------------------------------------------------- +// T M4.5 async bridge — Handle:await() path (task #9). +// +// These drive the real end-to-end surface: EditorState (runtime wired +// into the LSP manager + builtin lsp.lua loaded), `pmacs.lsp.spawn` +// with a fake-server mode, a `pmacs.async` coroutine that `:await()`s, +// and Rust ticking processes/lsp/async until the coroutine settles a +// `_G` flag. Mirrors `m9_1_lua_send_request_returns_awaitable_handle`. +// --------------------------------------------------------------------------- + +/// Tick processes → lsp → async until the Lua expression `flag` +/// evaluates true, or the deadline elapses. Returns whether it fired. +fn pump_lua_flag(state: &mut pmacs::editor::EditorState, flag: &str, secs: u64) -> bool { + let deadline = Instant::now() + Duration::from_secs(secs); + loop { + state.tick_processes(); + state.tick_lsp(); + state.tick_async(); + let done: bool = state + .lua_host + .lua() + .load(format!("return ({flag}) == true")) + .eval() + .unwrap_or(false); + if done { + return true; + } + if Instant::now() >= deadline { + return false; + } + std::thread::sleep(Duration::from_millis(10)); + } +} + +/// Spawn the fake LSP via the Lua surface (optionally in a test mode) +/// and pump until the manager reports it `initialized`. +fn spawn_lsp_and_init(state: &mut pmacs::editor::EditorState, mode: Option<&str>) { + let fake = fake_lsp_path(); + let env = match mode { + Some(m) => format!(", env = {{ PMACS_FAKE_LSP_MODE = '{m}' }}"), + None => String::new(), + }; + state + .lua_host + .lua() + .load(format!( + "_G._lsp = pmacs.lsp.spawn({{ label='await-test', language_id='rust', \ + command='{fake}', restart='never'{env} }})" + )) + .exec() + .expect("spawn lsp via Lua"); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + state.tick_processes(); + state.tick_lsp(); + state.tick_async(); + let init: bool = state + .lua_host + .lua() + .load( + "for _,r in ipairs(pmacs.lsp.list()) do \ + if r.state and r.state.kind=='initialized' then return true end end \ + return false", + ) + .eval() + .unwrap_or(false); + if init { + return; + } + assert!( + Instant::now() < deadline, + "fake LSP never reached initialized (mode {mode:?})" + ); + std::thread::sleep(Duration::from_millis(10)); + } +} + +/// Success: `request_completion():await()` returns the result table, +/// and the typed store is *also* populated (the hybrid model). +#[test] +fn m4_5_await_completion_returns_result_and_populates_store() { + use pmacs::editor::EditorState; + let mut state = EditorState::new(); + spawn_lsp_and_init(&mut state, None); + state + .lua_host + .lua() + .load( + "_G._done=false _G._res=nil + pmacs.async(function() + _G._res = pmacs.lsp.request_completion(_G._lsp,'file:///x.rs',0,0):await() + _G._items = pmacs.completion.items(_G._lsp,'file:///x.rs') + _G._done=true + end)", + ) + .exec() + .expect("dispatch await coroutine"); + assert!( + pump_lua_flag(&mut state, "_G._done", 5), + "await coroutine never completed" + ); + let (res_is_table, store_count): (bool, i64) = state + .lua_host + .lua() + .load("return type(_G._res)=='table', (_G._items and #_G._items) or 0") + .eval() + .expect("read result"); + assert!(res_is_table, "await() should return the result table"); + assert!( + store_count >= 3, + "hybrid: completion store must also be populated (got {store_count})" + ); +} + +/// Server JSON-RPC error → `:await()` raises `{ tag = 'failed' }`. +#[test] +fn m4_5_await_server_error_raises_failed() { + use pmacs::editor::EditorState; + let mut state = EditorState::new(); + spawn_lsp_and_init(&mut state, Some("error")); + state + .lua_host + .lua() + .load( + "_G._done=false _G._tag=nil _G._msg=nil + pmacs.async(function() + local ok,v = pcall(function() + return pmacs.lsp.request_hover(_G._lsp,'file:///x.rs',0,0):await() + end) + _G._tag = (not ok) and type(v)=='table' and v.tag or 'unexpected-ok' + _G._msg = (type(v)=='table' and v.message) or '' + _G._done=true + end)", + ) + .exec() + .expect("dispatch await coroutine"); + assert!( + pump_lua_flag(&mut state, "_G._done", 5), + "await coroutine never completed" + ); + let (tag, msg): (String, String) = state + .lua_host + .lua() + .load("return _G._tag, _G._msg") + .eval() + .expect("read tag"); + assert_eq!(tag, "failed", "server error must surface as failed"); + assert!( + msg.contains("synthetic error"), + "failure message should carry the server's error text; got {msg:?}" + ); +} + +/// Server stopped while a request is in flight → the teardown drain +/// wakes the awaiter with `{ tag = 'cancelled' }` (not a hang). +#[test] +fn m4_5_await_cancelled_when_server_stops_mid_request() { + use pmacs::editor::EditorState; + let mut state = EditorState::new(); + spawn_lsp_and_init(&mut state, Some("silent")); + state + .lua_host + .lua() + .load( + "_G._done=false _G._tag=nil + pmacs.async(function() + local ok,v = pcall(function() + return pmacs.lsp.request_definition(_G._lsp,'file:///x.rs',0,0):await() + end) + _G._tag = (not ok) and type(v)=='table' and v.tag or 'unexpected-ok' + _G._done=true + end) + -- Request is now in flight against a silent server; stop it. + pmacs.lsp.stop(_G._lsp)", + ) + .exec() + .expect("dispatch + stop"); + assert!( + pump_lua_flag(&mut state, "_G._done", 5), + "await coroutine never completed (server-stop drain didn't wake it)" + ); + let tag: String = state + .lua_host + .lua() + .load("return _G._tag") + .eval() + .expect("read tag"); + assert_eq!(tag, "cancelled", "server-gone must wake await as cancelled"); +} + +/// Alive-but-silent server → the per-request timeout sweep fails the +/// awaiter (`{ tag = 'failed', message ~ 'timed out' }`) so it can't +/// park forever. +#[test] +fn m4_5_await_times_out_against_silent_server() { + use pmacs::editor::EditorState; + let mut state = EditorState::new(); + spawn_lsp_and_init(&mut state, Some("silent")); + state + .lua_host + .lua() + .load( + "pmacs.lsp.set_request_timeout_ms(150) + _G._done=false _G._tag=nil _G._msg=nil + pmacs.async(function() + local ok,v = pcall(function() + return pmacs.lsp.request_hover(_G._lsp,'file:///x.rs',0,0):await() + end) + _G._tag = (not ok) and type(v)=='table' and v.tag or 'unexpected-ok' + _G._msg = (type(v)=='table' and v.message) or '' + _G._done=true + end)", + ) + .exec() + .expect("dispatch await coroutine"); + assert!( + pump_lua_flag(&mut state, "_G._done", 5), + "await coroutine never timed out" + ); + let (tag, msg): (String, String) = state + .lua_host + .lua() + .load("return _G._tag, _G._msg") + .eval() + .expect("read tag"); + assert_eq!(tag, "failed", "timeout must surface as failed"); + assert!( + msg.contains("timed out"), + "timeout message should say so; got {msg:?}" + ); +} + +/// A newer same-(server,method,uri) request supersedes the in-flight +/// one: the first handle's `:await()` raises `{ tag = 'cancelled' }`. +/// Silent server so the only way the first can settle is supersede. +#[test] +fn m4_5_await_superseded_request_is_cancelled() { + use pmacs::editor::EditorState; + let mut state = EditorState::new(); + spawn_lsp_and_init(&mut state, Some("silent")); + state + .lua_host + .lua() + .load( + "_G._done=false _G._h1=nil + pmacs.async(function() + local h1 = pmacs.lsp.request_completion(_G._lsp,'file:///x.rs',0,0) + local h2 = pmacs.lsp.request_completion(_G._lsp,'file:///x.rs',0,0) + local ok1,v1 = pcall(function() return h1:await() end) + _G._h1 = (not ok1) and type(v1)=='table' and v1.tag or 'unexpected-ok' + _G._done=true + -- h2 left in flight; the server stop below drains it. + end)", + ) + .exec() + .expect("dispatch supersede coroutine"); + assert!( + pump_lua_flag(&mut state, "_G._done", 5), + "supersede coroutine never completed" + ); + let h1: String = state + .lua_host + .lua() + .load("return _G._h1") + .eval() + .expect("read h1 tag"); + assert_eq!( + h1, "cancelled", + "the superseded (older) request must await-cancel" + ); + let _ = state.lua_host.lua().load("pmacs.lsp.stop(_G._lsp)").exec(); +} diff --git a/tests/m9_1_acceptance.rs b/tests/m9_1_acceptance.rs index b42f6ac..47c8a59 100644 --- a/tests/m9_1_acceptance.rs +++ b/tests/m9_1_acceptance.rs @@ -280,7 +280,7 @@ fn m9_1_multiple_mcps_coexist_on_one_manager() { fn m9_1_lsp_and_mcp_can_coexist_on_one_supervisor() { let sup = Rc::new(RefCell::new(ProcessSupervisor::new())); let runtime: SharedAsyncRuntime = Rc::new(AsyncRuntime::with_pool_size(1)); - let lsp_mgr = Rc::new(RefCell::new(LspManager::new(sup.clone()))); + let lsp_mgr = Rc::new(RefCell::new(LspManager::new(sup.clone(), runtime.clone()))); let mcp_mgr = Rc::new(RefCell::new(McpManager::new(sup.clone(), runtime.clone()))); // Spin up one LSP and one MCP through the same supervisor.