mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-27 06:14:48 -06:00
fix(bash): do not hang when a command backgrounds a long-lived process
A bash command that leaves a process running in the background (server &, a daemon) could wedge the whole workstream forever: the tool read stdout/stderr to EOF, which never arrives because the child inherits the pipe, and the timeout watchdog bailed the moment the tracked bash exited. Wait on the tracked process bounded by the tool timeout (keyed on process exit, not pipe EOF) and terminate its whole session group on every exit path, reaping any backgrounded survivor, forcing the drain threads to EOF, and leaving nothing to leak. Decode with errors=replace so undecodable output is preserved instead of dropped, and pre-bind proc so a Popen failure surfaces the real error. Behavior change: a process the command backgrounds no longer survives the call. First-class opt-in backgrounding is left as a separate change.
This commit is contained in:
Patrick Buckley
@@ -0,0 +1,184 @@
|
||||
"""Regression tests for the bash tool hanging on a backgrounded child.
|
||||
|
||||
A bash command that backgrounds a long-lived process (``server &``,
|
||||
``python -m http.server &``, any daemon) used to wedge the whole workstream
|
||||
forever: the child inherits the tool's stdout/stderr pipe, so the foreground
|
||||
read never hit EOF, and the timeout watchdog bailed the moment the tracked
|
||||
``bash`` exited. ``_exec_bash`` now waits on the tracked process (not pipe
|
||||
EOF) bounded by ``tool_timeout`` and kills the whole session group on exit, so
|
||||
the call always returns and never leaks the background child.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
import signal
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
|
||||
from tests._session_helpers import NullUI, make_session
|
||||
from turnstone.core.trajectory import EffectStatus
|
||||
|
||||
|
||||
def _pid_alive(pid: int) -> bool:
|
||||
try:
|
||||
os.kill(pid, 0)
|
||||
except ProcessLookupError:
|
||||
return False
|
||||
except PermissionError:
|
||||
return True
|
||||
return True
|
||||
|
||||
|
||||
def _kill_pid(pid: int) -> None:
|
||||
with contextlib.suppress(OSError):
|
||||
os.kill(pid, signal.SIGKILL)
|
||||
|
||||
|
||||
def _run_in_thread(fn, timeout):
|
||||
"""Run ``fn`` in a daemon thread; return ``(finished, result)``."""
|
||||
box = {}
|
||||
|
||||
def _target():
|
||||
box["result"] = fn()
|
||||
|
||||
t = threading.Thread(target=_target, daemon=True)
|
||||
t.start()
|
||||
t.join(timeout)
|
||||
return (not t.is_alive()), box.get("result")
|
||||
|
||||
|
||||
def test_backgrounded_child_does_not_hang_and_is_reaped():
|
||||
"""Foreground exits immediately but leaves ``sleep 60 &`` holding the pipe.
|
||||
|
||||
Old behaviour: infinite hang (EOF never arrives, watchdog bails once the
|
||||
tracked bash exits). New behaviour: returns promptly and the background
|
||||
child is reaped by the session-group kill.
|
||||
"""
|
||||
pidfile = tempfile.mktemp(suffix=".pid")
|
||||
# A generous tool_timeout proves the return comes from foreground-exit, not
|
||||
# from the deadline firing.
|
||||
session = make_session(tool_timeout=30)
|
||||
command = f"sleep 60 & echo $! > {pidfile}; echo done"
|
||||
bg_pid = None
|
||||
try:
|
||||
finished, result = _run_in_thread(
|
||||
lambda: session._exec_bash({"call_id": "c1", "command": command}),
|
||||
timeout=15,
|
||||
)
|
||||
assert finished, "_exec_bash hung on a backgrounded child"
|
||||
assert result is not None
|
||||
call_id, output = result
|
||||
assert call_id == "c1"
|
||||
assert "done" in output
|
||||
|
||||
# The backgrounded process must have been reaped by the group kill.
|
||||
with open(pidfile) as f:
|
||||
bg_pid = int(f.read().strip())
|
||||
deadline = time.monotonic() + 5
|
||||
while _pid_alive(bg_pid) and time.monotonic() < deadline:
|
||||
time.sleep(0.05)
|
||||
assert not _pid_alive(bg_pid), f"backgrounded child {bg_pid} leaked"
|
||||
finally:
|
||||
if bg_pid is not None:
|
||||
_kill_pid(bg_pid)
|
||||
if os.path.exists(pidfile):
|
||||
os.unlink(pidfile)
|
||||
|
||||
|
||||
def test_timeout_still_fires_with_backgrounded_child():
|
||||
"""A silent foreground command plus a backgrounded child still hits the
|
||||
deadline: the watchdog kills the whole group and the result reads UNKNOWN
|
||||
(the ``unknown, never none`` timeout discipline)."""
|
||||
session = make_session(tool_timeout=1)
|
||||
command = "sleep 60 & sleep 60"
|
||||
|
||||
finished, result = _run_in_thread(
|
||||
lambda: session._exec_bash({"call_id": "c1", "command": command}),
|
||||
timeout=10,
|
||||
)
|
||||
assert finished, "_exec_bash did not return at its deadline"
|
||||
assert result is not None
|
||||
call_id, output = result
|
||||
assert call_id == "c1"
|
||||
assert "timed out" in output.lower()
|
||||
assert "UNKNOWN" in output
|
||||
assert session._tool_status.get("c1") is EffectStatus.UNKNOWN
|
||||
|
||||
|
||||
def test_undecodable_output_is_preserved_not_swallowed():
|
||||
"""Undecodable bytes on stdout must not silently vanish.
|
||||
|
||||
The drain's broad ``except (ValueError, OSError)`` would otherwise catch the
|
||||
``UnicodeDecodeError`` (a ``ValueError``) and kill the thread before any line
|
||||
was yielded — dropping ALL output and reporting a clean success. ``Popen``
|
||||
now decodes with ``errors="replace"`` so output always survives.
|
||||
"""
|
||||
session = make_session(tool_timeout=30)
|
||||
# Valid lines bracketing a raw invalid-UTF-8 byte sequence.
|
||||
command = r"printf 'before\n'; printf '\xff\xfe'; printf 'after\n'"
|
||||
finished, result = _run_in_thread(
|
||||
lambda: session._exec_bash({"call_id": "c1", "command": command}),
|
||||
timeout=15,
|
||||
)
|
||||
assert finished
|
||||
assert result is not None
|
||||
_call_id, output = result
|
||||
assert output != "(no output)"
|
||||
assert "before" in output
|
||||
assert "after" in output
|
||||
|
||||
|
||||
def test_stdout_streams_to_ui_from_drain_thread():
|
||||
"""stdout chunks are now emitted from the drain thread; they must still reach
|
||||
``on_tool_output_chunk``."""
|
||||
chunks: list[str] = []
|
||||
|
||||
class RecordingUI(NullUI):
|
||||
def on_tool_output_chunk(self, call_id, chunk):
|
||||
chunks.append(chunk)
|
||||
|
||||
session = make_session(tool_timeout=30, ui=RecordingUI())
|
||||
finished, result = _run_in_thread(
|
||||
lambda: session._exec_bash({"call_id": "c1", "command": "echo streamed-line"}),
|
||||
timeout=15,
|
||||
)
|
||||
assert finished
|
||||
assert any("streamed-line" in c for c in chunks)
|
||||
|
||||
|
||||
def test_cancel_midbash_reports_unknown():
|
||||
"""An external ``cancel()`` during a running bash unblocks the process-bounded
|
||||
wait and reports UNKNOWN (unknown-never-none), not a clean result."""
|
||||
session = make_session(tool_timeout=30)
|
||||
|
||||
def _cancel_soon():
|
||||
time.sleep(0.5)
|
||||
session.cancel()
|
||||
|
||||
threading.Thread(target=_cancel_soon, daemon=True).start()
|
||||
finished, result = _run_in_thread(
|
||||
lambda: session._exec_bash({"call_id": "c1", "command": "sleep 30"}),
|
||||
timeout=15,
|
||||
)
|
||||
assert finished, "cancel did not unblock _exec_bash"
|
||||
assert result is not None
|
||||
_call_id, output = result
|
||||
assert "cancelled" in output.lower()
|
||||
assert session._tool_status.get("c1") is EffectStatus.UNKNOWN
|
||||
|
||||
|
||||
def test_popen_failure_reports_cleanly(monkeypatch):
|
||||
"""If ``Popen`` itself raises, the ``finally`` must not mask the real error
|
||||
with ``UnboundLocalError`` — ``proc`` is pre-bound to ``None``."""
|
||||
from turnstone.core import session as session_mod
|
||||
|
||||
session = make_session(tool_timeout=30)
|
||||
|
||||
def _boom(*args, **kwargs):
|
||||
raise OSError("cannot fork")
|
||||
|
||||
monkeypatch.setattr(session_mod.subprocess, "Popen", _boom)
|
||||
call_id, output = session._exec_bash({"call_id": "c1", "command": "echo hi"})
|
||||
assert call_id == "c1"
|
||||
assert "cannot fork" in output
|
||||
+102
-48
@@ -432,6 +432,10 @@ _SEARCH_DRAIN_CHUNK: int = 8192
|
||||
# exits. The thread reads from a closed pipe at that point; a small
|
||||
# timeout keeps shutdown bounded if the OS hasn't propagated EOF yet.
|
||||
_SEARCH_DRAIN_JOIN_TIMEOUT: float = 2.0
|
||||
|
||||
# Poll granularity while waiting for a bash command to exit — bounds how long a
|
||||
# cooperative cancel or the wall-clock ``timeout`` can go unnoticed.
|
||||
_BASH_WAIT_POLL_S: float = 0.1
|
||||
# Excluded directory patterns — hit by both backends. ripgrep also respects
|
||||
# ``.gitignore`` and skips hidden directories by default, so most of these
|
||||
# are belt-and-suspenders for the rg path; they're load-bearing for grep.
|
||||
@@ -14033,6 +14037,9 @@ class ChatSession:
|
||||
preamble += "set -e\n"
|
||||
f.write(preamble + command)
|
||||
script_path = f.name
|
||||
# Pre-bind so the ``finally`` can't raise ``UnboundLocalError`` and
|
||||
# mask the real error if ``Popen`` below fails.
|
||||
proc: subprocess.Popen[str] | None = None
|
||||
try:
|
||||
from turnstone.core.env import scrubbed_env
|
||||
|
||||
@@ -14041,67 +14048,114 @@ class ChatSession:
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
errors="replace",
|
||||
start_new_session=True,
|
||||
env=scrubbed_env(extra=self._skill_resource_env()),
|
||||
)
|
||||
with self._procs_lock:
|
||||
self._active_procs.add(proc)
|
||||
# Drain stderr in background thread to avoid pipe deadlock
|
||||
# Drain stdout and stderr in background threads. Reading the
|
||||
# pipes to EOF in the foreground is unsafe: a command that
|
||||
# backgrounds a long-lived child (``server &``) leaks the pipe
|
||||
# write-end to that child, so EOF never arrives and the read —
|
||||
# hence the whole tool call — would hang forever, with the
|
||||
# timeout defeated once the tracked ``bash`` has exited. Instead
|
||||
# we wait on the tracked process bounded by ``timeout`` and tear
|
||||
# down its whole session group on exit, reaping any such survivor.
|
||||
stdout_parts: list[str] = []
|
||||
stderr_lines: list[str] = []
|
||||
|
||||
def drain_stderr() -> None:
|
||||
assert proc.stderr is not None
|
||||
for line in proc.stderr:
|
||||
stderr_lines.append(line)
|
||||
def _drain(pipe: Any, sink: list[str], to_ui: bool) -> None:
|
||||
try:
|
||||
for line in pipe:
|
||||
sink.append(line)
|
||||
if to_ui:
|
||||
try:
|
||||
self.ui.on_tool_output_chunk(call_id, line)
|
||||
except Exception:
|
||||
log.debug(
|
||||
"UI callback error during tool output",
|
||||
exc_info=True,
|
||||
)
|
||||
except (ValueError, OSError):
|
||||
# Pipe torn down by the session-group kill below is the
|
||||
# expected case; anything else must not kill the drain
|
||||
# silently. (Undecodable bytes can't land here — the
|
||||
# ``errors="replace"`` on Popen pre-empts UnicodeDecodeError,
|
||||
# which is a ValueError that would otherwise drop all output.)
|
||||
log.debug("bash.drain_read_error", exc_info=True)
|
||||
|
||||
stderr_thread = threading.Thread(target=drain_stderr, daemon=True)
|
||||
assert proc.stdout is not None and proc.stderr is not None
|
||||
stdout_thread = threading.Thread(
|
||||
target=_drain,
|
||||
args=(proc.stdout, stdout_parts, True),
|
||||
name=f"bash-drain-out-{call_id}",
|
||||
daemon=True,
|
||||
)
|
||||
stderr_thread = threading.Thread(
|
||||
target=_drain,
|
||||
args=(proc.stderr, stderr_lines, False),
|
||||
name=f"bash-drain-err-{call_id}",
|
||||
daemon=True,
|
||||
)
|
||||
stdout_thread.start()
|
||||
stderr_thread.start()
|
||||
|
||||
# Stream stdout line-by-line with process-group timeout
|
||||
stdout_parts: list[str] = []
|
||||
# Snapshot the session-group id while the leader is alive
|
||||
# (``start_new_session=True`` makes ``pgid == proc.pid``). While
|
||||
# any member survives, the group names only our own descendants;
|
||||
# once they have all exited it is empty and the kill below is a
|
||||
# harmless no-op. (A pid-wraparound landing a fresh session
|
||||
# leader on this exact id in the microseconds after a normal-exit
|
||||
# reap is the standard accepted TOCTOU — negligible, and only when
|
||||
# nothing needs killing anyway.)
|
||||
try:
|
||||
pgid = os.getpgid(proc.pid)
|
||||
except OSError:
|
||||
pgid = proc.pid
|
||||
|
||||
# Wait for the tracked command, bounded by ``timeout`` and
|
||||
# cooperatively cancellable. Keyed on process exit, never pipe
|
||||
# EOF, so a leaked background child cannot extend the wait.
|
||||
timed_out = threading.Event()
|
||||
deadline = time.monotonic() + timeout
|
||||
while True:
|
||||
if cancel.is_set():
|
||||
break
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
timed_out.set()
|
||||
break
|
||||
try:
|
||||
proc.wait(timeout=min(remaining, _BASH_WAIT_POLL_S))
|
||||
break
|
||||
except subprocess.TimeoutExpired:
|
||||
continue
|
||||
|
||||
def _on_timeout() -> None:
|
||||
if proc.poll() is not None:
|
||||
return # process already exited
|
||||
timed_out.set()
|
||||
with contextlib.suppress(OSError, ProcessLookupError):
|
||||
try:
|
||||
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
|
||||
except OSError:
|
||||
with contextlib.suppress(OSError, ProcessLookupError):
|
||||
proc.kill()
|
||||
# Terminate the whole session group on every exit path: reaps a
|
||||
# backgrounded child that would otherwise leak (ports, PIDs) or
|
||||
# hold the output pipe open, and forces the drain threads to EOF.
|
||||
with contextlib.suppress(OSError, ProcessLookupError):
|
||||
os.killpg(pgid, signal.SIGKILL)
|
||||
|
||||
timer = threading.Timer(timeout, _on_timeout)
|
||||
timer.start()
|
||||
try:
|
||||
assert proc.stdout is not None
|
||||
for line in proc.stdout:
|
||||
stdout_parts.append(line)
|
||||
try:
|
||||
self.ui.on_tool_output_chunk(call_id, line)
|
||||
except Exception:
|
||||
log.debug("UI callback error during tool output", exc_info=True)
|
||||
# Check cancellation during long-running commands
|
||||
if cancel.is_set():
|
||||
with contextlib.suppress(OSError, ProcessLookupError):
|
||||
try:
|
||||
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
|
||||
except OSError:
|
||||
with contextlib.suppress(OSError, ProcessLookupError):
|
||||
proc.kill()
|
||||
raise GenerationCancelled()
|
||||
finally:
|
||||
timer.cancel()
|
||||
|
||||
try:
|
||||
with contextlib.suppress(subprocess.TimeoutExpired):
|
||||
proc.wait(timeout=10)
|
||||
except subprocess.TimeoutExpired:
|
||||
log.warning("Process did not exit after SIGKILL, pid=%d", proc.pid)
|
||||
if proc.returncode is None:
|
||||
# Survived SIGKILL (uninterruptible sleep — NFS, a driver).
|
||||
# Restores the diagnostic the old Timer path emitted.
|
||||
log.warning("bash.survived_sigkill", pid=proc.pid)
|
||||
# Writers are gone, so the drains hit EOF promptly. They are
|
||||
# daemon threads: a pathological double-``setsid`` grandchild
|
||||
# that escaped the group cannot wedge shutdown — it leaks its
|
||||
# drain (logged below) until it dies.
|
||||
stdout_thread.join(timeout=5)
|
||||
stderr_thread.join(timeout=5)
|
||||
if stdout_thread.is_alive() or stderr_thread.is_alive():
|
||||
log.warning("bash.drain_leaked", call_id=call_id, pid=proc.pid)
|
||||
finally:
|
||||
with self._procs_lock:
|
||||
self._active_procs.discard(proc)
|
||||
if proc is not None:
|
||||
with self._procs_lock:
|
||||
self._active_procs.discard(proc)
|
||||
os.unlink(script_path)
|
||||
|
||||
if timed_out.is_set():
|
||||
@@ -14110,9 +14164,9 @@ class ChatSession:
|
||||
# Distinguish user cancel from unexpected SIGKILL.
|
||||
# Popen.returncode is negative of the signal number when killed.
|
||||
if cancel.is_set() and proc.returncode == -signal.SIGKILL:
|
||||
# SIGKILL'd mid-flight (the command was parked on a silent
|
||||
# read, so the in-loop cooperative check never fired). Its
|
||||
# side effects are unobserved: record outcome UNKNOWN and
|
||||
# SIGKILL'd mid-flight by our session-group kill once the
|
||||
# bounded wait observed the cancel. Its side effects are
|
||||
# unobserved: record outcome UNKNOWN and
|
||||
# mark it an error, not a clean empty success — a destructive
|
||||
# command killed here must not read as "did not run" on
|
||||
# replay. Keep whatever partial stdout we captured.
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "bash",
|
||||
"description": "Execute a bash command and return stdout + stderr. Use for running programs, git, tests, system commands, installing packages, etc. Environment questions ('What Python version?', 'Is X installed?') are tool-use tasks — e.g. bash(command='python --version'). For file creation use write_file instead; for man pages use man instead. Long output is truncated (head+tail preserved, middle elided). Stderr lines prefixed with [stderr].",
|
||||
"description": "Execute a bash command and return stdout + stderr. Use for running programs, git, tests, system commands, installing packages, etc. Environment questions ('What Python version?', 'Is X installed?') are tool-use tasks — e.g. bash(command='python --version'). For file creation use write_file instead; for man pages use man instead. Long output is truncated (head+tail preserved, middle elided). Stderr lines prefixed with [stderr]. Runs to completion and returns: any process the command leaves running in the background (e.g. 'server &') is terminated when the command returns — nothing persists across calls.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
|
||||
Reference in New Issue
Block a user