fix: keep debugger operations off the request loop (#110)

## What this changes

- Offload synchronous debugger startup, command, and shutdown handlers
to AnyIO workers while preserving tool filtering and MCP error
conversion.
- Keep send_ctrl_break on the request loop so it can interrupt a command
even when that command occupies the only worker slot. Existing
debugger-level I/O synchronization is unchanged.
- Claim a session before blocking shutdown; concurrent closes cannot
shut it down twice or replace the first KD close's resume choice.
- Add nine hermetic tests through the server's public registered tool
handler and an Unreleased changelog entry.

## Why

Fixes #109.

The asynchronous request handler currently calls blocking debugger
operations directly. A waiting command prevents the same server from
dispatching the break-in tool intended to interrupt it. Startup and
shutdown also block unrelated tool calls. The existing wait_for_break
worker dispatch already follows this responsiveness model.

The expanded regression suite fails 7 cases and passes 2 against
unchanged develop at 35b6cdb49d, and
passes all 9 with this patch. This does not change public tool schemas
or timeouts.

## How it was verified

- uv run --no-sync pytest src/mcp_windbg/tests/ -q --tb=short: 110
passed, 12 skipped.
- Both-process coverage with MCP_WINDBG_COVERAGE=1, coverage run -m
pytest, coverage combine: 110 passed, 12 skipped, 87% coverage. The
required 88% floor is **not met** on this host. The same run against
unchanged develop is 101 passed, 12 skipped, 80% coverage and also fails
the floor. No threshold or exclusions were changed.
- uv lock --check: passed.
- Public MCP registry schema validation: passed.
- uv build and twine check of both generated distributions: passed.
- Markdown typography check of all 35 Markdown files and git diff
--check: passed.

This is intentionally a draft. No CDB executable, Git LFS crash dump, or
real kernel target is available here; the 12 debugger/kernel scenarios
skip. The new tests establish dispatch, error, and concurrent-close
behavior using fake sessions, not real debugger operation. Real CDB/KD
scenarios and the required coverage floor remain landing gates. No
kernel runtime, release version, tags, or target configuration were
changed.

---------

Co-authored-by: Sven Scharmentke <sven.scharmentke@liasoft.de>

CHANGELOG.md

@@ -5,6 +5,17 @@ All notable changes to the MCP Server for WinDbg Crash Analysis project will be
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- Debugger startup, commands, and shutdown run on worker threads so slow I/O
+  does not block other MCP calls. Break-in remains available even when the
+  debugger worker pool is full. Concurrent close calls claim a session once
+  and preserve the closing caller's kernel resume policy. Break-in keeps the
+  validated session record so a concurrent close cannot cause a registry error
+  (#109, #110). Thanks to @adity982 for the fix.
+
 ## [1.3.0] - 2026-09-10
 
 ### Added

src/mcp_windbg/debug_session.py

@@ -366,9 +366,9 @@ class DebuggerSession:
     def _exclusive(self):
         """Acquire the session's I/O lock, or explain why the session is busy.
 
-        Never waits. The server calls most tools straight from its event loop,
-        so blocking here would stall every other session too - and would stall
-        the very ``send_ctrl_break`` the failure message points at.
+        Never waits: queued operations on a busy session would occupy debugger
+        workers needed by other sessions. ``send_ctrl_break`` remains available
+        without acquiring this lock.
         """
         if not self._io_lock.acquire(blocking=False):
             raise DebuggerError(

src/mcp_windbg/server.py

@@ -98,7 +98,7 @@ def _require_session(session_id: str, kind: str):
 
 
 def _require_live_session(session_id: str, what: str):
-    """Return the session for ``session_id``, requiring a live (non-dump) target.
+    """Return the record for ``session_id``, requiring a live (non-dump) target.
 
     Shared by the tools that only make sense against something that runs -
     ``send_ctrl_break`` and ``wait_for_break`` - and accepts either kind, since a
@@ -113,20 +113,25 @@ def _require_live_session(session_id: str, what: str):
                 f"session_id {session_id!r} is a dump session; there is no "
                 f"running target to {what}."
             ))
-    return session
+    return record
 
 
-def _close_session(session_id: str, kind: str) -> bool:
+def _close_session(session_id: str, kind: str, resume: Optional[bool] = None) -> bool:
     """Shut down and forget a session; returns False if id/kind did not match."""
     record = _sessions.get(session_id)
     if record is None or record["kind"] != kind:
         return False
+    # Claim the session before blocking in shutdown. Concurrent close calls must
+    # not shut down the same debugger twice or overwrite its resume policy.
+    record = _sessions.pop(session_id, None)
+    if record is None:
+        return False
+    if resume is not None:
+        record["session"].resume_on_close = resume
     try:
         record["session"].shutdown()
     except Exception:
         pass
-    finally:
-        _sessions.pop(session_id, None)
     return True
 
 
@@ -452,6 +457,11 @@ def _create_server(
         content = await _dispatch_tool(params.name, params.arguments or {})
         return CallToolResult(content=content)
 
+    async def _run_debugger_handler(handler, *args):
+        # Debugger startup, commands, and shutdown can wait for seconds or minutes.
+        # Keep the event loop available for other tools, especially break-in.
+        return await anyio.to_thread.run_sync(functools.partial(handler, *args))
+
     async def _dispatch_tool(name: str, arguments: dict) -> list[TextContent]:
         try:
             call_id = uuid.uuid4().hex
@@ -461,41 +471,43 @@ def _create_server(
                 return filter_tool_content(name, _handle_list_dumps(arguments), call_id)
 
             if name == "open_cdb_dump":
-                return filter_tool_content(name, _handle_open_cdb_dump(
+                return filter_tool_content(name, await _run_debugger_handler(_handle_open_cdb_dump,
                     arguments, cdb_path, symbols_path, timeout, verbose, auto_dump_dir_symbols
                 ), call_id)
 
             if name == "open_cdb_remote":
-                return filter_tool_content(name, _handle_open_cdb_remote(
+                return filter_tool_content(name, await _run_debugger_handler(_handle_open_cdb_remote,
                     arguments, cdb_path, symbols_path, timeout, verbose
                 ), call_id)
 
             if name == "open_kd_session":
-                return filter_tool_content(name, _handle_open_kd_session(
+                return filter_tool_content(name, await _run_debugger_handler(_handle_open_kd_session,
                     arguments, kd_path, symbols_path, timeout, verbose
                 ), call_id)
 
             if name == "run_cdb_command":
-                return filter_tool_content(name, _handle_run_command(
+                return filter_tool_content(name, await _run_debugger_handler(_handle_run_command,
                     RunCdbCommand(**arguments), "cdb", CDB_COMMAND_TIMEOUT, timeout
                 ), call_id)
 
             if name == "run_kd_command":
-                return filter_tool_content(name, _handle_run_command(
+                return filter_tool_content(name, await _run_debugger_handler(_handle_run_command,
                     RunKdCommand(**arguments), "kd", KD_COMMAND_TIMEOUT, timeout
                 ), call_id)
 
             if name == "close_cdb_session":
-                return filter_tool_content(name, _handle_close(CloseCdbSession(**arguments).session_id, "cdb"), call_id)
+                return filter_tool_content(name, await _run_debugger_handler(
+                    _handle_close, CloseCdbSession(**arguments).session_id, "cdb"
+                ), call_id)
 
             if name == "close_kd_session":
                 close_args = CloseKdSession(**arguments)
-                record = _sessions.get(close_args.session_id)
-                if record and record["kind"] == "kd":
-                    record["session"].resume_on_close = close_args.resume
-                return filter_tool_content(name, _handle_close(close_args.session_id, "kd"), call_id)
+                return filter_tool_content(name, await _run_debugger_handler(
+                    _handle_close, close_args.session_id, "kd", close_args.resume
+                ), call_id)
 
             if name == "send_ctrl_break":
+                # This escape hatch must not queue behind blocked debugger workers.
                 return filter_tool_content(name, _handle_send_ctrl_break(SendCtrlBreak(**arguments).session_id), call_id)
 
             if name == "wait_for_break":
@@ -611,19 +623,19 @@ def _create_server(
         text = f"Command: {args.command}\n\nOutput:\n```\n" + "\n".join(output) + "\n```"
         return [TextContent(type="text", text=text)]
 
-    def _handle_close(session_id, kind) -> list[TextContent]:
-        if _close_session(session_id, kind):
+    def _handle_close(session_id, kind, resume=None) -> list[TextContent]:
+        if _close_session(session_id, kind, resume):
             return [TextContent(type="text", text=f"Successfully closed {kind} session {session_id}")]
         return [TextContent(type="text", text=f"No active {kind} session found for session_id {session_id}")]
 
     def _handle_send_ctrl_break(session_id) -> list[TextContent]:
-        session = _require_live_session(session_id, "break into")
-        label = _sessions[session_id]["label"]
-        session.send_ctrl_break()
+        record = _require_live_session(session_id, "break into")
+        label = record["label"]
+        record["session"].send_ctrl_break()
         return [TextContent(type="text", text=f"Sent CTRL+BREAK to session {session_id} ({label}).")]
 
     async def _handle_wait_for_break(args: WaitForBreak) -> list[TextContent]:
-        session = _require_live_session(args.session_id, "wait on")
+        session = _require_live_session(args.session_id, "wait on")["session"]
         # Deliberately not _effective_timeout: the --timeout floor is about how
         # long a *command* may take, and has nothing to say about how long you
         # are willing to sit on a target waiting for it to bugcheck.

src/mcp_windbg/tests/test_tool_concurrency.py

@@ -0,0 +1,280 @@
+"""The registered tool handler must not block unrelated calls on debugger I/O."""
+
+from __future__ import annotations
+
+import threading
+
+import anyio
+import pytest
+from mcp.types import CallToolRequestParams
+
+from mcp_windbg import server as server_module
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.fixture
+def tool_handler(monkeypatch):
+    captured = {}
+    real_server = server_module.Server
+
+    def record_handlers(*args, **kwargs):
+        captured.update(kwargs)
+        return real_server(*args, **kwargs)
+
+    monkeypatch.setattr(server_module, "Server", record_handlers)
+    monkeypatch.setattr(server_module, "_sessions", {})
+    server_module._create_server()
+    return captured["on_call_tool"]
+
+
+class BlockingSession:
+    is_live_session = True
+
+    def __init__(self):
+        self.started = threading.Event()
+        self.release = threading.Event()
+        self.thread_id = None
+
+    def send_command(self, command, timeout):
+        self.thread_id = threading.get_ident()
+        self.started.set()
+        if not self.release.wait(1):
+            raise RuntimeError("The event loop could not dispatch the break-in")
+        return ["completed"]
+
+    def send_ctrl_break(self):
+        self.release.set()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["cdb", "kd"])
+async def test_blocked_command_can_be_interrupted_by_another_tool(tool_handler, kind):
+    session = BlockingSession()
+    server_module._sessions["test-session"] = {
+        "session": session,
+        "kind": kind,
+        "label": "test target",
+    }
+    results = []
+    limiter = anyio.to_thread.current_default_thread_limiter()
+    old_limit = limiter.total_tokens
+    limiter.total_tokens = 1
+
+    async def command():
+        results.append(
+            await tool_handler(
+                None,
+                CallToolRequestParams(
+                    name=f"run_{kind}_command",
+                    arguments={
+                        "session_id": "test-session",
+                        "command": "kb",
+                        "timeout_seconds": 10,
+                    },
+                ),
+            )
+        )
+
+    try:
+        async with anyio.create_task_group() as tasks:
+            tasks.start_soon(command)
+            with anyio.fail_after(2):
+                while not session.started.is_set():
+                    await anyio.sleep(0)
+                interrupted = await tool_handler(
+                    None,
+                    CallToolRequestParams(
+                        name="send_ctrl_break",
+                        arguments={"session_id": "test-session"},
+                    ),
+                )
+                assert "Sent CTRL+BREAK" in interrupted.content[0].text
+    finally:
+        session.release.set()
+        limiter.total_tokens = old_limit
+
+    assert session.thread_id != threading.get_ident()
+    assert "completed" in results[0].content[0].text
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["cdb", "kd"])
+async def test_command_errors_keep_mcp_error_conversion(tool_handler, kind):
+    class FailingSession:
+        def send_command(self, command, timeout):
+            raise RuntimeError("test command failure")
+
+    server_module._sessions["test-session"] = {
+        "session": FailingSession(),
+        "kind": kind,
+        "label": "test target",
+    }
+    with pytest.raises(server_module.MCPError, match="test command failure"):
+        await tool_handler(
+            None,
+            CallToolRequestParams(
+                name=f"run_{kind}_command",
+                arguments={"session_id": "test-session", "command": "kb"},
+            ),
+        )
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "tool,arguments",
+    [
+        ("open_cdb_dump", {"dump_path": "test.dmp"}),
+        ("open_cdb_remote", {"connection_string": "test remote"}),
+        ("open_kd_session", {"connection_string": "test kernel"}),
+    ],
+)
+async def test_startup_keeps_other_tools_available(
+    tool_handler, monkeypatch, tmp_path, tool, arguments
+):
+    started = threading.Event()
+    release = threading.Event()
+    thread_ids = []
+
+    class StartupSession:
+        def __init__(self, **kwargs):
+            thread_ids.append(threading.get_ident())
+            started.set()
+            if not release.wait(1):
+                raise RuntimeError("Startup blocked the event loop")
+
+        def send_command(self, command, timeout):
+            return ["ready"]
+
+    monkeypatch.setattr(server_module, "CDBSession", StartupSession)
+    monkeypatch.setattr(server_module, "KDSession", StartupSession)
+    results = []
+
+    async def open_session():
+        results.append(
+            await tool_handler(
+                None, CallToolRequestParams(name=tool, arguments=arguments)
+            )
+        )
+
+    try:
+        async with anyio.create_task_group() as tasks:
+            tasks.start_soon(open_session)
+            with anyio.fail_after(2):
+                while not started.is_set():
+                    await anyio.sleep(0)
+                result = await tool_handler(
+                    None,
+                    CallToolRequestParams(
+                        name="list_dumps",
+                        arguments={"directory_path": str(tmp_path)},
+                    ),
+                )
+                assert "No crash dump" in result.content[0].text
+                release.set()
+    finally:
+        release.set()
+
+    assert len(thread_ids) == 1
+    assert thread_ids[0] != threading.get_ident()
+    assert "Opened" in results[0].content[0].text
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["cdb", "kd"])
+async def test_concurrent_close_claims_session_once(tool_handler, kind):
+    started = threading.Event()
+    release = threading.Event()
+    calls = []
+
+    class ClosingSession:
+        resume_on_close = True
+
+        def shutdown(self):
+            calls.append((self.resume_on_close, threading.get_ident()))
+            started.set()
+            if not release.wait(1):
+                raise RuntimeError("Shutdown blocked the event loop")
+
+    server_module._sessions["test-session"] = {
+        "session": ClosingSession(),
+        "kind": kind,
+        "label": "test target",
+    }
+    results = []
+
+    async def close_session():
+        arguments = {"session_id": "test-session"}
+        if kind == "kd":
+            arguments["resume"] = False
+        results.append(
+            await tool_handler(
+                None,
+                CallToolRequestParams(
+                    name=f"close_{kind}_session",
+                    arguments=arguments,
+                ),
+            )
+        )
+
+    try:
+        async with anyio.create_task_group() as tasks:
+            tasks.start_soon(close_session)
+            with anyio.fail_after(2):
+                while not started.is_set():
+                    await anyio.sleep(0)
+                second = await tool_handler(
+                    None,
+                    CallToolRequestParams(
+                        name=f"close_{kind}_session",
+                        arguments={"session_id": "test-session"},
+                    ),
+                )
+                assert "No active" in second.content[0].text
+                release.set()
+    finally:
+        release.set()
+
+    assert len(calls) == 1
+    assert calls[0][0] == (kind != "kd")
+    assert calls[0][1] != threading.get_ident()
+    assert "Successfully closed" in results[0].content[0].text
+    assert not server_module._sessions
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["cdb", "kd"])
+async def test_break_in_racing_with_close_reports_debugger_error(tool_handler, kind):
+    class ClosingSession:
+        @property
+        def is_live_session(self):
+            # Force shutdown between validating the session and signalling it.
+            closer = threading.Thread(
+                target=server_module._close_session, args=("test-session", kind)
+            )
+            closer.start()
+            closer.join(2)
+            assert not closer.is_alive()
+            return True
+
+        def shutdown(self):
+            pass
+
+        def send_ctrl_break(self):
+            raise RuntimeError("Debugger process is not running")
+
+    server_module._sessions["test-session"] = {
+        "session": ClosingSession(),
+        "kind": kind,
+        "label": "test target",
+    }
+    with pytest.raises(server_module.MCPError, match="Debugger process is not running"):
+        await tool_handler(
+            None,
+            CallToolRequestParams(
+                name="send_ctrl_break", arguments={"session_id": "test-session"}
+            ),
+        )

src/mcp_windbg/tests/test_tool_concurrency_live.py

@@ -0,0 +1,93 @@
+"""Concurrent tool calls over MCP against a disposable local CDB target."""
+
+from __future__ import annotations
+
+import os
+import re
+import uuid
+
+import anyio
+import pytest
+from mcp.client.session import ClientSession
+from mcp.client.stdio import StdioServerParameters, stdio_client
+from mcp.shared.exceptions import MCPError
+
+from e2e import harness
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.anyio
+@pytest.mark.live
+@pytest.mark.remote
+async def test_break_in_interrupts_command_over_mcp():
+    if not harness.cdb_available():
+        pytest.skip("cdb.exe not found")
+
+    with harness.RemoteCdbServer(
+        target=["waitfor.exe", "McpConcurrencyTestSignal"]
+    ) as remote:
+        command, args = harness.server_command([])
+        params = StdioServerParameters(command=command, args=args, env=dict(os.environ))
+        with anyio.fail_after(60):
+            async with stdio_client(params) as (read_stream, write_stream):
+                async with ClientSession(read_stream, write_stream) as client:
+                    await client.initialize()
+                    opened = await client.call_tool(
+                        "open_cdb_remote",
+                        {"connection_string": remote.connection_string},
+                    )
+                    assert not opened.is_error
+                    session_id = re.search(
+                        r"session_id:\s*(cdb-[0-9a-f]+)", opened.content[0].text
+                    ).group(1)
+                    results = []
+                    marker = "COMMAND_STARTED_" + uuid.uuid4().hex
+
+                    async def run_command():
+                        # A conditional resume waits for the next prompt, unlike
+                        # a bare 'g' which the server deliberately returns from.
+                        results.append(await client.call_tool(
+                            "run_cdb_command",
+                            {
+                                "session_id": session_id,
+                                "command": f".echo {marker}; .if (1) {{ g }}",
+                                "timeout_seconds": 30,
+                            },
+                        ))
+
+                    try:
+                        async with anyio.create_task_group() as tasks:
+                            tasks.start_soon(run_command)
+                            with anyio.fail_after(10):
+                                while not any(
+                                    line.strip() == marker for line in remote.output_lines
+                                ):
+                                    await anyio.sleep(0.01)
+                                # A separate request must finish while the
+                                # command is still waiting on the running target.
+                                await client.list_tools()
+                                assert not results
+                                with pytest.raises(MCPError, match="session is busy"):
+                                    await client.call_tool(
+                                        "run_cdb_command",
+                                        {"session_id": session_id, "command": "r"},
+                                    )
+                                interrupted = await client.call_tool(
+                                    "send_ctrl_break", {"session_id": session_id}
+                                )
+                                assert not interrupted.is_error
+
+                        assert not results[0].is_error
+                        assert "Break instruction exception" in results[0].content[0].text
+                        followup = await client.call_tool(
+                            "run_cdb_command",
+                            {"session_id": session_id, "command": ".echo STILL_USABLE"},
+                        )
+                        assert not followup.is_error
+                        assert "STILL_USABLE" in followup.content[0].text
+                    finally:
+                        await client.call_tool("close_cdb_session", {"session_id": session_id})