deer-flow/backend/tests/test_mcp_file_migration.py

"""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.

``ResourceLink`` conversion contract: links never become URL-sourced ``file``
blocks — Chat Completions message serialization rejects those, and a
checkpointed block would fail every later turn of the thread. ``http(s)``
image links stay image blocks; every other link becomes a text placeholder,
with the original link preserved in the artifact under ``resource_links``.
"""

import os
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.constants import MCP_TMP_SUBDIR
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, tmp_path: Path):
        src = tmp_path / "shot.png"
        assert mcp_tools._local_path_from_uri(src.as_uri()) == src

    def test_bare_absolute_path(self, tmp_path: Path):
        src = tmp_path / "data" / "out.pdf"
        assert mcp_tools._local_path_from_uri(str(src)) == src

    def test_file_uri_with_url_encoded_spaces(self, tmp_path: Path):
        src = tmp_path / "my shot.png"
        assert mcp_tools._local_path_from_uri(src.as_uri()) == src

    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_malformed_uri_is_ignored(self):
        assert mcp_tools._local_path_from_uri("//[::1/foo.png") 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, tmp_path: Path):
        # file://localhost/abs/path is the host form of file:///abs/path.
        src = tmp_path / "shot.png"
        assert mcp_tools._local_path_from_uri(src.as_uri().replace("file://", "file://localhost", 1)) == src

    def test_windows_drive_letter_path_is_resolved(self):
        # urlparse reads a Windows drive prefix ("C:/...") as the URI scheme.
        # On Windows hosts it must still resolve as a bare local path; on
        # POSIX it is not a local path at all.
        path = mcp_tools._local_path_from_uri("C:/Users/shot.png")
        if os.name == "nt":
            assert path == Path("C:/Users/shot.png")
        else:
            assert path is None

    @pytest.mark.skipif(os.name != "nt", reason="a raw '|' in a file URI path rejects with OSError only on Windows")
    def test_windows_url2pathname_oserror_is_left_untouched(self):
        assert mcp_tools._local_path_from_uri("file:///C:/tmp/a|b.png") is None

    @pytest.mark.skipif(os.name != "nt", reason="exercises the file://C:/… two-slash Windows drive URI form")
    def test_windows_two_slash_file_uri_resolves_drive(self):
        assert mcp_tools._local_path_from_uri("file://C:/Users/shot.png") == Path("C:/Users/shot.png")

    def test_remote_host_file_uri_is_ignored(self):
        assert mcp_tools._local_path_from_uri("file://example.com/a.png") is None

    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(src.as_uri(), thread_id="t1", user_id="u1")

        assert result == f"{VIRTUAL_PATH_PREFIX}/workspace/shot.png"

    @pytest.mark.skipif(os.name != "nt", reason="exercises the file:///C:/... drive-qualified URI form")
    def test_windows_file_uri_translates_to_virtual_path(self, paths: Paths):
        src = _workspace_file(paths, "shot.png")
        # The checkout and temp dirs may live on any drive, so derive the
        # expected drive instead of hardcoding "C:".
        assert src.as_uri().startswith(f"file:///{src.drive}/")

        with _patch_paths(paths):
            result = mcp_tools._local_uri_to_virtual_path(src.as_uri(), 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_new_absolute_path_with_spaces_is_rewritten(self, paths: Paths):
        workspace = paths.sandbox_work_dir("t1", user_id="u1")
        src = _workspace_file(paths, "reports/final report.txt")
        text = f'Saved as "{src}" and ready.'

        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/reports/final report.txt" and ready.'

    def test_new_relative_path_with_spaces_is_rewritten(self, paths: Paths):
        workspace = paths.sandbox_work_dir("t1", user_id="u1")
        src = _workspace_file(paths, "final reports/result.txt")

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(
                "Saved as final reports/result.txt.",
                thread_id="t1",
                user_id="u1",
                source_base_dir=workspace,
                changed_files=[src],
            )

        assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/final reports/result.txt."

    def test_new_dot_relative_path_with_spaces_is_rewritten(self, paths: Paths):
        workspace = paths.sandbox_work_dir("t1", user_id="u1")
        src = _workspace_file(paths, "final reports/result.txt")

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(
                "Saved as ./final reports/result.txt.",
                thread_id="t1",
                user_id="u1",
                source_base_dir=workspace,
                changed_files=[src],
            )

        assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/final reports/result.txt."

    def test_changed_path_with_spaces_outside_user_data_is_untouched(self, tmp_path: Path, paths: Paths):
        outside = tmp_path / "outside report.txt"
        outside.write_text("outside", encoding="utf-8")
        text = f"Saved as {outside}"

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(
                text,
                thread_id="t1",
                user_id="u1",
                changed_files=[outside],
            )

        assert result == text

    @pytest.mark.skipif(os.name == "nt", reason="exercises POSIX file URI spellings")
    def test_new_file_uri_with_literal_spaces_is_rewritten(self, paths: Paths):
        src = _workspace_file(paths, "final report.txt")
        text = f"Saved as file://{src}"

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(
                text,
                thread_id="t1",
                user_id="u1",
                changed_files=[src],
            )

        assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/final report.txt"

    def test_changed_path_with_spaces_does_not_rewrite_longer_name(self, paths: Paths):
        src = _workspace_file(paths, "final report.txt")
        text = f"Backup: {src}.bak"

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(
                text,
                thread_id="t1",
                user_id="u1",
                changed_files=[src],
            )

        assert result == text

    def test_changed_path_with_spaces_does_not_rewrite_whitespace_suffix(self, paths: Paths):
        src = _workspace_file(paths, "final report.txt")
        longer = _workspace_file(paths, "final report.txt copy")
        text = f"Backup: {longer}"

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(
                text,
                thread_id="t1",
                user_id="u1",
                changed_files=[src],
            )

        assert result == text

    def test_changed_path_with_spaces_followed_by_prose_is_untouched(self, paths: Paths):
        src = _workspace_file(paths, "final report.txt")
        text = f"Saved as {src} and ready."

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(
                text,
                thread_id="t1",
                user_id="u1",
                changed_files=[src],
            )

        assert result == text

    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"

    @pytest.mark.skipif(os.name != "nt", reason="exercises backslash drive-qualified paths in free text")
    def test_windows_backslash_drive_path_in_text_is_rewritten(self, paths: Paths):
        src = _workspace_file(paths, "shot.png")
        text = f"Saved as {src}"
        assert "\\" in text

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1")

        assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/shot.png"

    @pytest.mark.skipif(os.name != "nt", reason="exercises backslash relative paths in free text")
    def test_windows_backslash_relative_path_in_text_is_rewritten(self, paths: Paths):
        _workspace_file(paths, "temp/page.yml")
        workspace = paths.sandbox_work_dir("t1", user_id="u1")

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text("Saved as temp\\page.yml", thread_id="t1", user_id="u1", source_base_dir=workspace)

        assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/temp/page.yml"

    def test_single_slash_file_uri_is_matched_as_posix_absolute(self):
        # file:/… (single slash, as RFC 8089 and Java's File.toURI() produce)
        # must not be stolen mid-token by the drive-qualified alternative: the
        # engine has to fall through to the /… absolute alternative.
        match = mcp_tools._LOCAL_PATH_IN_TEXT_RE.search("Saved as file:/tmp/workspace/shot.png")
        assert match.group(0) == "/tmp/workspace/shot.png"

    @pytest.mark.skipif(os.name != "nt", reason="exercises the file://C:/… two-slash URI form in free text")
    def test_windows_two_slash_file_uri_in_text_is_rewritten(self, paths: Paths):
        src = _workspace_file(paths, "shot.png")
        two_slash_uri = src.as_uri().replace("file:///", "file://", 1)
        assert two_slash_uri.startswith(f"file://{src.drive}")

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(f"Saved as {two_slash_uri}", thread_id="t1", user_id="u1")

        assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/shot.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_malformed_path_like_text_is_left_untouched(self, paths: Paths):
        text = "Saved at //[::1/foo.png"

        with _patch_paths(paths):
            result = mcp_tools._rewrite_local_paths_in_text(text, thread_id="t1", user_id="u1")

        assert result == text

    def test_oversized_path_like_text_is_left_untouched(self, paths: Paths):
        workspace = paths.sandbox_work_dir("t1", user_id="u1")
        text = f"手术室/重症监护室（OR/ICU）整体解决方案{'说明' * 200}"

        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_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."

    @pytest.mark.parametrize(
        "other_path",
        [
            r"C:\outside\page.yml",
            pytest.param(
                r"\\server\share\page.yml",
                marks=pytest.mark.skipif(os.name == "nt", reason="resolving a UNC path on Windows can contact an SMB server"),
            ),
            r"missing\page.yml",
            r"page.yml\inner.txt",
            "missing/page.yml",
            "page.yml/inner.txt",
        ],
    )
    def test_bare_filename_does_not_rewrite_path_segments(self, paths: Paths, other_path: str):
        workspace = paths.sandbox_work_dir("t1", user_id="u1")
        src = _workspace_file(paths, "page.yml")
        text = f"Saved as page.yml. Other path: `{other_path}`."

        with _patch_paths(paths):
            content, _ = mcp_tools._convert_call_tool_result(
                CallToolResult(content=[TextContent(type="text", text=text)]),
                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.yml. Other path: `{other_path}`."

    def test_bare_filename_before_markdown_hard_break_is_left_untouched(self, paths: Paths):
        """A trailing backslash before a newline is treated as a path separator.

        That position is more often a Markdown hard line break, so a genuine bare
        filename there keeps its literal spelling instead of becoming a virtual
        path. A backslash is also a legal filename character on POSIX, so this
        forgoes one rewrite rather than risk corrupting a real path reference.
        """
        workspace = paths.sandbox_work_dir("t1", user_id="u1")
        src = _workspace_file(paths, "page.yml")
        text = "Saved as page.yml\\\nnext line"

        with _patch_paths(paths):
            content, _ = mcp_tools._convert_call_tool_result(
                CallToolResult(content=[TextContent(type="text", text=text)]),
                thread_id="t1",
                user_id="u1",
                source_base_dir=workspace,
                changed_files=[src],
            )

        assert content[0]["text"] == text

    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


@pytest.mark.skipif(os.name == "nt", reason="a literal backslash in a filename is POSIX-only")
class TestRewriteUniqueBareFilenames:
    """The correlated virtual path must be inserted verbatim, never as a template.

    The replacement is built from the real file's relative path, where a
    backslash is an ordinary character, so handing it to ``re.sub`` as a
    template reads it as a regex escape instead.
    """

    def test_backslash_in_replacement_is_inserted_literally(self, paths: Paths):
        workspace = paths.sandbox_work_dir("t1", user_id="u1")
        src = _workspace_file(paths, r"screenshots\raw.png")
        text = r"Saved as screenshots\raw.png"

        with _patch_paths(paths):
            result = mcp_tools._rewrite_unique_bare_filenames(
                text,
                changed_files=[src],
                thread_id="t1",
                user_id="u1",
                source_base_dir=workspace,
            )

        assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/screenshots\\raw.png"

    def test_unknown_regex_escape_in_replacement_does_not_raise(self, paths: Paths):
        workspace = paths.sandbox_work_dir("t1", user_id="u1")
        src = _workspace_file(paths, r"screenshots\q3.png")
        text = r"Saved as screenshots\q3.png"

        with _patch_paths(paths):
            result = mcp_tools._rewrite_unique_bare_filenames(
                text,
                changed_files=[src],
                thread_id="t1",
                user_id="u1",
                source_base_dir=workspace,
            )

        assert result == f"Saved as {VIRTUAL_PATH_PREFIX}/workspace/screenshots\\q3.png"


class TestWorkspaceSnapshots:
    def test_changed_workspace_files_detects_created_and_modified_files(self, paths: Paths):
        import time

        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)

        # Ensure the mtime advances so the change is detectable.  Without the
        # sleep, write_bytes(b"new") may land in the same nanosecond as the
        # snapshot, and since b"old" and b"new" have the same length, the
        # (mtime_ns, size) signature stays identical → _changed_workspace_files
        # misses the modification.
        time.sleep(0.05)
        existing.write_bytes(b"new_content")  # different length guarantees size change too
        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_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_local_image_resource_link_becomes_text_placeholder(self, paths: Paths):
        src = _workspace_file(paths, "page.png", content=b"png")
        result = CallToolResult(
            content=[ResourceLink(type="resource_link", name="page", uri=src.as_uri(), mimeType="image/png")],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        # Local images are downgraded too: the provider cannot fetch virtual
        # paths — the view-image tool is the intended path for local images.
        virtual = f"{VIRTUAL_PATH_PREFIX}/workspace/page.png"
        assert content[0]["type"] == "text"
        assert content[0]["text"] == f"[Resource: page (image/png) available at {virtual}]"
        assert artifact == {"resource_links": [{"name": "page", "uri": virtual, "mime_type": "image/png"}]}

    def test_local_file_resource_link_becomes_text_placeholder(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=src.as_uri(), mimeType="application/pdf")],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        virtual = f"{VIRTUAL_PATH_PREFIX}/outputs/doc.pdf"
        assert content[0]["type"] == "text"
        assert content[0]["text"] == f"[Resource: doc (application/pdf) available at {virtual}]"
        assert artifact == {"resource_links": [{"name": "doc", "uri": virtual, "mime_type": "application/pdf"}]}

    def test_resource_link_outside_user_data_becomes_text_placeholder(self, tmp_path: Path, paths: Paths):
        src = tmp_path / "page.png"
        src.write_bytes(b"png")
        uri = src.as_uri()
        result = CallToolResult(
            content=[ResourceLink(type="resource_link", name="page", uri=uri, mimeType="image/png")],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        # The URI stays unresolved outside the user-data tree, but the link is
        # still downgraded — a ``file://`` URL block is equally fatal. The raw
        # host path is withheld from model-visible text (US-16); it survives
        # only in the artifact channel, which the model never sees.
        assert content[0]["type"] == "text"
        assert content[0]["text"] == "[Resource: page (image/png)]"
        assert all("file://" not in block.get("text", "") for block in content)
        assert artifact == {"resource_links": [{"name": "page", "uri": uri, "mime_type": "image/png"}]}

    def test_remote_image_resource_link_stays_image_block(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, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        assert content[0]["type"] == "image"
        assert content[0]["url"] == url
        assert artifact is None

    def test_remote_image_resource_link_with_uppercase_scheme_stays_image_block(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, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        # Scheme matching is case-insensitive: the link stays an image block,
        # not a text placeholder. The block's URL validation normalizes the
        # scheme to lowercase; the passthrough itself is what matters.
        assert content[0]["type"] == "image"
        assert content[0]["url"] == "https://example.com/remote.png"
        assert artifact is None

    def test_ui_resource_link_becomes_text_placeholder(self, paths: Paths):
        # Regression guard: an MCP Apps UI card must not produce any block the
        # Chat Completions translator rejects (any URL-sourced file block).
        url = "ui://app/card.html"
        mime = "text/html;profile=mcp-app"
        result = CallToolResult(
            content=[ResourceLink(type="resource_link", name="card", uri=url, mimeType=mime)],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        assert content[0]["type"] == "text"
        assert content[0]["text"] == f"[Resource: card ({mime}) available at {url}]"
        assert all(not (block.get("type") == "file" and "url" in block) for block in content)
        assert artifact == {"resource_links": [{"name": "card", "uri": url, "mime_type": mime}]}

    def test_remote_non_image_resource_link_becomes_text_placeholder(self, paths: Paths):
        url = "https://example.com/report.pdf"
        result = CallToolResult(
            content=[ResourceLink(type="resource_link", name="report", uri=url, mimeType="application/pdf")],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        assert content[0]["type"] == "text"
        assert content[0]["text"] == f"[Resource: report (application/pdf) available at {url}]"
        assert artifact == {"resource_links": [{"name": "report", "uri": url, "mime_type": "application/pdf"}]}

    def test_unknown_scheme_resource_link_without_name_or_mime_uses_fallbacks(self, paths: Paths):
        url = "s3://bucket/report.pdf"
        result = CallToolResult(
            content=[ResourceLink(type="resource_link", name="", uri=url, mimeType=None)],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        assert content[0]["type"] == "text"
        assert content[0]["text"] == f"[Resource: unnamed (unknown type) available at {url}]"
        assert artifact == {"resource_links": [{"name": "", "uri": url, "mime_type": None}]}

    def test_data_uri_resource_link_never_enters_model_text_or_state(self, paths: Paths):
        payload = "QUFB" + "A" * 500
        url = f"data:application/pdf;base64,{payload}"
        result = CallToolResult(
            content=[
                ResourceLink(type="resource_link", name="report", uri=url, mimeType="application/pdf"),
                ResourceLink(type="resource_link", name="", uri="data:image/png;base64,QUJD", mimeType=None),
            ],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        # The URI embeds the whole payload: it must not be inlined into the
        # model-visible text nor checkpointed inside resource_links. The
        # placeholder is the canonical location-less form.
        assert content[0]["type"] == "text"
        assert content[0]["text"] == "[Resource: report (application/pdf)]"
        assert content[1]["type"] == "text"
        assert content[1]["text"] == "[Resource: unnamed (unknown type)]"
        assert all(payload not in block.get("text", "") for block in content)
        assert all("data:" not in block.get("text", "") for block in content)
        assert artifact is None

    def test_image_mime_data_uri_resource_link_is_downgraded(self, paths: Paths):
        """An ``image/*`` ``data:`` link is downgraded too, by design: the
        conversion layer keeps new inline payloads out of state, even though
        the read-time middleware lets persisted ``data:`` image blocks pass."""
        result = CallToolResult(
            content=[ResourceLink(type="resource_link", name="shot", uri="data:image/png;base64,QUJD", mimeType="image/png")],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        assert content[0]["type"] == "text"
        assert content[0]["text"] == "[Resource: shot (image/png)]"
        assert all("data:" not in block.get("text", "") for block in content)
        assert artifact is None

    def test_blob_uri_resource_link_never_enters_model_text_or_state(self, paths: Paths):
        url = "blob:https://example.com/550e8400-e29b-41d4-a716-446655440000"
        result = CallToolResult(
            content=[ResourceLink(type="resource_link", name="report", uri=url, mimeType="application/pdf")],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        # A ``blob:`` URI names a browser-local object nothing else can
        # dereference: like ``data:`` it never enters checkpointed state, and
        # the placeholder carries no location segment.
        assert content[0]["type"] == "text"
        assert content[0]["text"] == "[Resource: report (application/pdf)]"
        assert all("blob:" not in block.get("text", "") for block in content)
        assert artifact is None

    def test_resource_links_merge_with_structured_content_artifact(self, paths: Paths):
        url = "https://example.com/report.pdf"
        result = CallToolResult(
            content=[
                TextContent(type="text", text="done"),
                ResourceLink(type="resource_link", name="report", uri=url, mimeType="application/pdf"),
            ],
            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"},
            "resource_links": [{"name": "report", "uri": url, "mime_type": "application/pdf"}],
        }

    def test_multiple_resource_links_preserve_order(self, paths: Paths):
        first = "https://example.com/a.pdf"
        second = "ui://app/card.html"
        result = CallToolResult(
            content=[
                ResourceLink(type="resource_link", name="a", uri=first, mimeType="application/pdf"),
                ResourceLink(type="resource_link", name="b", uri=second, mimeType="text/html"),
            ],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        assert [block["type"] for block in content] == ["text", "text"]
        assert content[0]["text"] == f"[Resource: a (application/pdf) available at {first}]"
        assert content[1]["text"] == f"[Resource: b (text/html) available at {second}]"
        assert artifact == {
            "resource_links": [
                {"name": "a", "uri": first, "mime_type": "application/pdf"},
                {"name": "b", "uri": second, "mime_type": "text/html"},
            ]
        }

    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_text_path_with_spaces_rewritten_from_changed_files(self, paths: Paths):
        workspace = paths.sandbox_work_dir("t1", user_id="u1")
        src = _workspace_file(paths, "report with spaces.txt")
        result = CallToolResult(content=[TextContent(type="text", text=f"Saved as {src}")], 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/report with spaces.txt"

    def test_no_context_resource_link_still_becomes_text_placeholder(self, paths: Paths):
        src = _workspace_file(paths, "x.png", content=b"png")
        uri = src.as_uri()
        result = CallToolResult(
            content=[ResourceLink(type="resource_link", name="x", uri=uri, mimeType="image/png")],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result)

        # Without thread context the URI is not virtualized, but the downgrade
        # still applies — a ``file://`` URL image block is not fetchable. The
        # host path stays out of model-visible text (US-16).
        assert content[0]["type"] == "text"
        assert content[0]["text"] == "[Resource: x (image/png)]"
        assert all("file://" not in block.get("text", "") for block in content)
        assert artifact == {"resource_links": [{"name": "x", "uri": uri, "mime_type": "image/png"}]}

    def test_windows_drive_resource_link_omits_host_path_from_placeholder(self, paths: Paths):
        uri = "C:\\Users\\shots\\page.png"
        result = CallToolResult(
            content=[ResourceLink(type="resource_link", name="page", uri=uri, mimeType="image/png")],
            isError=False,
        )

        with _patch_paths(paths):
            content, artifact = mcp_tools._convert_call_tool_result(result, thread_id="t1", user_id="u1")

        # urlparse reads the drive prefix as a single-letter scheme; a bare
        # Windows host path must not reach model-visible text either.
        assert content[0]["type"] == "text"
        assert content[0]["text"] == "[Resource: page (image/png)]"
        assert all(":\\" not in block.get("text", "") for block in content)
        assert artifact == {"resource_links": [{"name": "page", "uri": "c:\\Users\\shots\\page.png", "mime_type": "image/png"}]}

    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_malformed_path_like_text_result_does_not_raise(self, paths: Paths):
        result = CallToolResult(content=[TextContent(type="text", text="Saved at //[::1/foo.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"] == "text"
        assert content[0]["text"] == "Saved at //[::1/foo.png"

    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"}}