From 19325320ba4ebddf08273a5052cfe32a997c78fe Mon Sep 17 00:00:00 2001 From: Fabio Scarsi <103823600+fabioscarsi@users.noreply.github.com> Date: Wed, 13 May 2026 02:36:24 +0800 Subject: [PATCH] fix(orchestration): bound timeouts and teardown subprocess trees Ensure local shell timeouts clean up subprocess trees without breaking Windows fallback paths, raise the run_shell sanity ceiling for long tasks, and bound post-task analysis so execution cannot hang indefinitely. Add focused regression coverage for timeout cleanup, retry guidance, timeout caps, and analyzer cancellation. Co-authored-by: GPT-5.5 --- .gitignore | 1 + openspace/grounding/backends/shell/session.py | 3 +- .../shell/transport/local_connector.py | 115 +++++++++++++++- openspace/prompts/grounding_agent_prompts.py | 11 ++ openspace/tool_layer.py | 23 +++- tests/test_orchestration_timeouts.py | 124 ++++++++++++++++++ 6 files changed, 267 insertions(+), 10 deletions(-) create mode 100644 tests/test_orchestration_timeouts.py diff --git a/.gitignore b/.gitignore index 4f5ec62..87c5978 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,7 @@ Desktop.ini .idea/ .pytest_cache/ .venv/ +.test-venv/ # Distribution / packaging dist/ diff --git a/openspace/grounding/backends/shell/session.py b/openspace/grounding/backends/shell/session.py index 7dad2e2..b7538fa 100644 --- a/openspace/grounding/backends/shell/session.py +++ b/openspace/grounding/backends/shell/session.py @@ -286,7 +286,8 @@ class RunShellTool(BaseTool): super().__init__() async def _arun(self, command: str, timeout: int = 30) -> ToolResult: - timeout = min(timeout, 120) + # Trust caller-provided long-running work, but keep a sanity ceiling. + timeout = min(timeout, 1800) try: result = await self._session.connector.run_bash_script( command, diff --git a/openspace/grounding/backends/shell/transport/local_connector.py b/openspace/grounding/backends/shell/transport/local_connector.py index 2db21b0..6d28b23 100644 --- a/openspace/grounding/backends/shell/transport/local_connector.py +++ b/openspace/grounding/backends/shell/transport/local_connector.py @@ -11,6 +11,8 @@ work without any changes. import asyncio import os import platform +import signal +import subprocess import tempfile import uuid from typing import Any, Optional, Dict @@ -94,6 +96,105 @@ def _wrap_script_with_conda(script: str, conda_env: str | None) -> str: return script +def _subprocess_group_kwargs() -> Dict[str, Any]: + """Return platform-specific kwargs that allow timeout cleanup.""" + if platform_name == "Windows": + creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + return {"creationflags": creationflags} if creationflags else {} + return {"start_new_session": True} + + +async def _wait_for_exit(proc: asyncio.subprocess.Process, timeout: float) -> bool: + try: + await asyncio.wait_for(proc.wait(), timeout=timeout) + return True + except asyncio.TimeoutError: + return False + + +async def _kill_single_process(proc: asyncio.subprocess.Process) -> bool: + """Fallback cleanup for platforms where process-tree cleanup is unavailable.""" + for action in (proc.terminate, proc.kill): + if proc.returncode is not None: + return True + try: + action() + except ProcessLookupError: + return True + except Exception as e: + logger.warning("Process cleanup action failed: %s", e) + if await _wait_for_exit(proc, timeout=5): + return True + return proc.returncode is not None + + +async def _kill_windows_process_tree(proc: asyncio.subprocess.Process, pid: int) -> bool: + """Best-effort Windows process-tree cleanup.""" + try: + taskkill = await asyncio.create_subprocess_exec( + "taskkill", + "/PID", + str(pid), + "/T", + "/F", + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + await asyncio.wait_for(taskkill.wait(), timeout=5) + except (FileNotFoundError, PermissionError, OSError, asyncio.TimeoutError) as e: + logger.warning("taskkill process-tree cleanup failed: %s", e) + + if await _wait_for_exit(proc, timeout=1): + return True + return await _kill_single_process(proc) + + +async def _kill_posix_process_tree(proc: asyncio.subprocess.Process, pid: int) -> bool: + """Send SIGTERM, then SIGKILL, to the POSIX process group.""" + try: + pgid = os.getpgid(pid) + os.killpg(pgid, signal.SIGTERM) + except ProcessLookupError: + return True + except PermissionError as e: + logger.warning("SIGTERM to process group failed: %s", e) + return False + except (AttributeError, OSError) as e: + logger.warning("Process-group cleanup unavailable: %s", e) + return await _kill_single_process(proc) + + if await _wait_for_exit(proc, timeout=5): + return True + + try: + os.killpg(os.getpgid(pid), signal.SIGKILL) + except ProcessLookupError: + return True + except PermissionError as e: + logger.warning("SIGKILL to process group failed: %s", e) + return False + except (AttributeError, OSError) as e: + logger.warning("Process-group SIGKILL unavailable: %s", e) + return await _kill_single_process(proc) + + return await _wait_for_exit(proc, timeout=5) + + +async def _kill_process_tree(proc: asyncio.subprocess.Process, timeout: int) -> bool: + """Terminate a timed-out subprocess and its descendants when possible.""" + if proc.returncode is not None: + return True + + pid = proc.pid + logger.warning( + "Subprocess timeout after %ss; killing process tree (pid=%s)", timeout, pid + ) + + if platform_name == "Windows": + return await _kill_windows_process_tree(proc, pid) + return await _kill_posix_process_tree(proc, pid) + + class LocalShellConnector(BaseConnector[Any]): """ Shell connector that runs scripts **locally** using asyncio subprocesses, @@ -156,6 +257,7 @@ class LocalShellConnector(BaseConnector[Any]): stderr=asyncio.subprocess.PIPE, cwd=cwd, env=exec_env, + **_subprocess_group_kwargs(), ) stdout_b, stderr_b = await asyncio.wait_for( proc.communicate(), timeout=timeout @@ -172,10 +274,12 @@ class LocalShellConnector(BaseConnector[Any]): "returncode": returncode, } except asyncio.TimeoutError: + killed = await _kill_process_tree(proc, timeout) + suffix = "(killed)" if killed else "(kill attempted)" return { "status": "error", - "output": f"Execution timed out after {timeout} seconds", - "content": f"Execution timed out after {timeout} seconds", + "output": f"Execution timed out after {timeout} seconds {suffix}", + "content": f"Execution timed out after {timeout} seconds {suffix}", "error": "", "returncode": -1, } @@ -210,6 +314,7 @@ class LocalShellConnector(BaseConnector[Any]): stderr=asyncio.subprocess.STDOUT, cwd=cwd, env=exec_env, + **_subprocess_group_kwargs(), ) stdout_b, _ = await asyncio.wait_for( proc.communicate(), timeout=timeout @@ -225,10 +330,12 @@ class LocalShellConnector(BaseConnector[Any]): "returncode": returncode, } except asyncio.TimeoutError: + killed = await _kill_process_tree(proc, timeout) + suffix = "(killed)" if killed else "(kill attempted)" return { "status": "error", - "output": f"Script execution timed out after {timeout} seconds", - "content": f"Script execution timed out after {timeout} seconds", + "output": f"Script execution timed out after {timeout} seconds {suffix}", + "content": f"Script execution timed out after {timeout} seconds {suffix}", "error": "", "returncode": -1, } diff --git a/openspace/prompts/grounding_agent_prompts.py b/openspace/prompts/grounding_agent_prompts.py index 3a68d0d..0c078b5 100644 --- a/openspace/prompts/grounding_agent_prompts.py +++ b/openspace/prompts/grounding_agent_prompts.py @@ -28,6 +28,17 @@ class GroundingAgentPrompts: "- If you need results to decide next action, wait for next iteration" ) + sections.append( + "# On Tool Timeouts\n\n" + "When a tool returns a timeout result, the operation's outcome may be " + "incomplete or UNKNOWN unless the result explicitly says the process " + "was killed.\n" + "- Do NOT call the same tool with the same arguments twice in this task.\n" + "- Treat the task as paused-pending-investigation, not failed-and-retry.\n" + "- Report the timeout to the user and stop so they can decide whether " + "to investigate, kill orphaned work, or retry with adjusted parameters." + ) + # Tool Selection Tips (only mention backends that exist) tips: List[str] = [] has_mcp = "mcp" in scope diff --git a/openspace/tool_layer.py b/openspace/tool_layer.py index 7ceb447..b09d514 100644 --- a/openspace/tool_layer.py +++ b/openspace/tool_layer.py @@ -61,6 +61,7 @@ class OpenSpaceConfig: # Skill Evolution evolution_max_concurrent: int = 3 # Max parallel evolutions per trigger + execution_analysis_timeout: float = 300.0 # Overall post-task analyzer/evolver bound # Logging Configuration log_level: str = "INFO" @@ -595,9 +596,21 @@ class OpenSpace: if cancelled_exc is None: # Run execution analysis + evolution BEFORE building the return # value, so evolved_skills is populated. - await self._maybe_analyze_execution( - task_id, recording_dir, result - ) + try: + await asyncio.wait_for( + self._maybe_analyze_execution( + task_id, recording_dir, result + ), + timeout=self.config.execution_analysis_timeout, + ) + except asyncio.TimeoutError: + result["analysis_timed_out"] = True + logger.warning( + "Analyzer/evolver exceeded %.1fs bound for task %s; " + "cancelled. Evolved skills (if any) may be incomplete.", + self.config.execution_analysis_timeout, + task_id, + ) # Trigger quality evolution periodically await self._maybe_evolve_quality() @@ -837,8 +850,8 @@ class OpenSpace: }) except Exception as e: - # Analysis failure must never break the main execution flow - logger.debug(f"Execution analysis skipped: {e}") + # Analysis failure must never break the main execution flow. + logger.warning(f"Execution analysis skipped: {e}") async def _maybe_evolve_quality(self) -> None: """Trigger quality evolution based on global execution count. diff --git a/tests/test_orchestration_timeouts.py b/tests/test_orchestration_timeouts.py new file mode 100644 index 0000000..feb2084 --- /dev/null +++ b/tests/test_orchestration_timeouts.py @@ -0,0 +1,124 @@ +import asyncio +import platform +import shlex +import subprocess +import sys +from types import SimpleNamespace + +import pytest + +from openspace.grounding.backends.shell.session import RunShellTool +from openspace.grounding.backends.shell.transport.local_connector import ( + LocalShellConnector, +) +from openspace.prompts.grounding_agent_prompts import GroundingAgentPrompts +from openspace.tool_layer import OpenSpace, OpenSpaceConfig + + +@pytest.mark.asyncio +@pytest.mark.skipif(platform.system() == "Windows", reason="POSIX process-group check") +async def test_run_subprocess_timeout_kills_child_process(tmp_path): + marker = tmp_path / "child-survived" + script = tmp_path / "spawn_child.py" + child_code = ( + "import pathlib, time; " + "time.sleep(1.0); " + f"pathlib.Path({str(marker)!r}).write_text('alive')" + ) + parent_code = ( + "import subprocess, sys, time; " + f"subprocess.Popen([sys.executable, '-c', {child_code!r}]); " + "time.sleep(30)" + ) + script.write_text(parent_code) + + connector = LocalShellConnector() + result = await connector._run_subprocess( + [sys.executable, str(script)], + timeout=0.2, + ) + + await asyncio.sleep(1.4) + + assert result["returncode"] == -1 + assert "timed out" in result["content"] + assert "(killed)" in result["content"] + assert not marker.exists() + + +@pytest.mark.asyncio +async def test_run_shell_command_timeout_reports_cleanup(): + connector = LocalShellConnector() + code = "import time; time.sleep(10)" + if platform.system() == "Windows": + command = subprocess.list2cmdline([sys.executable, "-c", code]) + else: + command = f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}" + + result = await connector._run_shell_command(command, timeout=0.2) + + assert result["returncode"] == -1 + assert "timed out" in result["content"] + assert "(killed)" in result["content"] or "(kill attempted)" in result["content"] + + +@pytest.mark.asyncio +async def test_run_shell_timeout_cap_allows_longer_work(): + class FakeConnector: + def __init__(self): + self.timeouts = [] + + async def run_bash_script(self, command, *, timeout): + self.timeouts.append(timeout) + return {"returncode": 0, "content": "ok", "error": ""} + + connector = FakeConnector() + tool = RunShellTool(SimpleNamespace(connector=connector)) + + await tool._arun("echo ok", timeout=180) + await tool._arun("echo ok", timeout=9999) + + assert connector.timeouts == [180, 1800] + + +def test_grounding_prompt_warns_against_timeout_retry(): + prompt = GroundingAgentPrompts.build_system_prompt(["shell"]) + + assert "# On Tool Timeouts" in prompt + assert "Do NOT call the same tool with the same arguments twice" in prompt + assert "UNKNOWN" in prompt + assert "paused-pending-investigation" in prompt + + +@pytest.mark.asyncio +async def test_execute_bounds_post_task_analysis(): + space = OpenSpace( + OpenSpaceConfig( + enable_recording=False, + execution_analysis_timeout=0.01, + ) + ) + space._initialized = True + space._grounding_agent = SimpleNamespace( + process=lambda context: asyncio.sleep( + 0, + result={ + "status": "success", + "response": "done", + "iterations": 1, + "tool_executions": [], + }, + ), + _last_tools=[], + ) + + async def never_returns(*args, **kwargs): + await asyncio.sleep(3600) + + space._maybe_analyze_execution = never_returns + + result = await space.execute("do a tiny task") + + assert result["status"] == "success" + assert result["analysis_timed_out"] is True + assert space.is_running() is False