fix(lsp): never report stale diagnostics — wait for fresh post-edit data

Slow language servers (tsserver on large projects especially) publish
diagnostics long after an edit. The client's wait/report path had three
holes that together surfaced the PREVIOUS edit's errors as if they were
current ("ghost diagnostics"), sending the agent chasing errors it had
already fixed:

1. open_file only cleared the diagnostic stores on first open — on the
   didChange path (every subsequent edit) stale push/pull entries
   survived.
2. wait_for_diagnostics' predicates were satisfiable by that leftover
   state (`path in _published`, `path in _pull_diagnostics`), so the
   "wait" often returned instantly with old data.
3. diagnostics_for merged the stale push store unconditionally, so even
   a fresh clean pull got the old error merged back in.

Fix: anchor freshness on a per-file didChange timestamp.

- Pull results record their request send-time and are dropped when a
  didChange raced past them; the pull store is invalidated on every
  change, not just first open.
- wait_for_diagnostics now returns bool (fresh data vs timeout), only
  counts pushes published at/after the change (and version >= ours when
  the server echoes versions), and accepts an explicit timeout — the
  user's lsp.wait_timeout config now actually controls the inner wait
  budget instead of only the outer thread-join.
- diagnostics_for(fresh_only=True) excludes stores that predate the
  latest change; all manager report paths use it.
- On timeout the manager returns [] ("no data") instead of stale
  state, logs a WARNING via eventlog, and does NOT mark the server
  broken — slow is not dead.
- seed-on-first-push no longer marks the file published, so the TS
  seed push can't satisfy a waiter.

Tests: new "stale" and "slow_push" mock-server scripts model the slow
tsserver, plus client- and service-level regression tests
(tests/agent/lsp/test_stale_diagnostics.py).

agent/lsp/client.py

@@ -198,6 +198,17 @@ class LSPClient:
         self._published_version: Dict[str, int] = {}
         # First-push seen flag, for typescript-style seed-on-first-push.
         self._first_push_seen: Set[str] = set()
+        # Per-path loop time of the most recent didOpen/didChange we
+        # sent.  Freshness anchor: pushes/pulls that predate this are
+        # results for OLD content and must never satisfy a waiter or
+        # be reported to the agent (the "ghost diagnostics" bug —
+        # slow servers like tsserver publish long after the edit, so
+        # leftovers from the previous edit look like current errors).
+        self._changed_at: Dict[str, float] = {}
+        # Per-path loop time at which the most recent *stored* pull
+        # request was sent.  Send-time (not receipt) so a didChange
+        # racing a pull in flight correctly invalidates the result.
+        self._pulled_at: Dict[str, float] = {}
         # Capability registrations — only diagnostic ones are tracked.
         self._diagnostic_registrations: Dict[str, Dict[str, Any]] = {}
 
@@ -650,15 +661,13 @@ class LSPClient:
         loop_time = asyncio.get_event_loop().time()
 
         if self._seed_first_push and path not in self._first_push_seen:
-            # First push: seed without firing the event so a waiter
+            # First push: seed the store without marking the file as
-            # doesn't resolve on the very first push (which arrives
+            # "published" and without firing the event.  The very first
-            # before the user-triggered didChange could've produced
+            # push arrives before the user-triggered didChange could've
-            # fresh diagnostics).
+            # produced fresh diagnostics, so it must never satisfy a
+            # waiter — it's baseline data only.
             self._first_push_seen.add(path)
             self._push_diagnostics[path] = diagnostics
-            self._published[path] = loop_time
-            if isinstance(version, int):
-                self._published_version[path] = version
             return
 
         self._push_diagnostics[path] = diagnostics
@@ -725,6 +734,12 @@ class LSPClient:
                 },
             )
             self._files[abs_path] = {"version": new_version, "text": text}
