fix(lsp): take _state_lock in _set_delta_baseline; pin the cap through the production call sites

The pop/insert/evict sequence mutates the process-wide baseline dict from arbitrary
gateway/subagent threads; on main the write was a single atomic assignment, so the cap
introduced a race (RuntimeError/KeyError past 256 entries) that would surface in the
write-tool path. Callers never hold the lock, so the helper takes it.

The cap test now drives snapshot_baseline and _apply_delta instead of the helper, so
reverting either production call site to a direct assignment is red, and asserts every
mutation runs under the lock.

agent/lsp/manager.py

@@ -226,11 +226,13 @@ class LSPService:
         self._set_delta_baseline(os.path.abspath(file_path), diags or [])
 
     def _set_delta_baseline(self, abs_path: str, diags: _Diags) -> None:
-        """Store a baseline, refreshing recency (pop + reinsert) so eviction tracks write order."""
-        self._delta_baseline.pop(abs_path, None)
-        self._delta_baseline[abs_path] = diags
-        while len(self._delta_baseline) > _DELTA_BASELINE_CAP:
-            del self._delta_baseline[next(iter(self._delta_baseline))]
+        """Store a baseline, refreshing recency (pop + reinsert) so eviction tracks write order.
+        Callers run on arbitrary threads; the multi-step mutation needs the lock (callers don't hold it)."""
+        with self._state_lock:
+            self._delta_baseline.pop(abs_path, None)
+            self._delta_baseline[abs_path] = diags
+            while len(self._delta_baseline) > _DELTA_BASELINE_CAP:
+                del self._delta_baseline[next(iter(self._delta_baseline))]
 
     def get_diagnostics_sync(
         self, file_path: str, *, delta: bool = True, timeout: Optional[float] = None,

tests/agent/lsp/test_manager_locking.py

@@ -59,16 +59,40 @@ def test_reused_multiroot_client_attaches_outside_state_lock(monkeypatch):
     assert result is client
 
 
-def test_delta_baseline_is_capped_by_write_recency():
-    """Baselines beyond _DELTA_BASELINE_CAP evict the path written longest ago, and rewriting a
-    path refreshes it instead of aging it out ahead of paths never touched again."""
+def test_delta_baseline_is_capped_by_write_recency(monkeypatch):
+    """Driven through the production entry points (``snapshot_baseline`` before a write, ``_apply_delta``
+    rolling the baseline forward after one): baselines beyond _DELTA_BASELINE_CAP evict the path
+    written longest ago, and re-touching a path refreshes it instead of aging it out."""
     service = LSPService(enabled=False, wait_mode="document", wait_timeout=0.1,
                          install_strategy="manual", idle_timeout=0)
+    monkeypatch.setattr(service, "enabled_for", lambda _p: True)
+    server_diags: list = []
+
+    class _Loop:  # stands in for the (unstarted) background loop; returns the server's current diagnostics
+        def run(self, coro, timeout=None):
+            coro.close()
+            return list(server_diags)
+
+    monkeypatch.setattr(service, "_loop", _Loop())
+
+    class _LockedDict(dict):  # every mutation of the shared baseline dict must run under _state_lock
+        def __setitem__(self, k, v):
+            assert service._state_lock.locked()
+            super().__setitem__(k, v)
+
+        def __delitem__(self, k):
+            assert service._state_lock.locked()
+            super().__delitem__(k)
+
+    service._delta_baseline = _LockedDict()
     cap = manager._DELTA_BASELINE_CAP
     for i in range(cap):
-        service._set_delta_baseline(f"/repo/f{i}.py", [])
-    service._set_delta_baseline("/repo/f0.py", [{"message": "rewritten"}])
-    service._set_delta_baseline("/repo/new.py", [])
+        service.snapshot_baseline(f"/repo/f{i}.py")
+    server_diags.append({"message": "rewritten"})
+    assert service._apply_delta("/repo/f0.py", [{"message": "rewritten"}], None) == [{"message": "rewritten"}]
+    server_diags.clear()
+    service.snapshot_baseline("/repo/new.py")
     assert len(service._delta_baseline) == cap
     assert "/repo/f1.py" not in service._delta_baseline
     assert service._delta_baseline["/repo/f0.py"] == [{"message": "rewritten"}]
+    assert "/repo/new.py" in service._delta_baseline