fix(mcp): preserve active calls during session eviction (#6900)

Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com>

README.md

@@ -843,6 +843,7 @@ For HTTP/SSE MCP servers, OAuth token flows are supported (`client_credentials`,
 Missing, malformed, or out-of-range token response `expires_in` values use a one-hour default lifetime. This includes lifetimes that cannot be added to the current time without overflowing the expiry timestamp.
 Durable HTTP/SSE task status and cancellation calls select configured `user_auth` credentials using the persisted task owner, including after restart; per-request secrets are not retained for background calls. If a request-scoped credential overrides submit authentication, both credentials must authorize access to the same remote task.
 For stdio MCP servers, per-tool call timeouts can be configured with `tool_call_timeout`; durable background-task calls honor the same setting for HTTP/SSE servers as well.
+The persistent stdio session pool evicts only idle sessions at capacity. If every pooled session is executing a tool, a new session request fails with a capacity error and can be retried after a call finishes.
 For stdio file outputs, a bare filename is linked to a uniquely matching file created or changed by that call. Filenames embedded in unrelated paths, including Windows backslash paths, are left intact.
 For HTTP/SSE background-task calls, `session_init_timeout` separately bounds connection setup (including the SSE endpoint event) and MCP initialization together; it stops applying once the tool call begins. Initialization deadline errors identify the server and configured time limit.
 Ordinary `task` subagents retain the parent run's captured thread incarnation for MCP calls, including legacy threads, so delegation preserves the same lifecycle scope.

backend/packages/harness/deerflow/mcp/AGENTS.md

@@ -36,8 +36,8 @@ The optional background-task `status` filter flows through `McpTaskSubmitter.lis
   teardown; an abandoned live-registry record remains bounded by `MAX_SESSIONS`
   and can be removed by LRU pressure or explicit cleanup. This cannot close the
   transport after its loop is gone. Normal `asyncio.run()` drains its owners.
-- **Persistent stdio session capacity**: `MCPSessionPool.MAX_SESSIONS` is a hard cap on the live LRU registry. Capacity is enforced both before creating a session and when an in-flight session is promoted, because different keys can finish initialization concurrently after all observing spare capacity. Promotion-time victims are signalled and drained by separately tracked teardown tasks on their owning loops; the newly promoted owner must never await a victim, so a blocked victim exit cannot prevent its own closure or disconnect recovery.
-- **Persistent stdio binding fence**: pooled session registry keys remain `(server_name, scope_key, owning_loop, ownership_domain)`. Separately, each `(ownership_domain, server_name)` has an opaque `ServerBinding` capability derived from a SHA-256 digest of the base stdio process identity (`transport`, `command`, `args`, configured `cwd`, configured `env`; the resolved values themselves are never stored in the binding repr). Discovery captures the exact pool plus binding before its first await, and durable stdio calls bind before per-call workspace `cwd`/`TMPDIR` augmentation. `get_session()` rejects bindings that are already superseded at admission, owner promotion, or the final post-await check. That final check is a check-before-return guard, not a lease across subsequent `call_tool()` use: a reconciliation racing after it detaches and closes the owner, and the caller then fails through the existing transport-disconnect handling. Binding reconciliation consumes already-classified active fingerprints/removals, atomically installs epochs/tombstones and detaches matching owners under the pool lock, and keeps those detached owners in the pool's teardown reaper so their `__aexit__` stays observable even when the caller drops the returned `PreparedRetirement`; configuration diffing remains a higher-layer responsibility.
+- **Persistent stdio session capacity**: `MCPSessionPool.MAX_SESSIONS` is a hard cap on the live LRU registry. Capacity is enforced both before creating a session and when an in-flight session is promoted, because different keys can finish initialization concurrently after all observing spare capacity. Both eviction points skip owners with active `call_pooled_session_tool()` calls (owner-specific counts, released in `finally`). An all-busy pool rejects new acquisitions/promotions; hits and joiners remain allowed. Explicit close and binding retirement still force closure. Promotion-time victims are signalled and drained by separately tracked teardown tasks on their owning loops; the newly promoted owner must never await a victim, so a blocked victim exit cannot prevent its own closure or disconnect recovery.
+- **Persistent stdio binding fence**: pooled session registry keys remain `(server_name, scope_key, owning_loop, ownership_domain)`. Separately, each `(ownership_domain, server_name)` has an opaque `ServerBinding` capability derived from a SHA-256 digest of the base stdio process identity (`transport`, `command`, `args`, configured `cwd`, configured `env`; the resolved values themselves are never stored in the binding repr). Discovery captures the exact pool plus binding before its first await, and durable stdio calls bind before per-call workspace `cwd`/`TMPDIR` augmentation. `get_session()` rejects bindings that are already superseded at admission, owner promotion, or the final post-await check. That final check is a check-before-return guard, not a lease across subsequent `call_tool()` use: a reconciliation racing after it detaches and closes the owner. The shared call helper rejects an already-detached session before executing the tool; retirement during a call still uses the existing transport-disconnect handling. Binding reconciliation consumes already-classified active fingerprints/removals, atomically installs epochs/tombstones and detaches matching owners under the pool lock, and keeps those detached owners in the pool's teardown reaper so their `__aexit__` stays observable even when the caller drops the returned `PreparedRetirement`; configuration diffing remains a higher-layer responsibility.
 - **Process-local selective reconciliation**: `mcp/cache.py` keeps an *applied reconciliation baseline* (`_applied_mcp_revision`) apart from the published tool-cache baseline: the effective MCP snapshot, the normalized interceptor chain, and a `normalized_connection_fingerprint()` per enabled stdio server derived from `build_server_params()` (not a whole-config `model_dump()`). A claim *installs* the revision it records — repairing a stale durable/failed binding before `ensure_binding()` fails closed, and tombstoning unlisted deployment servers with no previous baseline — after consulting the frozen durable-task snapshot (a rejection takes the conservative reset; a claim otherwise fails like `get_mcp_tools()`). The revision is recorded when discovery starts, so a failed/cancelled discovery can still retire a binding it seeded. Classification and application share one `_init_condition` section. An unchanged slice adopts the new path/signature and keeps the published tools; an interceptor change, an unreadable/unverifiable revision, a shared-reset-marker change, a frozen rejection, or an unbuildable baseline keeps the conservative whole-pool reset; otherwise `reconcile_bindings(active, removed, domain="deployment")` runs without replacing the pool, then the published tools are dropped and `_cache_generation` is bumped. Unchanged stdio servers keep their binding, session and owner; changed/removed ones advance an epoch or get a tombstone, with teardown left to the pool's reaper outside every lock. Selective transitions never retire personal state; full resets do. A new shared-reset generation retires local state first whenever published tools, an applied baseline, or deployment pooled state exist (an empty/personal-only process adopts it); explicit `reset_mcp_tools_cache()` / `POST /api/mcp/cache/reset` stay full resets. Gateway writers use `prepare_mcp_reconciliation()` / `finish_mcp_reconciliation()` under and after the config locks (delete→identical-re-add stays two transitions; a failed handoff returns 500). The shared `mcpLifecycle` ledger (`mcp/lifecycle.py`) makes this cross-worker: readers compare `_AppliedMcpRevision.lifecycle_tokens` before the snapshot shortcut, reader/claim/committed force-rebind changed servers, and every config write needs a lifecycle-aware writer (restrict writes to upgraded workers mid-upgrade). Per-server results are cached by name and reused only while their canonical snapshot, normalized global interceptor chain and current binding ownership match; selective reconciliation evicts the rest, every full reset clears the cache, and a server that failed discovery is left absent rather than cached as a success.
 - **Cache invalidation**: Staleness is detected in two stages. First, the resolved config path and a `(mtime, size, sha256)` content signature are compared against the values recorded when the cache was published, not a strict mtime `>` comparison — this catches same-second edits, mtime that stays put or moves backward (`git checkout`, `cp -p` / backup restore, `tar` / `rsync`, object-store / network mounts), and a switch to a different config file with an equal-or-older mtime. The signature helper (`config/file_signature.py::get_config_signature`) is shared with `config/app_config.py::get_app_config()` for the sibling runtime-editable config file, rather than each maintaining its own copy. Second, because the file also carries `skills` and middleware settings, a changed signature alone is not enough: `mcp/cache.py::_effective_mcp_config_snapshot()` serializes the effective MCP slice (enabled servers' parsed config in declaration order plus `mcpInterceptors`; the raw `transport` alias is dropped so equivalent `type`/`transport` spellings compare equal) and the cache is retired only when that slice changed. A skills-only or middleware-only edit therefore adopts the new signature and keeps the cached tools and pooled sessions, while a real MCP change is classified against the process-local applied-reconciliation baseline and applied selectively (see the selective-reconciliation bullet above) instead of unconditionally replacing the pool. The snapshot may embed resolved credentials, so it stays in process memory and is never logged or persisted. `_is_cache_stale()` adopts a new signature when the MCP slice is unchanged, so it is not a read-only predicate; production callers run it under `_init_condition`. `ExtensionsConfig.resolve_config_path()` raises `FileNotFoundError` for an explicit `config_path`/`DEER_FLOW_EXTENSIONS_CONFIG_PATH` that points at a missing file — an operator-asserted path going missing is a real misconfiguration, so this is intentionally loud for callers that load the config for actual use (e.g. `from_file()` via `get_mcp_tools()`); only the fallback search mode returns `None`. The MCP cache's own path resolution (`mcp/cache.py::_resolve_config_path`) is narrower: it catches that specific `FileNotFoundError` locally and treats it the same as "unconfigured", so this staleness check degrades to "not stale" instead of propagating an exception when a previously-valid explicit/env-var config disappears mid-run. Two no-config states are distinguished: `_initialized_without_config` makes a cache published while no config existed stale as soon as one appears, while a config deleted after a successful load keeps the last-known-good fail-soft behavior and stays "not stale", while a file that is present but whose content cannot be hashed is treated as stale instead, because the cache cannot prove the revision it published is still the current one. `initialize_mcp_tools()` reads one `ExtensionsConfig` instance, reuses only the per-server entries whose canonical snapshot, interceptor chain and current binding match, re-discovers the rest through `get_mcp_tools_by_server(extensions_config=..., server_names=missing)`, and publishes the flat list and per-server entries atomically only when the effective MCP snapshot, the `mcpLifecycle` tokens, the shared-reset marker and every candidate binding are still current; otherwise it resets or reconciles the tool cache as before. This closes the discovery window — tools built from a superseded MCP revision, or from a resolved path whose signature cannot be re-read, are never published — while a skills-only edit during discovery still publishes because the MCP slice is unchanged.
 - **Transports**: stdio (command-based), SSE, HTTP

backend/packages/harness/deerflow/mcp/session_pool.py

@@ -8,7 +8,8 @@ This module provides a session pool that maintains persistent MCP sessions,
 scoped by ``(server_name, scope_key, owning_loop, ownership_domain)``. Consecutive calls on
 the same loop share server-side state; independent loops use separate sessions.
 The sync wrapper uses a fresh loop per call, so it does not preserve that state.
-Sessions are evicted in LRU order when the pool reaches capacity.
+Idle sessions are evicted in LRU order when the pool reaches capacity.
+If every session is executing a tool, acquisition of a new session fails.
 
 Lifecycle model (owner task)
 ----------------------------
@@ -47,7 +48,8 @@ import json
 import logging
 import threading
 from collections import OrderedDict
-from collections.abc import Collection, Mapping
+from collections.abc import Collection, Iterator, Mapping
+from contextlib import contextmanager
 from dataclasses import dataclass, field
 from typing import Any, Literal
 
@@ -165,16 +167,17 @@ async def call_pooled_session_tool(
     call_kwargs: dict[str, Any],
     domain: MCPPoolDomain = "deployment",
 ) -> Any:
-    """Call a pooled session and evict it only after an explicit disconnect."""
-    try:
-        return await session.call_tool(tool_name, arguments, **call_kwargs)
-    except Exception as error:
-        if _is_mcp_transport_disconnect(error):
-            cleanup_call = pool.close_session_if_current(server_name, scope_key, session) if domain == "deployment" else pool.close_session_if_current(server_name, scope_key, session, domain=domain)
-            cleanup = asyncio.create_task(cleanup_call)
-            if await _finish_session_cleanup(cleanup, server_name):
-                raise asyncio.CancelledError
-        raise
+    """Protect a pooled tool call from LRU eviction and retire on disconnect."""
+    with pool._session_call(server_name, scope_key, session, domain=domain):
+        try:
+            return await session.call_tool(tool_name, arguments, **call_kwargs)
+        except Exception as error:
+            if _is_mcp_transport_disconnect(error):
+                cleanup_call = pool.close_session_if_current(server_name, scope_key, session) if domain == "deployment" else pool.close_session_if_current(server_name, scope_key, session, domain=domain)
+                cleanup = asyncio.create_task(cleanup_call)
+                if await _finish_session_cleanup(cleanup, server_name):
+                    raise asyncio.CancelledError
+            raise
 
 
 class MCPSessionPool:
@@ -213,6 +216,9 @@ class MCPSessionPool:
         # threading.Lock is not bound to any event loop, so it is safe to
         # acquire from both async paths and sync/worker-thread paths.
         self._lock = threading.Lock()
+        # Counts follow owner identity, not the reusable server/scope key.
+        # Explicit retirement may detach an owner while its calls unwind.
+        self._active_calls: dict[asyncio.Task[Any], int] = {}
         # Strong references to detached teardown reaper tasks. The event loop
         # only keeps weak references to tasks, so an unheld reaper — and with
         # it the owner it is awaiting, mid-__aexit__ — could be
@@ -377,6 +383,50 @@ class MCPSessionPool:
     # Session owner task
     # ------------------------------------------------------------------
 
+    @contextmanager
+    def _session_call(
+        self,
+        server_name: str,
+        scope_key: str,
+        session: ClientSession,
+        *,
+        domain: MCPPoolDomain,
+    ) -> Iterator[None]:
+        """Protect a tool call from capacity eviction, but not explicit close."""
+        key = (server_name, scope_key, asyncio.get_running_loop(), domain)
+        with self._lock:
+            entry = self._entries.get(key)
+            if entry is None or entry[0] is not session or entry[2].done():
+                raise RuntimeError(f"MCP session for '{server_name}' is no longer pooled; acquire a new session")
+            owner = entry[2]
+            self._active_calls[owner] = self._active_calls.get(owner, 0) + 1
+        try:
+            yield
+        finally:
+            with self._lock:
+                remaining = self._active_calls[owner] - 1
+                if remaining:
+                    self._active_calls[owner] = remaining
+                else:
+                    self._active_calls.pop(owner)
+
+    def _evict_idle_locked(self) -> list[tuple[asyncio.AbstractEventLoop, asyncio.Task[Any], asyncio.Event]]:
+        """Make room for one entry without detaching any active owners."""
+        needed = max(0, len(self._entries) - self.MAX_SESSIONS + 1)
+        if not needed:
+            return []
+        victims = []
+        for key, (_session, loop, task, close_evt) in self._entries.items():
+            if not self._active_calls.get(task, 0):
+                victims.append((key, loop, task, close_evt))
+                if len(victims) == needed:
+                    break
+        if len(victims) < needed:
+            raise RuntimeError("MCP session pool is at capacity with all sessions busy; retry after a tool call completes")
+        for key, _loop, _task, _close_evt in victims:
+            self._entries.pop(key)
+        return [(loop, task, close_evt) for _key, loop, task, close_evt in victims]
+
     def _discard_owner(
         self,
         key: tuple[str, str, asyncio.AbstractEventLoop, MCPPoolDomain],
@@ -428,11 +478,8 @@ class MCPSessionPool:
                 binding_ok = not self._retired and self._bindings.get(binding_key) is expected_binding
                 still_ours = self._inflight.get(key) == (loop, ready, task, close_evt)
                 if still_ours and binding_ok:
+                    promoted_evicted = self._evict_idle_locked()
                     self._inflight.pop(key)
-                    while len(self._entries) >= self.MAX_SESSIONS:
-                        oldest_key, (_, ent_loop, ent_task, ent_close) = next(iter(self._entries.items()))
-                        self._entries.pop(oldest_key)
-                        promoted_evicted.append((ent_loop, ent_task, ent_close))
                     self._entries[key] = (session, loop, task, close_evt)
                     if not ready.done():
                         ready.set_result(session)
@@ -533,17 +580,14 @@ class MCPSessionPool:
             if inflight is not None:
                 join = inflight[1]
             else:
+                # Reject before publishing an owner if there is no idle victim.
+                evicted = self._evict_idle_locked()
                 ready = current_loop.create_future()
                 close_evt = asyncio.Event()
                 task = current_loop.create_task(self._run_session(key, connection, ready, close_evt, expected_binding))
                 self._inflight[key] = (current_loop, ready, task, close_evt)
                 task.add_done_callback(lambda owner: self._discard_owner(key, owner))
 
-            while len(self._entries) >= self.MAX_SESSIONS:
-                oldest_key, (_, loop, ent_task, ent_close) = next(iter(self._entries.items()))
-                self._entries.pop(oldest_key)
-                evicted.append((loop, ent_task, ent_close))
-
         for loop, _ent_task, ent_close in evicted:
             self._signal_close(loop, ent_close)
         try:

backend/tests/test_mcp_binding_fencing.py

@@ -317,8 +317,11 @@ async def test_durable_stdio_call_binds_base_connection_before_workspace_augment
     )
     result = SimpleNamespace(structuredContent={"status": "running"}, isError=False)
     session = SimpleNamespace(call_tool=AsyncMock(return_value=result))
+    cm = _GatedSessionCm()
+    cm.session = session
+    session.initialize = cm.initialize
     pool = MCPSessionPool()
-    pool.get_session = AsyncMock(return_value=session)
+    pool.get_session = AsyncMock(wraps=pool.get_session)
 
     prepared = {
         "transport": "stdio",
@@ -329,27 +332,31 @@ async def test_durable_stdio_call_binds_base_connection_before_workspace_augment
     }
 
     caller = McpTaskToolCaller(config)
-    with (
-        patch("deerflow.mcp.task_tool_caller.get_session_pool", return_value=pool),
-        patch("deerflow.mcp.task_tool_caller._prepare_stdio_connection", return_value=prepared),
-    ):
-        actual = await caller.call_tool(
-            server_name="reports",
-            tool_name="status_report",
-            arguments={"task_id": "1"},
-            user_id="u1",
-            thread_id="t1",
-        )
+    try:
+        with (
+            patch("deerflow.mcp.task_tool_caller.get_session_pool", return_value=pool),
+            patch("deerflow.mcp.task_tool_caller._prepare_stdio_connection", return_value=prepared),
+            patch("langchain_mcp_adapters.sessions.create_session", return_value=cm),
+        ):
+            actual = await caller.call_tool(
+                server_name="reports",
+                tool_name="status_report",
+                arguments={"task_id": "1"},
+                user_id="u1",
+                thread_id="t1",
+            )
 
-    assert actual is result
-    binding = pool._bindings[("deployment", "reports")]
-    assert binding.fingerprint == normalized_connection_fingerprint({"transport": "stdio", "command": "report-mcp", "args": []})
-    pool.get_session.assert_awaited_once_with(
-        "reports",
-        "u1:t1",
-        prepared,
-        binding=binding,
-    )
+        assert actual is result
+        binding = pool._bindings[("deployment", "reports")]
+        assert binding.fingerprint == normalized_connection_fingerprint({"transport": "stdio", "command": "report-mcp", "args": []})
+        pool.get_session.assert_awaited_once_with(
+            "reports",
+            "u1:t1",
+            prepared,
+            binding=binding,
+        )
+    finally:
+        await pool.close_all()
 
 
 def test_ensure_binding_rejects_changed_fingerprint():

backend/tests/test_mcp_session_pool_active_calls.py

@@ -0,0 +1,332 @@
+"""Capacity pressure must not interrupt already-admitted MCP calls."""
+
+import asyncio
+from contextlib import asynccontextmanager
+
+import pytest
+from mcp.shared.exceptions import McpError
+from mcp.types import CONNECTION_CLOSED, CallToolResult, ErrorData, TextContent
+
+from deerflow.mcp.session_pool import MCPSessionPool, call_pooled_session_tool
+
+
+class ControlledSession:
+    def __init__(self, connection):
+        self.connection = connection
+        self.started = asyncio.Event()
+        self.release = asyncio.Event()
+        self.closed = connection.get("closed", asyncio.Event())
+
+    async def initialize(self):
+        if "initialize_started" in self.connection:
+            self.connection["initialize_started"].set()
+        if "initialize_gate" in self.connection:
+            await self.connection["initialize_gate"].wait()
+
+    async def call_tool(self, name, arguments, **kwargs):
+        self.started.set()
+        arguments.get("started", self.started).set()
+        await arguments.get("release", self.release).wait()
+        if self.closed.is_set():
+            raise McpError(ErrorData(code=CONNECTION_CLOSED, message="Connection closed"))
+        if "error" in arguments:
+            raise arguments["error"]
+        return CallToolResult(content=[TextContent(type="text", text="finished")], isError=arguments.get("is_error", False))
+
+
+@pytest.fixture
+def controlled_pool(monkeypatch):
+    pool = MCPSessionPool()
+
+    @asynccontextmanager
+    async def create_session(connection):
+        owner = asyncio.current_task()
+        session = ControlledSession(connection)
+        try:
+            yield session
+        finally:
+            assert asyncio.current_task() is owner
+            session.closed.set()
+
+    monkeypatch.setattr("langchain_mcp_adapters.sessions.create_session", create_session)
+    return pool
+
+
+async def call(session, pool, scope, *, domain="deployment", arguments=None):
+    return await call_pooled_session_tool(session, pool, server_name="server", scope_key=scope, tool_name="slow", arguments=arguments or {}, call_kwargs={}, domain=domain)
+
+
+@pytest.mark.asyncio
+async def test_capacity_evicts_idle_session_instead_of_inflight_call(controlled_pool):
+    pool = controlled_pool
+    pool.MAX_SESSIONS = 2
+    active = await pool.get_session("server", "active", {})
+    idle = await pool.get_session("server", "idle", {})
+    running = asyncio.create_task(call(active, pool, "active"))
+    try:
+        await asyncio.wait_for(active.started.wait(), 1)
+        replacement = await pool.get_session("server", "replacement", {})
+        assert not active.closed.is_set(), "LRU capacity eviction closed an in-flight call"
+        assert idle.closed.is_set()
+        active.release.set()
+        result = await asyncio.wait_for(running, 1)
+        assert result.content[0].text == "finished"
+        assert len(pool._entries) == 2
+        assert replacement is not idle
+    finally:
+        active.release.set()
+        await asyncio.gather(running, return_exceptions=True)
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+async def test_full_busy_pool_rejects_new_scope_without_closing_active_call(controlled_pool):
+    pool = controlled_pool
+    pool.MAX_SESSIONS = 1
+    active = await pool.get_session("server", "active", {})
+    running = asyncio.create_task(call(active, pool, "active"))
+    try:
+        await asyncio.wait_for(active.started.wait(), 1)
+        with pytest.raises(RuntimeError, match="capacity"):
+            await pool.get_session("server", "replacement", {})
+        assert not pool._inflight
+        assert len(pool._entries) == 1
+        # Reusing an existing busy session does not require a new slot.
+        assert await pool.get_session("server", "active", {}) is active
+        assert not active.closed.is_set()
+        active.release.set()
+        assert (await asyncio.wait_for(running, 1)).content[0].text == "finished"
+        replacement = await pool.get_session("server", "replacement", {})
+        assert active.closed.is_set()
+        assert replacement is not active
+    finally:
+        active.release.set()
+        await asyncio.gather(running, return_exceptions=True)
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["scope", "server", "session", "current", "all"])
+async def test_explicit_close_still_retires_an_active_call(controlled_pool, operation):
+    pool = controlled_pool
+    active = await pool.get_session("server", "active", {})
+    running = asyncio.create_task(call(active, pool, "active"))
+    try:
+        await asyncio.wait_for(active.started.wait(), 1)
+        close = {
+            "scope": lambda: pool.close_scope("active"),
+            "server": lambda: pool.close_server("server"),
+            "session": lambda: pool.close_session("server", "active"),
+            "current": lambda: pool.close_session_if_current("server", "active", active),
+            "all": pool.close_all,
+        }[operation]
+        await close()
+        assert active.closed.is_set()
+        active.release.set()
+        with pytest.raises(McpError, match="Connection closed"):
+            await asyncio.wait_for(running, 1)
+    finally:
+        active.release.set()
+        await asyncio.gather(running, return_exceptions=True)
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("outcome", ["cancel", "timeout", "protocol_error", "tool_error", "success"])
+async def test_call_exit_releases_capacity_protection(controlled_pool, outcome):
+    pool = controlled_pool
+    pool.MAX_SESSIONS = 1
+    session = await pool.get_session("server", "active", {})
+    arguments = {}
+    if outcome == "timeout":
+        arguments["error"] = TimeoutError("tool timeout")
+    elif outcome == "protocol_error":
+        arguments["error"] = McpError(ErrorData(code=408, message="request timed out"))
+    elif outcome == "tool_error":
+        arguments["is_error"] = True
+    running = asyncio.create_task(call(session, pool, "active", arguments=arguments))
+    try:
+        await asyncio.wait_for(session.started.wait(), 1)
+        if outcome == "cancel":
+            running.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await running
+        else:
+            session.release.set()
+            if "error" in arguments:
+                with pytest.raises(type(arguments["error"])):
+                    await running
+            else:
+                assert (await running).isError is (outcome == "tool_error")
+        assert not session.closed.is_set(), "nonfatal call exits must retain the session"
+        assert not pool._active_calls
+        await pool.get_session("server", "replacement", {})
+        assert session.closed.is_set()
+    finally:
+        session.release.set()
+        await asyncio.gather(running, return_exceptions=True)
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+async def test_shared_session_stays_busy_until_last_call_exits(controlled_pool):
+    pool = controlled_pool
+    pool.MAX_SESSIONS = 1
+    session = await pool.get_session("server", "active", {})
+    starts = [asyncio.Event(), asyncio.Event()]
+    releases = [asyncio.Event(), asyncio.Event()]
+    calls = [asyncio.create_task(call(session, pool, "active", arguments={"started": starts[i], "release": releases[i]})) for i in range(2)]
+    try:
+        await asyncio.wait_for(asyncio.gather(*(event.wait() for event in starts)), 1)
+        calls[0].cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await calls[0]
+        with pytest.raises(RuntimeError, match="capacity"):
+            await pool.get_session("server", "replacement", {})
+        assert not session.closed.is_set()
+        releases[1].set()
+        assert (await asyncio.wait_for(calls[1], 1)).content[0].text == "finished"
+        await pool.get_session("server", "replacement", {})
+        assert session.closed.is_set()
+        assert not pool._active_calls
+    finally:
+        for event in releases:
+            event.set()
+        await asyncio.gather(*calls, return_exceptions=True)
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("with_idle", [False, True])
+async def test_promotion_rechecks_busy_sessions_after_initialization(controlled_pool, with_idle):
+    pool = controlled_pool
+    pool.MAX_SESSIONS = 2 if with_idle else 1
+    initialized = asyncio.Event()
+    initialize_gate = asyncio.Event()
+    rejected_closed = asyncio.Event()
+    pending = asyncio.create_task(pool.get_session("server", "pending", {"initialize_started": initialized, "initialize_gate": initialize_gate, "closed": rejected_closed}))
+    running = None
+    joiner = None
+    try:
+        await asyncio.wait_for(initialized.wait(), 1)
+        active = await pool.get_session("server", "active", {})
+        running = asyncio.create_task(call(active, pool, "active"))
+        await asyncio.wait_for(active.started.wait(), 1)
+        idle = await pool.get_session("server", "idle", {}) if with_idle else None
+        # Join the already-admitted creation even though the registry is now full.
+        joined = asyncio.Event()
+
+        async def join():
+            joined.set()
+            return await pool.get_session("server", "pending", {})
+
+        joiner = asyncio.create_task(join())
+        await asyncio.wait_for(joined.wait(), 1)
+        initialize_gate.set()
+        if with_idle:
+            replacement, shared = await asyncio.wait_for(asyncio.gather(pending, joiner), 1)
+            assert replacement is shared
+            await asyncio.wait_for(idle.closed.wait(), 1)
+        else:
+            outcomes = await asyncio.wait_for(asyncio.gather(pending, joiner, return_exceptions=True), 1)
+            assert all(isinstance(error, RuntimeError) and "capacity" in str(error) for error in outcomes)
+            await asyncio.wait_for(rejected_closed.wait(), 1)
+        assert not active.closed.is_set()
+        assert not pool._inflight
+        assert len(pool._entries) == pool.MAX_SESSIONS
+        active.release.set()
+        assert (await asyncio.wait_for(running, 1)).content[0].text == "finished"
+    finally:
+        initialize_gate.set()
+        if running is not None:
+            active.release.set()
+        await asyncio.gather(pending, *([joiner] if joiner is not None else []), *([running] if running is not None else []), return_exceptions=True)
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+async def test_old_call_exit_cannot_unpin_replacement_owner(controlled_pool):
+    pool = controlled_pool
+    pool.MAX_SESSIONS = 1
+    old = await pool.get_session("server", "active", {})
+    old_call = asyncio.create_task(call(old, pool, "active"))
+    new_call = None
+    try:
+        await asyncio.wait_for(old.started.wait(), 1)
+        await pool.close_scope("active")
+        new = await pool.get_session("server", "active", {})
+        new_call = asyncio.create_task(call(new, pool, "active"))
+        await asyncio.wait_for(new.started.wait(), 1)
+        old.release.set()
+        with pytest.raises(McpError, match="Connection closed"):
+            await asyncio.wait_for(old_call, 1)
+        with pytest.raises(RuntimeError, match="capacity"):
+            await pool.get_session("server", "replacement", {})
+        assert not new.closed.is_set()
+        new.release.set()
+        assert (await asyncio.wait_for(new_call, 1)).content[0].text == "finished"
+        assert not pool._active_calls
+        await pool.get_session("server", "replacement", {})
+        assert new.closed.is_set()
+    finally:
+        old.release.set()
+        if new_call is not None:
+            new.release.set()
+        await asyncio.gather(old_call, *([new_call] if new_call is not None else []), return_exceptions=True)
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+async def test_busy_personal_session_survives_deployment_capacity_pressure(controlled_pool):
+    pool = controlled_pool
+    pool.MAX_SESSIONS = 2
+    personal = await pool.get_session("server", "shared", {}, domain="personal")
+    deployment = await pool.get_session("server", "shared", {})
+    running = asyncio.create_task(call(personal, pool, "shared", domain="personal"))
+    try:
+        await asyncio.wait_for(personal.started.wait(), 1)
+        await pool.get_session("server", "replacement", {})
+        assert not personal.closed.is_set()
+        assert deployment.closed.is_set()
+        personal.release.set()
+        assert (await asyncio.wait_for(running, 1)).content[0].text == "finished"
+    finally:
+        personal.release.set()
+        await asyncio.gather(running, return_exceptions=True)
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+async def test_rebinding_still_retires_active_calls(controlled_pool):
+    pool = controlled_pool
+    old_binding = pool.ensure_binding("server", "original-fingerprint")
+    session = await pool.get_session("server", "active", {}, binding=old_binding)
+    running = asyncio.create_task(call(session, pool, "active"))
+    try:
+        await asyncio.wait_for(session.started.wait(), 1)
+        pool.reconcile_bindings({"server": "new-fingerprint"}, ())
+        await asyncio.wait_for(session.closed.wait(), 1)
+        session.release.set()
+        with pytest.raises(McpError, match="Connection closed"):
+            await asyncio.wait_for(running, 1)
+        assert not pool._active_calls
+        assert not pool._entries
+    finally:
+        session.release.set()
+        await asyncio.gather(running, return_exceptions=True)
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+async def test_detached_session_cannot_start_an_unprotected_call(controlled_pool):
+    pool = controlled_pool
+    pool.MAX_SESSIONS = 1
+    old = await pool.get_session("server", "old", {})
+    try:
+        await pool.get_session("server", "replacement", {})
+        with pytest.raises(RuntimeError, match="no longer pooled"):
+            await call(old, pool, "old")
+        assert not old.started.is_set()
+        assert not pool._active_calls
+    finally:
+        await pool.close_all()

backend/tests/test_mcp_session_pool_loop_isolation.py

@@ -9,7 +9,7 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from deerflow.mcp.session_pool import MCPSessionPool
+from deerflow.mcp.session_pool import MCPSessionPool, call_pooled_session_tool
 from deerflow.tools.sync import make_sync_tool_wrapper
 
 
@@ -102,6 +102,48 @@ def test_owner_completion_does_not_remove_a_replacement(loop_pool):
     run(0, replace())
 
 
+@pytest.mark.parametrize("capacity", [1, 2])
+def test_capacity_pressure_on_foreign_loop_preserves_active_call(loop_pool, capacity):
+    pool, run, closed = loop_pool
+    pool.MAX_SESSIONS = capacity
+    active = run(0, pool.get_session("s", "u:t", {}))
+    idle = run(1, pool.get_session("s", "u:t", {})) if capacity == 2 else None
+    started = threading.Event()
+
+    async def start():
+        release = asyncio.Event()
+
+        async def slow(*args, **kwargs):
+            started.set()
+            await release.wait()
+            return "finished"
+
+        active.call_tool.side_effect = slow
+        task = asyncio.create_task(call_pooled_session_tool(active, pool, server_name="s", scope_key="u:t", tool_name="slow", arguments={}, call_kwargs={}))
+        return release, task
+
+    release, task = run(0, start())
+
+    async def finish():
+        release.set()
+        return await task
+
+    try:
+        assert started.wait(5)
+        if idle is not None:
+            run(1, pool.get_session("s", "new", {}))
+            assert idle in closed
+        else:
+            with pytest.raises(RuntimeError, match="capacity"):
+                run(1, pool.get_session("s", "new", {}))
+            assert not pool._inflight
+        assert active not in closed
+        assert run(0, finish()) == "finished"
+        assert not pool._active_calls
+    finally:
+        run(0, finish())
+
+
 @pytest.mark.parametrize("retirement", ["lru", "explicit"])
 def test_abandoned_closed_loop_entry_can_be_retired(loop_pool, retirement):
     """Model the registry left by unsupported loop.close() with pending owners.

backend/tests/test_mcp_task_tool_caller.py

@@ -204,7 +204,7 @@ async def test_task_without_applicable_user_auth_preserves_interceptor_context(t
 
     caller._interceptors.append(inspect_context)
     session = SimpleNamespace(initialize=AsyncMock(), call_tool=AsyncMock(return_value="result"))
-    pool = SimpleNamespace(get_session=AsyncMock(return_value=session))
+    pool = MCPSessionPool()
     user_token = set_current_user(ambient_user) if ambient_user is not None else None
     try:
         with (
@@ -223,6 +223,7 @@ async def test_task_without_applicable_user_auth_preserves_interceptor_context(t
         assert observed_users == [ambient_user]
         assert get_current_user() is ambient_user
     finally:
+        await pool.close_all()
         if user_token is not None:
             reset_current_user(user_token)