fix(middleware): isolate temporary files for concurrent tool outputs (#6150)

* fix(middleware): isolate temporary files for concurrent tool outputs

* docs: trim tool output middleware guidance

* fix(middleware): preserve output permissions and clarify crash cleanup

---------

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

README.md

@@ -1759,6 +1759,13 @@ 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.
+Mounted sandboxes running under another UID need read access through the shared
+storage permissions. An unclean shutdown can leave `.tool-output-*.tmp` files in
+`tool_output.storage_subdir` (default `.tool-results`) under thread outputs.
+Remove leftovers during thread-data maintenance with all Gateway writers stopped,
+or when deleting the corresponding inactive thread's data.
+
 `E2BSandboxProvider` uses `wait` as its default overflow policy. It waits for
 `acquire_timeout`, then fails the agent turn. DeerFlow does not retry the turn
 automatically. Clients can use the structured error to schedule a retry.

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 tool output per app config. Externalizes oversized results to `tool_output.storage_subdir` (default `.tool-results`, constant `TOOL_RESULTS_DIRNAME`) under thread outputs, leaving a typed synopsis + `read_file` reference. These process-feedback files are excluded from workspace-change scans and delivery verification. `wrap_model_call` elides successful `write_file` content only in model-bound requests (#5328) after a later successful same-path `read_file`/`write_file`/`str_replace`; the on-disk file becomes the reference. Preserves the newest `keep_recent_writes` writes. Pairs call occurrences via `tool_call_args.pair_tool_call_results` and rewrites through shared `tool_call_args` helpers; controls: `elide_superseded_writes`, `superseded_write_min_chars`.
+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). 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`.
 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

@@ -0,0 +1,26 @@
+# Host tool-output publication
+
+`_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.
+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
+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.
+
+## Unclean shutdown
+
+SIGKILL, OOM termination, and host failure bypass exception cleanup and can leave
+unique temporary files. There is no automatic stale-temp sweeper. These internal
+files are excluded from workspace-change and delivery scans; `keep_recent_writes`
+elides model-visible messages and does not prune files.
+
+Remove `.tool-output-*.tmp` leftovers during thread-data maintenance only when all
+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.

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

@@ -29,6 +29,7 @@ import logging
 import os
 import posixpath
 import shlex
+import uuid
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import replace as dc_replace
 from typing import TYPE_CHECKING, Any, override
@@ -197,20 +198,24 @@ def _externalize(
     if not os.path.abspath(filepath).startswith(os.path.abspath(storage_dir)):
         return None
 
-    # Publish through a sibling temp file: a write that fails part-way (disk
-    # full, interrupted request) used to leave a truncated file under the final
-    # name even though this function reported failure, so the outputs directory
-    # accumulated half-written files that nothing references.
-    tmp_path = f"{filepath}.tmp"
+    # 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:
-        with open(tmp_path, "w", encoding="utf-8") as f:
+        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.
+        with open(candidate_path, "x", encoding="utf-8") as f:
+            tmp_path = candidate_path
             f.write(content)
         os.replace(tmp_path, filepath)
     except OSError:
-        try:
-            os.unlink(tmp_path)
-        except OSError:
-            pass
+        if tmp_path is not None:
+            try:
+                os.unlink(tmp_path)
+            except OSError:
+                pass
         return None
 
     return f"{_VIRTUAL_OUTPUTS_BASE}/{storage_subdir}/{filename}"

backend/tests/test_externalize_failure_cleanup.py

@@ -1,6 +1,15 @@
 """A failed externalization must not leave a half-written output behind."""
 
+import json
+import os
 import pathlib
+import subprocess
+import sys
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+
+import pytest
 
 from deerflow.agents.middlewares import tool_output_budget_middleware as mw
 
@@ -10,6 +19,7 @@ class _FailsAfterHalfTheWrite:
 
     def __init__(self, handle):
         self._handle = handle
+        self.name = handle.name
 
     def write(self, data):
         self._handle.write(data[: len(data) // 2])
@@ -27,12 +37,10 @@ class _FailsAfterHalfTheWrite:
 def test_failed_write_leaves_no_file_behind(tmp_path, monkeypatch):
     real_open = open
 
-    def failing_open(path, mode="r", *args, **kwargs):
-        handle = real_open(path, mode, *args, **kwargs)
-        return _FailsAfterHalfTheWrite(handle) if "w" in mode else handle
+    def failing_open(*args, **kwargs):
+        return _FailsAfterHalfTheWrite(real_open(*args, **kwargs))
 
-    # open is a builtin, so it is not in the module dict to begin with.
-    monkeypatch.setitem(mw.__dict__, "open", failing_open)
+    monkeypatch.setattr(mw, "open", failing_open, raising=False)
 
     result = mw._externalize(
         "X" * 100,
@@ -59,3 +67,99 @@ def test_successful_write_publishes_the_whole_content(tmp_path):
     written = [p for p in pathlib.Path(tmp_path).rglob("*") if p.is_file()]
     assert len(written) == 1
     assert written[0].read_text(encoding="utf-8") == "Y" * 100
+
+
+@pytest.mark.parametrize("second_publish_fails", [False, True])
+def test_concurrent_externalizations_do_not_share_temporary_files(tmp_path, monkeypatch, second_publish_fails):
+    first_ready = threading.Event()
+    release_first = threading.Event()
+    real_replace = mw.os.replace
+
+    def ordered_replace(source, destination):
+        if not first_ready.is_set():
+            # The first writer has closed its file but has not published it.
+            first_ready.set()
+            assert release_first.wait(timeout=10)
+        elif second_publish_fails:
+            raise OSError("Publication failed")
+        return real_replace(source, destination)
+
+    monkeypatch.setattr(mw.os, "replace", ordered_replace)
+    kwargs = dict(tool_name="bash", tool_call_id="same_call", outputs_path=str(tmp_path), storage_subdir="sub")
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        first = executor.submit(mw._externalize, "A" * 1000, **kwargs)
+        try:
+            assert first_ready.wait(timeout=10)
+            second_path = mw._externalize("B" * 200, **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)
+    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
+
+
+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")
+
+    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")
+
+    assert result is None
+    assert list(storage_dir.iterdir()) == [published]
+    assert published.read_text(encoding="utf-8") == "Previous complete output"
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits are not available on Windows")
+@pytest.mark.parametrize("mask", [0o022, 0o002, 0o077], ids=["other-readable", "group-writable", "private"])
+def test_published_output_respects_umask(tmp_path, mask):
+    # Set umask only in a child; it is process-global and must not race other tests.
+    script = """
+import json
+import os
+import pathlib
+import stat
+import sys
+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"
+initial_mode = stat.S_IMODE(published.stat().st_mode)
+os.chmod(published, 0o600)
+assert mw._externalize("Second complete output", **kwargs) is not None
+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["modes"] == [0o666 & ~mask] * 2
+
+
+def test_exclusive_creation_collision_preserves_other_writers_temp(tmp_path, monkeypatch):
+    storage_dir = tmp_path / "sub"
+    storage_dir.mkdir()
+    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")
+    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")
+
+    assert result is None
+    assert set(storage_dir.iterdir()) == {owned_by_other_writer, published}
+    assert owned_by_other_writer.read_text(encoding="utf-8") == "Other writer's pending output"
+    assert published.read_text(encoding="utf-8") == "Previous complete output"