mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(middleware): publish externalized outputs atomically (#6109)
* fix(middleware): publish externalized outputs atomically _externalize reported failure with None but had already created the file, so a write that failed part-way (disk full, interrupted request) left a truncated file under the final name in the outputs directory. Write to a sibling temp file and os.replace it into place, removing the temp file when the write fails. * style: drop the unused pytest import from the externalize test
This commit is contained in:
parent
baae72d876
commit
5b83cb502c
@ -29,7 +29,6 @@ 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
|
||||
@ -151,6 +150,15 @@ 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:
|
||||
"""Build the on-disk filename for an externalized tool output.
|
||||
|
||||
@ -159,8 +167,11 @@ 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")
|
||||
short_id = uuid.uuid4().hex[:12]
|
||||
return f"{safe_name}-{short_id}.{ext}"
|
||||
# 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}"
|
||||
|
||||
|
||||
def _externalize(
|
||||
@ -186,10 +197,20 @@ 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"
|
||||
try:
|
||||
with open(filepath, "w", encoding="utf-8") as f:
|
||||
with open(tmp_path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
os.replace(tmp_path, filepath)
|
||||
except OSError:
|
||||
try:
|
||||
os.unlink(tmp_path)
|
||||
except OSError:
|
||||
pass
|
||||
return None
|
||||
|
||||
return f"{_VIRTUAL_OUTPUTS_BASE}/{storage_subdir}/{filename}"
|
||||
|
||||
61
backend/tests/test_externalize_failure_cleanup.py
Normal file
61
backend/tests/test_externalize_failure_cleanup.py
Normal file
@ -0,0 +1,61 @@
|
||||
"""A failed externalization must not leave a half-written output behind."""
|
||||
|
||||
import pathlib
|
||||
|
||||
from deerflow.agents.middlewares import tool_output_budget_middleware as mw
|
||||
|
||||
|
||||
class _FailsAfterHalfTheWrite:
|
||||
"""File object that fails part-way through the write, like a full disk."""
|
||||
|
||||
def __init__(self, handle):
|
||||
self._handle = handle
|
||||
|
||||
def write(self, data):
|
||||
self._handle.write(data[: len(data) // 2])
|
||||
self._handle.flush()
|
||||
raise OSError(28, "No space left on device")
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc):
|
||||
self._handle.close()
|
||||
return False
|
||||
|
||||
|
||||
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
|
||||
|
||||
# open is a builtin, so it is not in the module dict to begin with.
|
||||
monkeypatch.setitem(mw.__dict__, "open", failing_open)
|
||||
|
||||
result = mw._externalize(
|
||||
"X" * 100,
|
||||
tool_name="bash",
|
||||
tool_call_id="call_1",
|
||||
outputs_path=str(tmp_path),
|
||||
storage_subdir="sub",
|
||||
)
|
||||
|
||||
assert result is None
|
||||
assert [p for p in pathlib.Path(tmp_path).rglob("*") if p.is_file()] == []
|
||||
|
||||
|
||||
def test_successful_write_publishes_the_whole_content(tmp_path):
|
||||
result = mw._externalize(
|
||||
"Y" * 100,
|
||||
tool_name="bash",
|
||||
tool_call_id="call_2",
|
||||
outputs_path=str(tmp_path),
|
||||
storage_subdir="sub",
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
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
|
||||
Loading…
x
Reference in New Issue
Block a user