Fix sandbox runtime links and subprocess cleanup

plugins/home/src/reme_home/auto_bash.py

@@ -95,7 +95,11 @@ def _read_specs():
     """Build the fixed safecmd policy once, when read-only mode is used."""
     from safecmd import CmdSpec  # pylint: disable=import-outside-toplevel
 
-    return {CmdSpec(name, denied=options.get("denied")) for name, options in _READ_COMMANDS.items()}
+    return {
+        CmdSpec(f"{name} {subcommand}".strip(), denied=options.get("denied"))
+        for name, options in _READ_COMMANDS.items()
+        for subcommand in options.get("allow", [""])
+    }
 
 
 def _literal_shell(node: dict, command: str, quoted: bool = False) -> bool:
@@ -119,7 +123,7 @@ def _literal_shell(node: dict, command: str, quoted: bool = False) -> bool:
             continue
         if op in {">&", "<&"} and (destination == "-" or destination.rstrip("-").isdigit()):
             continue
-        if op not in {">", ">>", ">|", "<>", "&>", "&>>", ">&"} or destination != "/dev/null":
+        if op not in {">", ">>", ">|", "<>", "&>", "&>>", ">&"} or destination not in _SAFE_DESTINATIONS:
             return False
     for value in node.values():
         children = value if isinstance(value, list) else [value]
@@ -129,10 +133,8 @@ def _literal_shell(node: dict, command: str, quoted: bool = False) -> bool:
 
 
 def _allowed_arguments(tokens: list[str]) -> bool:
-    """Supplement CmdSpec with subcommands, abbreviated flags and Git formats."""
+    """Supplement CmdSpec with abbreviated flags and Git formats."""
     options = _READ_COMMANDS[tokens[0]]
-    if "allow" in options and (len(tokens) < 2 or tokens[1] not in options["allow"]):
-        return False
     # Signature placeholders invoke Git's configured GPG program.
     if tokens[0] == "git" and any("%G" in token for token in tokens[1:]):
         return False
@@ -253,20 +255,13 @@ class AutoBashStep(BaseStep):
         command = self.context.get("command", "")
         timeout = self.context.get("timeout", self.kwargs.get("timeout", 60000))
         is_read_only = self.context.get("is_read_only", self.kwargs.get("is_read_only", True))
-        if not isinstance(command, str) or not command.strip():
-            self.context.response.success = False
-            self.context.response.answer = "command is required"
-            return self.context.response
-        if not isinstance(timeout, int) or isinstance(timeout, bool) or timeout <= 0:
-            self.context.response.success = False
-            self.context.response.answer = "timeout must be a positive integer in milliseconds"
-            return self.context.response
-        if not isinstance(is_read_only, bool):
-            self.context.response.success = False
-            self.context.response.answer = "is_read_only must be a boolean"
-            return self.context.response
-
         try:
+            if not isinstance(command, str) or not command.strip():
+                raise ValueError("command is required")
+            if not isinstance(timeout, int) or isinstance(timeout, bool) or timeout <= 0:
+                raise ValueError("timeout must be a positive integer in milliseconds")
+            if not isinstance(is_read_only, bool):
+                raise ValueError("is_read_only must be a boolean")
             backend = file_sandbox_for_step(self)
         except (TypeError, ValueError, RuntimeError) as exc:
             self.context.response.success = False

plugins/home/tests/test_auto_bash.py

@@ -244,14 +244,28 @@ async def test_step_configuration_can_disable_read_only(tmp_path):
 
 
 @pytest.mark.asyncio
-async def test_step_rejects_non_boolean_mode(tmp_path):
-    """A string false value must not silently enable unrestricted execution."""
-    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+@pytest.mark.parametrize(
+    ("inputs", "message"),
+    [
+        ({"command": " "}, "command is required"),
+        ({"command": 1}, "command is required"),
+        ({"command": "echo ok", "timeout": 0}, "timeout must be a positive integer in milliseconds"),
+        ({"command": "echo ok", "timeout": True}, "timeout must be a positive integer in milliseconds"),
+        ({"command": "echo ok", "timeout": "1"}, "timeout must be a positive integer in milliseconds"),
+        ({"command": "touch denied.txt", "is_read_only": "false"}, "is_read_only must be a boolean"),
+        ({"command": "", "timeout": 0, "is_read_only": "false"}, "command is required"),
+    ],
+)
+async def test_step_parameter_errors_do_not_execute(tmp_path, monkeypatch, inputs, message):
+    """Parameter errors preserve their messages and precedence without executing."""
 
-    response = await step(command="touch denied.txt", is_read_only="false")
+    def unexpected_backend(_step):
+        pytest.fail("invalid parameters reached backend selection")
 
