fix(middleware): derive the externalized filename from the tool call id (#6107)

* fix(middleware): derive the externalized filename from the tool call id

_build_externalized_filename documents that the host-disk and sandbox paths
share it so both produce the identical naming scheme, but it used a random uuid
suffix and ignored tool_call_id: the same call got a different name every time,
so the two paths could not agree and externalizing one output twice left two
files behind. Derive the name from the call id instead, sanitised the same way
tool names are, and drop the now unused uuid import.

* fix(middleware): hash externalized call IDs and content

* fix(middleware): catch sandbox filename encoding failures

* fix(middleware): handle host output encoding failures

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -1878,6 +1878,11 @@ For example, independent read-only research can run concurrently when the wall-c
 ### Sandbox & File System
 
 Host-externalized tool outputs use the Gateway's normal file-creation umask.
+Host and sandbox outputs use deterministic filenames hashed from the raw tool
+call ID and output content. Missing, colliding, or oversized IDs cannot overwrite
+different output or make the filename too long; identical output can reuse its path.
+Host and remote sandbox persistence failures, including invalid Unicode during filename
+hashing, use the configured inline fallback.
 `tool_output` character/count budgets, including per-tool overrides, require
 non-negative integers; YAML booleans are rejected rather than treated as 0 or 1.
 An explicit zero per-tool override disables externalization while preserving

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -45,7 +45,7 @@ strict providers reject.
 1. **InputSanitizationMiddleware** - First, so it is the outermost `wrap_model_call` wrapper; every inner middleware (including LLM retries) sees sanitized messages. `additional_kwargs.original_user_content` is server-owned provenance: Gateway strips caller-supplied values for non-internal run requests, trusted IM calls may carry the string they captured before adding transport/file context, and the middleware replaces any non-string value before wrapping. Uploads and sanitization retain first-writer-wins only for validated strings. Caller markers are marked `untrusted_input`, never stripped; scope is every turn.
 
    **KnowledgeScopeMiddleware** follows input sanitization: it exposes only Gateway-admitted execution scope, removes scope/display data from model messages, and blocks `knowledge_search` when disabled without reading storage or RAGFlow.
-2. **ToolOutputBudgetMiddleware** - Caps model-bound outputs per app config. Externalizes oversized results under thread outputs in `tool_output.storage_subdir` (default `.tool-results`, `TOOL_RESULTS_DIRNAME`), leaving a typed synopsis + `read_file` reference. Host permissions, publication and crash cleanup: [contract](TOOL_OUTPUT_PERSISTENCE.md). Blob refs restore across hosts; failures use ≤30,000 inline chars. These files bypass workspace-change scans and delivery verification. `wrap_model_call` elides successful `write_file` content only in model requests after a later successful same-path `read_file`/`write_file`/`str_replace` (#5328); disk is the reference. Keeps `keep_recent_writes` newest writes. Shared `tool_call_args` helpers pair calls (`pair_tool_call_results`) and rewrite args. Controls: `elide_superseded_writes`, `superseded_write_min_chars`.
+2. **ToolOutputBudgetMiddleware** - Caps model-bound outputs; externalizes to thread outputs (`tool_output.storage_subdir`, default `.tool-results`, `TOOL_RESULTS_DIRNAME`) with a synopsis + `read_file` reference. Host/sandbox filename SHA-256 hashes a length-framed raw call ID and content. Blob refs restore across hosts; hash/write failures use inline fallback. Internal outputs bypass workspace/delivery scans. Publication and write elision: [contract](TOOL_OUTPUT_PERSISTENCE.md). Model-only write elision uses shared `tool_call_args`; retain `keep_recent_writes` newest writes. Controls: `elide_superseded_writes`, `superseded_write_min_chars`.
 3. **ToolResultSanitizationMiddleware** - Neutralizes framework/injection tags (e.g. `<system-reminder>`) and boundary markers in *remote-content* tool results (`web_fetch`/`web_search`/`image_search`/`web_capture`) so attacker-controlled fetched pages cannot forge trusted framework context. Mirrors `InputSanitizationMiddleware`'s user-input guardrail for the other untrusted-content entry point; sits inner of `ToolOutputBudgetMiddleware` (neutralizes the raw output, then the budget truncates). Local tool output (bash/read_file) is left untouched. Scope is a name-based allowlist for the first-party web tools, plus every MCP-sourced tool via its `deerflow_mcp` metadata tag, so an MCP server naming its fetcher `fetch_url` is still covered
 
    Result-rewriting middlewares between the raw callable boundary and the

backend/packages/harness/deerflow/agents/middlewares/TOOL_OUTPUT_PERSISTENCE.md

@@ -1,5 +1,15 @@
 # Host tool-output publication
 
+Host and sandbox outputs share a filename helper whose fixed-length SHA-256
+suffix hashes the raw call ID and UTF-8 content with an explicit ID-length
+boundary. Repeating the same call ID and content reuses one path; changed content
+or distinct raw IDs use distinct paths, including missing IDs and IDs that would
+collide after sanitization. Provider IDs never appear directly in filenames.
+
+For host storage and non-mounted sandboxes, filename hashing stays inside the
+write failure boundary. Invalid UTF-8 text or call IDs return `None` before
+file publication, allowing the tool wrappers to use the configured inline fallback.
+
 `_externalize` creates a unique sibling `.tool-output-*.tmp` with exclusive
 creation (`open(..., "x")`). Its mode is `0o666 & ~umask`, preserving ordinary
 file-creation permissions without reading or changing the process-wide umask.
@@ -7,13 +17,14 @@ This matters when a mounted sandbox reads the output under a different UID.
 Restrictive operator umasks remain restrictive.
 
 The writer closes its file before atomically replacing the deterministic final
-path. The last successful publisher wins. Ownership starts only after exclusive
+path. Concurrent publishers of identical output can safely reuse that path;
+different output retains its own path. Ownership starts only after exclusive
 creation succeeds: a collision or creation failure must not remove another
-writer's pending file. An observed `OSError` cleans only this invocation's temp
-and leaves previously published content intact.
+writer's pending file. An observed `OSError` or `UnicodeEncodeError` cleans only
+this invocation's temp and leaves previously published content intact.
 
 When blob storage is enabled, blob publication follows host publication. If the
-blob write fails, that message uses the safety-limited inline fallback and does
+blob encoding or write fails, that message uses the safety-limited inline fallback and does
 not advertise the host path. It also must not unlink the deterministic final
 path: a concurrent publisher may already have replaced it and durably
 checkpointed that content. Remove published files only as part of inactive
@@ -31,3 +42,12 @@ Gateway processes writing the shared storage are stopped, or when deleting the
 corresponding inactive thread's data. Age alone cannot prove a writer is dead,
 especially across workers or shared mounts, so publication must not delete other
 writers' files based on a TTL.
+
+## Model-bound write elision
+
+`wrap_model_call` elides successful `write_file` content only in model requests
+after a later successful same-path `read_file`, `write_file`, or `str_replace`
+(#5328); disk remains the reference. `keep_recent_writes` preserves the newest
+writes. Use the shared `tool_call_args` helpers (`pair_tool_call_results` and
+argument rewriting). Controls: `elide_superseded_writes` and
+`superseded_write_min_chars`.

backend/packages/harness/deerflow/agents/middlewares/tool_output_budget_middleware.py

@@ -173,16 +173,7 @@ def _sanitize_tool_name(name: str) -> str:
     return safe or "unknown"
 
 
-def _sanitize_tool_call_id(tool_call_id: str) -> str:
-    """Make a tool call id safe to use inside a filename.
-
-    The id reaches us from the model/provider, so it gets the same treatment as
-    a tool name: no separators and no traversal components.
-    """
-    return _sanitize_tool_name(tool_call_id)
-
-
-def _build_externalized_filename(*, tool_name: str, tool_call_id: str) -> str:
+def _build_externalized_filename(*, tool_name: str, tool_call_id: str, content: str) -> str:
     """Build the on-disk filename for an externalized tool output.
 
     Shared by the host-disk and sandbox externalization paths so both
@@ -190,11 +181,15 @@ def _build_externalized_filename(*, tool_name: str, tool_call_id: str) -> str:
     """
     safe_name = _sanitize_tool_name(tool_name)
     ext = _EXT_MAP.get(tool_name, "txt")
-    # Derived from the call id so the host-disk and sandbox paths agree on one
-    # name for a given call, and so externalizing the same output twice is
-    # idempotent instead of leaving two files behind.
-    safe_id = _sanitize_tool_call_id(tool_call_id)
-    return f"{safe_name}-{safe_id}.{ext}"
+    # Hash the raw ID and content so missing IDs, sanitization collisions and
+    # oversized IDs cannot overwrite distinct output. Frame the ID length to
+    # keep the boundary unambiguous, including when either value contains NUL.
+    call_id_bytes = tool_call_id.encode("utf-8")
+    digest = hashlib.sha256()
+    digest.update(len(call_id_bytes).to_bytes(8, "big"))
+    digest.update(call_id_bytes)
+    digest.update(content.encode("utf-8"))
+    return f"{safe_name}-{digest.hexdigest()}.{ext}"
 
 
 def _externalize(
@@ -214,17 +209,16 @@ def _externalize(
     except OSError:
         return None
 
-    filename = _build_externalized_filename(tool_name=tool_name, tool_call_id=tool_call_id)
-    filepath = os.path.join(storage_dir, filename)
-
-    if not os.path.abspath(filepath).startswith(os.path.abspath(storage_dir)):
-        return None
-
     # Each writer owns a unique sibling temp file, so concurrent calls cannot
     # truncate or clean up each other's pending output. Publish only after close
     # (also required on Windows), keeping the final filename deterministic.
     tmp_path = None
     try:
+        filename = _build_externalized_filename(tool_name=tool_name, tool_call_id=tool_call_id, content=content)
+        filepath = os.path.join(storage_dir, filename)
+        if not os.path.abspath(filepath).startswith(os.path.abspath(storage_dir)):
+            return None
+
         candidate_path = os.path.join(storage_dir, f".tool-output-{uuid.uuid4().hex}.tmp")
         # Exclusive creation keeps per-writer ownership while honoring umask,
         # unlike NamedTemporaryFile's fixed 0600 mode on mounted outputs.
@@ -232,7 +226,7 @@ def _externalize(
             tmp_path = candidate_path
             f.write(content)
         os.replace(tmp_path, filepath)
-    except OSError:
+    except (OSError, UnicodeEncodeError):
         if tmp_path is not None:
             try:
                 os.unlink(tmp_path)
@@ -299,10 +293,12 @@ def _externalize_to_sandbox(
     """
     if os.path.isabs(storage_subdir) or ".." in storage_subdir:
         return None
-    filename = _build_externalized_filename(tool_name=tool_name, tool_call_id=tool_call_id)
-    virtual_dir = f"{_VIRTUAL_OUTPUTS_BASE}/{storage_subdir}"
-    virtual_path = f"{virtual_dir}/{filename}"
     try:
+        # Hashing encodes provider text too; keep encoding errors within the
+        # same failure boundary as sandbox writes so callers can truncate.
+        filename = _build_externalized_filename(tool_name=tool_name, tool_call_id=tool_call_id, content=content)
+        virtual_dir = f"{_VIRTUAL_OUTPUTS_BASE}/{storage_subdir}"
+        virtual_path = f"{virtual_dir}/{filename}"
         # AIO sandbox write_file does NOT create parent directories, so create
         # them explicitly before writing. execute_command returns its stdout
         # verbatim (including an "Error: ..." string on failure) rather than
@@ -520,7 +516,12 @@ def _budget_content(
 
         if host_outputs_path is not None:
             blob_store = get_blob_store_if_enabled()
-            blob_bytes = content.encode("utf-8") if blob_store is not None else None
+            try:
+                blob_bytes = content.encode("utf-8") if blob_store is not None else None
+            except UnicodeEncodeError:
+                blob_bytes = None
+                durable_fallback_required = True
+                logger.warning("Tool output cannot be encoded as UTF-8 for durable blob externalization")
             if blob_bytes is not None and len(blob_bytes) > _MAX_TOOL_OUTPUT_BLOB_BYTES:
                 durable_fallback_required = True
                 logger.warning(
@@ -528,7 +529,7 @@ def _budget_content(
                     len(blob_bytes),
                     _MAX_TOOL_OUTPUT_BLOB_BYTES,
                 )
-            else:
+            elif not durable_fallback_required:
                 virtual_path = _externalize(
                     content,
                     tool_name=tool_name,

backend/tests/test_externalize_failure_cleanup.py

@@ -70,7 +70,8 @@ def test_successful_write_publishes_the_whole_content(tmp_path):
 
 
 @pytest.mark.parametrize("second_publish_fails", [False, True])
-def test_concurrent_externalizations_do_not_share_temporary_files(tmp_path, monkeypatch, second_publish_fails):
+@pytest.mark.parametrize("second_content", ["A" * 1000, "B" * 200], ids=["identical-output", "different-output"])
+def test_concurrent_externalizations_do_not_share_temporary_files(tmp_path, monkeypatch, second_publish_fails, second_content):
     first_ready = threading.Event()
     release_first = threading.Event()
     real_replace = mw.os.replace
@@ -90,31 +91,38 @@ def test_concurrent_externalizations_do_not_share_temporary_files(tmp_path, monk
         first = executor.submit(mw._externalize, "A" * 1000, **kwargs)
         try:
             assert first_ready.wait(timeout=10)
-            second_path = mw._externalize("B" * 200, **kwargs)
+            second_path = mw._externalize(second_content, **kwargs)
         finally:
             # Always release the worker, including when the second call fails.
             release_first.set()
         first_path = first.result(timeout=10)
 
-    assert first_path == "/mnt/user-data/outputs/sub/bash-same_call.log"
-    assert second_path == (None if second_publish_fails else first_path)
+    assert first_path is not None
+    if second_publish_fails:
+        assert second_path is None
+    else:
+        assert second_path is not None
+        assert (first_path == second_path) is (second_content == "A" * 1000)
     written = [p for p in tmp_path.rglob("*") if p.is_file()]
-    assert len(written) == 1
-    # The last successful publisher wins, without mixed bytes or leftover temps.
-    assert written[0].read_text(encoding="utf-8") == "A" * 1000
+    assert len(written) == (2 if second_path is not None and second_path != first_path else 1)
+    # Every advertised path retains its complete bytes, with no leftover temps.
+    assert (tmp_path / "sub" / first_path.rsplit("/", 1)[1]).read_text(encoding="utf-8") == "A" * 1000
+    if second_path is not None:
+        assert (tmp_path / "sub" / second_path.rsplit("/", 1)[1]).read_text(encoding="utf-8") == second_content
 
 
 def test_temporary_file_creation_failure_preserves_existing_output(tmp_path, monkeypatch):
     storage_dir = tmp_path / "sub"
-    storage_dir.mkdir()
-    published = storage_dir / "bash-call_1.log"
-    published.write_text("Previous complete output", encoding="utf-8")
+    content = "Previous complete output"
+    path = mw._externalize(content, tool_name="bash", tool_call_id="call_1", outputs_path=str(tmp_path), storage_subdir="sub")
+    assert path is not None
+    published = storage_dir / path.rsplit("/", 1)[1]
 
     def fail_to_create(*args, **kwargs):
         raise OSError("Cannot create temporary file")
 
     monkeypatch.setattr(mw, "open", fail_to_create, raising=False)
-    result = mw._externalize("New output", tool_name="bash", tool_call_id="call_1", outputs_path=str(tmp_path), storage_subdir="sub")
+    result = mw._externalize(content, tool_name="bash", tool_call_id="call_1", outputs_path=str(tmp_path), storage_subdir="sub")
 
     assert result is None
     assert list(storage_dir.iterdir()) == [published]
@@ -135,29 +143,32 @@ from deerflow.agents.middlewares import tool_output_budget_middleware as mw
 
 os.umask(int(sys.argv[2]))
 kwargs = dict(tool_name="bash", tool_call_id="mode", outputs_path=sys.argv[1], storage_subdir="sub")
-assert mw._externalize("First complete output", **kwargs) is not None
-published = pathlib.Path(sys.argv[1]) / "sub/bash-mode.log"
+content = "Complete output"
+first_path = mw._externalize(content, **kwargs)
+assert first_path is not None
+published = pathlib.Path(sys.argv[1]) / "sub" / first_path.rsplit("/", 1)[1]
 initial_mode = stat.S_IMODE(published.stat().st_mode)
 os.chmod(published, 0o600)
-assert mw._externalize("Second complete output", **kwargs) is not None
+assert mw._externalize(content, **kwargs) == first_path
 print(json.dumps({"modes": [initial_mode, stat.S_IMODE(published.stat().st_mode)], "content": published.read_text(encoding="utf-8")}))
 """
     completed = subprocess.run([sys.executable, "-c", script, str(tmp_path), str(mask)], capture_output=True, text=True, timeout=30, check=True)
     observed = json.loads(completed.stdout)
-    assert observed["content"] == "Second complete output"
+    assert observed["content"] == "Complete output"
     assert observed["modes"] == [0o666 & ~mask] * 2
 
 
 def test_exclusive_creation_collision_preserves_other_writers_temp(tmp_path, monkeypatch):
     storage_dir = tmp_path / "sub"
-    storage_dir.mkdir()
+    content = "Previous complete output"
+    path = mw._externalize(content, tool_name="bash", tool_call_id="call_1", outputs_path=str(tmp_path), storage_subdir="sub")
+    assert path is not None
     owned_by_other_writer = storage_dir / ".tool-output-collision.tmp"
     owned_by_other_writer.write_text("Other writer's pending output", encoding="utf-8")
-    published = storage_dir / "bash-call_1.log"
-    published.write_text("Previous complete output", encoding="utf-8")
+    published = storage_dir / path.rsplit("/", 1)[1]
     monkeypatch.setattr(mw, "uuid", SimpleNamespace(uuid4=lambda: SimpleNamespace(hex="collision")), raising=False)
 
-    result = mw._externalize("New output", tool_name="bash", tool_call_id="call_1", outputs_path=str(tmp_path), storage_subdir="sub")
+    result = mw._externalize(content, tool_name="bash", tool_call_id="call_1", outputs_path=str(tmp_path), storage_subdir="sub")
 
     assert result is None
     assert set(storage_dir.iterdir()) == {owned_by_other_writer, published}

backend/tests/test_tool_output_externalized_filename.py

@@ -0,0 +1,176 @@
+"""Externalized outputs retain stable, distinct paths for each call and content."""
+
+import re
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from langchain_core.messages import ToolMessage
+
+from deerflow.agents.middlewares.tool_output_budget_middleware import (
+    ToolOutputBudgetMiddleware,
+    _build_externalized_filename,
+    _externalize,
+    _externalize_to_sandbox,
+)
+from deerflow.config.tool_output_config import ToolOutputConfig
+from deerflow.sandbox.sandbox import Sandbox
+
+
+def test_the_same_tool_call_gets_the_same_filename():
+    """The host-disk and sandbox paths share this helper to agree on one name.
+
+    A random suffix cannot do that: the same call would be written under two
+    names, and externalizing one output twice would leave two files behind.
+    """
+    first = _build_externalized_filename(tool_name="bash", tool_call_id="call_abc123", content="output")
+    second = _build_externalized_filename(tool_name="bash", tool_call_id="call_abc123", content="output")
+
+    assert first == second
+    assert re.fullmatch(r"bash-[a-f0-9]{64}\.log", first)
+
+
+@pytest.mark.parametrize("first_id, second_id", [("call_a", "call_b"), ("a/call", "b/call"), ("../call", "call")])
+def test_different_tool_calls_get_different_filenames(first_id: str, second_id: str):
+    first = _build_externalized_filename(tool_name="bash", tool_call_id=first_id, content="output")
+    second = _build_externalized_filename(tool_name="bash", tool_call_id=second_id, content="output")
+
+    assert first != second
+
+
+@pytest.mark.parametrize("tool_call_id", ["../../etc/passwd", "..\\..\\etc\\passwd", "\x00", ""])
+def test_a_tool_call_id_cannot_escape_the_output_directory(tool_call_id: str):
+    name = _build_externalized_filename(tool_name="bash", tool_call_id=tool_call_id, content="output")
+
+    assert "/" not in name
+    assert ".." not in name
+    assert name.startswith("bash-")
+    assert re.fullmatch(r"bash-[a-f0-9]{64}\.log", name)
+
+
+@pytest.mark.parametrize("tool_call_id", ["", "call_abc123"])
+def test_changed_content_gets_a_distinct_filename(tool_call_id: str):
+    first = _build_externalized_filename(tool_name="bash", tool_call_id=tool_call_id, content="first output")
+    second = _build_externalized_filename(tool_name="bash", tool_call_id=tool_call_id, content="second output")
+
+    assert first != second
+
+
+@pytest.mark.parametrize("first, second", [(("a", "bc"), ("ab", "c")), (("a\x00b", "c"), ("a", "b\x00c"))])
+def test_call_id_and_content_are_unambiguously_combined(first: tuple[str, str], second: tuple[str, str]):
+    first_name = _build_externalized_filename(tool_name="bash", tool_call_id=first[0], content=first[1])
+    second_name = _build_externalized_filename(tool_name="bash", tool_call_id=second[0], content=second[1])
+
+    assert first_name != second_name
+
+
+@pytest.mark.parametrize(
+    "first_id, second_id, first_content, second_content",
+    [("", "", "first output", "second output"), ("a/call", "b/call", "output", "output"), ("call_1", "call_1", "first output", "second output")],
+    ids=["missing-ids", "sanitized-collision", "changed-content"],
+)
+def test_host_externalization_preserves_earlier_outputs(tmp_path: Path, first_id: str, second_id: str, first_content: str, second_content: str):
+    kwargs = dict(tool_name="bash", outputs_path=str(tmp_path), storage_subdir=".tool-results")
+    first_path = _externalize(first_content, tool_call_id=first_id, **kwargs)
+    second_path = _externalize(second_content, tool_call_id=second_id, **kwargs)
+
+    assert first_path is not None
+    assert second_path is not None
+    assert first_path != second_path
+    storage_dir = tmp_path / ".tool-results"
+    assert (storage_dir / first_path.rsplit("/", 1)[1]).read_text(encoding="utf-8") == first_content
+    assert (storage_dir / second_path.rsplit("/", 1)[1]).read_text(encoding="utf-8") == second_content
+
+
+@pytest.mark.parametrize("tool_call_id", ["", "a/call", "x" * 10_000, "调用" * 10_000], ids=["empty", "path", "long-ascii", "long-unicode"])
+def test_host_and_sandbox_use_the_same_bounded_filename(tmp_path: Path, tool_call_id: str):
+    content = "输出\nfull output"
+    kwargs = dict(tool_name="bash", tool_call_id=tool_call_id, storage_subdir=".tool-results")
+    sandbox = MagicMock(spec=Sandbox)
+    sandbox.execute_command.return_value = "OK"
+
+    host_path = _externalize(content, outputs_path=str(tmp_path), **kwargs)
+    sandbox_path = _externalize_to_sandbox(content, sandbox=sandbox, **kwargs)
+
+    assert host_path is not None
+    assert sandbox_path == host_path
+    filename = host_path.rsplit("/", 1)[1]
+    assert re.fullmatch(r"bash-[a-f0-9]{64}\.log", filename)
+    assert len(filename.encode("utf-8")) < 255
+    assert (tmp_path / ".tool-results" / filename).read_text(encoding="utf-8") == content
+    sandbox.write_file.assert_called_once_with(host_path, content)
+
+
+@pytest.mark.parametrize("invalid_field", ["content", "tool_call_id"])
+def test_host_externalization_handles_filename_encoding_failure(tmp_path: Path, invalid_field: str):
+    kwargs = dict(content="output", tool_call_id="call_1")
+    kwargs[invalid_field] += "\ud800"
+
+    result = _externalize(tool_name="bash", storage_subdir=".tool-results", outputs_path=str(tmp_path), **kwargs)
+
+    assert result is None
+    assert not any(path.is_file() for path in tmp_path.rglob("*"))
+
+
+@pytest.mark.parametrize("invalid_field", ["content", "tool_call_id"])
+def test_sandbox_externalization_handles_filename_encoding_failure(invalid_field: str):
+    sandbox = MagicMock(spec=Sandbox)
+    kwargs = dict(content="output", tool_call_id="call_1")
+    kwargs[invalid_field] += "\ud800"
+
+    result = _externalize_to_sandbox(tool_name="bash", storage_subdir=".tool-results", sandbox=sandbox, **kwargs)
+
+    assert result is None
+    sandbox.execute_command.assert_not_called()
+    sandbox.write_file.assert_not_called()
+
+
+@pytest.mark.parametrize("async_wrapper", [False, True], ids=["sync", "async"])
+@pytest.mark.parametrize("invalid_field", ["content", "tool_call_id"])
+@pytest.mark.parametrize("storage_target", ["sandbox", "host", "mounted-host", "blob"])
+@pytest.mark.asyncio
+async def test_filename_encoding_failure_uses_bounded_fallback(monkeypatch, tmp_path: Path, async_wrapper: bool, invalid_field: str, storage_target: str):
+    from deerflow.agents.middlewares import tool_output_budget_middleware as mw
+
+    sandbox = MagicMock(spec=Sandbox)
+    provider = SimpleNamespace(uses_thread_data_mounts=storage_target == "mounted-host", get=lambda _: sandbox)
+    monkeypatch.setattr(mw, "get_sandbox_provider", lambda: provider)
+    blob_store = MagicMock() if storage_target == "blob" else None
+    monkeypatch.setattr(mw, "get_blob_store_if_enabled", lambda: blob_store)
+    content = "A" * 16_000 + ("\ud800" if invalid_field == "content" else "X") + "B" * 16_000
+    call_id = "call_1" + ("\ud800" if invalid_field == "tool_call_id" else "")
+    message = ToolMessage(content=content, name="remote_executor", tool_call_id=call_id, id="message_1", artifact={"original": True})
+    state = {"thread_data": {"outputs_path": str(tmp_path)}}
+    if storage_target in {"sandbox", "mounted-host"}:
+        state["sandbox"] = {"sandbox_id": "sb-1"}
+    request = SimpleNamespace(
+        tool_call={"name": "remote_executor", "id": call_id},
+        runtime=SimpleNamespace(state=state, context={"thread_id": "thread-1"}),
+    )
+    config = ToolOutputConfig()
+    middleware = ToolOutputBudgetMiddleware(config=config)
+    if async_wrapper:
+
+        async def handler(_):
+            return message
+
+        result = await middleware.awrap_tool_call(request, handler)
+    else:
+        result = middleware.wrap_tool_call(request, lambda _: message)
+
+    assert isinstance(result, ToolMessage)
+    assert "Persistent storage unavailable" in result.content
+    assert result.content.startswith("A" * config.fallback_head_chars)
+    assert result.content.endswith("B" * config.fallback_tail_chars)
+    assert len(result.content) <= config.fallback_max_chars
+    result.content.encode("utf-8")
+    assert result.tool_call_id == call_id
+    assert result.id == message.id
+    assert result.artifact == message.artifact
+    assert message.content == content
+    sandbox.execute_command.assert_not_called()
+    sandbox.write_file.assert_not_called()
+    if blob_store is not None:
+        blob_store.put_bytes.assert_not_called()
+    assert not any(path.is_file() for path in tmp_path.rglob("*"))