fix(lsp): look up nested single-root clients and version docs before didChange send

Two LSP freshness bugs reported by @tobific (#108882, #108881):

- `_current_diags_async()` keyed the client lookup by the enclosing
  workspace root while `_get_or_spawn()` stores single-root servers under
  `srv.resolve_root(...)` (a nested package.json project). The lookup
  returned [] for a live client with diagnostics, so the delta baseline was
  refreshed from nothing. Use the same resolved-root key.

- `open_or_change()` published `_DocState.version` only after awaiting the
  didChange write. A versionless publishDiagnostics read during that await
  was credited with the OLD version and judged stale once the send resumed.
  Bump the version before the send; a failed send (swallowed by
  `_send_notification`) leaves a version nothing satisfies, i.e. "no
  verdict", which is the existing contract.

The mock server gains a push-only `versionless` script so the race is
reproducible without a real language server.

agent/lsp/client.py

@@ -501,12 +501,16 @@ class LSPClient:
         if self._sync_kind == 2:
             change["range"] = {"start": {"line": 0, "character": 0}, "end": _end_position(doc.text)}
         new_version = doc.version + 1
+        # Bumping the version is the whole invalidation story (see _DocState).  It happens
+        # BEFORE the send: the write awaits, and a versionless publishDiagnostics read during
+        # that await is credited with doc.version -- tagged with the old number it would be
+        # judged stale the moment the send resumes.  A failed send is swallowed by
+        # _send_notification, leaving a version nothing ever satisfies (= "no verdict").
+        doc.version, doc.text = new_version, text
         await self._send_notification(
             "textDocument/didChange",
             {"textDocument": {"uri": uri, "version": new_version}, "contentChanges": [change]},
         )
-        # Bumping the version is the whole invalidation story (see _DocState).
-        doc.version, doc.text = new_version, text
         return new_version
 
     async def save_file(self, path: str) -> None:

agent/lsp/manager.py

@@ -358,8 +358,13 @@ class LSPService:
         srv = find_server_for_file(file_path)
         if not (ws and gated and srv):
             return []
+        # Same key _get_or_spawn() stored under: single-root servers live under their
+        # resolved project root (a nested package.json), not the enclosing workspace.
+        root = srv.resolve_root(file_path, ws)
+        if root is None:
+            return []
         with self._state_lock:
-            client = self._clients.get(_client_key(srv, ws))
+            client = self._clients.get(_client_key(srv, root))
         return list(client.diagnostics_for(file_path, fresh_only=True)) if client else []
 
     async def _get_or_spawn(self, file_path: str) -> Optional[LSPClient]:

tests/agent/lsp/_mock_lsp_server.py

@@ -25,6 +25,10 @@ Behaviour (all behaviours selectable via env var ``MOCK_LSP_SCRIPT``):
   ``didChange`` sleeps ``MOCK_LSP_PUSH_DELAY`` seconds (default 1.0)
   and then pushes EMPTY diagnostics.  Models a server that fixes
   the ghost if you actually wait for it.  Pull endpoint rejects.
+- ``"versionless"`` — errors on ``didOpen``, clean on ``didChange``, and
+  no ``version`` field in any publishDiagnostics (the client credits
+  each push with its current document version at receipt).  Push-only:
+  the pull endpoint rejects.
 - ``"clean_eof"`` — closes stdout after ``didOpen`` but keeps the
   process and stdin alive.
 - ``"malformed_frame"`` — writes an invalid frame after ``didOpen``,
@@ -165,21 +169,18 @@ def main():
             diagnostics = []
             if script == "errors":
                 diagnostics = error_diag
-            write_message(
-                {
-                    "jsonrpc": "2.0",
-                    "method": "textDocument/publishDiagnostics",
-                    "params": {
-                        "uri": uri,
-                        "version": version,
-                        "diagnostics": diagnostics,
-                    },
-                }
-            )
+            if script == "versionless":
+                # Servers that never echo a document version: the client credits the
+                # push with its current version at receipt.
+                diagnostics = [] if is_change else error_diag
+            params = {"uri": uri, "version": version, "diagnostics": diagnostics}
+            if script == "versionless":
+                del params["version"]
+            write_message({"jsonrpc": "2.0", "method": "textDocument/publishDiagnostics", "params": params})
             continue
 
         if msg.get("method") == "textDocument/diagnostic":
-            if script in {"stale", "slow_push"}:
+            if script in {"stale", "slow_push", "versionless"}:
                 # These scripts model push-only servers so the ghost
                 # can't be papered over by the pull channel.
                 write_message(

tests/agent/lsp/test_nested_root_current_diagnostics.py

@@ -0,0 +1,45 @@
+"""``_current_diags_async`` must find a single-root client under the root it was spawned with.
+
+``_get_or_spawn`` keys single-root servers by ``srv.resolve_root(...)`` (a nested ``package.json``
+project), but the current-diagnostics lookup keyed by the enclosing workspace root, so the delta
+baseline was refreshed from ``[]`` while the live client held diagnostics.
+"""
+from __future__ import annotations
+
+import dataclasses
+import sys
+from pathlib import Path
+
+from agent.lsp import manager, servers
+
+MOCK_SERVER = str(Path(__file__).parent / "_mock_lsp_server.py")
+
+
+def test_nested_single_root_client_is_found_by_current_lookup(tmp_path, monkeypatch):
+    repo = tmp_path / "repo"
+    (repo / ".git").mkdir(parents=True)
+    nested = repo / "package"
+    nested.mkdir()
+    (nested / "package.json").write_text("{}", encoding="utf-8")
+    src = nested / "x.ts"
+    src.write_text("const x = 1;\n", encoding="utf-8")
+    monkeypatch.chdir(repo)
+
+    original = next(s for s in servers.SERVERS if s.server_id == "typescript")
+    assert original.resolve_root(str(src), str(repo)) == str(nested) and not original.multi_root
+
+    def spawn(root, ctx):
+        return servers.SpawnSpec(command=[sys.executable, MOCK_SERVER], workspace_root=root, cwd=root,
+                                 env={"MOCK_LSP_SCRIPT": "errors"}, initialization_options={})
+
+    mocked = dataclasses.replace(original, build_spawn=spawn)
+    monkeypatch.setattr(servers, "SERVERS", [mocked if s is original else s for s in servers.SERVERS])
+
+    svc = manager.LSPService(enabled=True, wait_mode="document", wait_timeout=5, install_strategy="manual")
+    try:
+        svc.snapshot_baseline(str(src))
+        live = svc.get_diagnostics_sync(str(src), delta=False)
+        assert live
+        assert svc._loop.run(svc._current_diags_async(str(src)), timeout=5) == live
+    finally:
+        svc.shutdown()

tests/agent/lsp/test_versionless_push_during_send.py

@@ -0,0 +1,50 @@
+"""A versionless publishDiagnostics read while didChange is still being written must count as fresh.
+
+``open_or_change`` used to bump ``_DocState.version`` only after awaiting the send.  Servers that omit
+``version`` are credited with ``doc.version`` at receipt, so a reply that landed during that await was
+tagged with the OLD version and then rejected as stale once the send resumed.  The mock replies
+versionless; the paused send wrapper holds the await open until its reply has been read.
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+from agent.lsp.client import LSPClient
+
+MOCK_SERVER = str(Path(__file__).parent / "_mock_lsp_server.py")
+
+
+@pytest.mark.asyncio
+async def test_versionless_push_read_during_didchange_send_is_fresh(tmp_path, monkeypatch):
+    src = tmp_path / "x.py"
+    src.write_text("bad\n", encoding="utf-8")
+    client = LSPClient(
+        server_id="mock-versionless", workspace_root=str(tmp_path),
+        command=[sys.executable, MOCK_SERVER], cwd=str(tmp_path),
+        env={"MOCK_LSP_SCRIPT": "versionless", "PYTHONPATH": os.environ.get("PYTHONPATH", "")},
+    )
+    await client.start()
+    try:
+        first = await client.open_file(str(src), language_id="python")
+        assert await client.wait_for_diagnostics(str(src), first, timeout=5)
+        real_send = client._send_notification
+
+        async def send_and_let_reply_land(method, params):
+            seen = client._push_counter
+            await real_send(method, params)
+            if method == "textDocument/didChange":
+                while client._push_counter == seen:
+                    await asyncio.sleep(0.001)
+
+        monkeypatch.setattr(client, "_send_notification", send_and_let_reply_land)
+        src.write_text("clean\n", encoding="utf-8")
+        version = await client.open_file(str(src), language_id="python")
+        assert await client.wait_for_diagnostics(str(src), version, timeout=1)
+        assert client.diagnostics_for(str(src), fresh_only=True) == []
+    finally:
+        await client.shutdown()