+    monkeypatch.setattr("reme_home.auto_bash.file_sandbox_for_step", unexpected_backend)
+    response = await AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))(**inputs)
     assert response.success is False
-    assert response.answer == "is_read_only must be a boolean"
+    assert response.answer == message
     assert not (tmp_path / "denied.txt").exists()
 
 
@@ -519,8 +533,11 @@ async def test_git_pretty_config_cannot_launch_gpg(tmp_path):
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("subcommand", ["status", "diff", "log", "show", "rev-parse", "ls-files"])
-@pytest.mark.parametrize("flag", ["--help", "--hel", "--he", "--h", "-h"])
+@pytest.mark.parametrize(
+    ("subcommand", "flag"),
+    [(name, "--help") for name in ("status", "diff", "log", "show", "rev-parse", "ls-files")]
+    + [("status", flag) for flag in ("--hel", "--he", "--h", "-h")],
+)
 async def test_git_help_flags_are_denied(tmp_path, monkeypatch, subcommand, flag):
     """All allowlisted Git commands reject help before starting a process."""
 
@@ -638,6 +655,8 @@ async def test_allow_and_denied_command_options_are_combined(tmp_path, monkeypat
     from reme_home import auto_bash
 
     commands = auto_bash._READ_COMMANDS  # pylint: disable=protected-access
+    # Rebuild specs from the temporary policy rather than the production cache.
+    monkeypatch.setattr(auto_bash, "_read_specs", auto_bash._read_specs.__wrapped__)  # pylint: disable=protected-access
     monkeypatch.setitem(commands, "echo", {"allow": ["hello"]})
     tool = AutoBash(cwd=tmp_path)
 
@@ -645,11 +664,6 @@ async def test_allow_and_denied_command_options_are_combined(tmp_path, monkeypat
     assert not await tool.check_read_only({"command": "echo goodbye"})
 
     monkeypatch.setitem(commands, "echo", {"allow": ["hello"], "denied": ["-n"]})
-    from safecmd import CmdSpec
-
-    specs = {spec for spec in auto_bash._read_specs() if "echo" not in spec.name}  # pylint: disable=protected-access
-    specs.add(CmdSpec("echo", denied=["-n"]))
-    monkeypatch.setattr(auto_bash, "_read_specs", lambda: specs)
     assert not await tool.check_read_only({"command": "echo hello -n"})
     assert await tool.check_read_only({"command": "find . -name note.txt"})
     assert not await tool.check_read_only({"command": "find . -delete"})

reme/utils/file_sandbox/file_sandbox_backend.py

@@ -4,6 +4,8 @@ import asyncio
 import os
 import sys
 from abc import abstractmethod
+from collections import deque
+from pathlib import Path
 
 from agentscope.tool import ExecResult
 
@@ -39,6 +41,35 @@ class FileSandboxBackend(LocalProcessBackend):
             raise ValueError("sandbox paths must not contain duplicates")
         self._path_identities = {path: self._identity(path) for path in paths}
 
+    @staticmethod
+    def _runtime_paths() -> tuple[set[str], set[str]]:
+        """Collect runtime symlinks and traversal paths without exposing parent trees."""
+        links = set()
+        metadata = set()
+        for runtime in (sys.prefix, sys.base_prefix, sys.executable):
+            pending = deque(Path(runtime).parts)
+            current = "/"
+            followed = 0
+            while pending:
+                part = pending.popleft()
+                if part == "/":
+                    current = "/"
+                    continue
+                if part == "..":
+                    current = os.path.dirname(current)
+                    continue
+                path = os.path.join(current, part)
+                if os.path.islink(path):
+                    followed += 1
+                    if followed > 40:
+                        raise RuntimeError(f"too many runtime symlinks: {runtime}")
+                    links.add(path)
+                    pending.extendleft(reversed(Path(os.readlink(path)).parts))
+                else:
+                    metadata.add(path)
+                    current = path
+        return links, metadata
+
     @staticmethod
     def _identity(path: str) -> tuple[int, int]:
         info = os.stat(path, follow_symlinks=False)

reme/utils/file_sandbox/path_bubble_backend.py

@@ -1,6 +1,7 @@
 """Bubblewrap execution with host-identical, explicitly mounted paths."""
 
 import asyncio
+import errno
 import os
 import shutil
 import uuid
@@ -8,7 +9,10 @@ from pathlib import Path
 
 import sys
 
-from .file_sandbox_backend import FileSandboxBackend
+from .file_sandbox_backend import FileSandboxBackend, contains
+from ..local_backend import finish_cleanup
+
+_CGROUP_CLEANUP_TIMEOUT = 0.25
 
 # Run after exec of a fresh interpreter: Python preexec_fn can deadlock when
 # the parent has threads. The requested command starts only after enrollment.
@@ -61,6 +65,15 @@ class PathBubbleBackend(FileSandboxBackend):
         mounts += [(path, "--bind") for path in self._writable_paths]
         for path, mode in sorted(mounts, key=lambda item: (item[0].count("/"), item[0])):
             args.extend((mode, path, path))
+        # Existing mounts already preserve their internal links. Recreate only
+        # runtime links missing from the sandbox, without mounting their parents.
+        visible_paths = [
+            args[index + 2] for index in range(len(args) - 2) if args[index] in {"--ro-bind", "--bind", "--symlink"}
+        ]
+        links, _ = self._runtime_paths()
+        for path in sorted(links):
+            if not any(contains(root, path) for root in visible_paths):
+                args.extend(("--symlink", os.readlink(path), path))
         binary_paths = [str(Path(sys.prefix) / "bin"), "/usr/local/bin", "/usr/bin", "/bin"]
         args.extend(("--clearenv", "--setenv", "HOME", self._cwd, "--setenv", "TMPDIR", "/tmp"))
         args.extend(("--setenv", "PATH", ":".join(binary_paths), "--chdir", cwd, "--"))
@@ -83,7 +96,7 @@ class PathBubbleBackend(FileSandboxBackend):
             )
         except BaseException:
             if group is not None:
-                group.rmdir()
+                await finish_cleanup(self._remove_cgroup(group))
             raise
         if group is not None:
             self._process_cgroups[process.pid] = group
@@ -109,16 +122,34 @@ class PathBubbleBackend(FileSandboxBackend):
         except (OSError, StopIteration) as exc:
             raise RuntimeError("PathBubbleBackend requires a writable delegated cgroup v2 for memory limits") from exc
 
-    def _release_process(self, process: asyncio.subprocess.Process) -> None:
+    async def _release_process(self, process: asyncio.subprocess.Process) -> None:
         """Release the per-execution group after its process has been reaped."""
-        group = self._process_cgroups.pop(process.pid, None)
+        group = self._process_cgroups.get(process.pid)
         if group is None:
             return
+        await self._remove_cgroup(group)
+        self._process_cgroups.pop(process.pid)
+
+    @staticmethod
+    async def _remove_cgroup(group: Path) -> None:
+        """Kill remaining members and yield until the group is empty, within a deadline."""
         try:
-            group.rmdir()
-        except OSError as exc:
             try:
-                (group / "cgroup.kill").write_text("1", encoding="ascii")
                 group.rmdir()
-            except OSError as cleanup_error:
-                raise RuntimeError(f"could not remove sandbox cgroup {group}: {cleanup_error}") from exc
+                return
+            except OSError as exc:
+                if exc.errno != errno.EBUSY:
+                    raise
+            (group / "cgroup.kill").write_text("1", encoding="ascii")
+            loop = asyncio.get_running_loop()
+            deadline = loop.time() + _CGROUP_CLEANUP_TIMEOUT
+            while True:
+                try:
+                    group.rmdir()
+                    return
+                except OSError as exc:
+                    if exc.errno != errno.EBUSY or loop.time() >= deadline:
+                        raise
+                await asyncio.sleep(0.01)
+        except OSError as exc:
+            raise RuntimeError(f"could not remove sandbox cgroup {group}: {exc}") from exc

reme/utils/file_sandbox/path_seatbelt_backend.py

@@ -61,6 +61,14 @@ class PathSeatbeltBackend(FileSandboxBackend):
         """Quote a path as one Seatbelt profile string, never as profile syntax."""
         return f"(subpath {json.dumps(path)})" if os.path.isdir(path) else f"(literal {json.dumps(path)})"
 
+    @classmethod
+    def _runtime_link_rules(cls) -> list[str]:
+        """Permit runtime symlink traversal without granting their parent trees."""
+        links, metadata = cls._runtime_paths()
+        return [f"(allow file-read* (literal {json.dumps(path)}))" for path in sorted(links)] + [
+            f"(allow file-read-metadata (literal {json.dumps(path)}))" for path in sorted(metadata - links)
+        ]
+
     def _profile(self, tempdir: str) -> str:
         """Allow only configured files plus the system runtime needed to launch tools."""
         system_roots = ("/System", "/usr", "/bin", "/sbin", "/private/etc", "/dev")
@@ -82,6 +90,7 @@ class PathSeatbeltBackend(FileSandboxBackend):
                 ancestors.add(parent)
                 parent = os.path.dirname(parent)
         lines.extend(f"(allow file-read-metadata (literal {json.dumps(path)}))" for path in sorted(ancestors))
+        lines.extend(self._runtime_link_rules())
         lines.extend(f"(allow file-read* {self._path_rule(path)})" for path in read_paths)
         # Shell redirection to the null device is safe and is accepted by
         # AutoBash's read-only policy; do not grant writes to the rest of /dev.
@@ -130,7 +139,7 @@ class PathSeatbeltBackend(FileSandboxBackend):
         self._process_tempdirs[process.pid] = tempdir
         return process
 
-    def _release_process(self, process: asyncio.subprocess.Process) -> None:
+    async def _release_process(self, process: asyncio.subprocess.Process) -> None:
         tempdir = self._process_tempdirs.pop(process.pid, None)
         if tempdir is not None:
             shutil.rmtree(tempdir)

reme/utils/local_backend.py

@@ -110,7 +110,7 @@ class LocalProcessBackend(LocalBackend):
             await finish_cleanup(rollback())
             raise
 
-    def _release_process(self, process: asyncio.subprocess.Process) -> None:
+    async def _release_process(self, process: asyncio.subprocess.Process) -> None:
         """Release any platform resources owned by a completed execution."""
 
     @staticmethod
@@ -192,12 +192,12 @@ class LocalProcessBackend(LocalBackend):
             self._collect_output(process, data, stdout_buffer, stderr_buffer, exceeded),
         )
 
