From 2acdcaa33952dee4d30c57c0090fba1db778f217 Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Sat, 31 Oct 2026 11:50:16 -0700 Subject: [PATCH 01/10] Keep sandbox HTTP command execution responsive --- README.md | 8 + .../harness/sandbox/_http_command.py | 79 ++++++++ .../harness/sandbox/docker_deployment.py | 11 ++ .../harness/sandbox/modal_deployment.py | 5 + .../harness/sandbox/server_command.py | 39 ++++ tests/core/harness/test_http_command.py | 181 ++++++++++++++++++ 6 files changed, 323 insertions(+) create mode 100644 src/olmo_eval/harness/sandbox/_http_command.py create mode 100644 src/olmo_eval/harness/sandbox/server_command.py create mode 100644 tests/core/harness/test_http_command.py diff --git a/README.md b/README.md index f9a623e63..34746eb91 100644 --- a/README.md +++ b/README.md @@ -1433,6 +1433,14 @@ result = await manager.execute_code("print('hello')", capabilities=Capability.PY await manager.stop() ``` +Independent `execute_command` and `execute_code` calls preserve the sandbox server's +initialized environment. Docker/Podman uses container execution when process +supervision is available; other container and Modal paths use asynchronous HTTP +commands so a long command does not block health checks or file requests. +Command deadlines terminate the command process group. Commands that finish +normally can leave background processes running. Persistent bash sessions retain +shell state between calls and currently use SWE-ReX's session execution path. + ### Sandbox Modes | Mode | Description | diff --git a/src/olmo_eval/harness/sandbox/_http_command.py b/src/olmo_eval/harness/sandbox/_http_command.py new file mode 100644 index 000000000..18319201d --- /dev/null +++ b/src/olmo_eval/harness/sandbox/_http_command.py @@ -0,0 +1,79 @@ +"""Asynchronous SWE-ReX command execution, embedded into remote server startup.""" + +from __future__ import annotations + +import asyncio +import contextlib +import os +import signal +import tempfile +from typing import Any + +from swerex.exceptions import CommandTimeoutError, NonZeroExitCodeError +from swerex.runtime import local +from swerex.runtime.abstract import Command, CommandResponse +from swerex.runtime.local import LocalRuntime + + +async def execute(self: LocalRuntime, command: Command) -> CommandResponse: + """Execute independent commands without occupying the HTTP server loop.""" + with tempfile.TemporaryFile() as stdout_file, tempfile.TemporaryFile() as stderr_file: + kwargs: dict[str, Any] = { + "env": command.env, + "cwd": command.cwd, + "stdout": stdout_file, + "stderr": stdout_file if command.merge_output_streams else stderr_file, + "start_new_session": True, + } + args = [command.command] if isinstance(command.command, str) else command.command + if command.shell: + # Match subprocess.run's sequence handling with shell=True. + process = await asyncio.create_subprocess_exec("/bin/sh", "-c", *args, **kwargs) + else: + process = await asyncio.create_subprocess_exec(*args, **kwargs) + try: + await asyncio.wait_for(process.wait(), command.timeout) + except (TimeoutError, asyncio.CancelledError) as exc: + with contextlib.suppress(ProcessLookupError): + os.killpg(process.pid, signal.SIGKILL) + await asyncio.wait_for(process.wait(), 2.0) + if isinstance(exc, asyncio.CancelledError): + raise + raise CommandTimeoutError( + f"Timeout ({command.timeout}s) exceeded while running command" + ) from exc + + def read_output() -> tuple[bytes, bytes]: + stdout_file.seek(0) + stderr_file.seek(0) + return ( + stdout_file.read(os.fstat(stdout_file.fileno()).st_size), + stderr_file.read(os.fstat(stderr_file.fileno()).st_size), + ) + + # Regular files keep a detached child from holding communicate() open. + stdout, stderr = await asyncio.to_thread(read_output) + response = CommandResponse( + stdout=stdout.decode(errors="backslashreplace"), + stderr=stderr.decode(errors="backslashreplace"), + exit_code=process.returncode, + ) + if command.check and response.exit_code != 0: + msg = ( + f"Command {command.command!r} failed with exit code {response.exit_code}. " + f"Stdout:\n{response.stdout!r}\nStderr:\n{response.stderr!r}" + ) + if command.error_msg: + msg = f"{command.error_msg}: {msg}" + raise NonZeroExitCodeError(msg) + return response + + +class ResponsiveLocalRuntime(LocalRuntime): + async def execute(self, command: Command) -> CommandResponse: + return await execute(self, command) + + +def install() -> None: + """Install command execution before the server constructs its runtime.""" + vars(local).update(LocalRuntime=ResponsiveLocalRuntime) diff --git a/src/olmo_eval/harness/sandbox/docker_deployment.py b/src/olmo_eval/harness/sandbox/docker_deployment.py index 82d56a9da..cc929afd6 100644 --- a/src/olmo_eval/harness/sandbox/docker_deployment.py +++ b/src/olmo_eval/harness/sandbox/docker_deployment.py @@ -104,6 +104,17 @@ def __init__(self, **kwargs: Any): self._captures: list[_CapturedPipe] = [] self._shutdown_task: asyncio.Task[None] | None = None + def _get_swerex_start_cmd(self, token: str, port: int) -> list[str]: + from .server_command import server_command + + executable = "swerex-remote" + if self._config.python_standalone_dir: + executable = f"{self._config.python_standalone_dir}/python3.11/bin/swerex-remote" + return [ + *self._config.exec_shell, + server_command(executable, token, port, install_pipx=True), + ] + async def start(self) -> None: if self._shutdown_task is not None: if not self._shutdown_task.done(): diff --git a/src/olmo_eval/harness/sandbox/modal_deployment.py b/src/olmo_eval/harness/sandbox/modal_deployment.py index dfcfc1e96..0fa54cc4d 100644 --- a/src/olmo_eval/harness/sandbox/modal_deployment.py +++ b/src/olmo_eval/harness/sandbox/modal_deployment.py @@ -25,6 +25,11 @@ async def stop_modal_app(app_name: str) -> None: class ManagedModalDeployment(ModalDeployment): """SWE-ReX Modal deployment with managed lifecycle and encrypted transport.""" + def _start_swerex_cmd(self, token: str) -> str: + from .server_command import server_command + + return server_command("swerex-remote", token, self._port) + async def start(self) -> None: if self._runtime is not None and self._sandbox is not None: self.logger.warning("Deployment is already started. Ignoring duplicate start() call.") diff --git a/src/olmo_eval/harness/sandbox/server_command.py b/src/olmo_eval/harness/sandbox/server_command.py new file mode 100644 index 000000000..1c0eec4f3 --- /dev/null +++ b/src/olmo_eval/harness/sandbox/server_command.py @@ -0,0 +1,39 @@ +"""Launch the installed SWE-ReX server with responsive command execution.""" + +import base64 +import shlex +from pathlib import Path + + +def server_command(executable: str, token: str, port: int, *, install_pipx: bool = False) -> str: + """Preserve the server interpreter and shell environment while installing its adapter.""" + source = Path(__file__).with_name("_http_command.py").read_bytes() + payload = base64.b64encode(source + b"\ninstall()\n").decode() + patch = ( + f"import base64; exec(compile(base64.b64decode({payload!r}), " + "'', 'exec'))\n" + ) + encoded = base64.b64encode(patch.encode()).decode() + # A temporary console script retains the installed entrypoint's shebang. + # This works for an isolated SWE-ReX venv as well as an activated task venv. + fallback = "exit 1" + if executable == "swerex-remote": + setup = ( + "python3 -m pip install pipx && python3 -m pipx ensurepath && " if install_pipx else "" + ) + pipx = "python3 -m pipx" if install_pipx else "pipx" + fallback = ( + setup + f"{pipx} install swe-rex && " + f'original="$({pipx} environment --value PIPX_BIN_DIR)/swerex-remote"' + ) + return ( + f"original=$(command -v {shlex.quote(executable)}) || " + f"{{ {fallback}; }} || exit 1; " + "bootstrap=$(mktemp -d) || exit 1; " + "trap 'rm -rf \"$bootstrap\"' EXIT; " + '{ head -n 1 "$original"; ' + f"printf %s {shlex.quote(encoded)} | base64 -d; " + 'tail -n +2 "$original"; } > "$bootstrap/server" && ' + 'chmod 700 "$bootstrap/server" && ' + f'"$bootstrap/server" --auth-token {shlex.quote(token)} --port {port}' + ) diff --git a/tests/core/harness/test_http_command.py b/tests/core/harness/test_http_command.py new file mode 100644 index 000000000..1d506f70e --- /dev/null +++ b/tests/core/harness/test_http_command.py @@ -0,0 +1,181 @@ +"""SWE-ReX command adapter and deployment bootstrap regressions.""" + +import asyncio +import os +import subprocess +import sys + +import httpx +import pytest + +pytest.importorskip("swerex") + +from swerex.exceptions import CommandTimeoutError, NonZeroExitCodeError +from swerex.runtime.abstract import Command +from swerex.runtime.local import LocalRuntime + +from olmo_eval.harness.sandbox._http_command import execute +from olmo_eval.harness.sandbox.server_command import server_command + + +@pytest.mark.anyio +async def test_http_command_keeps_health_and_file_requests_responsive(monkeypatch, tmp_path): + import swerex.server as server + + monkeypatch.setattr(LocalRuntime, "execute", execute) + marker = tmp_path / "started" + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=server.app), base_url="http://sandbox" + ) as client: + task = asyncio.create_task( + client.post( + "/execute", + json={ + "command": [ + sys.executable, + "-c", + f"from pathlib import Path; import time; Path({str(marker)!r}).touch(); " + "time.sleep(1)", + ], + "timeout": 5, + }, + ) + ) + try: + async with asyncio.timeout(3): + while not marker.exists(): + await asyncio.sleep(0.01) + health = await client.get("/is_alive") + written = await client.post( + "/write_file", json={"path": str(tmp_path / "file"), "content": "ready"} + ) + assert health.status_code == written.status_code == 200 + assert not task.done() + assert (tmp_path / "file").read_text() == "ready" + assert (await task).json()["exit_code"] == 0 + finally: + await task + + +@pytest.mark.anyio +@pytest.mark.parametrize("shell", [False, True]) +async def test_command_preserves_environment_cwd_streams_and_status(tmp_path, shell): + code = 'printf "%s:%s" "$TOKEN" "$PWD"; printf error >&2; exit 7' + command = Command( + command=code if shell else ["/bin/sh", "-c", code], + shell=shell, + env={"TOKEN": "space value"}, + cwd=str(tmp_path), + merge_output_streams=True, + ) + response = await execute(LocalRuntime(), command) + assert response.exit_code == 7 + assert response.stdout == f"space value:{tmp_path}error" + assert response.stderr == "" + command.check = True + command.error_msg = "custom" + with pytest.raises(NonZeroExitCodeError, match="custom"): + await execute(LocalRuntime(), command) + + +@pytest.mark.anyio +@pytest.mark.parametrize("cancel", [False, True]) +async def test_deadline_and_cancellation_kill_descendants_holding_output(tmp_path, cancel): + marker = tmp_path / "orphan" + started = tmp_path / "started" + script = f"(sleep 0.7; touch {marker}) & touch {started}; sleep 30" + task = asyncio.create_task( + execute( + LocalRuntime(), + Command( + command=["/bin/sh", "-c", script], + timeout=None if cancel else 0.1, + ), + ) + ) + if cancel: + async with asyncio.timeout(3): + while not started.exists(): + await asyncio.sleep(0.01) + task.cancel() + with pytest.raises(asyncio.CancelledError if cancel else CommandTimeoutError): + await asyncio.wait_for(task, 3) + await asyncio.sleep(0.8) + assert not marker.exists() + response = await execute(LocalRuntime(), Command(command=["/bin/echo", "recovered"])) + assert response.stdout == "recovered\n" + + +@pytest.mark.parametrize("use_path", [False, True]) +def test_bootstrap_uses_installed_shebang_and_preserves_shell_environment(tmp_path, use_path): + entrypoint = tmp_path / "swerex-remote" + entrypoint.write_text( + f"#!{sys.executable}\n" + "import os, sys\n" + "from swerex.runtime.local import LocalRuntime\n" + "print(LocalRuntime.execute.__module__, os.environ['BOOT_ENV'], sys.argv[1:])\n" + ) + entrypoint.chmod(0o700) + executable = "swerex-remote" if use_path else str(entrypoint) + result = subprocess.run( + ["/bin/bash", "-c", server_command(executable, "token with ' quote", 8123)], + env=os.environ | {"PATH": f"{tmp_path}:{os.environ['PATH']}", "BOOT_ENV": "preserved"}, + text=True, + capture_output=True, + timeout=5, + ) + assert result.returncode == 0, result.stderr + assert result.stdout.startswith("__main__ preserved") + assert "token with" in result.stdout + assert "--port', '8123'" in result.stdout + assert entrypoint.read_text().startswith(f"#!{sys.executable}\nimport os") + + +def test_bootstrap_installs_with_module_when_pipx_is_not_on_path(tmp_path): + install = tmp_path / "install" + install.mkdir() + entrypoint = install / "swerex-remote" + entrypoint.write_text(f"#!{sys.executable}\nprint('bootstrapped')\n") + calls = tmp_path / "calls" + python = tmp_path / "python3" + python.write_text( + f"#!/bin/sh\nprintf '%s\\n' \"$*\" >> {calls}\n" + f"if [ \"$2 $3\" = 'pipx environment' ]; then echo {install}; fi\n" + ) + python.chmod(0o700) + result = subprocess.run( + ["/bin/bash", "-c", server_command("swerex-remote", "token", 8123, install_pipx=True)], + env=os.environ | {"PATH": f"{tmp_path}:/usr/bin:/bin"}, + text=True, + capture_output=True, + timeout=5, + ) + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "bootstrapped" + assert calls.read_text().splitlines() == [ + "-m pip install pipx", + "-m pipx ensurepath", + "-m pipx install swe-rex", + "-m pipx environment --value PIPX_BIN_DIR", + ] + + +@pytest.mark.anyio +async def test_background_child_does_not_hold_command_output_open(tmp_path): + child = tmp_path / "done" + response = await asyncio.wait_for( + execute( + LocalRuntime(), + Command( + command=[ + "/bin/sh", + "-c", + f"(sleep 0.3; touch {child}) & printf foreground", + ] + ), + ), + 0.2, + ) + assert response.stdout == "foreground" + await asyncio.sleep(0.5) + assert child.exists() From f45c977a4bff2e93842871add39a6fc464781585 Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Sat, 31 Oct 2026 11:53:20 -0700 Subject: [PATCH 02/10] Settle output readers before command cancellation closes their files --- .../harness/sandbox/_http_command.py | 11 +++- tests/core/harness/test_http_command.py | 54 +++++++++++++++++-- 2 files changed, 60 insertions(+), 5 deletions(-) diff --git a/src/olmo_eval/harness/sandbox/_http_command.py b/src/olmo_eval/harness/sandbox/_http_command.py index 18319201d..f466eade8 100644 --- a/src/olmo_eval/harness/sandbox/_http_command.py +++ b/src/olmo_eval/harness/sandbox/_http_command.py @@ -52,7 +52,16 @@ def read_output() -> tuple[bytes, bytes]: ) # Regular files keep a detached child from holding communicate() open. - stdout, stderr = await asyncio.to_thread(read_output) + reader = asyncio.create_task(asyncio.to_thread(read_output)) + cancelled: asyncio.CancelledError | None = None + while True: + try: + stdout, stderr = await asyncio.shield(reader) + break + except asyncio.CancelledError as exc: + cancelled = exc + if cancelled is not None: + raise cancelled response = CommandResponse( stdout=stdout.decode(errors="backslashreplace"), stderr=stderr.decode(errors="backslashreplace"), diff --git a/tests/core/harness/test_http_command.py b/tests/core/harness/test_http_command.py index 1d506f70e..7ec16b4c8 100644 --- a/tests/core/harness/test_http_command.py +++ b/tests/core/harness/test_http_command.py @@ -4,6 +4,7 @@ import os import subprocess import sys +import threading import httpx import pytest @@ -163,6 +164,7 @@ def test_bootstrap_installs_with_module_when_pipx_is_not_on_path(tmp_path): @pytest.mark.anyio async def test_background_child_does_not_hold_command_output_open(tmp_path): child = tmp_path / "done" + release = tmp_path / "release" response = await asyncio.wait_for( execute( LocalRuntime(), @@ -170,12 +172,56 @@ async def test_background_child_does_not_hold_command_output_open(tmp_path): command=[ "/bin/sh", "-c", - f"(sleep 0.3; touch {child}) & printf foreground", + f"(while [ ! -f {release} ]; do sleep 0.05; done; touch {child}) & printf foreground", ] ), ), - 0.2, + 3, ) assert response.stdout == "foreground" - await asyncio.sleep(0.5) - assert child.exists() + release.touch() + async with asyncio.timeout(3): + while not child.exists(): + await asyncio.sleep(0.01) + + +@pytest.mark.anyio +async def test_cancellation_waits_for_output_reader_before_closing_files(monkeypatch): + import olmo_eval.harness.sandbox._http_command as adapter + + started = threading.Event() + release = threading.Event() + files = [] + original_file = adapter.tempfile.TemporaryFile + original_thread = asyncio.to_thread + + def temporary_file(): + file = original_file() + files.append(file) + return file + + async def to_thread(function, *args, **kwargs): + def blocked_read(): + started.set() + assert release.wait(5) + assert all(not file.closed for file in files) + return function(*args, **kwargs) + + return await original_thread(blocked_read) + + monkeypatch.setattr(adapter.tempfile, "TemporaryFile", temporary_file) + monkeypatch.setattr(adapter.asyncio, "to_thread", to_thread) + task = asyncio.create_task(execute(LocalRuntime(), Command(command=["/bin/echo", "output"]))) + try: + async with asyncio.timeout(3): + while not started.is_set(): + await asyncio.sleep(0.01) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + assert all(not file.closed for file in files) + finally: + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert all(file.closed for file in files) From cee8b8c5efba742a5f858e43870cc3ebea0ae9cf Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Sat, 31 Oct 2026 11:53:53 -0700 Subject: [PATCH 03/10] Wrap the background command regression fixture --- tests/core/harness/test_http_command.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/core/harness/test_http_command.py b/tests/core/harness/test_http_command.py index 7ec16b4c8..ec76de1eb 100644 --- a/tests/core/harness/test_http_command.py +++ b/tests/core/harness/test_http_command.py @@ -172,7 +172,8 @@ async def test_background_child_does_not_hold_command_output_open(tmp_path): command=[ "/bin/sh", "-c", - f"(while [ ! -f {release} ]; do sleep 0.05; done; touch {child}) & printf foreground", + f"(while [ ! -f {release} ]; do sleep 0.05; done; " + f"touch {child}) & printf foreground", ] ), ), From 820a21a253a096d87681568dcc9e3c2193739860 Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Sat, 31 Oct 2026 12:01:18 -0700 Subject: [PATCH 04/10] Harden remote command startup and deadlines across Python runtimes --- .../harness/sandbox/_http_command.py | 44 ++++-- .../harness/sandbox/server_command.py | 2 + tests/core/harness/test_http_command.py | 136 ++++++++++++++++-- 3 files changed, 164 insertions(+), 18 deletions(-) diff --git a/src/olmo_eval/harness/sandbox/_http_command.py b/src/olmo_eval/harness/sandbox/_http_command.py index f466eade8..e5fe6c8d0 100644 --- a/src/olmo_eval/harness/sandbox/_http_command.py +++ b/src/olmo_eval/harness/sandbox/_http_command.py @@ -4,6 +4,7 @@ import asyncio import contextlib +import fcntl import os import signal import tempfile @@ -18,6 +19,11 @@ async def execute(self: LocalRuntime, command: Command) -> CommandResponse: """Execute independent commands without occupying the HTTP server loop.""" with tempfile.TemporaryFile() as stdout_file, tempfile.TemporaryFile() as stderr_file: + # The child shares these offsets; append prevents background writes from + # overwriting captured output when the reader seeks back to the start. + for output in (stdout_file, stderr_file): + flags = fcntl.fcntl(output, fcntl.F_GETFL) + fcntl.fcntl(output, fcntl.F_SETFL, flags | os.O_APPEND) kwargs: dict[str, Any] = { "env": command.env, "cwd": command.cwd, @@ -28,15 +34,35 @@ async def execute(self: LocalRuntime, command: Command) -> CommandResponse: args = [command.command] if isinstance(command.command, str) else command.command if command.shell: # Match subprocess.run's sequence handling with shell=True. - process = await asyncio.create_subprocess_exec("/bin/sh", "-c", *args, **kwargs) - else: - process = await asyncio.create_subprocess_exec(*args, **kwargs) + args = ["/bin/sh", "-c", *args] + spawning = asyncio.create_task(asyncio.create_subprocess_exec(*args, **kwargs)) + cancelled: asyncio.CancelledError | None = None + while True: + try: + process = await asyncio.shield(spawning) + break + except asyncio.CancelledError as exc: + if spawning.cancelled(): + raise + cancelled = exc try: + if cancelled is not None: + raise cancelled await asyncio.wait_for(process.wait(), command.timeout) - except (TimeoutError, asyncio.CancelledError) as exc: + # SWE-ReX can run on Python 3.10, where this is not built-in TimeoutError. + except (asyncio.TimeoutError, asyncio.CancelledError) as exc: # noqa: UP041 with contextlib.suppress(ProcessLookupError): os.killpg(process.pid, signal.SIGKILL) - await asyncio.wait_for(process.wait(), 2.0) + cleanup = asyncio.create_task(asyncio.wait_for(process.wait(), 2.0)) + while not cleanup.done(): + try: + await asyncio.shield(cleanup) + except asyncio.CancelledError: + pass + except asyncio.TimeoutError: # noqa: UP041 + break + with contextlib.suppress(asyncio.TimeoutError): # noqa: UP041 + cleanup.result() if isinstance(exc, asyncio.CancelledError): raise raise CommandTimeoutError( @@ -44,11 +70,9 @@ async def execute(self: LocalRuntime, command: Command) -> CommandResponse: ) from exc def read_output() -> tuple[bytes, bytes]: - stdout_file.seek(0) - stderr_file.seek(0) return ( - stdout_file.read(os.fstat(stdout_file.fileno()).st_size), - stderr_file.read(os.fstat(stderr_file.fileno()).st_size), + os.pread(stdout_file.fileno(), os.fstat(stdout_file.fileno()).st_size, 0), + os.pread(stderr_file.fileno(), os.fstat(stderr_file.fileno()).st_size, 0), ) # Regular files keep a detached child from holding communicate() open. @@ -59,6 +83,8 @@ def read_output() -> tuple[bytes, bytes]: stdout, stderr = await asyncio.shield(reader) break except asyncio.CancelledError as exc: + if reader.cancelled(): + raise cancelled = exc if cancelled is not None: raise cancelled diff --git a/src/olmo_eval/harness/sandbox/server_command.py b/src/olmo_eval/harness/sandbox/server_command.py index 1c0eec4f3..047e98583 100644 --- a/src/olmo_eval/harness/sandbox/server_command.py +++ b/src/olmo_eval/harness/sandbox/server_command.py @@ -29,6 +29,8 @@ def server_command(executable: str, token: str, port: int, *, install_pipx: bool return ( f"original=$(command -v {shlex.quote(executable)}) || " f"{{ {fallback}; }} || exit 1; " + 'case "$(head -n 1 "$original")" in *python*) ;; ' + "*) echo 'SWE-ReX bootstrap requires a Python-shebang entrypoint' >&2; exit 1;; esac; " "bootstrap=$(mktemp -d) || exit 1; " "trap 'rm -rf \"$bootstrap\"' EXIT; " '{ head -n 1 "$original"; ' diff --git a/tests/core/harness/test_http_command.py b/tests/core/harness/test_http_command.py index ec76de1eb..c169c16e6 100644 --- a/tests/core/harness/test_http_command.py +++ b/tests/core/harness/test_http_command.py @@ -2,9 +2,11 @@ import asyncio import os +import signal import subprocess import sys import threading +from pathlib import Path import httpx import pytest @@ -84,24 +86,24 @@ async def test_command_preserves_environment_cwd_streams_and_status(tmp_path, sh async def test_deadline_and_cancellation_kill_descendants_holding_output(tmp_path, cancel): marker = tmp_path / "orphan" started = tmp_path / "started" - script = f"(sleep 0.7; touch {marker}) & touch {started}; sleep 30" + script = f"(sleep 2; touch {marker}) & touch {started}; sleep 30" task = asyncio.create_task( execute( LocalRuntime(), Command( command=["/bin/sh", "-c", script], - timeout=None if cancel else 0.1, + timeout=None if cancel else 1.0, ), ) ) + async with asyncio.timeout(3): + while not started.exists(): + await asyncio.sleep(0.01) if cancel: - async with asyncio.timeout(3): - while not started.exists(): - await asyncio.sleep(0.01) task.cancel() with pytest.raises(asyncio.CancelledError if cancel else CommandTimeoutError): await asyncio.wait_for(task, 3) - await asyncio.sleep(0.8) + await asyncio.sleep(2.1) assert not marker.exists() response = await execute(LocalRuntime(), Command(command=["/bin/echo", "recovered"])) assert response.stdout == "recovered\n" @@ -113,8 +115,9 @@ def test_bootstrap_uses_installed_shebang_and_preserves_shell_environment(tmp_pa entrypoint.write_text( f"#!{sys.executable}\n" "import os, sys\n" - "from swerex.runtime.local import LocalRuntime\n" - "print(LocalRuntime.execute.__module__, os.environ['BOOT_ENV'], sys.argv[1:])\n" + "from swerex.server import runtime\n" + "print(any(c.__name__ == 'ResponsiveLocalRuntime' for c in type(runtime).__mro__), " + "os.environ['BOOT_ENV'], sys.argv[1:])\n" ) entrypoint.chmod(0o700) executable = "swerex-remote" if use_path else str(entrypoint) @@ -126,7 +129,7 @@ def test_bootstrap_uses_installed_shebang_and_preserves_shell_environment(tmp_pa timeout=5, ) assert result.returncode == 0, result.stderr - assert result.stdout.startswith("__main__ preserved") + assert result.stdout.startswith("True preserved") assert "token with" in result.stdout assert "--port', '8123'" in result.stdout assert entrypoint.read_text().startswith(f"#!{sys.executable}\nimport os") @@ -226,3 +229,118 @@ def blocked_read(): with pytest.raises(asyncio.CancelledError): await task assert all(file.closed for file in files) + + +@pytest.mark.anyio +async def test_cancel_during_spawn_settles_and_kills_created_process(monkeypatch): + import olmo_eval.harness.sandbox._http_command as adapter + + created = asyncio.Event() + release = asyncio.Event() + original_spawn = asyncio.create_subprocess_exec + processes = [] + + async def spawn(*args, **kwargs): + process = await original_spawn(*args, **kwargs) + processes.append(process) + created.set() + await release.wait() + return process + + monkeypatch.setattr(adapter.asyncio, "create_subprocess_exec", spawn) + task = asyncio.create_task(execute(LocalRuntime(), Command(command=["/bin/sleep", "30"]))) + try: + await asyncio.wait_for(created.wait(), 3) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert processes[0].returncode is not None + finally: + release.set() + for process in processes: + if process.returncode is None: + os.killpg(process.pid, signal.SIGKILL) + await process.wait() + + +@pytest.mark.anyio +async def test_embedded_source_handles_legacy_asyncio_timeout(monkeypatch): + import olmo_eval.harness.sandbox._http_command as adapter + + class LegacyTimeoutError(Exception): + pass + + original_wait = asyncio.wait_for + + async def wait_for(awaitable, timeout): + if timeout == 0.01: + awaitable.close() + raise LegacyTimeoutError + return await original_wait(awaitable, timeout) + + namespace = {"__name__": "embedded_adapter"} + exec(compile(Path(adapter.__file__).read_text(), "embedded_adapter", "exec"), namespace) + monkeypatch.setattr(adapter.asyncio, "TimeoutError", LegacyTimeoutError) + monkeypatch.setattr(adapter.asyncio, "wait_for", wait_for) + with pytest.raises(CommandTimeoutError): + await namespace["execute"]( + LocalRuntime(), Command(command=["/bin/sleep", "30"], timeout=0.01) + ) + + +def test_bootstrap_rejects_shell_entrypoint_with_clear_error(tmp_path): + entrypoint = tmp_path / "swerex-remote" + entrypoint.write_text("#!/bin/sh\necho should-not-run\n") + entrypoint.chmod(0o700) + result = subprocess.run( + ["/bin/sh", "-c", server_command(str(entrypoint), "token", 8123)], + text=True, + capture_output=True, + timeout=3, + ) + assert result.returncode == 1 + assert "requires a Python-shebang entrypoint" in result.stderr + assert not result.stdout + + +@pytest.mark.anyio +async def test_background_writes_do_not_move_the_readers_position(monkeypatch, tmp_path): + import olmo_eval.harness.sandbox._http_command as adapter + + release = tmp_path / "release" + done = tmp_path / "done" + original_pread = adapter.os.pread + + def pread(descriptor, size, offset): + if size: + release.touch() + # The real child appends between the reader's size snapshot and read. + import time + + until = time.monotonic() + 3 + while not done.exists(): + assert time.monotonic() < until + time.sleep(0.01) + return original_pread(descriptor, size, offset) + + monkeypatch.setattr(adapter.os, "pread", pread) + try: + response = await execute( + LocalRuntime(), + Command( + command=[ + "/bin/sh", + "-c", + "printf foreground; " + f"(while [ ! -f {release} ]; do sleep 0.01; done; " + f"printf background; touch {done}) &", + ] + ), + ) + assert done.exists() + assert response.stdout == "foreground" + finally: + release.touch() From 493f4e2276f2ffd6b417a53dc5a3d21dfe97afbf Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Sat, 31 Oct 2026 12:01:47 -0700 Subject: [PATCH 05/10] Run persistent shell operations without blocking sandbox HTTP --- README.md | 3 +- .../harness/sandbox/_http_session.py | 73 +++++++++ .../harness/sandbox/server_command.py | 12 +- tests/core/harness/test_http_session.py | 142 ++++++++++++++++++ 4 files changed, 223 insertions(+), 7 deletions(-) create mode 100644 src/olmo_eval/harness/sandbox/_http_session.py create mode 100644 tests/core/harness/test_http_session.py diff --git a/README.md b/README.md index 34746eb91..e74287dff 100644 --- a/README.md +++ b/README.md @@ -1439,7 +1439,8 @@ supervision is available; other container and Modal paths use asynchronous HTTP commands so a long command does not block health checks or file requests. Command deadlines terminate the command process group. Commands that finish normally can leave background processes running. Persistent bash sessions retain -shell state between calls and currently use SWE-ReX's session execution path. +shell state between calls. Each session serializes shell operations in a worker +thread, so waiting for its prompt leaves health checks and file requests responsive. ### Sandbox Modes diff --git a/src/olmo_eval/harness/sandbox/_http_session.py b/src/olmo_eval/harness/sandbox/_http_session.py new file mode 100644 index 000000000..b68798970 --- /dev/null +++ b/src/olmo_eval/harness/sandbox/_http_session.py @@ -0,0 +1,73 @@ +"""Serialize persistent shell operations without blocking the HTTP server.""" + +from __future__ import annotations + +import asyncio +import contextlib +from collections.abc import Awaitable, Callable +from typing import Any, TypeVar + +from swerex.runtime import local +from swerex.runtime.abstract import ( + BashAction, + BashInterruptAction, + BashObservation, + CloseSessionResponse, + CreateBashSessionResponse, +) +from swerex.runtime.local import BashSession + +_T = TypeVar("_T") + + +class ResponsiveBashSession(BashSession): + """Keep one pexpect reader per session while other HTTP requests run.""" + + def __init__(self, *args: Any, **kwargs: Any): + super().__init__(*args, **kwargs) + self._operation_lock = asyncio.Lock() + + async def _run_worker(self, operation: Callable[[], Awaitable[_T]]) -> _T: + async with self._operation_lock: + + def run() -> _T: + async def wait() -> _T: + return await operation() + + return asyncio.run(wait()) + + task = asyncio.create_task(asyncio.to_thread(run)) + cancelled = False + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError: + cancelled = True + # Sending a signal does not consume pexpect output. Only the + # worker may read the shell until it reaches its deadline. + with contextlib.suppress(RuntimeError, OSError): + self.shell.sendintr() + except Exception: + break + if cancelled: + # Retrieve any worker exception before propagating cancellation. + with contextlib.suppress(Exception): + task.result() + raise asyncio.CancelledError + return task.result() + + async def start(self) -> CreateBashSessionResponse: + return await self._run_worker(super().start) + + async def run(self, action: BashAction | BashInterruptAction) -> BashObservation: + if isinstance(action, BashInterruptAction) and self._operation_lock.locked(): + self.shell.sendintr() + return await self._run_worker(lambda: super(ResponsiveBashSession, self).run(action)) + + async def close(self) -> CloseSessionResponse: + return await self._run_worker(super().close) + + +def install() -> None: + """Install persistent sessions before the runtime creates any shell.""" + vars(local).update(BashSession=ResponsiveBashSession) diff --git a/src/olmo_eval/harness/sandbox/server_command.py b/src/olmo_eval/harness/sandbox/server_command.py index 1c0eec4f3..7aacd3d9b 100644 --- a/src/olmo_eval/harness/sandbox/server_command.py +++ b/src/olmo_eval/harness/sandbox/server_command.py @@ -7,12 +7,12 @@ def server_command(executable: str, token: str, port: int, *, install_pipx: bool = False) -> str: """Preserve the server interpreter and shell environment while installing its adapter.""" - source = Path(__file__).with_name("_http_command.py").read_bytes() - payload = base64.b64encode(source + b"\ninstall()\n").decode() - patch = ( - f"import base64; exec(compile(base64.b64decode({payload!r}), " - "'', 'exec'))\n" - ) + patches = [] + for module in ("_http_command.py", "_http_session.py"): + source = Path(__file__).with_name(module).read_bytes() + payload = base64.b64encode(source + b"\ninstall()\n").decode() + patches.append(f"exec(compile(base64.b64decode({payload!r}), {module!r}, 'exec'))") + patch = "import base64; " + "; ".join(patches) + "\n" encoded = base64.b64encode(patch.encode()).decode() # A temporary console script retains the installed entrypoint's shebang. # This works for an isolated SWE-ReX venv as well as an activated task venv. diff --git a/tests/core/harness/test_http_session.py b/tests/core/harness/test_http_session.py new file mode 100644 index 000000000..8a58c93e3 --- /dev/null +++ b/tests/core/harness/test_http_session.py @@ -0,0 +1,142 @@ +"""Persistent shell concurrency and deadline regressions using real pexpect.""" + +import asyncio + +import httpx +import pytest +from swerex.exceptions import CommandTimeoutError +from swerex.runtime import local +from swerex.runtime.abstract import BashAction, BashInterruptAction, CreateBashSessionRequest + +from olmo_eval.harness.sandbox._http_session import ResponsiveBashSession + + +@pytest.mark.anyio +async def test_http_session_keeps_file_requests_responsive(monkeypatch, tmp_path): + import swerex.server as server + + monkeypatch.setattr(local, "BashSession", ResponsiveBashSession) + runtime = local.LocalRuntime() + monkeypatch.setattr(server, "runtime", runtime) + marker = tmp_path / "started" + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=server.app), base_url="http://sandbox" + ) as client: + created = await client.post( + "/create_session", json=CreateBashSessionRequest(session="test").model_dump() + ) + assert created.status_code == 200 + task = asyncio.create_task( + client.post( + "/run_in_session", + json={ + "session": "test", + "command": f"touch {marker}; sleep 1", + "timeout": 3, + "check": "silent", + "action_type": "bash", + }, + ) + ) + try: + async with asyncio.timeout(3): + while not marker.exists(): + await asyncio.sleep(0.01) + response = await client.post( + "/write_file", json={"path": str(tmp_path / "file"), "content": "ready"} + ) + assert response.status_code == 200 and not task.done() + assert (await task).status_code == 200 + finally: + await task + await runtime.close() + + +@pytest.mark.anyio +async def test_session_serializes_commands_and_retains_environment(tmp_path): + session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) + await session.start() + try: + first = asyncio.create_task( + session.run( + BashAction( + command=f"sleep 0.1; cd {tmp_path}; export PERSIST=retained", + timeout=3, + ) + ) + ) + await asyncio.sleep(0.01) + second = asyncio.create_task( + session.run(BashAction(command='printf "%s:%s" "$PERSIST" "$PWD"', timeout=3)) + ) + await first + assert (await second).output == f"retained:{tmp_path}" + finally: + await session.close() + + +@pytest.mark.anyio +async def test_native_deadline_and_interrupt_recover_session(): + session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) + await session.start() + try: + await session.run(BashAction(command="echo ready", timeout=3, check="silent")) + with pytest.raises(CommandTimeoutError): + await session.run(BashAction(command="sleep 30", timeout=0.5, check="silent")) + await session.run(BashInterruptAction()) + assert ( + "recovered" + in (await session.run(BashAction(command="echo recovered", timeout=3))).output + ) + finally: + await session.close() + + +@pytest.mark.anyio +async def test_cancellation_settles_reader_before_next_command(tmp_path): + session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) + await session.start() + marker = tmp_path / "started" + try: + task = asyncio.create_task( + session.run(BashAction(command=f"touch {marker}; sleep 30", timeout=3)) + ) + async with asyncio.timeout(3): + while not marker.exists(): + await asyncio.sleep(0.01) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 5) + assert ( + "recovered" + in (await session.run(BashAction(command="echo recovered", timeout=3))).output + ) + finally: + await session.close() + + +@pytest.mark.anyio +async def test_interrupt_can_signal_an_active_serialized_reader(tmp_path): + session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) + await session.start() + marker = tmp_path / "started" + try: + task = asyncio.create_task( + session.run( + BashAction( + command=f"touch {marker}; sleep 30", + timeout=3, + check="silent", + ) + ) + ) + async with asyncio.timeout(3): + while not marker.exists(): + await asyncio.sleep(0.01) + await asyncio.wait_for(session.run(BashInterruptAction()), 3) + assert task.done() + await task + response = await session.run(BashAction(command="echo recovered", timeout=3)) + assert "recovered" in response.output + finally: + await session.close() From 8da9322c9887c779e4be5c73e2f4368e1965dea0 Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Sat, 31 Oct 2026 12:11:28 -0700 Subject: [PATCH 06/10] Bound session interruption and serialize lifecycle mutations --- .../harness/sandbox/_http_session.py | 105 +++++++++++++++--- tests/core/harness/test_http_session.py | 103 ++++++++++++++++- 2 files changed, 191 insertions(+), 17 deletions(-) diff --git a/src/olmo_eval/harness/sandbox/_http_session.py b/src/olmo_eval/harness/sandbox/_http_session.py index b68798970..38445ec34 100644 --- a/src/olmo_eval/harness/sandbox/_http_session.py +++ b/src/olmo_eval/harness/sandbox/_http_session.py @@ -4,6 +4,8 @@ import asyncio import contextlib +import os +import signal from collections.abc import Awaitable, Callable from typing import Any, TypeVar @@ -12,8 +14,12 @@ BashAction, BashInterruptAction, BashObservation, + CloseResponse, + CloseSessionRequest, CloseSessionResponse, CreateBashSessionResponse, + CreateSessionRequest, + CreateSessionResponse, ) from swerex.runtime.local import BashSession @@ -26,31 +32,76 @@ class ResponsiveBashSession(BashSession): def __init__(self, *args: Any, **kwargs: Any): super().__init__(*args, **kwargs) self._operation_lock = asyncio.Lock() + self._active_worker: asyncio.Task[Any] | None = None + self._closing = False - async def _run_worker(self, operation: Callable[[], Awaitable[_T]]) -> _T: + async def _interrupt_active(self, task: asyncio.Task[Any]) -> None: + if self._active_worker is not task or task.done(): + return + with contextlib.suppress(RuntimeError, OSError): + self.shell.sendintr() + try: + await asyncio.wait_for(asyncio.shield(task), 0.5) + return + except asyncio.TimeoutError: # noqa: UP041 -- remote Python 3.10 + pass + except Exception: + return + if self._active_worker is not task or task.done(): + return + # A foreground job may ignore SIGINT. Signal only this terminal's + # foreground group, leaving the server and other sessions untouched. + with contextlib.suppress(RuntimeError, OSError): + foreground = os.tcgetpgrp(self.shell.child_fd) + if foreground > 0 and foreground != os.getpgrp(): + os.killpg(foreground, signal.SIGKILL) + try: + await asyncio.wait_for(asyncio.shield(task), 1.0) + return + except asyncio.TimeoutError: # noqa: UP041 -- remote Python 3.10 + pass + except Exception: + return + + # Closing an unresponsive terminal wakes its remaining pexpect reader. + def close_terminal() -> None: + if self._active_worker is task and not task.done(): + self.shell.close(force=True) + + await asyncio.to_thread(close_terminal) + with contextlib.suppress(Exception): + await asyncio.shield(task) + + async def _run_worker( + self, operation: Callable[[], Awaitable[_T]], *, allow_closing: bool = False + ) -> _T: async with self._operation_lock: + if self._closing and not allow_closing: + raise RuntimeError("Session is closing") - def run() -> _T: - async def wait() -> _T: - return await operation() + async def run_operation() -> _T: + return await operation() - return asyncio.run(wait()) + def run() -> _T: + return asyncio.run(run_operation()) task = asyncio.create_task(asyncio.to_thread(run)) + self._active_worker = task cancelled = False while not task.done(): try: await asyncio.shield(task) except asyncio.CancelledError: cancelled = True - # Sending a signal does not consume pexpect output. Only the - # worker may read the shell until it reaches its deadline. - with contextlib.suppress(RuntimeError, OSError): - self.shell.sendintr() + # Keep the lock until the sole reader has settled. + try: + await self._interrupt_active(task) + except asyncio.CancelledError: + continue except Exception: break + self._active_worker = None if cancelled: - # Retrieve any worker exception before propagating cancellation. with contextlib.suppress(Exception): task.result() raise asyncio.CancelledError @@ -60,14 +111,40 @@ async def start(self) -> CreateBashSessionResponse: return await self._run_worker(super().start) async def run(self, action: BashAction | BashInterruptAction) -> BashObservation: - if isinstance(action, BashInterruptAction) and self._operation_lock.locked(): - self.shell.sendintr() + if isinstance(action, BashInterruptAction) and self._active_worker is not None: + await self._interrupt_active(self._active_worker) + # The existing reader consumed the prompt. Reading it again could + # block when the shell itself ignores SIGINT. + return BashObservation(output="", exit_code=0) return await self._run_worker(lambda: super(ResponsiveBashSession, self).run(action)) async def close(self) -> CloseSessionResponse: - return await self._run_worker(super().close) + self._closing = True + if self._active_worker is not None: + await self._interrupt_active(self._active_worker) + return await self._run_worker(super().close, allow_closing=True) + + +class ResponsiveSessionRuntime(local.LocalRuntime): + """Serialize registry mutations across offloaded shell lifecycles.""" + + def __init__(self, **kwargs: Any): + super().__init__(**kwargs) + self._lifecycle_lock = asyncio.Lock() + + async def create_session(self, request: CreateSessionRequest) -> CreateSessionResponse: + async with self._lifecycle_lock: + return await super().create_session(request) + + async def close_session(self, request: CloseSessionRequest) -> CloseSessionResponse: + async with self._lifecycle_lock: + return await super().close_session(request) + + async def close(self) -> CloseResponse: + async with self._lifecycle_lock: + return await super().close() def install() -> None: """Install persistent sessions before the runtime creates any shell.""" - vars(local).update(BashSession=ResponsiveBashSession) + vars(local).update(BashSession=ResponsiveBashSession, LocalRuntime=ResponsiveSessionRuntime) diff --git a/tests/core/harness/test_http_session.py b/tests/core/harness/test_http_session.py index 8a58c93e3..330b0c0dd 100644 --- a/tests/core/harness/test_http_session.py +++ b/tests/core/harness/test_http_session.py @@ -4,11 +4,16 @@ import httpx import pytest -from swerex.exceptions import CommandTimeoutError +from swerex.exceptions import CommandTimeoutError, SessionDoesNotExistError from swerex.runtime import local -from swerex.runtime.abstract import BashAction, BashInterruptAction, CreateBashSessionRequest +from swerex.runtime.abstract import ( + BashAction, + BashInterruptAction, + CloseSessionRequest, + CreateBashSessionRequest, +) -from olmo_eval.harness.sandbox._http_session import ResponsiveBashSession +from olmo_eval.harness.sandbox._http_session import ResponsiveBashSession, ResponsiveSessionRuntime @pytest.mark.anyio @@ -140,3 +145,95 @@ async def test_interrupt_can_signal_an_active_serialized_reader(tmp_path): assert "recovered" in response.output finally: await session.close() + + +@pytest.mark.anyio +@pytest.mark.parametrize("operation", ["cancel", "interrupt", "close"]) +async def test_ignored_interrupt_without_deadline_does_not_lock_session(tmp_path, operation): + session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) + await session.start() + marker = tmp_path / "started" + task = asyncio.create_task( + session.run( + BashAction( + command=f"trap '' INT; touch {marker}; sleep 30", + timeout=None, + check="silent", + ) + ) + ) + try: + async with asyncio.timeout(3): + while not marker.exists(): + await asyncio.sleep(0.01) + if operation == "cancel": + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 4) + elif operation == "interrupt": + await asyncio.wait_for(session.run(BashInterruptAction()), 4) + await task + else: + await asyncio.wait_for(session.close(), 4) + await task + assert task.done() + if operation != "close": + response = await session.run( + BashAction(command="trap - INT; echo recovered", timeout=3) + ) + assert "recovered" in response.output + finally: + await asyncio.wait_for(session.close(), 4) + + +@pytest.mark.anyio +async def test_concurrent_close_uses_domain_error(monkeypatch): + monkeypatch.setattr(local, "BashSession", ResponsiveBashSession) + runtime = ResponsiveSessionRuntime() + await runtime.create_session(CreateBashSessionRequest(session="test")) + responses = await asyncio.gather( + runtime.close_session(CloseSessionRequest(session="test")), + runtime.close_session(CloseSessionRequest(session="test")), + return_exceptions=True, + ) + assert sum(isinstance(value, SessionDoesNotExistError) for value in responses) == 1 + assert sum(not isinstance(value, Exception) for value in responses) == 1 + + +@pytest.mark.anyio +async def test_runtime_close_and_create_do_not_mutate_registry_during_iteration(monkeypatch): + monkeypatch.setattr(local, "BashSession", ResponsiveBashSession) + runtime = ResponsiveSessionRuntime() + await runtime.create_session(CreateBashSessionRequest(session="first")) + try: + closing = asyncio.create_task(runtime.close()) + await asyncio.sleep(0) + creating = asyncio.create_task( + runtime.create_session(CreateBashSessionRequest(session="second")) + ) + await asyncio.gather(closing, creating) + assert "second" in runtime.sessions + finally: + await runtime.close() + + +@pytest.mark.anyio +async def test_close_prevents_a_queued_unbounded_command_from_starting(tmp_path): + session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) + await session.start() + marker = tmp_path / "started" + first = asyncio.create_task( + session.run(BashAction(command=f"touch {marker}; sleep 30", timeout=None, check="silent")) + ) + try: + async with asyncio.timeout(3): + while not marker.exists(): + await asyncio.sleep(0.01) + queued = asyncio.create_task(session.run(BashAction(command="sleep 30", timeout=None))) + await asyncio.sleep(0) + await asyncio.wait_for(session.close(), 4) + await first + with pytest.raises(RuntimeError, match="closing"): + await queued + finally: + await asyncio.wait_for(session.close(), 4) From 7905755ae29aaaa7eaeb8f75184bac0d748e91d7 Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Sat, 31 Oct 2026 12:14:50 -0700 Subject: [PATCH 07/10] Share sandbox container-routing predicates --- .../harness/sandbox/_http_command.py | 3 +- src/olmo_eval/harness/sandbox/executor.py | 39 ++++++++++--------- 2 files changed, 21 insertions(+), 21 deletions(-) diff --git a/src/olmo_eval/harness/sandbox/_http_command.py b/src/olmo_eval/harness/sandbox/_http_command.py index e5fe6c8d0..381ba70f8 100644 --- a/src/olmo_eval/harness/sandbox/_http_command.py +++ b/src/olmo_eval/harness/sandbox/_http_command.py @@ -19,8 +19,7 @@ async def execute(self: LocalRuntime, command: Command) -> CommandResponse: """Execute independent commands without occupying the HTTP server loop.""" with tempfile.TemporaryFile() as stdout_file, tempfile.TemporaryFile() as stderr_file: - # The child shares these offsets; append prevents background writes from - # overwriting captured output when the reader seeks back to the start. + # Keep inherited output append-only even if a descriptor's offset changes. for output in (stdout_file, stderr_file): flags = fcntl.fcntl(output, fcntl.F_GETFL) fcntl.fcntl(output, fcntl.F_SETFL, flags | os.O_APPEND) diff --git a/src/olmo_eval/harness/sandbox/executor.py b/src/olmo_eval/harness/sandbox/executor.py index 94e6a3482..84b704589 100644 --- a/src/olmo_eval/harness/sandbox/executor.py +++ b/src/olmo_eval/harness/sandbox/executor.py @@ -210,6 +210,20 @@ def __init__( self._health_check_lock: asyncio.Lock = asyncio.Lock() self._local_command_supported: bool | None = None + @property + def _has_container_cli(self) -> bool: + return self.config.mode == SandboxMode.DOCKER and bool( + getattr(self._deployment, "container_name", None) + ) + + @property + def _can_supervise_local_commands(self) -> bool: + return ( + self._has_container_cli + and not self.config.exec_shell + and self._local_command_supported is not False + ) + def _log(self, level: int, msg: str) -> None: """Log a message with optional name prefix.""" if self.name: @@ -284,9 +298,7 @@ def patched_lookup(name: str, *args, **kwargs): self._quarantined_reason = None if self.config.exec_shell: self._local_command_supported = False - elif self.config.mode == SandboxMode.DOCKER and getattr( - self._deployment, "container_name", None - ): + elif self._has_container_cli: probe = await self._execute_stream_control( [ "sh", @@ -607,12 +619,7 @@ async def execute_command( if stream: return await self._execute_streaming(command, effective_timeout, prefix) - if ( - self.config.mode == SandboxMode.DOCKER - and getattr(self._deployment, "container_name", None) - and not self.config.exec_shell - and self._local_command_supported is not False - ): + if self._can_supervise_local_commands: return await self._execute_local_command(command, effective_timeout) return await self._execute_remote_command(["bash", "-c", command], effective_timeout) @@ -1032,7 +1039,8 @@ async def _execute_stream_control( """Run a sandbox command with bounded captured output and a client deadline.""" argv = ["bash", "-c", command] if isinstance(command, str) else list(command) container_name = getattr(self._deployment, "container_name", None) - if self.config.mode == SandboxMode.DOCKER and container_name: + if self._has_container_cli: + assert container_name is not None process = await asyncio.create_subprocess_exec( self.config.container_runtime, "exec", @@ -1259,12 +1267,7 @@ async def execute_code( try: effective_timeout = timeout if timeout is not None else self.config.command_timeout argv = [interpreter, "-c", code] - if ( - self.config.mode == SandboxMode.DOCKER - and getattr(self._deployment, "container_name", None) - and not self.config.exec_shell - and self._local_command_supported is not False - ): + if self._can_supervise_local_commands: return await self._execute_local_command(argv, effective_timeout) return await self._execute_remote_command(argv, effective_timeout) except SandboxTransportError: @@ -1468,9 +1471,7 @@ async def _quarantine_if_unresponsive(self, transport_error: Exception) -> None: last_health_error = "health check returned false" for attempt in range(1, max_attempts + 1): try: - if self.config.mode == SandboxMode.DOCKER and getattr( - self._deployment, "container_name", None - ): + if self._has_container_cli: probe = await self._execute_stream_control( "true", timeout=_HEALTH_CHECK_TIMEOUT ) From a4be0e17f26c694bd44d9c32d58e06ce907d23c8 Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Sat, 31 Oct 2026 12:55:01 -0700 Subject: [PATCH 08/10] Finish session interrupt cleanup after reader deadlines --- .../harness/sandbox/_http_session.py | 102 +++++- tests/core/harness/test_http_session.py | 299 +++++++++++++++++- 2 files changed, 391 insertions(+), 10 deletions(-) diff --git a/src/olmo_eval/harness/sandbox/_http_session.py b/src/olmo_eval/harness/sandbox/_http_session.py index 38445ec34..4a66f6aab 100644 --- a/src/olmo_eval/harness/sandbox/_http_session.py +++ b/src/olmo_eval/harness/sandbox/_http_session.py @@ -9,6 +9,7 @@ from collections.abc import Awaitable, Callable from typing import Any, TypeVar +from swerex.exceptions import CommandTimeoutError from swerex.runtime import local from swerex.runtime.abstract import ( BashAction, @@ -34,30 +35,89 @@ def __init__(self, *args: Any, **kwargs: Any): self._operation_lock = asyncio.Lock() self._active_worker: asyncio.Task[Any] | None = None self._closing = False + self._interrupt_task: asyncio.Task[None] | None = None + + def _kill_foreground(self, *, allow_shell: bool = True) -> None: + with contextlib.suppress(RuntimeError, OSError): + foreground = os.tcgetpgrp(self.shell.child_fd) + if ( + foreground > 0 + and foreground != os.getpgrp() + and (allow_shell or foreground != self.shell.pid) + ): + os.killpg(foreground, signal.SIGKILL) + + async def _finish_reader_timeout(self) -> None: + # Ownership is retained, so a non-shell foreground group still belongs + # to this action, even if its command advanced to another external job. + self._kill_foreground(allow_shell=False) + + def drain_prompt() -> None: + self.shell.expect(self._ps1, timeout=1.0) + + # Another SIGINT could flush the completed job's prompt without creating + # a replacement when the persistent shell ignores SIGINT. + try: + await asyncio.to_thread(drain_prompt) + except Exception as error: + self._closing = True + self._kill_foreground(allow_shell=False) + + def close_terminal() -> None: + shell = self.shell + try: + shell.close(force=True) + finally: + if shell.isalive() and not shell.terminate(force=True): + raise RuntimeError("Session terminal could not be terminated") + + try: + await asyncio.to_thread(close_terminal) + except Exception as cleanup_error: + raise error from cleanup_error + raise async def _interrupt_active(self, task: asyncio.Task[Any]) -> None: - if self._active_worker is not task or task.done(): + if self._active_worker is not task: + return + if self._interrupt_task is None: + self._interrupt_task = asyncio.create_task(self._stop_active(task)) + await asyncio.shield(self._interrupt_task) + + async def _stop_active(self, task: asyncio.Task[Any]) -> None: + if self._active_worker is not task: + return + if task.done(): + if not task.cancelled() and isinstance(task.exception(), CommandTimeoutError): + await self._finish_reader_timeout() return with contextlib.suppress(RuntimeError, OSError): self.shell.sendintr() try: await asyncio.wait_for(asyncio.shield(task), 0.5) return + except CommandTimeoutError: + await self._finish_reader_timeout() + return except asyncio.TimeoutError: # noqa: UP041 -- remote Python 3.10 pass except Exception: return - if self._active_worker is not task or task.done(): + if self._active_worker is not task: + return + if task.done(): + if not task.cancelled() and isinstance(task.exception(), CommandTimeoutError): + await self._finish_reader_timeout() return # A foreground job may ignore SIGINT. Signal only this terminal's # foreground group, leaving the server and other sessions untouched. - with contextlib.suppress(RuntimeError, OSError): - foreground = os.tcgetpgrp(self.shell.child_fd) - if foreground > 0 and foreground != os.getpgrp(): - os.killpg(foreground, signal.SIGKILL) + self._kill_foreground() try: await asyncio.wait_for(asyncio.shield(task), 1.0) return + except CommandTimeoutError: + await self._finish_reader_timeout() + return except asyncio.TimeoutError: # noqa: UP041 -- remote Python 3.10 pass except Exception: @@ -69,8 +129,12 @@ def close_terminal() -> None: self.shell.close(force=True) await asyncio.to_thread(close_terminal) - with contextlib.suppress(Exception): + try: await asyncio.shield(task) + except CommandTimeoutError: + await self._finish_reader_timeout() + except Exception: + pass async def _run_worker( self, operation: Callable[[], Awaitable[_T]], *, allow_closing: bool = False @@ -100,11 +164,30 @@ def run() -> _T: continue except Exception: break + # A reader deadline does not stop its command. Keep ownership until + # the shared interrupt cleanup and its prompt read have settled. + control = self._interrupt_task + if control is not None: + while not control.done(): + try: + await asyncio.shield(control) + except asyncio.CancelledError: + cancelled = True + except Exception: + break + error = control.exception() + if error is not None: + self._closing = True + else: + error = None + self._interrupt_task = None self._active_worker = None if cancelled: with contextlib.suppress(Exception): task.result() raise asyncio.CancelledError + if error is not None: + raise error return task.result() async def start(self) -> CreateBashSessionResponse: @@ -121,7 +204,10 @@ async def run(self, action: BashAction | BashInterruptAction) -> BashObservation async def close(self) -> CloseSessionResponse: self._closing = True if self._active_worker is not None: - await self._interrupt_active(self._active_worker) + try: + await self._interrupt_active(self._active_worker) + except Exception as exc: + self.logger.warning("Session interrupt cleanup failed during close: %s", exc) return await self._run_worker(super().close, allow_closing=True) diff --git a/tests/core/harness/test_http_session.py b/tests/core/harness/test_http_session.py index 330b0c0dd..a72ec91ce 100644 --- a/tests/core/harness/test_http_session.py +++ b/tests/core/harness/test_http_session.py @@ -139,8 +139,7 @@ async def test_interrupt_can_signal_an_active_serialized_reader(tmp_path): while not marker.exists(): await asyncio.sleep(0.01) await asyncio.wait_for(session.run(BashInterruptAction()), 3) - assert task.done() - await task + await asyncio.wait_for(task, 1) response = await session.run(BashAction(command="echo recovered", timeout=3)) assert "recovered" in response.output finally: @@ -237,3 +236,299 @@ async def test_close_prevents_a_queued_unbounded_command_from_starting(tmp_path) await queued finally: await asyncio.wait_for(session.close(), 4) + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "operation", + [ + "interrupt", + "cancel", + "both", + "cancel-during-cleanup", + "natural-exit", + "late-cleanup", + "grace-boundary", + "terminal-close-boundary", + ], +) +async def test_deadline_during_interrupt_grace_keeps_queued_action_safe( + monkeypatch, tmp_path, operation +): + import contextlib + import os + import signal + import threading + import time + + import pexpect + + session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) + await session.start() + for _ in range(3): + ready = await session.run(BashAction(command="echo warmed-up", timeout=3, check="silent")) + if "warmed-up" in ready.output: + break + assert "warmed-up" in ready.output + marker = tmp_path / "started" + finish = tmp_path / "finish" + finished = tmp_path / "finished" + second_started = tmp_path / "second-started" + expire = threading.Event() + interrupted = threading.Event() + draining = threading.Event() + cancel_requested = threading.Event() + readers_lock = threading.Lock() + readers = 0 + peak_readers = 0 + original_expect = session.shell.expect + original_sendintr = session.shell.sendintr + + def expect(pattern, timeout=-1, **kwargs): + nonlocal readers, peak_readers + with readers_lock: + readers += 1 + peak_readers = max(readers, peak_readers) + try: + if timeout == 0.1: + # Start the real deadline after SIGINT, inside its grace period. + assert interrupted.wait(3) + if operation == "terminal-close-boundary": + assert expire.wait(3) + elif timeout == 1.0 and interrupted.is_set(): + draining.set() + if operation == "cancel-during-cleanup": + assert cancel_requested.wait(3) + try: + return original_expect(pattern, timeout=timeout, **kwargs) + except pexpect.TIMEOUT: + if operation == "natural-exit" and timeout == 0.1: + finish.touch() + until = time.monotonic() + 3 + while not finished.exists(): + assert time.monotonic() < until + time.sleep(0.01) + raise + finally: + with readers_lock: + readers -= 1 + + def sendintr(): + original_sendintr() + interrupted.set() + + monkeypatch.setattr(session.shell, "expect", expect) + monkeypatch.setattr(session.shell, "sendintr", sendintr) + if operation == "late-cleanup": + original_stop = session._stop_active + + async def stop(*args): + # Let the real reader expire before its scheduled cleanup starts. + interrupted.set() + while not args[0].done(): + await asyncio.sleep(0.01) + await original_stop(*args) + + monkeypatch.setattr(session, "_stop_active", stop) + if operation in ("grace-boundary", "terminal-close-boundary"): + original_wait_for = asyncio.wait_for + + async def wait_for(awaitable, timeout): + if operation == "terminal-close-boundary" and timeout in (0.5, 1.0): + awaitable.cancel() + raise TimeoutError + try: + return await original_wait_for(awaitable, timeout) + except CommandTimeoutError: + if timeout == 0.5: + # The grace timer and reader deadline can become ready together. + raise TimeoutError from None + raise + + monkeypatch.setattr(asyncio, "wait_for", wait_for) + if operation == "terminal-close-boundary": + original_to_thread = asyncio.to_thread + + async def to_thread(function, *args, **kwargs): + if function.__name__ == "close_terminal": + async with asyncio.timeout(3): + while not second_started.exists(): + await asyncio.sleep(0.01) + expire.set() + while not session._active_worker.done(): + await asyncio.sleep(0.01) + return await original_to_thread(function, *args, **kwargs) + + monkeypatch.setattr(asyncio, "to_thread", to_thread) + work = ( + f"while [ ! -f {finish} ]; do sleep 0.01; done; touch {finished}" + if operation == "natural-exit" + else "sleep 30" + ) + command = f"trap '' INT; /bin/sh -c 'touch {marker}; {work}'" + if operation == "terminal-close-boundary": + command += f"; /bin/sh -c 'echo $$ > {second_started}; sleep 30'" + first = asyncio.create_task( + session.run(BashAction(command=command, timeout=0.1, check="silent")) + ) + foreground = None + queued = None + interrupt_task = None + try: + async with asyncio.timeout(3): + while not marker.exists(): + await asyncio.sleep(0.01) + foreground = os.tcgetpgrp(session.shell.child_fd) + assert foreground != os.getpgrp() + queued = asyncio.create_task( + session.run( + BashAction( + command="trap - INT; echo queued-recovered", + timeout=2, + check="silent", + ) + ) + ) + await asyncio.sleep(0) + if operation in ( + "interrupt", + "natural-exit", + "late-cleanup", + "grace-boundary", + "terminal-close-boundary", + ): + await asyncio.wait_for(session.run(BashInterruptAction()), 4) + with pytest.raises(CommandTimeoutError): + await first + else: + if operation != "cancel": + interrupt_task = asyncio.create_task(session.run(BashInterruptAction())) + signal_seen = interrupted if operation == "both" else draining + async with asyncio.timeout(3): + while not signal_seen.is_set(): + await asyncio.sleep(0.01) + first.cancel() + cancel_requested.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(first, 4) + if interrupt_task is not None: + await asyncio.wait_for(interrupt_task, 4) + response = await asyncio.wait_for(queued, 4) + assert response.output.strip() == "queued-recovered" + assert peak_readers == 1 + finally: + interrupted.set() + cancel_requested.set() + expire.set() + finish.touch() + if second_started.exists(): + with contextlib.suppress(ProcessLookupError): + os.killpg(int(second_started.read_text()), signal.SIGKILL) + if foreground is not None: + with contextlib.suppress(ProcessLookupError): + os.killpg(foreground, signal.SIGKILL) + for task in (first, queued, interrupt_task): + if task is not None: + with contextlib.suppress(Exception, asyncio.CancelledError): + await task + await session.close() + + +@pytest.mark.anyio +@pytest.mark.parametrize("concurrent_close", [False, True]) +async def test_failed_prompt_drain_terminates_compound_action( + monkeypatch, tmp_path, concurrent_close +): + import contextlib + import os + import signal + import subprocess + import threading + + import pexpect + + session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) + await session.start() + for _ in range(3): + ready = await session.run(BashAction(command="echo warmed-up", timeout=3, check="silent")) + if "warmed-up" in ready.output: + break + assert "warmed-up" in ready.output + started = tmp_path / "started" + second_pid = tmp_path / "second-pid" + interrupted = threading.Event() + draining = threading.Event() + release = threading.Event() + original_expect = session.shell.expect + original_sendintr = session.shell.sendintr + + def expect(pattern, timeout=-1, **kwargs): + if timeout == 0.1: + assert interrupted.wait(3) + elif timeout == 1.0 and interrupted.is_set(): + draining.set() + assert release.wait(3) + return original_expect(pattern, timeout=timeout, **kwargs) + + def sendintr(): + original_sendintr() + interrupted.set() + + monkeypatch.setattr(session.shell, "expect", expect) + monkeypatch.setattr(session.shell, "sendintr", sendintr) + action = asyncio.create_task( + session.run( + BashAction( + command=f"trap '' INT; /bin/sh -c 'touch {started}; sleep 30'; " + f"/bin/sh -c 'echo $$ > {second_pid}; sleep 30'", + timeout=0.1, + check="silent", + ) + ) + ) + interrupt = None + closing = None + try: + async with asyncio.timeout(3): + while not started.exists(): + await asyncio.sleep(0.01) + interrupt = asyncio.create_task(session.run(BashInterruptAction())) + async with asyncio.timeout(3): + while not draining.is_set() or not second_pid.exists(): + await asyncio.sleep(0.01) + if concurrent_close: + closing = asyncio.create_task(session.close()) + await asyncio.sleep(0) + release.set() + with pytest.raises(pexpect.TIMEOUT): + await asyncio.wait_for(interrupt, 4) + with pytest.raises(pexpect.TIMEOUT): + await action + if closing is not None: + await asyncio.wait_for(closing, 3) + else: + assert not session.shell.isalive() + await session.close() + with pytest.raises(RuntimeError, match="shell not initialized"): + _ = session.shell + state = subprocess.run( + ["ps", "-p", second_pid.read_text().strip(), "-o", "stat="], + capture_output=True, + text=True, + check=False, + ).stdout.strip() + assert not state or state.startswith("Z") + finally: + interrupted.set() + release.set() + if second_pid.exists(): + with contextlib.suppress(ProcessLookupError): + os.killpg(int(second_pid.read_text()), signal.SIGKILL) + for task in (action, interrupt, closing): + if task is not None: + with contextlib.suppress(Exception, asyncio.CancelledError): + await task + with contextlib.suppress(Exception): + await session.close() + with contextlib.suppress(RuntimeError): + await asyncio.to_thread(session.shell.close, force=True) From 9db5e95853df3925ffee60ccd90e9ed7736ad39f Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Mon, 9 Nov 2026 17:52:29 -0800 Subject: [PATCH 09/10] Keep cancellation and session state when interrupt cleanup fails Co-Authored-By: Claude Opus 5.5 --- .../harness/sandbox/_http_session.py | 5 ++++ tests/core/harness/test_http_session.py | 29 +++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/src/olmo_eval/harness/sandbox/_http_session.py b/src/olmo_eval/harness/sandbox/_http_session.py index 4a66f6aab..83cb142a9 100644 --- a/src/olmo_eval/harness/sandbox/_http_session.py +++ b/src/olmo_eval/harness/sandbox/_http_session.py @@ -162,6 +162,11 @@ def run() -> _T: await self._interrupt_active(task) except asyncio.CancelledError: continue + except Exception: + # Failed cleanup leaves the terminal unusable; the + # control-task handling below surfaces the error. + self._closing = True + break except Exception: break # A reader deadline does not stop its command. Keep ownership until diff --git a/tests/core/harness/test_http_session.py b/tests/core/harness/test_http_session.py index a72ec91ce..06f196c9b 100644 --- a/tests/core/harness/test_http_session.py +++ b/tests/core/harness/test_http_session.py @@ -120,6 +120,35 @@ async def test_cancellation_settles_reader_before_next_command(tmp_path): await session.close() +@pytest.mark.anyio +async def test_failed_cancellation_cleanup_still_cancels_and_closes(monkeypatch, tmp_path): + session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) + await session.start() + marker = tmp_path / "started" + + async def failing_stop(task): + raise RuntimeError("cleanup failed") + + monkeypatch.setattr(session, "_stop_active", failing_stop) + try: + task = asyncio.create_task( + session.run(BashAction(command=f"touch {marker}; sleep 1", timeout=5)) + ) + async with asyncio.timeout(3): + while not marker.exists(): + await asyncio.sleep(0.01) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 5) + assert session._active_worker is None + assert session._interrupt_task is None + with pytest.raises(RuntimeError, match="Session is closing"): + await session.run(BashAction(command="echo late", timeout=3)) + finally: + await asyncio.sleep(1.5) + await session.close() + + @pytest.mark.anyio async def test_interrupt_can_signal_an_active_serialized_reader(tmp_path): session = ResponsiveBashSession(CreateBashSessionRequest(session="test")) From ab60211ca63153bcc202399153dfc7d1524ede79 Mon Sep 17 00:00:00 2001 From: Abhishek Rao Date: Thu, 12 Nov 2026 10:31:45 -0800 Subject: [PATCH 10/10] Log the background-process change in the result compatibility log Co-Authored-By: Claude Opus 5.5 --- docs/result-compatibility.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/docs/result-compatibility.md b/docs/result-compatibility.md index 5afa307be..4b138ed6d 100644 --- a/docs/result-compatibility.md +++ b/docs/result-compatibility.md @@ -16,6 +16,22 @@ Add an entry in the PR that makes the change, newest first. Each entry states: - **Comparability**: which earlier results remain comparable and what replaces the old behavior, if anything. +## 2026-11-12: Sandbox HTTP commands stop waiting for background processes (#527) + +- **Affected**: independent `execute_command` and `execute_code` calls that go through the + SWE-ReX HTTP server: Modal sandboxes, and Docker/Podman sandboxes that set `exec_shell` or + whose image fails the process-supervision probe. Persistent bash sessions are not affected. +- **Change**: a command that starts a background process still writing to the command's + output (`python server.py &` with no redirect) used to block until `command_timeout` and + end in a timeout error. It now returns when its shell exits, with the output written so + far, and the background process keeps running; its later output goes to an unlinked file + in the sandbox's `/tmp` until it exits. Commands that start no such process are unchanged. +- **Identity**: no task hash moves, so results from before and after can only be told apart + by commit or date. +- **Comparability**: agentic results on those sandbox paths can differ wherever the agent + ran such a command, because the agent now sees its output instead of a timeout. Other + results are unchanged. + ## 2026-11-03: Terminal-Bench and TBLite leave harness failures out of their scores (#539) - **Affected**: the `terminal_bench_2` and `openthoughts_tblite` external evals, for