From bde5f2d09017e2f74b49148b26ee2a5b7c5307ea Mon Sep 17 00:00:00 2001 From: haochen <3033940900@qq.com> Date: Sat, 24 Oct 2026 16:39:22 +0800 Subject: [PATCH] fix(sandbox): close pipe write fds once when Popen fails (#7394) * fix(sandbox): close pipe write fds once when Popen fails The command runners closed the pipe write descriptors twice on the spawn-failure path: the except block released all four pipe fds, then the finally block closed the two write fds again. In a busy gateway another thread can claim a freed descriptor number between the two closes, so the second os.close silently closed an unrelated descriptor (a concurrent command's capture pipe or an HTTP socket). Guard the parent's write-end close on a successful spawn. * fix(sandbox): release pipe fds on BaseException and scope the Windows runner test Review follow-up on #7394: - Catch BaseException in both command runners' spawn-failure cleanup so a KeyboardInterrupt mid-spawn still releases each pipe fd exactly once; the guarded finally keeps the no-double-close guarantee. - Skip the _run_windows_command failure-path param on non-Windows runners: it reads subprocess.CREATE_NEW_PROCESS_GROUP when building the call, which raises AttributeError on POSIX and would fail Linux CI. - Add a KeyboardInterrupt regression test (fd-per-fd close accounting). --------- Co-authored-by: Zhangchenhao-111 <285911435+Zhangchenhao-111@users.noreply.github.com> --- .../deerflow/sandbox/local/local_sandbox.py | 44 +++++--- .../test_local_sandbox_pipe_fd_lifecycle.py | 104 ++++++++++++++++++ 2 files changed, 134 insertions(+), 14 deletions(-) create mode 100644 backend/tests/test_local_sandbox_pipe_fd_lifecycle.py diff --git a/backend/packages/harness/deerflow/sandbox/local/local_sandbox.py b/backend/packages/harness/deerflow/sandbox/local/local_sandbox.py index e43bf29f5a..99f432ef5e 100644 --- a/backend/packages/harness/deerflow/sandbox/local/local_sandbox.py +++ b/backend/packages/harness/deerflow/sandbox/local/local_sandbox.py @@ -887,6 +887,7 @@ class LocalSandbox(Sandbox): timed_out = False stdout_read_fd, stdout_write_fd = os.pipe() stderr_read_fd, stderr_write_fd = os.pipe() + process: subprocess.Popen | None = None try: process = subprocess.Popen( args, @@ -897,7 +898,10 @@ class LocalSandbox(Sandbox): creationflags=subprocess.CREATE_NEW_PROCESS_GROUP, env=env, ) - except Exception: + except BaseException: + # BaseException, not Exception: a KeyboardInterrupt mid-spawn must + # still release the fds; each fd is closed exactly once here, and + # the guarded finally below never closes them a second time. for fd in (stdout_read_fd, stdout_write_fd, stderr_read_fd, stderr_write_fd): try: os.close(fd) @@ -906,12 +910,16 @@ class LocalSandbox(Sandbox): pass raise finally: - for fd in (stdout_write_fd, stderr_write_fd): - try: - os.close(fd) - except OSError: - # The write fd may already be closed by the exception cleanup above. - pass + # Only the successful-spawn path owes the parent's write-end close; + # on failure the except block already released all four fds, and a + # second close could hit a descriptor number another thread claimed + # in between. + if process is not None: + for fd in (stdout_write_fd, stderr_write_fd): + try: + os.close(fd) + except OSError: + pass if encoding is None: encoding = locale.getpreferredencoding(False) @@ -1011,6 +1019,7 @@ class LocalSandbox(Sandbox): timed_out = False stdout_read_fd, stdout_write_fd = os.pipe() stderr_read_fd, stderr_write_fd = os.pipe() + process: subprocess.Popen | None = None try: process = subprocess.Popen( args, @@ -1024,7 +1033,10 @@ class LocalSandbox(Sandbox): # With start_new_session=True, the new process is also the # process-group leader, so its PID is the stable group identifier. process_group_id = process.pid - except Exception: + except BaseException: + # BaseException, not Exception: a KeyboardInterrupt mid-spawn must + # still release the fds; each fd is closed exactly once here, and + # the guarded finally below never closes them a second time. for fd in (stdout_read_fd, stdout_write_fd, stderr_read_fd, stderr_write_fd): try: os.close(fd) @@ -1033,12 +1045,16 @@ class LocalSandbox(Sandbox): pass raise finally: - for fd in (stdout_write_fd, stderr_write_fd): - try: - os.close(fd) - except OSError: - # The write fd may already be closed by the exception cleanup above. - pass + # Only the successful-spawn path owes the parent's write-end close; + # on failure the except block already released all four fds, and a + # second close could hit a descriptor number another thread claimed + # in between. + if process is not None: + for fd in (stdout_write_fd, stderr_write_fd): + try: + os.close(fd) + except OSError: + pass stdout_capture, stdout_thread = LocalSandbox._start_pipe_drain(stdout_read_fd, "deerflow-bash-stdout-drain") stderr_capture, stderr_thread = LocalSandbox._start_pipe_drain(stderr_read_fd, "deerflow-bash-stderr-drain") diff --git a/backend/tests/test_local_sandbox_pipe_fd_lifecycle.py b/backend/tests/test_local_sandbox_pipe_fd_lifecycle.py new file mode 100644 index 0000000000..0c03aa40f5 --- /dev/null +++ b/backend/tests/test_local_sandbox_pipe_fd_lifecycle.py @@ -0,0 +1,104 @@ +"""Regression tests for the pipe fd lifecycle in LocalSandbox command runners. + +``_run_windows_command`` and ``_run_posix_command`` closed the pipe write fds +twice when ``subprocess.Popen`` raised: once in the ``except`` block (all four +pipe fds) and once more in the ``finally`` block. Between the two closes +another thread can claim the freed descriptor number (a concurrent command's +capture pipe, an HTTP socket), and the second ``os.close`` then silently +closes that unrelated descriptor. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +from unittest.mock import patch + +import pytest + +from deerflow.sandbox.local.local_sandbox import LocalSandbox + + +def _record_os_close(monkeypatch: pytest.MonkeyPatch) -> list[int]: + """Route os.close through a recorder that delegates to the real close.""" + closed: list[int] = [] + real_close = os.close + + def recording_close(fd: int) -> None: + closed.append(fd) + real_close(fd) + + monkeypatch.setattr(os, "close", recording_close) + return closed + + +@pytest.mark.parametrize( + "runner_name", + [ + pytest.param( + "_run_windows_command", + marks=pytest.mark.skipif(sys.platform != "win32", reason="the runner reads subprocess.CREATE_NEW_PROCESS_GROUP, which only exists on Windows builds"), + ), + "_run_posix_command", + ], +) +def test_spawn_failure_closes_each_pipe_fd_exactly_once(monkeypatch: pytest.MonkeyPatch, runner_name: str) -> None: + runner = getattr(LocalSandbox, runner_name) + closed = _record_os_close(monkeypatch) + + def failing_popen(*args: object, **kwargs: object) -> subprocess.Popen: + raise OSError("spawn failed (simulated)") + + with patch.object(subprocess, "Popen", side_effect=failing_popen): + with pytest.raises(OSError, match="spawn failed"): + runner(["whatever"], 1.0, None) + + duplicated = sorted({fd for fd in closed if closed.count(fd) > 1}) + assert not duplicated, f"fds {duplicated} were closed more than once; a concurrent thread may have claimed them in between: {closed}" + assert len(closed) == 4, f"expected exactly one close per pipe end, got {closed}" + + +def test_spawn_interrupt_closes_each_pipe_fd_exactly_once(monkeypatch: pytest.MonkeyPatch) -> None: + """A BaseException (KeyboardInterrupt mid-spawn) must still release all four fds once.""" + runner = LocalSandbox._run_posix_command + closed = _record_os_close(monkeypatch) + + def interrupted_popen(*args: object, **kwargs: object) -> subprocess.Popen: + raise KeyboardInterrupt + + with patch.object(subprocess, "Popen", side_effect=interrupted_popen): + with pytest.raises(KeyboardInterrupt): + runner(["whatever"], 1.0, None) + + duplicated = sorted({fd for fd in closed if closed.count(fd) > 1}) + assert not duplicated, f"fds {duplicated} were closed more than once: {closed}" + assert len(closed) == 4, f"expected exactly one close per pipe end, got {closed}" + + +def test_successful_spawn_closes_write_fds_exactly_once(monkeypatch: pytest.MonkeyPatch) -> None: + """The guard must not skip the parent's write-end close after a real spawn.""" + if sys.platform == "win32": + runner = LocalSandbox._run_windows_command + command = ["cmd", "/c", "echo hi"] + else: + runner = LocalSandbox._run_posix_command + command = ["echo", "hi"] + + closed = _record_os_close(monkeypatch) + write_fds: list[int] = [] + real_popen = subprocess.Popen + + def spying_popen(args: object, *, stdout: int | None = None, stderr: int | None = None, **kwargs: object) -> subprocess.Popen: + write_fds.extend(fd for fd in (stdout, stderr) if fd is not None) + assert real_popen is not None + return real_popen(args, stdout=stdout, stderr=stderr, **kwargs) # type: ignore[arg-type] + + with patch.object(subprocess, "Popen", side_effect=spying_popen): + stdout_text, _, _, timed_out = runner(command, 30.0, None) + + assert not timed_out + assert "hi" in stdout_text + assert len(write_fds) == 2, "expected the runner to hand both pipe write ends to Popen" + for fd in write_fds: + assert closed.count(fd) == 1, f"write fd {fd} closed {closed.count(fd)} times: {closed}"