-        async def cleanup(completed):
+        async def cleanup(_completed):
             try:
-                if not completed:
-                    await self._stop_process(process, communication)
+                # Closed pipes do not imply that redirected background children exited.
+                await self._stop_process(process, communication)
             finally:
-                self._release_process(process)
+                await self._release_process(process)
 
         try:
             stdout, stderr = await self._execute(

tests/integration/test_seatbelt_backend.py

@@ -1,8 +1,12 @@
 """Exercise macOS sandbox startup and file isolation using real subprocesses."""
 
 import asyncio
+import os
 import sys
+import sysconfig
+import tempfile
 import venv
+from pathlib import Path
 
 import psutil
 
@@ -50,6 +54,52 @@ async def test_seatbelt_starts_virtual_environment_python(tmp_path):
     assert result.stdout.decode().strip() == str(workspace)
 
 
+@pytest.mark.asyncio
+async def test_seatbelt_external_virtual_environment_through_symlinks(tmp_path, monkeypatch):
+    """An external venv reached through /tmp keeps its packages and file isolation."""
+    workspace = tmp_path.resolve()
+    with tempfile.TemporaryDirectory(prefix="reme-venv-test-", dir="/private/tmp") as directory:
+        root = Path(directory)
+        environment = root / "venv"
+        venv.EnvBuilder(with_pip=False, symlinks=True).create(environment)
+        (root / "alias").symlink_to("venv", target_is_directory=True)
+        alias = Path("/tmp") / root.name / "alias"
+        interpreter = alias / "bin" / "python"
+        packages = Path(sysconfig.get_path("purelib", vars={"base": str(environment), "platbase": str(environment)}))
+        (packages / "reme_venv_marker.py").write_text("VALUE = 'venv-package'\n", encoding="utf-8")
+        secret = root / "outside.txt"
+        secret.write_text("private", encoding="utf-8")
+        # Match the runtime paths reported when ReMe is launched through this venv.
+        monkeypatch.setattr(sys, "prefix", str(alias))
+        monkeypatch.setattr(sys, "executable", str(interpreter))
+        backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[])
+        code = f"""
+import os, sys
+from pathlib import Path
+import reme_venv_marker
+assert os.path.realpath(sys.prefix) == {str(environment)!r}
+assert sys.prefix != sys.base_prefix
+assert reme_venv_marker.VALUE == 'venv-package'
+for path in [{str(secret)!r}, {str(Path('/tmp') / root.name / 'outside.txt')!r}]:
+    try:
+        Path(path).read_text()
+    except PermissionError:
+        pass
+    else:
+        raise AssertionError(f'outside file readable: {{path}}')
+print('ready')
+"""
+        # Exercise real Seatbelt directly, independent of GUI launchd availability.
+        process = await backend._start_process(  # pylint: disable=protected-access
+            [str(interpreter), "-c", code],
+            str(workspace),
+        )
+        result = await backend._communicate(process, data=None, timeout=10)  # pylint: disable=protected-access
+        assert result.exit_code == 0, result.stderr.decode(errors="replace")
+        assert result.stdout == b"ready\n"
+        assert os.path.realpath(interpreter) != str(interpreter)
+
+
 @pytest.mark.asyncio
 async def test_seatbelt_allows_null_device_redirection(tmp_path):
     """The shell can discard output through /dev/null inside a file sandbox."""