+            # Anchor freshness at this change: any pull result stored
+            # before now describes the PREVIOUS content and must not be
+            # reported (it would resurrect just-fixed errors as ghosts).
+            self._changed_at[abs_path] = asyncio.get_event_loop().time()
+            self._pull_diagnostics.pop(abs_path, None)
+            self._pulled_at.pop(abs_path, None)
             return new_version
 
         # First open: didChangeWatchedFiles CREATED + didOpen.
@@ -738,6 +753,8 @@ class LSPClient:
         self._pull_diagnostics.pop(abs_path, None)
         self._published.pop(abs_path, None)
         self._published_version.pop(abs_path, None)
+        self._pulled_at.pop(abs_path, None)
+        self._changed_at[abs_path] = asyncio.get_event_loop().time()
         await self._send_notification(
             "textDocument/didOpen",
             {
@@ -771,10 +788,17 @@ class LSPClient:
 
         Stores results into :attr:`_pull_diagnostics`.  Silently
         no-ops on errors (server may not support the pull endpoint).
+
+        The request's *send time* is recorded in :attr:`_pulled_at`
+        so freshness checks can compare against :attr:`_changed_at`.
+        A result whose request predates the latest didChange is
+        dropped — it describes content that no longer exists.
         """
+        abs_path = os.path.abspath(path)
+        sent_at = asyncio.get_event_loop().time()
         try:
             params: Dict[str, Any] = {
-                "textDocument": {"uri": file_uri(os.path.abspath(path))}
+                "textDocument": {"uri": file_uri(abs_path)}
             }
             result = await self._send_request_with_retry(
                 "textDocument/diagnostic",
@@ -786,9 +810,18 @@ class LSPClient:
             return
         if not isinstance(result, dict):
             return
+        if sent_at < self._changed_at.get(abs_path, 0.0):
+            # The document changed while this pull was in flight — the
+            # server answered for the OLD text.  Storing it would
+            # resurrect ghost diagnostics.
+            logger.debug(
+                "[%s] dropping stale pull result for %s", self.server_id, abs_path
+            )
+            return
         items = result.get("items")
         if isinstance(items, list):
-            self._pull_diagnostics[os.path.abspath(path)] = items
+            self._pull_diagnostics[abs_path] = items
+            self._pulled_at[abs_path] = sent_at
         related = result.get("relatedDocuments")
         if isinstance(related, dict):
             for uri, sub in related.items():
@@ -796,7 +829,36 @@ class LSPClient:
                     continue
                 sub_items = sub.get("items")
                 if isinstance(sub_items, list):
-                    self._pull_diagnostics[uri_to_path(uri)] = sub_items
+                    rel_path = uri_to_path(uri)
+                    if sent_at < self._changed_at.get(rel_path, 0.0):
+                        continue
+                    self._pull_diagnostics[rel_path] = sub_items
+                    self._pulled_at[rel_path] = sent_at
+
+    def _has_fresh_push(self, path: str, version: int) -> bool:
+        """True iff a *fresh* publishDiagnostics has arrived for ``path``.
+
+        Fresh means: published at/after the latest didOpen/didChange we
+        sent, AND (when the server echoes versions) for a version >= the
+        one we're waiting on.  A stale push left over from the previous
+        edit cycle satisfies neither and must not end a wait early —
+        that's exactly the "ghost diagnostics" failure mode with slow
+        servers like tsserver.
+        """
+        published_at = self._published.get(path)
+        if published_at is None:
+            return False
+        if published_at < self._changed_at.get(path, 0.0):
+            return False
+        current_v = self._published_version.get(path)
+        return current_v is None or current_v >= version
+
+    def _has_fresh_pull(self, path: str) -> bool:
+        """True iff the stored pull result postdates the latest change."""
+        pulled_at = self._pulled_at.get(path)
+        if pulled_at is None:
+            return path in self._pull_diagnostics
+        return pulled_at >= self._changed_at.get(path, 0.0)
 
     async def wait_for_diagnostics(
         self,
@@ -804,22 +866,36 @@ class LSPClient:
         version: int,
         *,
         mode: str = "document",
-    ) -> None:
+        timeout: Optional[float] = None,
+    ) -> bool:
         """Wait for the server to publish diagnostics for ``path`` at ``version``.
 
         ``mode`` is ``"document"`` (5s budget, document pulls) or
-        ``"full"`` (10s budget, also workspace pulls).  Best-effort —
+        ``"full"`` (10s budget, also workspace pulls).  ``timeout``
-        returns silently on timeout.  Does NOT throw if the server
+        overrides the mode's default budget when provided — this is
-        doesn't support pull diagnostics; we still get the push side.
+        how the user's ``lsp.wait_timeout`` config reaches the wait
+        loop (slow servers like tsserver on big projects need more
+        than the 5s default).
+
+        Returns ``True`` when *fresh* diagnostics arrived (a push at
+        or after our didChange, or a pull answered after it) and
+        ``False`` on timeout.  Callers must treat ``False`` as "no
+        data", NOT as "no errors" — the diagnostic stores may still
+        hold stale entries from the previous edit at that point.
+        Best-effort — never throws if the server doesn't support pull
+        diagnostics; we still get the push side.
         """
-        budget = DIAGNOSTICS_FULL_WAIT if mode == "full" else DIAGNOSTICS_DOCUMENT_WAIT
+        if timeout is not None and timeout > 0:
+            budget = timeout
+        else:
+            budget = DIAGNOSTICS_FULL_WAIT if mode == "full" else DIAGNOSTICS_DOCUMENT_WAIT
         deadline = asyncio.get_event_loop().time() + budget
         abs_path = os.path.abspath(path)
 
         while True:
             remaining = deadline - asyncio.get_event_loop().time()
             if remaining <= 0:
-                return
+                return False
 
             # Concurrent: document pull + push wait.
             pull_task = asyncio.create_task(self._pull_document_diagnostics(abs_path))
@@ -838,26 +914,22 @@ class LSPClient:
                     pass
 
             # If we got a fresh push for our version, we're done.
-            current_v = self._published_version.get(abs_path)
+            if self._has_fresh_push(abs_path, version):
-            if abs_path in self._published and (
+                return True
-                current_v is None or current_v >= version
-            ):
-                return
 
-            # Pull may have populated _pull_diagnostics — that's also
+            # Pull may have populated _pull_diagnostics with a
-            # success.
+            # post-change answer — that's also success.
-            if abs_path in self._pull_diagnostics:
+            if self._has_fresh_pull(abs_path):
-                return
+                return True
 
             # Loop until budget runs out.
 
     async def _wait_for_fresh_push(self, path: str, version: int, timeout: float) -> None:
-        """Wait until a publishDiagnostics arrives for ``path`` at ``version``+."""
+        """Wait until a fresh publishDiagnostics arrives for ``path`` at ``version``+."""
         deadline = asyncio.get_event_loop().time() + timeout
         baseline = self._push_counter
         while True:
-            current_v = self._published_version.get(path)
+            if self._has_fresh_push(path, version):
-            if path in self._published and (current_v is None or current_v >= version):
                 # Debounce — wait a tick in case more diagnostics arrive
                 # immediately after.  TS often emits in pairs.  We
                 # snapshot the counter so we wake on a *new* push, not
@@ -888,14 +960,28 @@ class LSPClient:
             except asyncio.TimeoutError:
                 continue
 
-    def diagnostics_for(self, path: str) -> List[Dict[str, Any]]:
+    def diagnostics_for(self, path: str, *, fresh_only: bool = False) -> List[Dict[str, Any]]:
         """Return current merged + deduped diagnostics for one file.
 
         Diagnostics from push and pull stores are concatenated and
         deduplicated by ``(severity, code, message, range)`` content
         key.  Empty list if the server hasn't published anything.
+
+        With ``fresh_only=True``, a store only contributes when its
+        data postdates the latest didOpen/didChange for the file —
+        stale leftovers from the previous edit cycle are excluded.
+        This is what report paths should use: after an edit, "stale
+        errors" and "no errors" must not be conflated.
         """
         abs_path = os.path.abspath(path)
+        if fresh_only:
+            changed_at = self._changed_at.get(abs_path, 0.0)
+            push_fresh = self._published.get(abs_path, -1.0) >= changed_at
+            pulled_at = self._pulled_at.get(abs_path)
+            pull_fresh = pulled_at is not None and pulled_at >= changed_at
+            push = (self._push_diagnostics.get(abs_path) or []) if push_fresh else []
+            pull = (self._pull_diagnostics.get(abs_path) or []) if pull_fresh else []
+            return _dedupe(push, pull)
         push = self._push_diagnostics.get(abs_path) or []
         pull = self._pull_diagnostics.get(abs_path) or []
         return _dedupe(push, pull)

agent/lsp/manager.py

@@ -292,7 +292,10 @@ class LSPService:
         if not self.enabled_for(file_path):
             return
         try:
-            diags = self._loop.run(self._snapshot_async(file_path), timeout=8.0)
+            # Outer join budget must exceed the inner wait budget or a
+            # slow-but-alive server gets falsely marked broken.
+            t = max(8.0, self._wait_timeout + 3.0)
+            diags = self._loop.run(self._snapshot_async(file_path), timeout=t)
             self._delta_baseline[os.path.abspath(file_path)] = diags or []
         except Exception as e:  # noqa: BLE001
             logger.debug("baseline snapshot failed for %s: %s", file_path, e)
@@ -341,7 +344,7 @@ class LSPService:
 
         try:
             t = timeout if timeout is not None else self._wait_timeout + 2.0
-            diags = self._loop.run(self._open_and_wait_async(file_path), timeout=t) or []
+            diags = self._loop.run(self._open_and_wait_async(file_path), timeout=t)
         except asyncio.TimeoutError as e:
             eventlog.log_timeout(server_id, file_path)
             logger.debug("LSP diagnostics timeout for %s: %s", file_path, e)
@@ -353,6 +356,17 @@ class LSPService:
             self._mark_broken_for_file(file_path, e)
             return []
 
+        if diags is None:
+            # The server is alive but never produced diagnostics for the
+            # post-edit content within the wait budget (common for
+            # tsserver on large projects).  Report "no data" rather than
+            # whatever stale state is in the stores — surfacing the
+            # previous edit's errors as if they were current is the
+            # ghost-diagnostics bug.  The server is NOT marked broken:
+            # slow is not dead, and the next edit may well succeed.
+            eventlog.log_timeout(server_id, file_path, kind="fresh diagnostics")
+            return []
+
         abs_path = os.path.abspath(file_path)
         if delta:
             baseline = self._delta_baseline.get(abs_path) or []
@@ -452,26 +466,43 @@ class LSPService:
             return []
         try:
             version = await client.open_file(file_path, language_id=language_id_for(file_path))
-            await client.wait_for_diagnostics(file_path, version, mode=self._wait_mode)
+            fresh = await client.wait_for_diagnostics(file_path, version, mode=self._wait_mode)
         except Exception as e:  # noqa: BLE001
             logger.debug("snapshot open/wait failed: %s", e)
             return []
         self._last_used[(client.server_id, client.workspace_root)] = time.time()
-        return list(client.diagnostics_for(file_path))
+        if not fresh:
+            # No fresh data for the pre-edit content — an empty baseline
+            # is safe: worst case the delta filter removes less, never
+            # more.  Never seed the baseline from stale stores.
+            return []
+        return list(client.diagnostics_for(file_path, fresh_only=True))
 
-    async def _open_and_wait_async(self, file_path: str) -> List[Dict[str, Any]]:
+    async def _open_and_wait_async(self, file_path: str) -> Optional[List[Dict[str, Any]]]:
+        """Open + wait for FRESH diagnostics.
+
+        Returns the fresh diagnostic list, or ``None`` when the server
+        never produced post-change data within the wait budget.  The
+        distinction matters: ``[]`` means "server checked the new
+        content, it's clean", ``None`` means "no verdict" — the caller
+        must not substitute stale data for either.
+        """
         client = await self._get_or_spawn(file_path)
         if client is None:
-            return []
+            return None
         try:
             version = await client.open_file(file_path, language_id=language_id_for(file_path))
             await client.save_file(file_path)
-            await client.wait_for_diagnostics(file_path, version, mode=self._wait_mode)
+            fresh = await client.wait_for_diagnostics(
+                file_path, version, mode=self._wait_mode, timeout=self._wait_timeout
+            )
         except Exception as e:  # noqa: BLE001
             logger.debug("open/wait failed for %s: %s", file_path, e)
-            return []
+            return None
         self._last_used[(client.server_id, client.workspace_root)] = time.time()
-        return list(client.diagnostics_for(file_path))
+        if not fresh:
+            return None
+        return list(client.diagnostics_for(file_path, fresh_only=True))
 
     async def _current_diags_async(self, file_path: str) -> List[Dict[str, Any]]:
         ws, gated = resolve_workspace_for_file(file_path)
@@ -482,7 +513,7 @@ class LSPService:
             client = self._clients.get((srv.server_id, ws))
         if client is None:
             return []
-        return list(client.diagnostics_for(file_path))
+        return list(client.diagnostics_for(file_path, fresh_only=True))
 
     async def _get_or_spawn(self, file_path: str) -> Optional[LSPClient]:
         srv = find_server_for_file(file_path)

tests/agent/lsp/_mock_lsp_server.py

@@ -17,6 +17,14 @@ Behaviour (all behaviours selectable via env var ``MOCK_LSP_SCRIPT``):
   (simulates a crashing server).
 - ``"slow"`` — same as ``clean`` but sleeps 1s before responding to
   ``initialize`` (lets us test timeout behaviour).
+- ``"stale"`` — pushes one error on ``didOpen``, then goes SILENT on
+  ``didChange`` (no push) and rejects the pull endpoint with
+  method-not-found.  Models a slow tsserver that hasn't re-checked
+  the edited content yet — the ghost-diagnostics scenario.
+- ``"slow_push"`` — like ``stale`` on didOpen (one error) but on
+  ``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.
 
 The script writes JSON-RPC framed messages to stdout and reads from
 stdin.  No third-party dependencies — uses only stdlib so it runs
@@ -96,20 +104,47 @@ def main():
             td = params.get("textDocument") or {}
             uri = td.get("uri", "")
             version = td.get("version", 0)
+            is_change = msg.get("method") == "textDocument/didChange"
+            error_diag = [
+                {
+                    "range": {
+                        "start": {"line": 0, "character": 0},
+                        "end": {"line": 0, "character": 5},
+                    },
+                    "severity": 1,
+                    "code": "MOCK001",
+                    "source": "mock-lsp",
+                    "message": "synthetic error from mock-lsp",
+                }
+            ]
+            if script == "stale":
+                # Ghost scenario: publish an error for the ORIGINAL
+                # content, then never publish again after edits.
+                if not is_change:
+                    write_message(
+                        {
+                            "jsonrpc": "2.0",
+                            "method": "textDocument/publishDiagnostics",
+                            "params": {"uri": uri, "version": version, "diagnostics": error_diag},
+                        }
+                    )
+                continue
+            if script == "slow_push":
+                diagnostics = error_diag
+                if is_change:
+                    time.sleep(float(os.environ.get("MOCK_LSP_PUSH_DELAY", "1.0")))
+                    diagnostics = []
+                write_message(
+                    {
+                        "jsonrpc": "2.0",
+                        "method": "textDocument/publishDiagnostics",
+                        "params": {"uri": uri, "version": version, "diagnostics": diagnostics},
+                    }
+                )
+                continue
             diagnostics = []
             if script == "errors":
-                diagnostics = [
+                diagnostics = error_diag
-                    {
-                        "range": {
-                            "start": {"line": 0, "character": 0},
-                            "end": {"line": 0, "character": 5},
-                        },
-                        "severity": 1,
-                        "code": "MOCK001",
-                        "source": "mock-lsp",
-                        "message": "synthetic error from mock-lsp",
-                    }
-                ]
             write_message(
                 {
                     "jsonrpc": "2.0",
@@ -124,6 +159,17 @@ def main():
             continue
 
         if msg.get("method") == "textDocument/diagnostic":
+            if script in {"stale", "slow_push"}:
+                # These scripts model push-only servers so the ghost
+                # can't be papered over by the pull channel.
+                write_message(
+                    {
+                        "jsonrpc": "2.0",
+                        "id": msg["id"],
+                        "error": {"code": -32601, "message": "method not found"},
+                    }
+                )
+                continue
             # Pull endpoint — return empty.
             write_message(
                 {

tests/agent/lsp/test_stale_diagnostics.py

@@ -0,0 +1,249 @@
+"""Regression tests for the "ghost diagnostics" staleness bug.
+
+Scenario: the agent edits a TypeScript file, tsserver takes a long
+time to re-check it, and the old diagnostics (for the PRE-edit
+content) were reported as if they were current — the agent then
+chases errors it already fixed.
+
+The contract under test:
+
+- ``wait_for_diagnostics`` must NOT be satisfied by diagnostics left
+  over from a previous edit cycle; it returns True only when fresh
+  (post-didChange) data arrived, False on timeout.
+- ``diagnostics_for(fresh_only=True)`` must exclude stale stores.
+- ``LSPService.get_diagnostics_sync`` must return [] ("no data")
+  rather than the stale diagnostics when the server never re-checks
+  within the wait budget, and must NOT mark the server broken.
+- A slow-but-eventually-correct server ("slow_push") is waited on,
+  honouring the configured ``lsp.wait_timeout``.
+"""
+from __future__ import annotations
+
+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")
+
+
+def _client(workspace: Path, script: str, **env_extra: str) -> LSPClient:
+    env = {
+        "MOCK_LSP_SCRIPT": script,
+        "PYTHONPATH": os.environ.get("PYTHONPATH", ""),
+        **env_extra,
+    }
+    return LSPClient(
+        server_id=f"mock-{script}",
+        workspace_root=str(workspace),
+        command=[sys.executable, MOCK_SERVER],
+        env=env,
+        cwd=str(workspace),
+    )
+
+
+@pytest.mark.asyncio
+async def test_stale_push_does_not_satisfy_wait(tmp_path: Path):
+    """A push from the previous edit cycle must not end the wait early.
+
+    The 'stale' mock publishes an error for the original content and
+    then goes silent — the wait after the edit must time out (False),
+    not return instantly on the leftover push.
+    """
+    f = tmp_path / "x.py"
+    f.write_text("bad code\n")
+
+    client = _client(tmp_path, "stale")
+    await client.start()
+    try:
+        v0 = await client.open_file(str(f), language_id="python")
+        assert await client.wait_for_diagnostics(str(f), v0, mode="document", timeout=2.0)
+        assert len(client.diagnostics_for(str(f))) == 1  # pre-edit error is real
+
+        # Fix the file.  The stale server never re-checks.
+        f.write_text("good code\n")
+        v1 = await client.open_file(str(f), language_id="python")
+        fresh = await client.wait_for_diagnostics(str(f), v1, mode="document", timeout=1.0)
+        assert fresh is False, "wait must not be satisfied by pre-edit leftovers"
+    finally:
+        await client.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_fresh_only_excludes_stale_stores(tmp_path: Path):
+    f = tmp_path / "x.py"
+    f.write_text("bad code\n")
+
+    client = _client(tmp_path, "stale")
+    await client.start()
+    try:
+        v0 = await client.open_file(str(f), language_id="python")
+        await client.wait_for_diagnostics(str(f), v0, mode="document", timeout=2.0)
+
+        f.write_text("good code\n")
+        await client.open_file(str(f), language_id="python")
+        # Merged legacy view still exposes the leftover push...
+        assert len(client.diagnostics_for(str(f))) == 1
+        # ...but the fresh-only view correctly reports no verdict yet.
+        assert client.diagnostics_for(str(f), fresh_only=True) == []
+    finally:
+        await client.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_slow_push_is_waited_for(tmp_path: Path):
+    """A server that re-checks slowly (but within budget) gets waited on,
+    and the fresh (clean) result replaces the old error."""
+    f = tmp_path / "x.py"
+    f.write_text("bad code\n")
+
+    client = _client(tmp_path, "slow_push", MOCK_LSP_PUSH_DELAY="0.8")
+    await client.start()
+    try:
+        v0 = await client.open_file(str(f), language_id="python")
+        assert await client.wait_for_diagnostics(str(f), v0, mode="document", timeout=2.0)
+        assert len(client.diagnostics_for(str(f), fresh_only=True)) == 1
+
+        f.write_text("good code\n")
+        v1 = await client.open_file(str(f), language_id="python")
+        fresh = await client.wait_for_diagnostics(str(f), v1, mode="document", timeout=5.0)
+        assert fresh is True, "slow push within budget must satisfy the wait"
+        assert client.diagnostics_for(str(f), fresh_only=True) == []
+    finally:
+        await client.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_wait_timeout_param_overrides_mode_budget(tmp_path: Path):
+    """The explicit timeout must control the wait budget (config plumb)."""
+    import asyncio
+
+    f = tmp_path / "x.py"
+    f.write_text("bad code\n")
+
+    client = _client(tmp_path, "stale")
+    await client.start()
+    try:
+        v0 = await client.open_file(str(f), language_id="python")
+        await client.wait_for_diagnostics(str(f), v0, mode="document", timeout=2.0)
+        f.write_text("good code\n")
+        v1 = await client.open_file(str(f), language_id="python")
+
+        loop = asyncio.get_event_loop()
+        start = loop.time()
+        fresh = await client.wait_for_diagnostics(str(f), v1, mode="document", timeout=0.5)
+        elapsed = loop.time() - start
+        assert fresh is False
+        # Must respect ~0.5s, not the 5s document default.
+        assert elapsed < 3.0
+    finally:
+        await client.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_stale_pull_result_dropped_when_change_races(tmp_path: Path):
+    """A pull answered for pre-edit content must not be stored after a
+    didChange raced past it (send-time anchoring)."""
+    f = tmp_path / "x.py"
+    f.write_text("bad code\n")
+
+    client = _client(tmp_path, "clean")
+    await client.start()
+    try:
+        v0 = await client.open_file(str(f), language_id="python")
+        await client.wait_for_diagnostics(str(f), v0, mode="document", timeout=2.0)
+        assert client._has_fresh_pull(os.path.abspath(str(f)))
+
+        # Simulate an edit racing in: the change invalidates the pull.
+        f.write_text("good code\n")
+        await client.open_file(str(f), language_id="python")
+        assert not client._has_fresh_pull(os.path.abspath(str(f)))
+        assert os.path.abspath(str(f)) not in client._pull_diagnostics
+    finally:
+        await client.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# Service-level: stale data must surface as "no data", never as errors
+# ---------------------------------------------------------------------------
+
+
+def _install_mock_server(script: str, server_id: str = "pyright"):
+    """Replace one registered server with a wrapper spawning the mock.
+
+    Mirrors the helper in test_service.py — reuse pyright so .py files
+    route to the mock without a real toolchain.
+    """
+    from agent.lsp.servers import SERVERS, ServerContext, ServerDef, SpawnSpec
+
+    target_index = next(i for i, s in enumerate(SERVERS) if s.server_id == server_id)
+    original = SERVERS[target_index]
+
+    def _spawn(root: str, ctx: ServerContext) -> SpawnSpec:
+        return SpawnSpec(
+            command=[sys.executable, MOCK_SERVER],
+            workspace_root=root,
+            cwd=root,
+            env={"MOCK_LSP_SCRIPT": script},
+            initialization_options={},
+        )
+
+    SERVERS[target_index] = ServerDef(
+        server_id=server_id,
+        extensions=original.extensions,
+        resolve_root=lambda fp, ws: ws,
+        build_spawn=_spawn,
+        seed_first_push=False,
+        description="mock " + server_id,
+    )
+    return target_index, original
+
+
+@pytest.fixture
+def stale_repo(monkeypatch, tmp_path):
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / ".git").mkdir()
+    (repo / "pyproject.toml").write_text("")
+    monkeypatch.chdir(str(repo))
+    idx, original = _install_mock_server("stale")
+    yield repo
+    from agent.lsp.servers import SERVERS
+
+    SERVERS[idx] = original
+
+
+def test_service_reports_no_data_not_stale_errors(stale_repo):
+    """When the server never re-checks the edited content in budget,
+    get_diagnostics_sync must return [] and keep the server usable."""
+    from agent.lsp.manager import LSPService
+
+    f = stale_repo / "x.py"
+    f.write_text("bad code\n")
+
+    svc = LSPService(
+        enabled=True,
+        wait_mode="document",
+        wait_timeout=1.0,
+        install_strategy="manual",
+    )
+    try:
+        # First contact: didOpen gets the (real) pre-edit error push.
+        first = svc.get_diagnostics_sync(str(f), delta=False)
+        assert len(first) == 1
+
+        # Edit the file — mock never re-publishes (slow tsserver model).
+        f.write_text("good code\n")
+        ghost = svc.get_diagnostics_sync(str(f), delta=False)
+        assert ghost == [], "stale pre-edit error must not be reported as current"
+
+        # Not marked broken: slow is not dead.
+        assert svc.enabled_for(str(f))
+        status = svc.get_status()
+        assert status["broken"] == []
+    finally:
+        svc.shutdown()

website/docs/user-guide/features/lsp.md

@@ -151,6 +151,12 @@ lsp:
 
   # How long to wait for diagnostics after each write.
   wait_mode: document      # "document" or "full"
+  # Max seconds to wait for the server to re-check the file after an
+  # edit. Only *fresh* diagnostics (produced for the post-edit
+  # content) are ever reported; if the server doesn't finish within
+  # this budget, the edit reports "no LSP data" rather than stale
+  # errors from before the edit. Raise this for slow servers on big
+  # projects (tsserver, rust-analyzer mid-indexing).
   wait_timeout: 5.0
 
   # How to handle missing server binaries.
@@ -209,6 +215,13 @@ budget is `wait_timeout` seconds — typically the server responds in
 tens of milliseconds for pyright/tsserver and a few seconds for
 rust-analyzer mid-indexing.
 
+Diagnostics are **freshness-gated**: a result only counts when the
+server produced it for the content of the current edit (a
+`publishDiagnostics` push at/after the change, or a pull request
+answered after it). Slow servers that haven't re-checked yet result
+in "no data" for that edit — never in yesterday's errors being
+re-reported as current.
+
 Servers are kept alive for the life of the Hermes process. There's
 no idle-timeout reaper — the cost of restarting the server's index
 on every write would be far higher than holding the daemon.