neuwirth
/
pmacs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #9 from levineuwirth/v1.1-semantic 

T M4.5 async bridge: LSP requests settle async-runtime jobs
This commit is contained in:
Levi Neuwirth 2026-05-18 23:37:49 +00:00 committed by GitHub
parent d3fa63290a ad530a5beb
commit dbaacad755
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 789 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

216
builtin/runtime/lsp.lua
View File

@ -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 --------------------------------------

9
src/async_runtime.rs
View File

@ -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",
}
}
}

21
src/bin/pmacs_fake_lsp.rs
View File

@ -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!({

9
src/editor.rs
View File

@ -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).

328
src/lsp.rs
View File

@ -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<Instant>,
/// 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<u64>,
}
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<Awaiter>,
/// 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<JobId> = Vec::new();
let mut timed_out_jobs: Vec<(JobId, String)> = Vec::new();
let mut abandoned_rids: Vec<u64> = 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<Awaiter> = 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<String>,
line: u32,
col: u32,
) -> Result<u64, String> {
) -> Result<JobId, String> {
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<String>,
line: u32,
col: u32,
) -> Result<u64, String> {
) -> Result<JobId, String> {
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<String>,
line: u32,
col: u32,
) -> Result<u64, String> {
) -> Result<JobId, String> {
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<String>,
line: u32,
col: u32,
) -> Result<u64, String> {
) -> Result<JobId, String> {
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<String>,
tab_size: u32,
insert_spaces: bool,
) -> Result<u64, String> {
) -> Result<JobId, String> {
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<LspServerId> = 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.

48
src/lua_bindings.rs
View File

@ -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<bool>,
)| {
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<SharedLspManager> {
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)?;

280
tests/m4_acceptance.rs
View File

@ -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();
}

2
tests/m9_1_acceptance.rs
View File

@ -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.