tests/unit/test_file_sandbox_backend.py

@@ -4,7 +4,6 @@ import asyncio
 import os
 import plistlib
 import sys
-from contextlib import nullcontext
 from pathlib import Path
 
 import pytest
@@ -93,6 +92,29 @@ def test_sandbox_reads_virtual_environment_base_runtime(tmp_path, monkeypatch):
     assert f'(allow file-write* (literal "{interpreter}"))' not in profile
 
 
+def test_seatbelt_runtime_links_have_only_literal_read_grants(tmp_path, monkeypatch):
+    """Follow relative link chains while keeping user-configured symlinks invalid."""
+    workspace = tmp_path / "workspace"
+    environment = tmp_path / "venv"
+    workspace.mkdir()
+    (environment / "bin").mkdir(parents=True)
+    (environment / "bin" / "python").symlink_to("python3")
+    (environment / "bin" / "python3").symlink_to(sys.executable)
+    alias = tmp_path / "alias"
+    alias.symlink_to("venv", target_is_directory=True)
+    monkeypatch.setattr(sys, "prefix", str(alias))
+    monkeypatch.setattr(sys, "executable", str(alias / "bin" / "python"))
+
+    backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[])
+    profile = backend._profile(str(workspace / "temporary"))  # pylint: disable=protected-access
+    for link in (alias, environment / "bin" / "python", environment / "bin" / "python3"):
+        assert f'(allow file-read* (literal "{link}"))' in profile
+        assert f'(allow file-read* (subpath "{link}"))' not in profile
+    assert f'(allow file-read* (subpath "{tmp_path}"))' not in profile
+    with pytest.raises(ValueError, match="without symlinks"):
+        PathSeatbeltBackend(cwd=workspace, read_only_paths=[str(alias)], writable_paths=[])
+
+
 def test_seatbelt_excludes_read_only_child_from_writable_parent(tmp_path):
     """A parent write grant must not override a nested read-only restriction."""
     protected = tmp_path / "protected"
