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 <noreply@cursor.com>

.gitignore

@@ -18,6 +18,7 @@ Desktop.ini
 .idea/
 .pytest_cache/
 .venv/
+.test-venv/
 
 # Distribution / packaging
 dist/

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,

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,
             }

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

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.

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