fix(mcp): preserve remote URLs during local file rewriting (#6878)

* fix(mcp): preserve remote URLs during local file rewriting

* fix(mcp): honor escaped delimiters in remote URLs

README.md

@@ -865,7 +865,7 @@ Missing, malformed, or out-of-range token response `expires_in` values use a one
 Durable HTTP/SSE task status and cancellation calls select configured `user_auth` credentials using the persisted task owner, including after restart; per-request secrets are not retained for background calls. If a request-scoped credential overrides submit authentication, both credentials must authorize access to the same remote task.
 For stdio MCP servers, per-tool call timeouts can be configured with `tool_call_timeout`; durable background-task calls honor the same setting for HTTP/SSE servers as well.
 The persistent stdio session pool evicts only idle sessions at capacity. If every pooled session is executing a tool, a new session request fails with a capacity error and can be retried after a call finishes.
-For stdio file outputs, a bare filename is linked to a uniquely matching file created or changed by that call. Filenames embedded in unrelated paths, including Windows backslash paths, are left intact.
+For stdio file outputs, a bare filename is linked to a uniquely matching file created or changed by that call. Filenames embedded in unrelated paths, including Windows backslash paths, are left intact. Explicit remote URLs (`scheme://...`) retain their original query parameters and fragments even when these contain a generated file's name or path, including URLs with backslash-escaped quotes or markup delimiters; local references outside the URL still translate normally.
 For HTTP/SSE background-task calls, `session_init_timeout` separately bounds connection setup (including the SSE endpoint event) and MCP initialization together; it stops applying once the tool call begins. Initialization deadline errors identify the server and configured time limit.
 Ordinary `task` subagents retain the parent run's captured thread incarnation for MCP calls, including legacy threads, so delegation preserves the same lifecycle scope.
 MCP tool names are prefixed with `<server_name>_` by default to prevent collisions across servers. If a server already namespaces its own tools, set `tool_name_prefix: false` on that server in `extensions_config.json` to keep the original names. Disable the prefix only when the resulting names remain unique across all enabled servers.

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

@@ -58,7 +58,7 @@ The optional background-task `status` filter flows through `McpTaskSubmitter.lis
 - **Stdio file outputs**: Persistent stdio sessions use the canonical versioned user/thread/incarnation scope (legacy NULL incarnations use the exact `user_id:thread_id` key). 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. `.mcp` is a DeerFlow-owned internal namespace: its temporary/debug files remain addressable when returned by a tool but are excluded from run workspace-change summaries — by directory name at any depth, consistent with the other reserved names in `EXCLUDED_DIR_NAMES` (`.git`, `node_modules`, …) and robust if a server ever creates a relative `.mcp` from a different cwd. Both launch paths pin it at the workspace root today. SSE/HTTP transports skip this filesystem prep entirely.
 - **Configured stdio working directory**: A non-empty `McpServerConfig.cwd` is forwarded by `build_server_params` for discovery, pooled Agent calls, and durable task calls. Omit the connection key when the setting is `None` or empty (including unresolved environment references), so the runtime's thread-workspace default still applies; HTTP/SSE connections must not receive it. `tests/test_mcp_cwd.py` exercises the JSON-config-to-subprocess path with relative entrypoints, relative tool inputs, and discovery/pool defaults for omitted, null, empty, and unresolved values.
 - **Stdio disconnect recovery**: Ordinary Agent tool calls and durable task submit/status/cancel calls that receive the MCP SDK's explicit `Connection closed` error or an AnyIO closed-stream error evict only that `(ownership domain, server_name, incarnation scope)` session when the registered entry is still the same `ClientSession` that failed. A late error from an old concurrent call cannot evict its replacement or a new in-flight creation. The failing call still surfaces its original error and is never replayed automatically; a later retry creates a fresh subprocess/session. Protocol timeouts, normal `isError=true` tool results, and interceptor failures do not evict a healthy stateful session.
-- **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. Bare-filename correlation must not replace a segment of a longer path: both `/` and `\` are separators at either boundary, even when the full-path pass leaves the reference unresolved. A trailing `\` immediately before a newline therefore counts as a separator too; that position is more often a Markdown hard break, but a backslash is also a legal filename character on POSIX, so forgoing that single rewrite is the conservative tradeoff. `tests/test_mcp_file_migration.py` covers this through the result converter with real workspace files.
+- **Stdio path translation**: Resolve local `ResourceLink`/free-text references only to existing files inside the thread's user-data tree; translate to `/mnt/user-data/...`, never copy or rewrite outside files. Bare-filename correlation must respect `/` and `\` at both boundaries, even after an unresolved full-path pass. Treat `\` before newline as a separator despite Markdown hard-break ambiguity: POSIX permits literal backslashes. Preserve explicit non-file `scheme://` URLs in every pass, including query/fragment paths and filenames, but not adjacent Markdown destinations or quoted fields. Escaped delimiters stay within URLs; whitespace always ends them. `file:` and Windows drive paths stay local. `tests/test_mcp_file_migration.py` uses real files and embedded text resources.
 - **ResourceLink conversion contract**: `_convert_call_tool_result` never emits URL-sourced `file` blocks from `ResourceLink` items — Chat Completions serialization rejects them (langchain-core), and a checkpointed block would fail every later turn. An `http(s)` `image/*` link stays an image block; `data:`/`blob:` URIs are non-referenceable (one embeds its payload, the other names a browser-local object), so they — `image/*` `data:` included — become a location-less `[Resource: name (mime)]` placeholder and stay out of `resource_links` (new inline payloads never enter state; the read-time middleware only preserves persisted `data:` image blocks). Everything else (`ui://` cards, virtual paths, remote non-image links, unknown schemes) becomes a `[Resource: name (mime) available at url]` placeholder, with the link preserved as `{"name", "uri", "mime_type"}` under the artifact's `resource_links` key. Placeholders come from the shared `resource_placeholder_text` formatter, the location gated by the shared `model_visible_location` (both in `deerflow.tools.resource_placeholder`, shared with the read-time middleware): the exact virtual root or a slash-delimited descendant and remote schemes keep the location (a sibling like `/mnt/user-data-backups/...` shares the prefix but is a host path); a raw `file://` URI, bare/Windows host path, or `data:`/`blob:` URI is withheld — the unresolved URI still lands in the artifact channel the model never sees. `tools/artifact_registry.py::extract_artifacts_from_result` consumes `resource_links` through the same referenceability gate. Embedded base64 payloads, text rewriting, and the error-to-`ToolException` path are unchanged.
 - **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. The five Gateway mutation paths validate the frozen durable-task snapshot before `atomic_write_extensions_config()` and return 409 on rejection. Runtime updates are picked up by the writer's committed `prepare_mcp_reconciliation()` handoff (or a conservative reset on handoff failure); another worker still picks them up through the two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which applies the same classification (selective retirement of changed bindings, or the conservative full retirement), voids an in-flight initialization whose revision was superseded, and performs no tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) additionally calls `publish_mcp_tools_cache_reset()`: it writes a random generation to a hidden JSON marker beside the resolved config under the same cross-process lock (`MCP_CACHE_RESET_MARKER`, a `deerflow.config.shared_reset_marker.SharedResetMarker` shared with the skills prompt-cache reset), then retires local state. Every worker mounting that directory records the marker signature when tools publish and invalidates on a later signature, so a remote `tools/list` change converges without a fake config edit; marker drift during discovery fences the stale result. The response reports this transport as `scope=shared_config` rather than claiming an implicit broadcast to replicas with independent filesystems. No resolvable config path keeps the legacy process-local behavior and is reported as `scope=process`.
 - **Stdio launch policy at the HTTP boundary** (`routers/mcp.py::_validate_mcp_update_request`, shared by `PUT` and the enable branch of `PATCH`): a config file may express anything, but the API is untrusted input, so an API-registered stdio server must (a) name a bare executable from the allowlist — `_DEFAULT_MCP_STDIO_COMMAND_ALLOWLIST` = `{npx, uvx}`, extended by `DEER_FLOW_MCP_STDIO_COMMAND_ALLOWLIST`, with path separators, whitespace, and shell metacharacters rejected in `command`; (b) carry no `args` flag in `_ARBITRARY_EXEC_ARGS`; and (c) set no `env` name in `_CODE_INJECTING_ENV_VARS`. Checks (b) and (c) exist because the command check alone names a binary without constraining what that binary runs. The `env` denylist applies to **every** allowlisted command, and both denylists match `--flag=value` as well as `--flag value`. The `args` denylist's **scope depends on the command**, because where a launcher stops parsing its own flags is what decides whether a token is an exec flag at all:

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

@@ -5,7 +5,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import re
-from collections.abc import Collection, Iterable, Mapping
+from collections.abc import Callable, Collection, Iterable, Mapping
 from dataclasses import dataclass, field
 from datetime import timedelta
 from pathlib import Path
@@ -66,7 +66,8 @@ _VALID_MCP_TOOL_NAME = re.compile(r"^[A-Za-z0-9_-]+$")
 # 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).
+# user-data tree. URL spans are excluded before applying the path matchers so
+# a URL query/fragment cannot be mistaken for a generated local file.
 _LOCAL_PATH_IN_TEXT_RE = re.compile(
     r"(?:file://)?/[^\s'\"<>|*?]+"  # POSIX absolute path or file:// URI
     r"|file://[A-Za-z]:[^\s'\"<>|*?]+"  # file://C:/… — some Windows tools skip the third slash
@@ -77,6 +78,10 @@ _LOCAL_PATH_IN_TEXT_RE = re.compile(
     r"|(?:\.{0,2}[\\/]|[\w.-]+[\\/])[^\s'\"<>|*?]+"
 )
 
+# Require :// so labels such as "Saved:temp/page.yml" remain local references.
+# At least two scheme characters leave Windows drive paths to local handling.
+_URL_SCHEME_IN_TEXT_RE = re.compile(r"(?<![\w.+/\\-])(?P<scheme>[A-Za-z][A-Za-z0-9+.-]+)://")
+
 # Trailing characters that are punctuation/markup rather than part of a path.
 _TEXT_PATH_TRAILING_CHARS = ".,;:!?)]}>\"'`"
 
@@ -249,6 +254,51 @@ def _result_has_text_content(call_tool_result: Any) -> bool:
     return False
 
 
+def _sub_outside_remote_urls(pattern: re.Pattern[str], replacement: Callable[[re.Match[str]], str], text: str) -> tuple[str, int]:
+    """Apply a local-file substitution outside explicit non-file URLs."""
+    parts: list[str] = []
+    previous_end = 0
+    count = 0
+    search_from = 0
+    while match := _URL_SCHEME_IN_TEXT_RE.search(text, search_from):
+        end = match.end()
+        prefix_start = match.start()
+        single_quoted = prefix_start > 0 and text[prefix_start - 1] == "'"
+        while prefix_start > 0 and text[prefix_start - 1].isspace():
+            prefix_start -= 1
+        in_markdown_link = text[max(0, prefix_start - 2) : prefix_start] == "]("
+        depth = 0
+        escaped = False
+        while end < len(text):
+            char = text[end]
+            if char.isspace() or (char in '<>"`|' and not escaped):
+                break
+            # An apostrophe is a delimiter only for a single-quoted URL.
+            if single_quoted and char == "'" and not escaped:
+                break
+            # Stop at the Markdown destination's closing paren, not at balanced
+            # or escaped parentheses inside the URL. Adjacent links stay local.
+            if in_markdown_link and not escaped:
+                if char == "(":
+                    depth += 1
+                elif char == ")":
+                    if depth == 0:
+                        break
+                    depth -= 1
+            escaped = char == "\\" and not escaped
+            end += 1
+        search_from = end
+        if match.group("scheme").lower() == "file":
+            continue
+        rewritten, substitutions = pattern.subn(replacement, text[previous_end : match.start()])
+        parts.extend((rewritten, text[match.start() : end]))
+        count += substitutions
+        previous_end = end
+    rewritten, substitutions = pattern.subn(replacement, text[previous_end:])
+    parts.append(rewritten)
+    return "".join(parts), count + substitutions
+
+
 def _rewrite_unique_bare_filenames(
     text: str,
     *,
@@ -292,7 +342,7 @@ def _rewrite_unique_bare_filenames(
         # A callable replacement, not a template: the virtual path is built from
         # the real file's relative path, where a backslash is an ordinary
         # character, so it must never be read as a regex escape.
-        rewritten_text, count = pattern.subn(lambda _match: unique[name], rewritten)
+        rewritten_text, count = _sub_outside_remote_urls(pattern, lambda _match: unique[name], rewritten)
         if count:
             logger.debug("MCP bare filename rewrite: %s -> %s", name, unique[name])
         rewritten = rewritten_text
@@ -339,7 +389,7 @@ def _rewrite_changed_paths_with_spaces(
         # Require punctuation or the end of the text instead of rewriting a prefix.
         pattern = re.compile(rf"(?<![\w./\\-]){re.escape(spelling)}(?!(?:[\w\s/\\-]|\.[\w]))")
         replacement = next(iter(destinations))
-        rewritten = pattern.sub(lambda _match: replacement, rewritten)
+        rewritten, _ = _sub_outside_remote_urls(pattern, lambda _match: replacement, rewritten)
     return rewritten
 
 
@@ -360,7 +410,8 @@ def _rewrite_local_paths_in_text(
     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.
+    were. Explicit non-file URLs are preserved in every pass, even if a query or
+    fragment happens to contain a path or filename that exists locally.
     """
     translated_by_source: dict[str, str | None] = {}
 
@@ -392,7 +443,7 @@ def _rewrite_local_paths_in_text(
             source_base_dir=source_base_dir,
         )
 
-    rewritten = _LOCAL_PATH_IN_TEXT_RE.sub(_replace, text)
+    rewritten, _ = _sub_outside_remote_urls(_LOCAL_PATH_IN_TEXT_RE, _replace, text)
     if changed_files is None:
         return rewritten
     return _rewrite_unique_bare_filenames(

backend/tests/test_mcp_file_migration.py

@@ -589,6 +589,288 @@ class TestRewriteLocalPathsInText:
 
         assert result == text
 
+    @pytest.mark.parametrize("filename", ["report.pdf", "report with spaces.pdf"])
+    @pytest.mark.parametrize("spelling", ["relative", "bare", "absolute", "file_uri"])
+    def test_colon_label_does_not_hide_local_file_reference(self, paths: Paths, filename: str, spelling: str):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, f"temp/{filename}")
+        reference = {"relative": f"temp/{filename}", "bare": filename, "absolute": str(src), "file_uri": src.as_uri()}[spelling]
+
+        with _patch_paths(paths):
+            content, _ = mcp_tools._convert_call_tool_result(
+                CallToolResult(content=[TextContent(type="text", text=f"Saved:{reference}")]),
+                thread_id="t1",
+                user_id="u1",
+                source_base_dir=workspace,
+                changed_files=[src],
+            )
+
+        assert content[0]["text"] == f"Saved:{VIRTUAL_PATH_PREFIX}/workspace/temp/{filename}"
+
+    @pytest.mark.parametrize("separator", ["|", ""])
+    @pytest.mark.parametrize("leading_whitespace", ["", "  ", "\n"])
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "https://example.com/download",
+            "https://example.com/download?filename=temp/report.pdf",
+            "https://example.com/download?filter=(original(nested))&filename=report.pdf",
+            r"https://example.com/download?filter=original\)&filename=report.pdf",
+        ],
+    )
+    def test_remote_markdown_link_does_not_hide_adjacent_local_link(self, paths: Paths, url: str, separator: str, leading_whitespace: str):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "temp/report.pdf")
+        text = f"[remote]({leading_whitespace}{url}){separator}[local](temp/report.pdf)"
+
+        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"[remote]({leading_whitespace}{url}){separator}[local]({VIRTUAL_PATH_PREFIX}/workspace/temp/report.pdf)"
+
+    @pytest.mark.parametrize("quote", ["'", '"'])
+    @pytest.mark.parametrize("query", ["filename=report.pdf", "label=owner's&filename=report.pdf"])
+    def test_quoted_remote_url_does_not_hide_local_file_reference(self, paths: Paths, quote: str, query: str):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "temp/report.pdf")
+        url = f"https://example.com/download?{query}".replace(quote, f"\\{quote}")
+        remote_field = f"{quote}url{quote}:{quote}{url}{quote}"
+        text = f"{{{remote_field},{quote}file{quote}:{quote}temp/report.pdf{quote}}}"
+
+        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"{{{remote_field},{quote}file{quote}:{quote}{VIRTUAL_PATH_PREFIX}/workspace/temp/report.pdf{quote}}}"
+
+    @pytest.mark.parametrize(
+        "url, markup",
+        [
+            (r"https://example.com/download?label=\"q\"&filename=report.pdf", '{{"url":"{}","file":"temp/report.pdf"}}'),
+            (r"https://example.com/a\|b?filename=report.pdf", "| {} | temp/report.pdf |"),
+        ],
+    )
+    def test_escaped_remote_url_delimiters_do_not_hide_local_file_reference(self, paths: Paths, url: str, markup: str):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "temp/report.pdf")
+
+        with _patch_paths(paths):
+            content, _ = mcp_tools._convert_call_tool_result(
+                CallToolResult(content=[TextContent(type="text", text=markup.format(url))]),
+                thread_id="t1",
+                user_id="u1",
+                source_base_dir=workspace,
+                changed_files=[src],
+            )
+
+        assert content[0]["text"] == markup.replace("temp/report.pdf", f"{VIRTUAL_PATH_PREFIX}/workspace/temp/report.pdf").format(url)
+
+    @pytest.mark.parametrize("delimiter", ['"', "<", ">", "`", "|"])
+    @pytest.mark.parametrize("backslash_count", [0, 1, 2, 3])
+    def test_remote_url_delimiter_respects_escape_parity(self, paths: Paths, delimiter: str, backslash_count: int):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "temp/report.pdf")
+        prefix = "https://example.com/download" + "\\" * backslash_count + delimiter
+
+        with _patch_paths(paths):
+            content, _ = mcp_tools._convert_call_tool_result(
+                CallToolResult(content=[TextContent(type="text", text=f"{prefix}report.pdf")]),
+                thread_id="t1",
+                user_id="u1",
+                source_base_dir=workspace,
+                changed_files=[src],
+            )
+
+        expected_reference = "report.pdf" if backslash_count % 2 else f"{VIRTUAL_PATH_PREFIX}/workspace/temp/report.pdf"
+        assert content[0]["text"] == f"{prefix}{expected_reference}"
+
+    @pytest.mark.parametrize("whitespace", [" ", "\t", "\n"])
+    def test_remote_url_whitespace_boundary_is_not_escaped(self, paths: Paths, whitespace: str):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "temp/report.pdf")
+        prefix = f"https://example.com/download\\{whitespace}"
+
+        with _patch_paths(paths):
+            content, _ = mcp_tools._convert_call_tool_result(
+                CallToolResult(content=[TextContent(type="text", text=f"{prefix}report.pdf")]),
+                thread_id="t1",
+                user_id="u1",
+                source_base_dir=workspace,
+                changed_files=[src],
+            )
+
+        assert content[0]["text"] == f"{prefix}{VIRTUAL_PATH_PREFIX}/workspace/temp/report.pdf"
+
+    def test_adjacent_remote_markdown_links_preserve_each_destination(self, paths: Paths):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "temp/report.pdf")
+        links = "[source](https://example.com/download?filename=report.pdf)[mirror](https://example.org/download?filename=temp/report.pdf)"
+        text = f"{links}|[local](temp/report.pdf)"
+
+        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"{links}|[local]({VIRTUAL_PATH_PREFIX}/workspace/temp/report.pdf)"
+
+    def test_remote_markdown_link_does_not_hide_adjacent_spaced_local_path(self, paths: Paths):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "final reports/report.pdf")
+        link = "[remote](https://example.com/download?filename=final%20reports/report.pdf)"
+        text = f"{link}|[local](final reports/report.pdf)"
+
+        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"{link}|[local]({VIRTUAL_PATH_PREFIX}/workspace/final reports/report.pdf)"
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "https://example.com/download?filename=report.pdf",
+            "https://example.com/download#report.pdf",
+            "https://example.com/download?filename=temp/report.pdf",
+            "https://example.com/download?filename=temp/report.pdf&signature=unchanged",
+            "https://example.com/download#temp/report.pdf",
+            "https://example.com/download?filter=(original)&filename=report.pdf",
+            "https://example.com/download?label=owner's&filename=report.pdf",
+            "https://example.com/download?label=owner's&filename=temp/report.pdf",
+            "HTTPS://example.com/download?filename=report.pdf",
+            "s3://example-bucket/download?filename=report.pdf",
+            "blob:https://example.com/report.pdf?filename=report.pdf",
+        ],
+    )
+    @pytest.mark.parametrize("markup", ["{}", "[download]({})", "<{}>", "`{}`"])
+    def test_remote_url_is_not_rewritten_as_a_generated_file(self, paths: Paths, url: str, markup: str):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "temp/report.pdf")
+        link = markup.format(url)
+        text = f"Source: {link}\nSaved as report.pdf. Local copy: [download](temp/report.pdf)"
+
+        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],
+            )
+
+        virtual_path = f"{VIRTUAL_PATH_PREFIX}/workspace/temp/report.pdf"
+        assert content[0]["text"] == f"Source: {link}\nSaved as {virtual_path}. Local copy: [download]({virtual_path})"
+
+    def test_remote_url_query_is_preserved_without_changed_files(self, paths: Paths):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        _workspace_file(paths, "temp/report.pdf")
+        url = "https://example.com/download?filename=temp/report.pdf"
+        text = f"Source: {url}\nSaved as temp/report.pdf"
+
+        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,
+            )
+
+        assert content[0]["text"] == f"Source: {url}\nSaved as {VIRTUAL_PATH_PREFIX}/workspace/temp/report.pdf"
+
+    def test_multiple_remote_urls_preserve_nested_file_uri_queries(self, paths: Paths):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "report.pdf")
+        first_url = f"https://example.com/download?source={src.as_uri()}"
+        second_url = "https://example.org/download?filename=report.pdf"
+        text = f"{first_url}\nSaved as report.pdf; mirror: <{second_url}>\nLocal copy: {src.as_uri()}"
+
+        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],
+            )
+
+        virtual_path = f"{VIRTUAL_PATH_PREFIX}/workspace/report.pdf"
+        assert content[0]["text"] == f"{first_url}\nSaved as {virtual_path}; mirror: <{second_url}>\nLocal copy: {virtual_path}"
+
+    def test_remote_url_keeps_a_path_prefix_containing_encoded_spaces(self, paths: Paths):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        spaced = _workspace_file(paths, "final reports/report.pdf")
+        url = "https://example.com/download?filename=final%20reports/report.pdf"
+        text = f"Source: {url}\nSaved as final reports/report.pdf."
+
+        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=[spaced],
+            )
+
+        assert content[0]["text"] == f"Source: {url}\nSaved as {VIRTUAL_PATH_PREFIX}/workspace/final reports/report.pdf."
+
+    def test_remote_url_keeps_a_filename_containing_percent_encoded_spaces(self, paths: Paths):
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "report%20final.pdf")
+        url = "https://example.com/download?filename=report%20final.pdf"
+        text = f"Source: {url}\nSaved as report%20final.pdf"
+
+        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"Source: {url}\nSaved as {VIRTUAL_PATH_PREFIX}/workspace/report%20final.pdf"
+
+    def test_embedded_text_preserves_remote_urls_and_local_file_uris(self, paths: Paths):
+        from mcp.types import EmbeddedResource, TextResourceContents
+
+        workspace = paths.sandbox_work_dir("t1", user_id="u1")
+        src = _workspace_file(paths, "report with spaces.pdf")
+        url = "https://example.com/download?filename=report%20with%20spaces.pdf"
+        text = f"Source: {url}\nLocal copy: {src.as_uri()}\nSaved as report with spaces.pdf"
+        result = CallToolResult(content=[EmbeddedResource(type="resource", resource=TextResourceContents(uri="resource://download-result", mimeType="text/plain", text=text))])
+
+        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],
+            )
+
+        virtual_path = f"{VIRTUAL_PATH_PREFIX}/workspace/report with spaces.pdf"
+        assert content[0]["text"] == f"Source: {url}\nLocal copy: {virtual_path}\nSaved as {virtual_path}"
+
     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")