@@ -112,95 +134,23 @@ def test_seatbelt_excludes_read_only_child_from_writable_parent(tmp_path):
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("backend_class", [PathBubbleBackend, PathSeatbeltBackend])
 @pytest.mark.parametrize("data", [b"", b"hello\x00\n\xff"])
-async def test_sandbox_file_io_round_trip(tmp_path, monkeypatch, backend_class, data):
+async def test_sandbox_file_io_round_trip(tmp_path, monkeypatch, data):
     """The common file helpers preserve binary stdin, literal paths and write errors."""
-    from unittest.mock import AsyncMock
-
-    backend = backend_class(cwd=tmp_path, read_only_paths=[], writable_paths=[str(tmp_path)])
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[str(tmp_path)])
     # Exercise shared file I/O with real pipes; OS isolation is checked separately.
     monkeypatch.setattr(
         backend,
         "_start_process",
         LocalProcessBackend()._start_process,  # pylint: disable=protected-access
     )
-    if backend_class is PathSeatbeltBackend:
-        monkeypatch.setattr(sys, "platform", "darwin")
-        monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.shutil.which", lambda _: "sandbox-exec")
-        monkeypatch.setattr(backend, "_has_gui_launchd", AsyncMock(return_value=False))
-    warning = (
-        pytest.warns(RuntimeWarning, match="No macOS GUI launchd domain")
-        if backend_class is PathSeatbeltBackend
-        else nullcontext()
-    )
-    with warning:
-        path = str(tmp_path / "nested directory" / "literal ' ; name.bin")
-        await backend.write_file(path, data)
-        assert await backend.read_file(path) == data
-        with pytest.raises(TypeError, match="data must be bytes"):
-            await backend.write_file(path, "invalid")
-        with pytest.raises(OSError):
-            await backend.write_file(str(tmp_path), data)
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("sandbox", [False, True])
-@pytest.mark.parametrize("cancel", [False, True])
-async def test_inherited_pipe_cleanup_is_bounded(tmp_path, monkeypatch, sandbox, cancel):
-    """A surviving child with an output pipe must not hold the timeout response open."""
-    from unittest.mock import Mock
-
-    backend = (
-        PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=0)
-        if sandbox
-        else LocalProcessBackend()
-    )
-    started = asyncio.Event()
-    cancelled = asyncio.Event()
-
-    class StuckPipe:
-        """Model an inherited pipe held open after termination."""
-
-        async def read(self, _size):
-            """Wait forever until the caller cancels pipe draining."""
-            started.set()
-            try:
-                await asyncio.Event().wait()
-            finally:
-                cancelled.set()
-
-    class StuckProcess:
-        """Model a child that keeps its output pipe open after termination."""
-
-        pid = 12345
-        _transport = Mock()
-        stdout = StuckPipe()
-        stderr = StuckPipe()
-        stdin = None
-
-        async def wait(self):
-            """A stale process transport must also have a bounded wait."""
-            await asyncio.Event().wait()
-
-    async def start(*_args, **_kwargs):
-        return StuckProcess()
-
-    monkeypatch.setattr(backend, "_start_process", start)
-    monkeypatch.setattr(backend, "_kill_process", lambda _process: None)
-    monkeypatch.setattr("reme.utils.local_backend._CLEANUP_TIMEOUT", 0.01)
-
-    task = asyncio.create_task(backend.exec_shell(["true"], timeout=None if cancel else 0.01))
-    await asyncio.wait_for(started.wait(), timeout=1)
-    if cancel:
-        task.cancel()
-        with pytest.raises(asyncio.CancelledError):
-            await asyncio.wait_for(task, timeout=1)
-    else:
-        result = await asyncio.wait_for(task, timeout=1)
-        assert result.exit_code == -1 and result.stderr == b"timed out"
-    assert cancelled.is_set()
-    StuckProcess._transport.close.assert_called_once()  # pylint: disable=protected-access
+    path = str(tmp_path / "nested directory" / "literal ' ; name.bin")
+    await backend.write_file(path, data)
+    assert await backend.read_file(path) == data
+    with pytest.raises(TypeError, match="data must be bytes"):
+        await backend.write_file(path, "invalid")
+    with pytest.raises(OSError):
+        await backend.write_file(str(tmp_path), data)
 
 
 @pytest.mark.asyncio

