diff --git a/backend/CLAUDE.md b/backend/CLAUDE.md index bce366da1..48cf942a7 100644 --- a/backend/CLAUDE.md +++ b/backend/CLAUDE.md @@ -350,6 +350,8 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti - **Cache invalidation**: Detects config file changes via mtime comparison - **Transports**: stdio (command-based), SSE, HTTP - **OAuth (HTTP/SSE)**: Supports token endpoint flows (`client_credentials`, `refresh_token`) with automatic token refresh + Authorization header injection +- **Stdio file outputs**: Persistent stdio sessions are scoped by `user_id:thread_id`. For stdio transports only, DeerFlow pins the subprocess default `cwd` to the thread workspace and `TMPDIR`/`TMP`/`TEMP` to `workspace/.mcp/tmp/`, unless the operator explicitly configured `cwd` or temp env values. SSE/HTTP transports skip this filesystem prep entirely. +- **Stdio path translation**: MCP-returned local file references are not copied. If a `ResourceLink` or conservative free-text path resolves to an existing file inside the thread's mounted user-data tree, it is translated deterministically to `/mnt/user-data/...`; paths outside that tree remain unchanged. - **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via mtime ### Skills System (`packages/harness/deerflow/skills/`) diff --git a/backend/packages/harness/deerflow/mcp/tools.py b/backend/packages/harness/deerflow/mcp/tools.py index e425efe0c..e4dc4f623 100644 --- a/backend/packages/harness/deerflow/mcp/tools.py +++ b/backend/packages/harness/deerflow/mcp/tools.py @@ -2,23 +2,291 @@ from __future__ import annotations +import asyncio import logging -from collections.abc import Mapping +import re +from collections.abc import Iterable, Mapping +from pathlib import Path from typing import Any +from urllib.parse import unquote, urlparse from langchain_core.tools import BaseTool, StructuredTool from langgraph.config import get_config from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.config.paths import VIRTUAL_PATH_PREFIX, Paths, get_paths from deerflow.mcp.client import build_servers_config from deerflow.mcp.oauth import build_oauth_tool_interceptor, get_initial_oauth_headers from deerflow.mcp.session_pool import get_session_pool from deerflow.reflection import resolve_variable +from deerflow.runtime.user_context import resolve_runtime_user_id from deerflow.tools.sync import make_sync_tool_wrapper from deerflow.tools.types import Runtime logger = logging.getLogger(__name__) +# Subdirectory under the thread's workspace used as the temp dir for stdio MCP +# subprocesses. Pinning the process temp dir here (alongside its cwd) makes +# tools that write to ``os.tmpdir()`` / ``tempfile.gettempdir()`` land inside +# the mounted user-data tree, where their output is resolvable by the +# sandbox/artifact API — instead of on an unreachable host temp path. +_MCP_TMP_SUBDIR = ".mcp/tmp" + +# Matches local-file references embedded in free text returned by an MCP server. +# Some servers (notably Playwright's ``browser_take_screenshot``) report saved +# files only as text/markdown links rather than ``ResourceLink`` blocks. Those +# references may be absolute paths, ``file://`` URIs, or paths relative to the +# server process cwd (e.g. ``temp/page.yml``, ``./shot.png``). Each match is +# only rewritten when it resolves to an existing file inside the thread's +# user-data tree, so an over-eager match is harmless (left untouched). +_LOCAL_PATH_IN_TEXT_RE = re.compile(r"(?:file://)?/[^\s'\"<>|*?]+|(?:\.{0,2}/|[\w.-]+/)[^\s'\"<>|*?]+") + +# Trailing characters that are punctuation/markup rather than part of a path. +_TEXT_PATH_TRAILING_CHARS = ".,;:!?)]}>\"'`" + +_FILE_SNAPSHOT = dict[Path, tuple[int, int]] + + +def _local_path_from_uri(uri: str, *, base_dir: Path | None = None) -> Path | None: + """Return an absolute local filesystem ``Path`` if *uri* points to a local + file, otherwise ``None``. + + Accepts bare paths and ``file://`` URIs. Remote URIs + (``http``/``https``/``data``/...) return ``None`` so the caller leaves them + untouched. Relative paths are resolved only when *base_dir* is supplied. + """ + if not uri: + return None + parsed = urlparse(uri) + if parsed.scheme == "file": + raw = unquote(parsed.path) + elif parsed.scheme == "": + raw = uri + else: + return None + if not raw: + return None + path = Path(raw) + if not path.is_absolute(): + if base_dir is None: + return None + path = base_dir / path + return path + + +def _local_uri_to_virtual_path( + uri: str, + *, + thread_id: str, + user_id: str, + source_base_dir: Path | None = None, +) -> str | None: + """Translate a local file reference into its ``/mnt/user-data/...`` virtual path. + + Stdio MCP servers run with their cwd and temp dir pinned inside the thread's + mounted user-data tree (see :func:`_make_session_pool_tool`), so the files + they produce already live somewhere the sandbox/artifact API can serve — the + only thing missing is the virtual prefix the rest of DeerFlow addresses them + by. This performs that purely deterministic host→virtual mapping: no copy, no + trusted-root list, and no exposure of files outside the thread's own tree. + + Returns ``None`` (so the caller leaves the reference untouched) when the URI + is remote, cannot be resolved, points outside this thread's user-data tree, + or does not name an existing file. Relative references are resolved against + *source_base_dir* (the server's cwd). + """ + src = _local_path_from_uri(uri, base_dir=source_base_dir) + if src is None: + return None + + try: + real = src.resolve() + except OSError: + return None + if not real.is_file(): + return None + + try: + user_data_root = get_paths().sandbox_user_data_dir(thread_id, user_id=user_id).resolve() + except OSError: + return None + + try: + relative = real.relative_to(user_data_root) + except ValueError: + # The file lives outside this thread's user-data mount; we cannot + # express it as a virtual path, so leave the original reference as-is. + logger.debug("MCP path rewrite skipped outside user-data tree: %s", real) + return None + + virtual_path = f"{VIRTUAL_PATH_PREFIX}/{relative.as_posix()}" + logger.debug("MCP path rewrite: %s -> %s", real, virtual_path) + return virtual_path + + +def _snapshot_workspace_files(root: Path) -> _FILE_SNAPSHOT: + """Return a lightweight snapshot of regular files under *root*.""" + snapshot: _FILE_SNAPSHOT = {} + if not root.exists(): + return snapshot + + try: + candidates = root.rglob("*") + for path in candidates: + try: + stat = path.stat() + except OSError: + continue + if path.is_file(): + snapshot[path] = (stat.st_mtime_ns, stat.st_size) + except OSError: + return snapshot + return snapshot + + +def _changed_workspace_files(root: Path, before: _FILE_SNAPSHOT) -> list[Path]: + """Return files under *root* that were created or modified since *before*.""" + after = _snapshot_workspace_files(root) + return [path for path, signature in after.items() if before.get(path) != signature] + + +def _prepare_stdio_workspace(paths: Paths, *, thread_id: str, user_id: str) -> tuple[Path, Path, _FILE_SNAPSHOT]: + """Prepare the thread workspace for a pinned stdio MCP subprocess. + + Bundles all the synchronous filesystem work (dir creation, temp-dir prep, + and the pre-call snapshot) into one helper so the caller can run it off the + event loop via :func:`asyncio.to_thread`. Returns the workspace cwd, the + pinned temp dir, and the pre-call file snapshot. + """ + paths.ensure_thread_dirs(thread_id, user_id=user_id) + source_base_dir = paths.sandbox_work_dir(thread_id, user_id=user_id) + tmp_dir = source_base_dir / _MCP_TMP_SUBDIR + try: + tmp_dir.mkdir(parents=True, exist_ok=True) + tmp_dir.chmod(0o700) + except OSError: + logger.warning("Failed to prepare MCP temp dir: %s", tmp_dir, exc_info=True) + before_files = _snapshot_workspace_files(source_base_dir) + return source_base_dir, tmp_dir, before_files + + +def _result_has_text_content(call_tool_result: Any) -> bool: + """Return ``True`` when the MCP result carries any text content. + + The after-call snapshot diff only feeds bare-filename correlation in free + text. When the result has no text blocks there is nothing to rewrite, so the + caller can skip the second recursive walk entirely. + """ + from mcp.types import EmbeddedResource, TextContent, TextResourceContents + + content = getattr(call_tool_result, "content", None) + if not content: + return False + for item in content: + if isinstance(item, TextContent): + return True + if isinstance(item, EmbeddedResource) and isinstance(item.resource, TextResourceContents): + return True + return False + + +def _rewrite_unique_bare_filenames( + text: str, + *, + changed_files: Iterable[Path], + thread_id: str, + user_id: str, + source_base_dir: Path | None = None, +) -> str: + """Rewrite bare filenames only when this call produced a unique match. + + A response like ``Saved as page-2026.yml`` is not structurally a path. The + only safe way to interpret it is to correlate the filename with files + created/modified by this exact tool call, and rewrite only when the basename + maps to exactly one file inside this thread's mounted user-data tree. + """ + candidates: dict[str, list[str]] = {} + for path in changed_files: + virtual_path = _local_uri_to_virtual_path( + str(path), + thread_id=thread_id, + user_id=user_id, + source_base_dir=source_base_dir, + ) + if virtual_path is None: + continue + candidates.setdefault(path.name, []).append(virtual_path) + + unique = {name: paths[0] for name, paths in candidates.items() if len(set(paths)) == 1} + if not unique: + if candidates: + logger.debug("MCP bare filename rewrite skipped: no unique candidate in %s", sorted(candidates)) + else: + logger.debug("MCP bare filename rewrite skipped: no snapshot candidates") + return text + + rewritten = text + for name in sorted(unique, key=len, reverse=True): + # Do not rewrite inside longer paths/words. A final sentence period is + # allowed, but ".bak" or another path segment is not. + pattern = re.compile(rf"(? %s", name, unique[name]) + rewritten = rewritten_text + return rewritten + + +def _rewrite_local_paths_in_text( + text: str, + *, + thread_id: str, + user_id: str, + source_base_dir: Path | None = None, + changed_files: Iterable[Path] | None = None, +) -> str: + """Best-effort rewrite of local file references found in free text. + + Some MCP servers (notably Playwright's ``browser_take_screenshot``) report + the saved file only as free text — e.g. ``Took the screenshot and saved it + as temp/page-2026.png`` — instead of a ``ResourceLink``. Free text is not a + reliable protocol, so this is deliberately conservative: every candidate + token is handed to :func:`_local_uri_to_virtual_path`, which only rewrites + it when it resolves to an existing file inside this thread's user-data tree. + Tokens that are not real paths (or point elsewhere) are left exactly as they + were, so an over-eager regex match has no harmful effect. + """ + translated_by_source: dict[str, str | None] = {} + + def _replace(match: re.Match[str]) -> str: + token = match.group(0) + # A path can end a sentence ("saved as temp/a.png."); strip trailing + # punctuation and restore it after the (possibly rewritten) path. + stripped = token.rstrip(_TEXT_PATH_TRAILING_CHARS) + trailing = token[len(stripped) :] + if stripped not in translated_by_source: + translated_by_source[stripped] = _local_uri_to_virtual_path( + stripped, + thread_id=thread_id, + user_id=user_id, + source_base_dir=source_base_dir, + ) + rewritten = translated_by_source[stripped] + if rewritten is None: + return token + return f"{rewritten}{trailing}" + + rewritten = _LOCAL_PATH_IN_TEXT_RE.sub(_replace, text) + if changed_files is None: + return rewritten + return _rewrite_unique_bare_filenames( + rewritten, + changed_files=changed_files, + thread_id=thread_id, + user_id=user_id, + source_base_dir=source_base_dir, + ) + def _extract_thread_id(runtime: Runtime | None) -> str: """Extract thread_id from the injected tool runtime or LangGraph config.""" @@ -38,11 +306,27 @@ def _extract_thread_id(runtime: Runtime | None) -> str: return "default" -def _convert_call_tool_result(call_tool_result: Any) -> Any: +def _convert_call_tool_result( + call_tool_result: Any, + *, + thread_id: str | None = None, + user_id: str | None = None, + source_base_dir: Path | None = None, + changed_files: Iterable[Path] | None = None, +) -> Any: """Convert an MCP CallToolResult to the LangChain ``content_and_artifact`` format. Implements the same conversion logic as the adapter without relying on the private ``langchain_mcp_adapters.tools._convert_call_tool_result`` symbol. + + When ``thread_id`` and ``user_id`` are provided, local files referenced by + ``ResourceLink`` blocks or plain text (e.g. screenshots saved by Playwright + MCP) have their references translated from the host path to the + ``/mnt/user-data/...`` virtual path so they can be resolved by the sandbox + and artifact API. The files themselves are not copied — stdio servers run + with their cwd/temp pinned inside the mounted tree, so they already live in + a servable location. Remote URIs and files outside the thread's user-data + tree are left untouched. """ from langchain_core.messages import ToolMessage from langchain_core.messages.content import create_file_block, create_image_block, create_text_block @@ -63,25 +347,46 @@ def _convert_call_tool_result(call_tool_result: Any) -> Any: # langgraph is optional; if unavailable, continue with standard MCP content conversion. pass + def _resolve_link_url(uri: str) -> str: + if thread_id is None or user_id is None: + return uri + rewritten = _local_uri_to_virtual_path(uri, thread_id=thread_id, user_id=user_id, source_base_dir=source_base_dir) + return rewritten if rewritten is not None else uri + + def _resolve_text(text: str) -> str: + # Servers like Playwright report saved files only as plain text, with no + # ResourceLink to hook into. Scan the text for local paths and translate + # them so the produced files are readable through the sandbox/artifact API. + if thread_id is None or user_id is None: + return text + return _rewrite_local_paths_in_text( + text, + thread_id=thread_id, + user_id=user_id, + source_base_dir=source_base_dir, + changed_files=changed_files, + ) + # Convert MCP content blocks to LangChain content blocks. lc_content = [] for item in call_tool_result.content: if isinstance(item, TextContent): - lc_content.append(create_text_block(text=item.text)) + lc_content.append(create_text_block(text=_resolve_text(item.text))) elif isinstance(item, ImageContent): lc_content.append(create_image_block(base64=item.data, mime_type=item.mimeType)) elif isinstance(item, ResourceLink): mime = item.mimeType or None + url = _resolve_link_url(str(item.uri)) if mime and mime.startswith("image/"): - lc_content.append(create_image_block(url=str(item.uri), mime_type=mime)) + lc_content.append(create_image_block(url=url, mime_type=mime)) else: - lc_content.append(create_file_block(url=str(item.uri), mime_type=mime)) + lc_content.append(create_file_block(url=url, mime_type=mime)) elif isinstance(item, EmbeddedResource): from mcp.types import BlobResourceContents res = item.resource if isinstance(res, TextResourceContents): - lc_content.append(create_text_block(text=res.text)) + lc_content.append(create_text_block(text=_resolve_text(res.text))) elif isinstance(res, BlobResourceContents): mime = res.mimeType or None if mime and mime.startswith("image/"): @@ -113,8 +418,9 @@ def _make_session_pool_tool( """Wrap an MCP tool so it reuses a persistent session from the pool. Replaces the per-call session creation with pool-managed sessions scoped - by ``(server_name, thread_id)``. This ensures stateful MCP servers (e.g. - Playwright) keep their state across tool calls within the same thread. + by ``(server_name, user_id:thread_id)``. This ensures stateful MCP servers + (e.g. Playwright) keep their state across tool calls within the same thread + while staying isolated per user. The configured ``tool_interceptors`` (OAuth, custom) are preserved and applied on every call before invoking the pooled session. @@ -132,7 +438,45 @@ def _make_session_pool_tool( **arguments: Any, ) -> Any: thread_id = _extract_thread_id(runtime) - session = await pool.get_session(server_name, thread_id, connection) + user_id = resolve_runtime_user_id(runtime) + # Scope the pooled session by user *and* thread. Filesystem isolation is + # per-(user_id, thread_id), so a thread_id alone could otherwise let two + # users with a colliding thread_id share one stateful MCP session. + scope_key = f"{user_id}:{thread_id}" + session_connection = dict(connection) + # cwd/temp pinning and the workspace snapshot only matter for stdio + # servers, which run as local subprocesses writing to a real filesystem. + # SSE/HTTP servers have no local cwd to pin, so skip the filesystem work + # entirely for them (avoids needless dir creation and recursive walks). + is_stdio = session_connection.get("transport", "stdio") == "stdio" + source_base_dir: Path | None = None + process_cwd: Path | None = None + before_files: _FILE_SNAPSHOT | None = None + if is_stdio: + paths = get_paths() + # Bundle the synchronous filesystem prep (dir creation, temp-dir + # setup, pre-call snapshot) and run it off the event loop — the + # snapshot walks the whole workspace and would otherwise block. + source_base_dir, tmp_dir, before_files = await asyncio.to_thread(_prepare_stdio_workspace, paths, thread_id=thread_id, user_id=user_id) + # Stdio MCP servers resolve relative output links against their + # process cwd. Keep that cwd inside the thread's mounted user-data + # tree so files produced by tools like Playwright land where the + # sandbox/artifact API can serve them and their references can be + # translated to virtual paths. + configured_cwd = session_connection.get("cwd", str(source_base_dir)) + session_connection["cwd"] = str(configured_cwd) + process_cwd = Path(configured_cwd) + # Pin the subprocess temp dir under the same mounted tree. Tools that + # default to the OS temp dir (Node's os.tmpdir(), Python's tempfile, + # many CLIs) then write inside user-data instead of an unreachable + # host path — the tool-agnostic counterpart to fixing the cwd. Merge + # rather than replace any operator-provided env. + session_env = dict(session_connection.get("env") or {}) + session_env.setdefault("TMPDIR", str(tmp_dir)) + session_env.setdefault("TMP", str(tmp_dir)) + session_env.setdefault("TEMP", str(tmp_dir)) + session_connection["env"] = session_env + session = await pool.get_session(server_name, scope_key, session_connection) if tool_interceptors: from langchain_mcp_adapters.interceptors import MCPToolCallRequest @@ -167,7 +511,22 @@ def _make_session_pool_tool( else: call_tool_result = await session.call_tool(original_name, arguments) - return _convert_call_tool_result(call_tool_result) + # The after-call snapshot diff only feeds bare-filename correlation in + # free text, so skip the second recursive walk when there is no text + # content to rewrite. Both the diff and the per-token path resolution + # inside _convert_call_tool_result touch the filesystem, so run them off + # the event loop. + changed_files: list[Path] | None = None + if is_stdio and before_files is not None and _result_has_text_content(call_tool_result): + changed_files = await asyncio.to_thread(_changed_workspace_files, source_base_dir, before_files) + return await asyncio.to_thread( + _convert_call_tool_result, + call_tool_result, + thread_id=thread_id, + user_id=user_id, + source_base_dir=process_cwd, + changed_files=changed_files, + ) return StructuredTool( name=tool.name, diff --git a/backend/tests/test_mcp_file_migration.py b/backend/tests/test_mcp_file_migration.py new file mode 100644 index 000000000..ccdcadaaf --- /dev/null +++ b/backend/tests/test_mcp_file_migration.py @@ -0,0 +1,575 @@ +"""Tests for translating MCP-produced local files into virtual sandbox paths. + +Regression coverage for GitHub issue #3597: Playwright MCP (and similar stdio +servers) write files to a path the sandbox/artifact API cannot resolve. The MCP +tool wrapper pins stdio cwd/temp under the thread's mounted user-data tree and +rewrites returned file references to ``/mnt/user-data/...`` virtual paths. +""" + +from pathlib import Path +from unittest.mock import patch + +import pytest +from mcp.types import CallToolResult, ResourceLink, TextContent + +from deerflow.config.paths import VIRTUAL_PATH_PREFIX, Paths +from deerflow.mcp import tools as mcp_tools + + +@pytest.fixture +def paths(tmp_path: Path) -> Paths: + return Paths(tmp_path) + + +def _patch_paths(paths: Paths): + return patch("deerflow.mcp.tools.get_paths", return_value=paths) + + +def _workspace_file(paths: Paths, relative_path: str, *, content: bytes = b"data") -> Path: + file_path = paths.sandbox_work_dir("t1", user_id="u1") / relative_path + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_bytes(content) + return file_path + + +class TestLocalPathFromUri: + def test_file_uri(self): + assert mcp_tools._local_path_from_uri("file:///tmp/shot.png") == Path("/tmp/shot.png") + + def test_bare_absolute_path(self): + assert mcp_tools._local_path_from_uri("/var/data/out.pdf") == Path("/var/data/out.pdf") + + def test_file_uri_with_url_encoded_spaces(self): + assert mcp_tools._local_path_from_uri("file:///tmp/my%20shot.png") == Path("/tmp/my shot.png") + + def test_remote_uri_is_ignored(self): + assert mcp_tools._local_path_from_uri("https://example.com/a.png") is None + assert mcp_tools._local_path_from_uri("data:image/png;base64,AAAA") is None + + def test_relative_path_is_ignored_without_base_dir(self): + assert mcp_tools._local_path_from_uri("relative/path.txt") is None + + def test_relative_path_uses_base_dir_when_provided(self, tmp_path: Path): + assert mcp_tools._local_path_from_uri("./shot.png", base_dir=tmp_path) == tmp_path / "shot.png" + assert mcp_tools._local_path_from_uri("temp/page.yml", base_dir=tmp_path) == tmp_path / "temp/page.yml" + + def test_file_uri_with_relative_path_is_ignored(self): + assert mcp_tools._local_path_from_uri("file:relative.txt") is None + + def test_file_uri_with_empty_path_is_ignored(self): + assert mcp_tools._local_path_from_uri("file://") is None + + def test_file_uri_with_localhost_host(self): + # file://localhost/abs/path is the host form of file:///abs/path. + assert mcp_tools._local_path_from_uri("file://localhost/tmp/shot.png") == Path("/tmp/shot.png") + + def test_empty_is_ignored(self): + assert mcp_tools._local_path_from_uri("") is None + + +class TestLocalUriToVirtualPath: + def test_workspace_file_translates_to_virtual_workspace_path(self, paths: Paths): + src = _workspace_file(paths, "temp/page.yml") + + with _patch_paths(paths): + result = mcp_tools._local_uri_to_virtual_path(str(src), thread_id="t1", user_id="u1") + + assert result == f"{VIRTUAL_PATH_PREFIX}/workspace/temp/page.yml" + + def test_outputs_file_translates_without_copy(self, paths: Paths): + outputs = paths.sandbox_outputs_dir("t1", user_id="u1") + outputs.mkdir(parents=True) + src = outputs / "report.pdf" + src.write_bytes(b"pdf") + + with _patch_paths(paths): + result = mcp_tools._local_uri_to_virtual_path(str(src), thread_id="t1", user_id="u1") + + assert result == f"{VIRTUAL_PATH_PREFIX}/outputs/report.pdf" + assert list(outputs.iterdir()) == [src] + + def test_relative_review_case_translates_against_cwd(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + _workspace_file(paths, "temp/page-2026-06-16T10-21-46-864Z.yml") + + with _patch_paths(paths): + result = mcp_tools._local_uri_to_virtual_path( + "temp/page-2026-06-16T10-21-46-864Z.yml", + thread_id="t1", + user_id="u1", + source_base_dir=workspace, + ) + + assert result == f"{VIRTUAL_PATH_PREFIX}/workspace/temp/page-2026-06-16T10-21-46-864Z.yml" + + def test_file_uri_inside_user_data_translates(self, paths: Paths): + src = _workspace_file(paths, "shot.png") + + with _patch_paths(paths): + result = mcp_tools._local_uri_to_virtual_path(f"file://{src}", thread_id="t1", user_id="u1") + + assert result == f"{VIRTUAL_PATH_PREFIX}/workspace/shot.png" + + def test_file_outside_user_data_is_not_exposed(self, tmp_path: Path, paths: Paths): + src = tmp_path / "outside.txt" + src.write_text("secret") + + with _patch_paths(paths): + result = mcp_tools._local_uri_to_virtual_path(str(src), thread_id="t1", user_id="u1") + + assert result is None + assert not paths.sandbox_outputs_dir("t1", user_id="u1").exists() + + def test_missing_file_directory_and_remote_uri_are_ignored(self, tmp_path: Path, paths: Paths): + with _patch_paths(paths): + assert mcp_tools._local_uri_to_virtual_path(str(tmp_path / "missing.png"), thread_id="t1", user_id="u1") is None + assert mcp_tools._local_uri_to_virtual_path(str(tmp_path), thread_id="t1", user_id="u1") is None + assert mcp_tools._local_uri_to_virtual_path("https://example.com/a.png", thread_id="t1", user_id="u1") is None + + def test_symlink_escape_is_not_exposed(self, tmp_path: Path, paths: Paths): + outside = tmp_path / "outside.txt" + outside.write_text("secret") + link = paths.sandbox_work_dir("t1", user_id="u1") / "link.txt" + link.parent.mkdir(parents=True) + try: + link.symlink_to(outside) + except (OSError, NotImplementedError): + pytest.skip("symlinks not supported on this platform") + + with _patch_paths(paths): + result = mcp_tools._local_uri_to_virtual_path(str(link), thread_id="t1", user_id="u1") + + assert result is None + + +class TestRewriteLocalPathsInText: + def test_review_case_temp_relative_path_is_rewritten(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + _workspace_file(paths, "temp/page-2026-06-16T10-21-46-864Z.yml") + text = "Saved as temp/page-2026-06-16T10-21-46-864Z.yml." + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1", source_base_dir=workspace) + + assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/temp/page-2026-06-16T10-21-46-864Z.yml." + + def test_relative_output_dir_path_is_rewritten(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + _workspace_file(paths, "artifacts/page.png") + text = "Screenshot saved to artifacts/page.png" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1", source_base_dir=workspace) + + assert result == f"Screenshot saved to {VIRTUAL_PATH_PREFIX}/workspace/artifacts/page.png" + + def test_absolute_output_dir_path_inside_user_data_is_rewritten(self, paths: Paths): + src = _workspace_file(paths, "absolute-output/page.png") + text = f"Screenshot saved to {src}" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1") + + assert result == f"Screenshot saved to {VIRTUAL_PATH_PREFIX}/workspace/absolute-output/page.png" + + def test_tmpdir_output_under_workspace_is_rewritten(self, paths: Paths): + src = _workspace_file(paths, ".mcp/tmp/page.png") + text = f"Saved to {src}" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1") + + assert result == f"Saved to {VIRTUAL_PATH_PREFIX}/workspace/.mcp/tmp/page.png" + + def test_old_tmp_path_outside_user_data_is_left_untouched(self, tmp_path: Path, paths: Paths): + src = tmp_path / "playwright-mcp-output" / "page.png" + src.parent.mkdir() + src.write_bytes(b"png") + text = f"Saved to {src}" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1") + + assert result == text + + def test_playwright_markdown_path_is_rewritten_twice_without_copy(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + _workspace_file(paths, ".playwright-mcp/page.png", content=b"png") + text = "### Result\n- [Screenshot](.playwright-mcp/page.png)\npath: '.playwright-mcp/page.png'" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1", source_base_dir=workspace) + + assert result.count(f"{VIRTUAL_PATH_PREFIX}/workspace/.playwright-mcp/page.png") == 2 + assert not paths.sandbox_outputs_dir("t1", user_id="u1").exists() + + def test_bare_filename_is_rewritten_only_when_changed_file_matches_uniquely(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + src = _workspace_file(paths, "page-2026-06-16T10-21-46-864Z.yml") + text = "Saved as page-2026-06-16T10-21-46-864Z.yml." + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text( + text, + thread_id="t1", + user_id="u1", + source_base_dir=workspace, + changed_files=[src], + ) + + assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/page-2026-06-16T10-21-46-864Z.yml." + + def test_bare_filename_without_changed_file_is_left_untouched(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + _workspace_file(paths, "page.yml") + text = "Saved as page.yml" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1", source_base_dir=workspace) + + assert result == text + + def test_bare_filename_with_multiple_changed_matches_is_left_untouched(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + a = _workspace_file(paths, "a/page.yml") + b = _workspace_file(paths, "b/page.yml") + text = "Saved as page.yml" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text( + text, + thread_id="t1", + user_id="u1", + source_base_dir=workspace, + changed_files=[a, b], + ) + + assert result == text + + def test_bare_filename_does_not_rewrite_longer_filename(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + src = _workspace_file(paths, "page.yml") + text = "Backup is page.yml.bak" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text( + text, + thread_id="t1", + user_id="u1", + source_base_dir=workspace, + changed_files=[src], + ) + + assert result == text + + def test_multiple_distinct_paths_in_one_message_all_rewritten(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + _workspace_file(paths, "temp/a.png") + _workspace_file(paths, "temp/b.png") + text = "Saved temp/a.png and temp/b.png together." + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1", source_base_dir=workspace) + + assert result == (f"Saved {VIRTUAL_PATH_PREFIX}/workspace/temp/a.png and {VIRTUAL_PATH_PREFIX}/workspace/temp/b.png together.") + + def test_markdown_link_in_parentheses_is_rewritten_without_eating_paren(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + _workspace_file(paths, "temp/shot.png") + text = "See ![shot](temp/shot.png) now" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1", source_base_dir=workspace) + + assert result == f"See ![shot]({VIRTUAL_PATH_PREFIX}/workspace/temp/shot.png) now" + + def test_path_for_nonexistent_relative_file_is_left_untouched(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + text = "Saved as temp/never-created.png" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1", source_base_dir=workspace) + + assert result == text + + def test_bare_filename_is_case_sensitive(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + src = _workspace_file(paths, "Page.yml") + text = "saved as page.yml" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text( + text, + thread_id="t1", + user_id="u1", + source_base_dir=workspace, + changed_files=[src], + ) + + assert result == text + + def test_bare_filename_not_rewritten_when_used_as_directory_segment(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + src = _workspace_file(paths, "page.yml") + text = "nested page.yml/inner.txt path" + + with _patch_paths(paths): + result = mcp_tools._rewrite_local_paths_in_text( + text, + thread_id="t1", + user_id="u1", + source_base_dir=workspace, + changed_files=[src], + ) + + assert result == text + + +class TestWorkspaceSnapshots: + def test_changed_workspace_files_detects_created_and_modified_files(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + existing = _workspace_file(paths, "existing.txt", content=b"old") + before = mcp_tools._snapshot_workspace_files(workspace) + + existing.write_bytes(b"new") + created = _workspace_file(paths, "created.txt", content=b"created") + + changed = set(mcp_tools._changed_workspace_files(workspace, before)) + + assert changed == {existing, created} + + def test_snapshot_of_missing_directory_is_empty(self, tmp_path: Path): + assert mcp_tools._snapshot_workspace_files(tmp_path / "does-not-exist") == {} + + def test_no_change_yields_no_changed_files(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + _workspace_file(paths, "stable.txt") + before = mcp_tools._snapshot_workspace_files(workspace) + + assert mcp_tools._changed_workspace_files(workspace, before) == [] + + def test_deleted_file_is_not_reported_as_changed(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + victim = _workspace_file(paths, "victim.txt") + before = mcp_tools._snapshot_workspace_files(workspace) + + victim.unlink() + + assert mcp_tools._changed_workspace_files(workspace, before) == [] + + +class TestPrepareStdioWorkspace: + def test_creates_dirs_and_returns_snapshot(self, paths: Paths): + existing = _workspace_file(paths, "existing.txt", content=b"old") + + source_base_dir, tmp_dir, before = mcp_tools._prepare_stdio_workspace(paths, thread_id="t1", user_id="u1") + + assert source_base_dir == paths.sandbox_work_dir("t1", user_id="u1") + assert tmp_dir == source_base_dir / mcp_tools._MCP_TMP_SUBDIR + assert tmp_dir.is_dir() + assert before == {existing: (existing.stat().st_mtime_ns, existing.stat().st_size)} + + +class TestResultHasTextContent: + def test_text_content_is_detected(self): + result = CallToolResult(content=[TextContent(type="text", text="hi")], isError=False) + assert mcp_tools._result_has_text_content(result) is True + + def test_embedded_text_resource_is_detected(self): + from mcp.types import EmbeddedResource, TextResourceContents + + res = TextResourceContents(uri="mem://n.txt", text="n", mimeType="text/plain") + result = CallToolResult(content=[EmbeddedResource(type="resource", resource=res)], isError=False) + assert mcp_tools._result_has_text_content(result) is True + + def test_image_only_result_has_no_text(self): + from mcp.types import ImageContent + + result = CallToolResult(content=[ImageContent(type="image", data="QUJD", mimeType="image/png")], isError=False) + assert mcp_tools._result_has_text_content(result) is False + + def test_empty_content_has_no_text(self): + result = CallToolResult(content=[], isError=False) + assert mcp_tools._result_has_text_content(result) is False + + +class TestConvertCallToolResultRewrites: + def test_resource_link_image_inside_workspace_rewritten(self, paths: Paths): + src = _workspace_file(paths, "page.png", content=b"png") + result = CallToolResult( + content=[ResourceLink(type="resource_link", name="page", uri=f"file://{src}", mimeType="image/png")], + isError=False, + ) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["type"] == "image" + assert content[0]["url"] == f"{VIRTUAL_PATH_PREFIX}/workspace/page.png" + + def test_resource_link_file_inside_outputs_rewritten(self, paths: Paths): + outputs = paths.sandbox_outputs_dir("t1", user_id="u1") + outputs.mkdir(parents=True) + src = outputs / "doc.pdf" + src.write_bytes(b"pdf") + result = CallToolResult( + content=[ResourceLink(type="resource_link", name="doc", uri=f"file://{src}", mimeType="application/pdf")], + isError=False, + ) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["type"] == "file" + assert content[0]["url"] == f"{VIRTUAL_PATH_PREFIX}/outputs/doc.pdf" + + def test_resource_link_outside_user_data_untouched(self, tmp_path: Path, paths: Paths): + src = tmp_path / "page.png" + src.write_bytes(b"png") + uri = f"file://{src}" + result = CallToolResult( + content=[ResourceLink(type="resource_link", name="page", uri=uri, mimeType="image/png")], + isError=False, + ) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["url"] == uri + + def test_remote_resource_link_untouched(self, paths: Paths): + url = "https://example.com/remote.png" + result = CallToolResult( + content=[ResourceLink(type="resource_link", name="r", uri=url, mimeType="image/png")], + isError=False, + ) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["url"] == url + + def test_text_review_case_rewritten(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + _workspace_file(paths, "temp/page-2026-06-16T10-21-46-864Z.yml") + result = CallToolResult( + content=[TextContent(type="text", text="Saved as temp/page-2026-06-16T10-21-46-864Z.yml")], + isError=False, + ) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1", source_base_dir=workspace) + + assert content[0]["text"] == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/temp/page-2026-06-16T10-21-46-864Z.yml" + + def test_text_bare_filename_rewritten_from_changed_files(self, paths: Paths): + workspace = paths.sandbox_work_dir("t1", user_id="u1") + src = _workspace_file(paths, "page-2026.yml") + result = CallToolResult(content=[TextContent(type="text", text="Saved as page-2026.yml")], isError=False) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result( + result, + thread_id="t1", + user_id="u1", + source_base_dir=workspace, + changed_files=[src], + ) + + assert content[0]["text"] == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/page-2026.yml" + + def test_no_context_does_not_rewrite(self, paths: Paths): + src = _workspace_file(paths, "x.png", content=b"png") + uri = f"file://{src}" + result = CallToolResult( + content=[ResourceLink(type="resource_link", name="x", uri=uri, mimeType="image/png")], + isError=False, + ) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result) + + assert content[0]["url"] == uri + + def test_text_content_passthrough(self, paths: Paths): + result = CallToolResult(content=[TextContent(type="text", text="hello")], isError=False) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["type"] == "text" + assert content[0]["text"] == "hello" + + def test_image_content_passthrough(self, paths: Paths): + from mcp.types import ImageContent + + result = CallToolResult(content=[ImageContent(type="image", data="QUJD", mimeType="image/png")], isError=False) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["type"] == "image" + + def test_embedded_text_resource(self, paths: Paths): + from mcp.types import EmbeddedResource, TextResourceContents + + res = TextResourceContents(uri="mem://note.txt", text="note", mimeType="text/plain") + result = CallToolResult(content=[EmbeddedResource(type="resource", resource=res)], isError=False) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["type"] == "text" + assert content[0]["text"] == "note" + + def test_embedded_blob_image_resource(self, paths: Paths): + from mcp.types import BlobResourceContents, EmbeddedResource + + res = BlobResourceContents(uri="mem://img.png", blob="QUJD", mimeType="image/png") + result = CallToolResult(content=[EmbeddedResource(type="resource", resource=res)], isError=False) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["type"] == "image" + + def test_embedded_blob_file_resource(self, paths: Paths): + from mcp.types import BlobResourceContents, EmbeddedResource + + res = BlobResourceContents(uri="mem://doc.pdf", blob="QUJD", mimeType="application/pdf") + result = CallToolResult(content=[EmbeddedResource(type="resource", resource=res)], isError=False) + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["type"] == "file" + + def test_unknown_content_item_stringified(self, paths: Paths): + class _Weird: + def __str__(self) -> str: + return "weird-item" + + result = CallToolResult(content=[TextContent(type="text", text="x")], isError=False) + result.content = [_Weird()] # bypass pydantic validation on the union + + with _patch_paths(paths): + content, _ = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert content[0]["type"] == "text" + assert content[0]["text"] == "weird-item" + + def test_error_result_raises_tool_exception(self, paths: Paths): + from langchain_core.tools import ToolException + + result = CallToolResult(content=[TextContent(type="text", text="boom")], isError=True) + + with _patch_paths(paths), pytest.raises(ToolException, match="boom"): + mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + def test_structured_content_becomes_artifact(self, paths: Paths): + result = CallToolResult(content=[TextContent(type="text", text="ok")], structuredContent={"k": "v"}, isError=False) + + with _patch_paths(paths): + _, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1") + + assert artifact == {"structured_content": {"k": "v"}} diff --git a/backend/tests/test_mcp_session_pool.py b/backend/tests/test_mcp_session_pool.py index 852cfd861..9bc84fb10 100644 --- a/backend/tests/test_mcp_session_pool.py +++ b/backend/tests/test_mcp_session_pool.py @@ -1,6 +1,7 @@ """Tests for the MCP persistent-session pool.""" import asyncio +import stat import threading from unittest.mock import AsyncMock, MagicMock, patch @@ -258,6 +259,293 @@ async def test_session_pool_tool_wrapping(): mock_session.call_tool.assert_awaited_once_with("navigate", {"url": "https://example.com"}) +@pytest.mark.asyncio +async def test_session_pool_tool_pins_cwd_and_temp_env(tmp_path): + """Stdio MCP subprocesses should write relative and temp outputs under user-data.""" + from langchain_core.tools import StructuredTool + from pydantic import BaseModel, Field + + from deerflow.config.paths import Paths + from deerflow.mcp.tools import _MCP_TMP_SUBDIR, _make_session_pool_tool + + class Args(BaseModel): + url: str = Field(..., description="url") + + original_tool = StructuredTool( + name="playwright_navigate", + description="Navigate browser", + args_schema=Args, + coroutine=AsyncMock(), + response_format="content_and_artifact", + ) + + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=MagicMock(content=[], isError=False, structuredContent=None)) + mock_cm = MagicMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_session) + mock_cm.__aexit__ = AsyncMock(return_value=False) + + paths = Paths(tmp_path) + connection = {"transport": "stdio", "command": "pw", "args": [], "env": {"KEEP": "1"}} + mock_runtime = MagicMock() + mock_runtime.context = {"thread_id": "thread-42", "user_id": "user-7"} + mock_runtime.config = {} + + with ( + patch("deerflow.mcp.tools.get_paths", return_value=paths), + patch("langchain_mcp_adapters.sessions.create_session", return_value=mock_cm) as create_session, + ): + wrapped = _make_session_pool_tool(original_tool, "playwright", connection) + await wrapped.coroutine(runtime=mock_runtime, url="https://example.com") + + session_connection = create_session.call_args.args[0] + workspace = paths.sandbox_work_dir("thread-42", user_id="user-7") + tmp_dir = workspace / _MCP_TMP_SUBDIR + + assert session_connection["cwd"] == str(workspace) + assert session_connection["env"]["KEEP"] == "1" + assert session_connection["env"]["TMPDIR"] == str(tmp_dir) + assert session_connection["env"]["TMP"] == str(tmp_dir) + assert session_connection["env"]["TEMP"] == str(tmp_dir) + assert tmp_dir.is_dir() + assert stat.S_IMODE(tmp_dir.stat().st_mode) == 0o700 + + +@pytest.mark.asyncio +async def test_session_pool_tool_does_not_override_explicit_tmpdir(tmp_path): + """An operator-provided TMPDIR must win over our injected default.""" + from langchain_core.tools import StructuredTool + from pydantic import BaseModel, Field + + from deerflow.config.paths import Paths + from deerflow.mcp.tools import _MCP_TMP_SUBDIR, _make_session_pool_tool + + class Args(BaseModel): + url: str = Field(..., description="url") + + original_tool = StructuredTool( + name="playwright_navigate", + description="Navigate browser", + args_schema=Args, + coroutine=AsyncMock(), + response_format="content_and_artifact", + ) + + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=MagicMock(content=[], isError=False, structuredContent=None)) + mock_cm = MagicMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_session) + mock_cm.__aexit__ = AsyncMock(return_value=False) + + paths = Paths(tmp_path) + connection = {"transport": "stdio", "command": "pw", "args": [], "env": {"TMPDIR": "/operator/tmp"}} + mock_runtime = MagicMock() + mock_runtime.context = {"thread_id": "thread-42", "user_id": "user-7"} + mock_runtime.config = {} + + with ( + patch("deerflow.mcp.tools.get_paths", return_value=paths), + patch("langchain_mcp_adapters.sessions.create_session", return_value=mock_cm) as create_session, + ): + wrapped = _make_session_pool_tool(original_tool, "playwright", connection) + await wrapped.coroutine(runtime=mock_runtime, url="https://example.com") + + session_connection = create_session.call_args.args[0] + # Operator-provided TMPDIR is preserved; TMP/TEMP still get our default. + assert session_connection["env"]["TMPDIR"] == "/operator/tmp" + assert session_connection["env"]["TMP"].endswith(_MCP_TMP_SUBDIR) + + +@pytest.mark.asyncio +async def test_session_pool_tool_does_not_override_explicit_cwd(tmp_path): + """An operator-provided cwd must win over our injected workspace default.""" + from langchain_core.tools import StructuredTool + from pydantic import BaseModel, Field + + from deerflow.config.paths import Paths + from deerflow.mcp.tools import _MCP_TMP_SUBDIR, _make_session_pool_tool + + class Args(BaseModel): + url: str = Field(..., description="url") + + original_tool = StructuredTool( + name="playwright_navigate", + description="Navigate browser", + args_schema=Args, + coroutine=AsyncMock(), + response_format="content_and_artifact", + ) + + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=MagicMock(content=[], isError=False, structuredContent=None)) + mock_cm = MagicMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_session) + mock_cm.__aexit__ = AsyncMock(return_value=False) + + operator_cwd = str(tmp_path / "operator-cwd") + paths = Paths(tmp_path) + connection = {"transport": "stdio", "command": "pw", "args": [], "cwd": operator_cwd} + mock_runtime = MagicMock() + mock_runtime.context = {"thread_id": "thread-42", "user_id": "user-7"} + mock_runtime.config = {} + + with ( + patch("deerflow.mcp.tools.get_paths", return_value=paths), + patch("langchain_mcp_adapters.sessions.create_session", return_value=mock_cm) as create_session, + ): + wrapped = _make_session_pool_tool(original_tool, "playwright", connection) + await wrapped.coroutine(runtime=mock_runtime, url="https://example.com") + + session_connection = create_session.call_args.args[0] + workspace = paths.sandbox_work_dir("thread-42", user_id="user-7") + tmp_dir = workspace / _MCP_TMP_SUBDIR + + assert session_connection["cwd"] == operator_cwd + assert session_connection["env"]["TMPDIR"] == str(tmp_dir) + + +@pytest.mark.asyncio +async def test_session_pool_tool_skips_fs_work_for_non_stdio_transport(tmp_path): + """SSE/HTTP transports must not get a pinned cwd/temp env or workspace dirs.""" + from langchain_core.tools import StructuredTool + from pydantic import BaseModel, Field + + from deerflow.config.paths import Paths + from deerflow.mcp.tools import _make_session_pool_tool + + class Args(BaseModel): + url: str = Field(..., description="url") + + original_tool = StructuredTool( + name="srv_act", + description="test", + args_schema=Args, + coroutine=AsyncMock(), + response_format="content_and_artifact", + ) + + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=MagicMock(content=[], isError=False, structuredContent=None)) + mock_cm = MagicMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_session) + mock_cm.__aexit__ = AsyncMock(return_value=False) + + paths = Paths(tmp_path) + connection = {"transport": "sse", "url": "http://localhost:9000/sse", "env": {"KEEP": "1"}} + mock_runtime = MagicMock() + mock_runtime.context = {"thread_id": "thread-42", "user_id": "user-7"} + mock_runtime.config = {} + + with ( + patch("deerflow.mcp.tools.get_paths", return_value=paths) as get_paths, + patch("langchain_mcp_adapters.sessions.create_session", return_value=mock_cm) as create_session, + ): + wrapped = _make_session_pool_tool(original_tool, "srv", connection) + await wrapped.coroutine(runtime=mock_runtime, url="https://example.com") + + session_connection = create_session.call_args.args[0] + assert "cwd" not in session_connection + assert session_connection["env"] == {"KEEP": "1"} + # No filesystem work at all: get_paths() is never consulted and no thread + # workspace directory is created for non-stdio transports. + get_paths.assert_not_called() + assert not paths.sandbox_work_dir("thread-42", user_id="user-7").exists() + + +@pytest.mark.asyncio +async def test_session_pool_tool_skips_after_walk_when_no_text_content(tmp_path): + """With no text content to rewrite, the post-call snapshot diff must be skipped.""" + from langchain_core.tools import StructuredTool + from pydantic import BaseModel, Field + + from deerflow.config.paths import Paths + from deerflow.mcp.tools import _make_session_pool_tool + + class Args(BaseModel): + url: str = Field(..., description="url") + + original_tool = StructuredTool( + name="playwright_navigate", + description="Navigate browser", + args_schema=Args, + coroutine=AsyncMock(), + response_format="content_and_artifact", + ) + + # An image-only result carries no text, so bare-filename correlation has + # nothing to do and the second recursive walk should not run. + from mcp.types import ImageContent + + image_result = MagicMock(content=[ImageContent(type="image", data="QUJD", mimeType="image/png")], isError=False, structuredContent=None) + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=image_result) + mock_cm = MagicMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_session) + mock_cm.__aexit__ = AsyncMock(return_value=False) + + paths = Paths(tmp_path) + connection = {"transport": "stdio", "command": "pw", "args": []} + mock_runtime = MagicMock() + mock_runtime.context = {"thread_id": "thread-42", "user_id": "user-7"} + mock_runtime.config = {} + + with ( + patch("deerflow.mcp.tools.get_paths", return_value=paths), + patch("langchain_mcp_adapters.sessions.create_session", return_value=mock_cm), + patch("deerflow.mcp.tools._changed_workspace_files") as changed_files, + ): + wrapped = _make_session_pool_tool(original_tool, "playwright", connection) + await wrapped.coroutine(runtime=mock_runtime, url="https://example.com") + + changed_files.assert_not_called() + + +@pytest.mark.asyncio +async def test_session_pool_tool_runs_after_walk_when_text_content_present(tmp_path): + """A text result must trigger the post-call snapshot diff for path rewriting.""" + from langchain_core.tools import StructuredTool + from pydantic import BaseModel, Field + + from deerflow.config.paths import Paths + from deerflow.mcp.tools import _make_session_pool_tool + + class Args(BaseModel): + url: str = Field(..., description="url") + + original_tool = StructuredTool( + name="playwright_navigate", + description="Navigate browser", + args_schema=Args, + coroutine=AsyncMock(), + response_format="content_and_artifact", + ) + + from mcp.types import TextContent + + text_result = MagicMock(content=[TextContent(type="text", text="Saved as shot.png")], isError=False, structuredContent=None) + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=text_result) + mock_cm = MagicMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_session) + mock_cm.__aexit__ = AsyncMock(return_value=False) + + paths = Paths(tmp_path) + connection = {"transport": "stdio", "command": "pw", "args": []} + mock_runtime = MagicMock() + mock_runtime.context = {"thread_id": "thread-42", "user_id": "user-7"} + mock_runtime.config = {} + + with ( + patch("deerflow.mcp.tools.get_paths", return_value=paths), + patch("langchain_mcp_adapters.sessions.create_session", return_value=mock_cm), + patch("deerflow.mcp.tools._changed_workspace_files", return_value=[]) as changed_files, + ): + wrapped = _make_session_pool_tool(original_tool, "playwright", connection) + await wrapped.coroutine(runtime=mock_runtime, url="https://example.com") + + changed_files.assert_called_once() + + @pytest.mark.asyncio async def test_session_pool_tool_forwards_interceptor_headers(): """Regression for PR #3294: when an interceptor sets ``request.headers``, the @@ -423,8 +711,10 @@ async def test_session_pool_tool_extracts_thread_id(): await wrapped.coroutine(runtime=mock_runtime, x=1) # Verify the session was created with the correct scope key. + # The scope key is "{user_id}:{thread_id}"; the autouse fixture sets + # the effective user to "test-user-autouse". pool = get_session_pool() - assert ("server", "from-config") in pool._entries + assert ("server", "test-user-autouse:from-config") in pool._entries @pytest.mark.asyncio @@ -459,7 +749,7 @@ async def test_session_pool_tool_default_scope(): await wrapped.coroutine(runtime=None, x=1) pool = get_session_pool() - assert ("server", "default") in pool._entries + assert ("server", "test-user-autouse:default") in pool._entries @pytest.mark.asyncio @@ -499,7 +789,7 @@ async def test_session_pool_tool_get_config_fallback(): await wrapped.coroutine(runtime=None, x=1) pool = get_session_pool() - assert ("server", "from-langgraph-config") in pool._entries + assert ("server", "test-user-autouse:from-langgraph-config") in pool._entries def test_session_pool_tool_sync_wrapper_path_is_safe():