tests/unit/test_local_backend.py

@@ -0,0 +1,106 @@
+"""Local executions must not leave same-group background processes running."""
+
+import asyncio
+import os
+
+import psutil
+import pytest
+
+from reme.utils.local_backend import LocalProcessBackend
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(os.name != "posix", reason="requires POSIX shell process groups")
+@pytest.mark.parametrize("exit_code", [0, 7])
+async def test_completed_command_stops_redirected_background_process(tmp_path, exit_code):
+    """Completion preserves the result and kills children even after all pipes close."""
+    (tmp_path / "note.txt").write_text("note\n", encoding="utf-8")
+    pid_file = tmp_path / "child.pid"
+    command = (
+        "tail -f note.txt < /dev/null > /dev/null 2>&1 &\n"
+        "echo $! > child.pid\n"
+        "printf 'output\\n'\n"
+        "printf 'diagnostic\\n' >&2\n"
+        f"exit {exit_code}"
+    )
+    try:
+        result = await asyncio.wait_for(
+            LocalProcessBackend().exec_shell(["/bin/sh", "-c", command], cwd=str(tmp_path), timeout=100),
+            timeout=3,
+        )
+        assert result.exit_code == exit_code
+        assert result.stdout == b"output\n"
+        assert result.stderr == b"diagnostic\n"
+        pid = int(pid_file.read_text(encoding="ascii"))
+
+        async def wait_stopped():
+            while True:
+                try:
+                    if psutil.Process(pid).status() == psutil.STATUS_ZOMBIE:
+                        return
+                except psutil.NoSuchProcess:
+                    return
+                await asyncio.sleep(0.01)
+
+        # Signal delivery and orphan reaping can lag the executor's return.
+        await asyncio.wait_for(wait_stopped(), timeout=1)
+    finally:
+        if pid_file.exists():
+            try:
+                psutil.Process(int(pid_file.read_text(encoding="ascii"))).kill()
+            except psutil.NoSuchProcess:
+                pass
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("cancel", [False, True])
+async def test_inherited_pipe_cleanup_is_bounded(monkeypatch, cancel):
+    """A surviving child with an output pipe must not hold the timeout response open."""
+    from unittest.mock import Mock
+
+    backend = LocalProcessBackend()
+    started = asyncio.Event()
+    cancelled = asyncio.Event()
+
+    class StuckPipe:
+        """Model an inherited pipe held open after termination."""
+
+        async def read(self, _size):
+            """Wait forever until the caller cancels pipe draining."""
+            started.set()
+            try:
+                await asyncio.Event().wait()
+            finally:
+                cancelled.set()
+
+    class StuckProcess:
+        """Model a child that keeps its output pipe open after termination."""
+
+        pid = 12345
+        _transport = Mock()
+        stdout = StuckPipe()
+        stderr = StuckPipe()
+        stdin = None
+
+        async def wait(self):
+            """A stale process transport must also have a bounded wait."""
+            await asyncio.Event().wait()
+
+    async def start(*_args, **_kwargs):
+        return StuckProcess()
+
+    monkeypatch.setattr(backend, "_start_process", start)
+    monkeypatch.setattr(backend, "_kill_process", lambda _process: None)
+    monkeypatch.setattr("reme.utils.local_backend._CLEANUP_TIMEOUT", 0.01)
+
+    task = asyncio.create_task(backend.exec_shell(["true"], timeout=None if cancel else 0.01))
+    await asyncio.wait_for(started.wait(), timeout=1)
+    if cancel:
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=1)
+    else:
+        result = await asyncio.wait_for(task, timeout=1)
+        assert result.exit_code == -1 and result.stderr == b"timed out"
+    assert cancelled.is_set()
+    StuckProcess._transport.close.assert_called_once()  # pylint: disable=protected-access

tests/unit/test_path_bubble_backend.py

@@ -1,8 +1,11 @@
 """Path-preserving Bubblewrap mounts and job integration."""
 
+import asyncio
+import errno
 import sys
 from types import SimpleNamespace
 from pathlib import Path
+from unittest.mock import AsyncMock
 
 import pytest
 import yaml
@@ -54,6 +57,46 @@ def test_path_bubble_backend_adds_workspace_read_only(tmp_path):
     assert ("--ro-bind", str(tmp_path), str(tmp_path)) in _mounts(argv)
 
 
+@pytest.mark.parametrize("relative", [False, True])
+def test_runtime_symlink_chains_preserve_interpreter_paths(tmp_path, monkeypatch, relative):
+    """External runtime aliases work without exposing siblings or duplicating mounted links."""
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    environment = tmp_path / "venv"
+    runtime = tmp_path / "runtime"
+    (environment / "bin").mkdir(parents=True)
+    (runtime / "bin").mkdir(parents=True)
+    (runtime / "bin" / "python").touch()
+    targets = {
+        tmp_path / "alias": "venv-alias",
+        tmp_path / "venv-alias": "venv",
+        tmp_path / "base-alias": "runtime",
+        tmp_path / "launcher": "runtime/bin/python",
+    }
+    for path, target in targets.items():
+        path.symlink_to(target if relative else tmp_path / target)
+    # This link is already available through the venv's read-only mount, but
+    # its target requires another link outside that mount.
+    (environment / "bin" / "python").symlink_to("../../launcher")
+    interpreter = str(tmp_path / "alias" / "bin" / "python")
+    monkeypatch.setattr(sys, "prefix", str(tmp_path / "alias"))
+    monkeypatch.setattr(sys, "base_prefix", str(tmp_path / "base-alias"))
+    monkeypatch.setattr(sys, "executable", interpreter)
+    backend = PathBubbleBackend(cwd=workspace, read_only_paths=[], writable_paths=[])
+
+    argv = backend._bwrap_argv([interpreter, "-c", "print(1)"], str(workspace))  # pylint: disable=protected-access
+    links = {argv[index + 2]: argv[index + 1] for index in range(len(argv) - 2) if argv[index] == "--symlink"}
+
+    for path, target in targets.items():
+        assert links[str(path)] == (target if relative else str(tmp_path / target))
+    assert str(environment / "bin" / "python") not in links
+    assert ("--ro-bind", str(environment), str(environment)) in _mounts(argv)
+    assert ("--ro-bind", str(runtime), str(runtime)) in _mounts(argv)
+    assert all(source != str(tmp_path) for _, source, _ in _mounts(argv))
+    assert not any(mode == "--bind" for mode, _, _ in _mounts(argv))
+    assert argv[-3:] == [interpreter, "-c", "print(1)"]
+
+
 @pytest.mark.parametrize("limit", [-1, True, 1.5, "4096"])
 def test_path_bubble_backend_rejects_invalid_memory_limit(tmp_path, limit):
     """Memory limits must be non-negative whole MiB."""
@@ -113,11 +156,73 @@ async def test_each_execution_gets_separate_memory_cgroup(tmp_path, monkeypatch)
     for argv, group in zip(subprocess_commands, groups):
         assert argv[:4] == (sys.executable, "-I", "-S", "-c")
         assert argv[5:7] == (str(group / "cgroup.procs"), "bwrap")
-    backend._release_process(first)  # pylint: disable=protected-access
-    backend._release_process(second)  # pylint: disable=protected-access
+    await backend._release_process(first)  # pylint: disable=protected-access
+    await backend._release_process(second)  # pylint: disable=protected-access
     assert not any(group.exists() for group in groups)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("outcome", ["delayed_exit", "cancel", "busy", "permission", "kill_error"])
+async def test_cgroup_cleanup_waits_without_blocking_and_reports_failures(tmp_path, monkeypatch, outcome):
+    """Model kernel EBUSY until members exit, including repeated cancellation during cleanup."""
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    group = tmp_path / "cgroup"
+    group.mkdir()
+    process = SimpleNamespace(pid=123, returncode=0)
+    backend._process_cgroups[process.pid] = group  # pylint: disable=protected-access
+    entered, empty = asyncio.Event(), asyncio.Event()
+    kills = []
+    original_rmdir, original_write = Path.rmdir, Path.write_text
+
+    def rmdir(path):
+        if path == group:
+            if outcome == "permission":
+                raise PermissionError(errno.EACCES, "permission denied")
+            if not empty.is_set():
+                entered.set()
+                raise OSError(errno.EBUSY, "members still exiting")
+        original_rmdir(path)
+
+    def write(path, data, **kwargs):
+        if path == group / "cgroup.kill":
+            kills.append(data)
+            if outcome == "kill_error":
+                raise PermissionError(errno.EACCES, "kill denied")
+            return len(data)
+        return original_write(path, data, **kwargs)
+
+    monkeypatch.setattr(Path, "rmdir", rmdir)
+    monkeypatch.setattr(Path, "write_text", write)
+    cleanup_timeout = 1 if outcome in ("delayed_exit", "cancel") else 0.03
+    monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend._CGROUP_CLEANUP_TIMEOUT", cleanup_timeout)
+    monkeypatch.setattr(backend, "_kill_process", lambda _: None)
+    monkeypatch.setattr(backend, "_collect_output", AsyncMock(return_value=(b"ok", b"")))
+    task = asyncio.create_task(backend._communicate(process))  # pylint: disable=protected-access
+    if outcome in ("delayed_exit", "cancel"):
+        await asyncio.wait_for(entered.wait(), timeout=1)
+        if outcome == "cancel":
+            task.cancel()
+            await asyncio.sleep(0)
+            task.cancel()
+        await asyncio.sleep(0.01)
+        assert not task.done()
+        empty.set()
+        if outcome == "cancel":
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(task, timeout=1)
+        else:
+            assert (await asyncio.wait_for(task, timeout=1)).stdout == b"ok"
+        assert not group.exists()
+        assert not backend._process_cgroups  # pylint: disable=protected-access
+    else:
+        with pytest.raises(RuntimeError, match="could not remove sandbox cgroup") as error:
+            await asyncio.wait_for(task, timeout=1)
+        assert error.value.__cause__.errno == (errno.EBUSY if outcome == "busy" else errno.EACCES)
+        assert group.exists()
+        assert backend._process_cgroups[process.pid] == group  # pylint: disable=protected-access
+    assert kills == ([] if outcome == "permission" else ["1"])
+
+
 @pytest.mark.asyncio
 async def test_zero_memory_limit_skips_cgroup(tmp_path, monkeypatch):
     """A zero limit bypasses cgroup creation and child enrollment."""
@@ -168,7 +273,8 @@ def test_path_bubble_backend_rejects_replaced_mount_source(tmp_path):
 
 
 @pytest.mark.asyncio
-async def test_python_execute_uses_path_backend_when_paths_supplied(tmp_path, monkeypatch):
+@pytest.mark.parametrize("memory_options", [{}, {"memory_limit_mib": 0}])
+async def test_python_execute_uses_path_backend_when_paths_supplied(tmp_path, monkeypatch, memory_options):
     """Python execution uses the same backend and timeout units."""
     seen = {}
 
@@ -178,30 +284,15 @@ async def test_python_execute_uses_path_backend_when_paths_supplied(tmp_path, mo
 
     monkeypatch.setattr(PathBubbleBackend, "exec_shell", fake_exec)
     step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
-    response = await step(code="print('ok')", read_only_paths=[], writable_paths=[str(tmp_path)])
+    response = await step(code="print('ok')", read_only_paths=[], writable_paths=[str(tmp_path)], **memory_options)
 
     assert response.success is True
     assert response.answer == "ok\n"
     assert seen["command"] == [sys.executable, "-c", "print('ok')"]
     assert seen["cwd"] == str(tmp_path)
     assert seen["timeout"] == 60
-    assert seen["backend"]._memory_limit_mib == 4096  # pylint: disable=protected-access
-
-
-@pytest.mark.asyncio
-async def test_python_execute_passes_custom_memory_limit(tmp_path, monkeypatch):
-    """Job parameters reach the backend without changing path selection."""
-    seen = {}
-
-    async def fake_exec(self, _command, *, cwd=None, timeout=None):  # pylint: disable=unused-argument
-        seen["limit"] = self._memory_limit_mib  # pylint: disable=protected-access
-        return ExecResult(exit_code=0, stdout=b"", stderr=b"")
-
-    monkeypatch.setattr(PathBubbleBackend, "exec_shell", fake_exec)
-    step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
-    response = await step(code="pass", read_only_paths=[], memory_limit_mib=0)
-    assert response.success is True
-    assert seen["limit"] == 0
+    expected_limit = memory_options.get("memory_limit_mib", 4096)
+    assert seen["backend"]._memory_limit_mib == expected_limit  # pylint: disable=protected-access
 
 
 @pytest.mark.asyncio