Add sandboxed Python execution and home AutoBash plugin

plugins/home/README.md

@@ -0,0 +1,20 @@
+# Home plugin
+
+Provides the `auto_bash` job through an independently enabled ReMe plugin.
+
+Install it into the same Python environment as ReMe:
+
+```bash
+reme plugins install plugins/home
+```
+
+Enable it in an application config with `plugins: [home]`. The plugin then
+contributes the `auto_bash` job, which can be run with
+`reme start config=<your-config.yaml> job=auto_bash command=pwd`.
+
+The job defaults to read-only command validation and a 60-second timeout. Set
+`is_read_only=false` to disable command validation. Supplying `read_only_paths`
+or `writable_paths` enables a file sandbox (Bubblewrap on Linux, Seatbelt on
+macOS), with sandbox paths identical to their host paths. Linux defaults to a
+4 GiB cgroup memory limit; macOS has no memory limit and warns when a nonzero
+`memory_limit_bytes` value is supplied.

plugins/home/pyproject.toml

@@ -0,0 +1,29 @@
+[project]
+name = "reme-home"
+version = "0.1.0"
+description = "Local command execution jobs for ReMe."
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "reme-ai[as]>=0.4.1.12",
+    "safecmd>=0.1.16,<0.2",
+]
+
+[project.entry-points."reme.plugins"]
+home = "reme_home"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+include = ["reme_home*"]
+
+[tool.setuptools.package-data]
+reme_home = ["plugin.yaml"]
+
+[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "function"
+pythonpath = ["src", "../.."]
+testpaths = ["tests"]
+
+[build-system]
+requires = ["setuptools>=77", "wheel"]
+build-backend = "setuptools.build_meta"

plugins/home/src/reme_home/__init__.py

@@ -0,0 +1,5 @@
+"""Home plugin: local command execution jobs for ReMe."""
+
+from .auto_bash import AutoBash, AutoBashStep
+
+__all__ = ["AutoBash", "AutoBashStep"]

plugins/home/src/reme_home/auto_bash.py

@@ -0,0 +1,170 @@
+"""Run allowlisted shell commands without an approval prompt."""
+
+from agentscope.message import TextBlock, ToolResultState
+from agentscope.permission import PermissionBehavior, PermissionContext, PermissionDecision
+from agentscope.tool import Bash, ToolChunk
+
+from reme.utils.file_sandbox import create_file_sandbox_backend
+from reme.steps.base_step import BaseStep
+
+_READ_COMMANDS = {
+    ":": {},
+    "cat": {},
+    "cd": {},
+    "cmp": {},
+    "diff": {"denied": ["--output"]},
+    "echo": {},
+    "false": {},
+    "file": {},
+    "find": {"denied": ["-delete", "-exec", "-execdir", "-fls", "-fprint", "-fprint0", "-fprintf", "-ok", "-okdir"]},
+    "git": {
+        "allow": ["status", "diff", "log", "show", "rev-parse", "ls-files"],
+        "denied": ["--output", "--ext-diff", "--textconv", "--paginate"],
+    },
+    "grep": {},
+    "head": {},
+    "ls": {},
+    "pwd": {},
+    "rg": {"denied": ["--pre", "--pre-glob", "--replace"]},
+    "sort": {"denied": ["-o", "--output", "--compress-program", "--temporary-directory"]},
+    "stat": {},
+    "tail": {},
+    "true": {},
+    "wc": {},
+    "which": {},
+}
+_SAFE_DESTINATIONS = {"/dev/null"}
+
+
+class AutoBash(Bash):
+    """AgentScope Bash with an optional read-only execution guard."""
+
+    name = "AutoBash"
+    description = "Execute shell commands; read-only mode denies commands that may write or run other commands."
+    input_schema = {
+        **Bash.input_schema,
+        "properties": {
+            **Bash.input_schema["properties"],
+            "timeout": {
+                **Bash.input_schema["properties"]["timeout"],
+                "description": "Optional timeout in milliseconds (default: 60000, max: 600000)",
+                "default": 60000,
+            },
+        },
+    }
+
+    def __init__(self, *, is_read_only: bool = True, **kwargs):
+        super().__init__(**kwargs)
+        self.is_read_only = is_read_only
+
+    @staticmethod
+    def _is_read_only(command: str) -> bool:
+        """Use safecmd's Bash parser and a strict command policy."""
+        if not isinstance(command, str) or not command.strip():
+            return False
+        # Variable and command expansion can hide a second operation in an
+        # argument. Keep this first version literal while safecmd checks the
+        # remaining shell syntax and nested commands.
+        if "$" in command or "`" in command:
+            return False
+        from safecmd import CmdSpec, validate  # pylint: disable=import-outside-toplevel
+        from safecmd.bashxtract import extract_commands  # pylint: disable=import-outside-toplevel
+        from safecmd.core import normalize_dest  # pylint: disable=import-outside-toplevel
+
+        specs = {CmdSpec(name, denied=options.get("denied")) for name, options in _READ_COMMANDS.items()}
+        try:
+            validate(command, cmds=specs, dests=_SAFE_DESTINATIONS)
+            commands, operators, redirects = extract_commands(command)
+            for tokens in commands:
+                options = _READ_COMMANDS[tokens[0]]
+                # An allow list constrains the first argument (for git, its subcommand).
+                # CmdSpec independently rejects every denied argument.
+                if "allow" in options and (len(tokens) < 2 or tokens[1] not in options["allow"]):
+                    return False
+            if "=" in operators or any(normalize_dest(destination) != "/dev/null" for _, destination in redirects):
+                return False
+        except Exception:  # pylint: disable=broad-exception-caught
+            # Parser failures and missing shfmt must never allow execution.
+            return False
+        return True
+
+    async def check_read_only(self, tool_input: dict) -> bool:
+        """Report the same policy used by the execution guard."""
+        return self.is_read_only and self._is_read_only(tool_input.get("command", ""))
+
+    async def check_permissions(self, tool_input: dict, _context: PermissionContext) -> PermissionDecision:
+        """Resolve permission automatically, without an ASK decision."""
+        if not self.is_read_only:
+            return PermissionDecision(behavior=PermissionBehavior.ALLOW, message="Command allowed")
+        allowed = await self.check_read_only(tool_input)
+        return PermissionDecision(
+            behavior=PermissionBehavior.ALLOW if allowed else PermissionBehavior.DENY,
+            message=(
+                "Read-only command allowed" if allowed else "Permission denied: only read-only commands are allowed"
+            ),
+        )
+
+    async def call(self, command: str, description: str = "", timeout: int = 60000):
+        """Apply the read-only guard when configured, then run the command."""
+        if self.is_read_only and (not isinstance(command, str) or not self._is_read_only(command)):
+            yield ToolChunk(
+                content=[TextBlock(text="Permission denied: only read-only shell commands are allowed.")],
+                state=ToolResultState.ERROR,
+                is_last=True,
+            )
+            return
+        # Git status may otherwise refresh the index, and Git's pager may run
+        # a configured program. These exports apply only in read-only mode.
+        if self.is_read_only:
+            command = "export GIT_OPTIONAL_LOCKS=0 GIT_PAGER=/bin/cat; " + command
+        async for chunk in super().call(command=command, description=description, timeout=timeout):
+            yield chunk
+
+
+class AutoBashStep(BaseStep):
+    """Expose AgentScope Bash with configurable read-only mode as a ReMe job."""
+
+    async def execute(self):
+        assert self.context is not None
+        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))
+        read_only_paths = self.context.get("read_only_paths", self.kwargs.get("read_only_paths"))
+        writable_paths = self.context.get("writable_paths", self.kwargs.get("writable_paths"))
+        memory_limit_bytes = self.context.get("memory_limit_bytes", self.kwargs.get("memory_limit_bytes"))
+        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:
+            backend = (
+                create_file_sandbox_backend(
+                    cwd=self.workspace_path,
+                    read_only_paths=[] if read_only_paths is None else read_only_paths,
+                    writable_paths=[] if writable_paths is None else writable_paths,
+                    memory_limit_bytes=memory_limit_bytes,
+                )
+                if read_only_paths is not None or writable_paths is not None
+                else None
+            )
+        except (TypeError, ValueError, RuntimeError) as exc:
+            self.context.response.success = False
+            self.context.response.answer = str(exc)
+            return self.context.response
+
+        tool = AutoBash(cwd=self.workspace_path, is_read_only=is_read_only, backend=backend)
+        chunks = [chunk async for chunk in tool.call(command=command, timeout=timeout)]
+        self.context.response.success = all(chunk.state != ToolResultState.ERROR for chunk in chunks)
+        self.context.response.answer = "".join(
+            block.text for chunk in chunks for block in chunk.content if isinstance(block, TextBlock)
+        )
+        return self.context.response

plugins/home/src/reme_home/plugin.yaml

@@ -0,0 +1,40 @@
+backends:
+  auto_bash_step: reme_home.auto_bash:AutoBashStep
+
+application_defaults:
+  jobs:
+    auto_bash:
+      backend: base
+      description: "Run a shell command in the workspace; read-only mode is enabled by default"
+      parameters:
+        type: object
+        properties:
+          command:
+            type: string
+            description: "shell command to execute"
+          is_read_only:
+            type: boolean
+            description: "restrict execution to read-only commands"
+            default: true
+          timeout:
+            type: integer
+            description: "maximum execution time in milliseconds (default 60000, max 600000)"
+            default: 60000
+            minimum: 1
+            maximum: 600000
+          read_only_paths:
+            type: array
+            items: { type: string }
+            description: "Read-only host paths; providing either path list enables the platform file sandbox"
+          writable_paths:
+            type: array
+            items: { type: string }
+            description: "Writable host paths; sandbox paths match host paths"
+          memory_limit_bytes:
+            type: integer
+            minimum: 0
+            description: "Linux cgroup memory limit in bytes (default 4 GiB); macOS warns and ignores nonzero values"
+        required:
+          - command
+      steps:
+        - backend: auto_bash_step

plugins/home/tests/test_auto_bash.py

@@ -0,0 +1,297 @@
+"""Shell job and its optional read-only execution guard."""
+
+import inspect
+from importlib.metadata import EntryPoint
+from pathlib import Path
+
+import pytest
+import yaml
+
+from agentscope.message import ToolResultState
+from agentscope.permission import PermissionBehavior
+from agentscope.tool import ExecResult
+from reme_home.auto_bash import AutoBash, AutoBashStep
+
+from reme.utils.path_bubble_backend import PathBubbleBackend
+from reme.components.application_context import ApplicationContext
+from reme.config.config_parser import _load_config
+from reme.enumeration import ComponentEnum
+from reme.plugin import resolve_plugin_runtime
+
+
+@pytest.fixture(autouse=True)
+def linux_backend_selection(monkeypatch):
+    """Keep existing backend integration assertions on the Linux implementation."""
+    monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "linux")
+
+
+@pytest.mark.asyncio
+async def test_auto_bash_runs_in_workspace(tmp_path):
+    """Approved commands run relative to the configured workspace."""
+    (tmp_path / "note.txt").write_text("hello", encoding="utf-8")
+    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+
+    response = await step(command="cat note.txt")
+
+    assert response.success is True
+    assert response.answer == "hello"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "command",
+    [
+        "touch denied.txt",
+        "cat note.txt > denied.txt",
+        "echo hello > /dev/null-extra",
+        "cat $(touch denied.txt)",
+        "echo ${X:-$(touch denied.txt)}",
+        "cat note.txt; touch denied.txt",
+        "cat note.txt\ntouch denied.txt",
+        "tee denied.txt",
+        "find . -delete",
+        "find . -exec touch denied.txt {} \\;",
+        "find . -fprint denied.txt",
+        "sort -o denied.txt note.txt",
+        "sort --compress-program='touch denied.txt' note.txt",
+        "diff --output=denied.txt note.txt note.txt",
+        "git add note.txt",
+        "git branch new-branch",
+        "git diff --output=denied.txt",
+        "git log --ext-diff",
+        "git -c alias.status=add status",
+        "GIT_OPTIONAL_LOCKS=1 git status",
+        "GIT_PAGER=sh git log",
+        "nop",
+        "rg --pre='touch denied.txt' pattern .",
+    ],
+)
+async def test_auto_bash_denies_mutation_before_execution(tmp_path, command):
+    """Writing and dynamic shell forms fail before a child process starts."""
+    (tmp_path / "note.txt").write_text("hello", encoding="utf-8")
+    tool = AutoBash(cwd=tmp_path)
+
+    chunks = [chunk async for chunk in tool.call(command)]
+
+    assert len(chunks) == 1
+    assert chunks[0].state == ToolResultState.ERROR
+    assert "Permission denied" in chunks[0].content[0].text
+    assert not (tmp_path / "denied.txt").exists()
+
+
+@pytest.mark.asyncio
+async def test_step_returns_permission_error(tmp_path):
+    """The job exposes a denied command as an unsuccessful response."""
+    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+
+    response = await step(command="touch denied.txt")
+
+    assert response.success is False
+    assert "Permission denied" in response.answer
+    assert not (tmp_path / "denied.txt").exists()
+
+
+@pytest.mark.asyncio
+async def test_step_can_run_unrestricted_commands(tmp_path):
+    """Job input can disable the guard for commands that write files."""
+    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+
+    response = await step(command="echo hello > written.txt", is_read_only=False)
+
+    assert response.success is True
+    assert (tmp_path / "written.txt").read_text(encoding="utf-8") == "hello\n"
+
+
+@pytest.mark.asyncio
+async def test_unsupported_platform_runs_without_paths_but_rejects_path_policy(tmp_path, monkeypatch):
+    """AutoBash never drops an explicitly requested sandbox on unsupported systems."""
+    monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "unsupported")
+    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+
+    plain = await step(command="echo host", is_read_only=False)
+    assert plain.success is True
+    assert plain.answer.strip() == "host"
+
+    restricted = await step(command="echo blocked", is_read_only=False, read_only_paths=[])
+    assert restricted.success is False
+    assert restricted.answer == "No file sandbox backend for unsupported"
+
+
+@pytest.mark.asyncio
+async def test_step_configuration_can_disable_read_only(tmp_path):
+    """Step configuration applies when the job does not supply the flag."""
+    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)), is_read_only=False)
+
+    response = await step(command="touch written.txt")
+
+    assert response.success is True
+    assert (tmp_path / "written.txt").exists()
+
+
+@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)))
+
+    response = await step(command="touch denied.txt", is_read_only="false")
+
+    assert response.success is False
+    assert response.answer == "is_read_only must be a boolean"
+    assert not (tmp_path / "denied.txt").exists()
+
+
+@pytest.mark.asyncio
+async def test_permission_decisions_never_ask(tmp_path):
+    """Direct AgentScope use makes allow/deny decisions without ASK."""
+    tool = AutoBash(cwd=tmp_path)
+
+    allowed = await tool.check_permissions({"command": "ls"}, None)
+    denied = await tool.check_permissions({"command": "touch denied.txt"}, None)
+
+    assert allowed.behavior == PermissionBehavior.ALLOW
+    assert denied.behavior == PermissionBehavior.DENY
+
+
+@pytest.mark.asyncio
+async def test_unrestricted_tool_allows_commands_without_read_only_check(tmp_path, monkeypatch):
+    """The unrestricted tool never invokes the read-only parser."""
+    tool = AutoBash(cwd=tmp_path, is_read_only=False)
+    monkeypatch.setattr(tool, "_is_read_only", lambda _command: pytest.fail("read-only parser was called"))
+
+    decision = await tool.check_permissions({"command": "touch written.txt"}, None)
+    chunks = [chunk async for chunk in tool.call("touch written.txt")]
+
+    assert decision.behavior == PermissionBehavior.ALLOW
+    assert all(chunk.state != ToolResultState.ERROR for chunk in chunks)
+    assert (tmp_path / "written.txt").exists()
+
+
+def test_auto_bash_defaults_to_sixty_seconds():
+    """Tool and default job expose the same 60-second timeout."""
+    manifest = Path(__file__).parents[1] / "src" / "reme_home" / "plugin.yaml"
+    with manifest.open(encoding="utf-8") as config_file:
+        job = yaml.safe_load(config_file)["application_defaults"]["jobs"]["auto_bash"]
+
+    assert inspect.signature(AutoBash.call).parameters["timeout"].default == 60000
+    assert AutoBash.input_schema["properties"]["timeout"]["default"] == 60000
+    assert job["parameters"]["properties"]["timeout"]["default"] == 60000
+    assert job["parameters"]["properties"]["is_read_only"]["default"] is True
+
+
+@pytest.mark.asyncio
+async def test_auto_bash_uses_path_backend_when_paths_supplied(tmp_path, monkeypatch):
+    """AutoBash routes an injected path policy through the backend."""
+    seen = {}
+
+    async def fake_exec(self, command, *, cwd=None, timeout=None):
+        seen.update(command=command, cwd=cwd, timeout=timeout, backend=self)
+        return ExecResult(exit_code=0, stdout=b"ok", stderr=b"")
+
+    monkeypatch.setattr(PathBubbleBackend, "exec_shell", fake_exec)
+    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    response = await step(command="printf ok", is_read_only=False, read_only_paths=[], writable_paths=[str(tmp_path)])
+
+    assert response.success is True
+    assert response.answer == "ok"
+    assert seen["command"] == ["/bin/sh", "-c", "printf ok"]
+    assert seen["cwd"] == str(tmp_path)
+    assert seen["timeout"] == 60
+
+
+@pytest.mark.asyncio
+async def test_sandbox_start_failure_does_not_run_on_host(tmp_path, monkeypatch):
+    """The plugin reports a sandbox failure instead of executing on the host."""
+
+    async def unavailable(*_args, **_kwargs):
+        raise RuntimeError("bwrap unavailable")
+
+    monkeypatch.setattr(PathBubbleBackend, "_start_process", unavailable)
+    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    result = await step(command="printf host", is_read_only=False, read_only_paths=[])
+
+    assert result.success is False
+    assert "bwrap unavailable" in result.answer
+
+
+def test_home_manifest_declares_step_and_job():
+    """The installed manifest is the source for backend and job registration."""
+    manifest = Path(__file__).parents[1] / "src" / "reme_home" / "plugin.yaml"
+    config = yaml.safe_load(manifest.read_text(encoding="utf-8"))
+    assert config["backends"]["auto_bash_step"] == "reme_home.auto_bash:AutoBashStep"
+    job = config["application_defaults"]["jobs"]["auto_bash"]
+    assert job["steps"] == [{"backend": "auto_bash_step"}]
+    assert job["parameters"]["properties"]["read_only_paths"]["type"] == "array"
+    assert job["parameters"]["properties"]["writable_paths"]["type"] == "array"
+
+
+def test_home_plugin_registers_job_and_step(monkeypatch):
+    """Enabling home adds its job and Step to one application runtime."""
+    entry = EntryPoint(name="home", value="reme_home", group="reme.plugins")
+    monkeypatch.setattr("reme.plugin.find_entry_points", lambda group, name: [entry] if name == "home" else [])
+    config = _load_config("default")
+    assert "auto_bash" not in config["jobs"]
+    config["plugins"] = ["home"]
+
+    runtime = resolve_plugin_runtime(config)
+
+    assert runtime.registry.get(ComponentEnum.STEP, "auto_bash_step") is AutoBashStep
+    assert runtime.config["jobs"]["auto_bash"]["steps"] == [{"backend": "auto_bash_step"}]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "command",
+    [
+        ":",
+        "cd . && pwd",
+        "echo hello",
+        "true",
+        "false",
+        "grep hello note.txt",
+        "rg hello note.txt",
+        "find . -name note.txt",
+        "sort note.txt",
+        "diff note.txt note.txt",
+        "cmp note.txt note.txt",
+        "git status --short",
+        "git diff --stat",
+        "git log --oneline",
+        "git show HEAD",
+        "git rev-parse HEAD",
+        "git ls-files",
+    ],
+)
+async def test_safe_command_forms_are_allowed(tmp_path, command):
+    """Common non-writing commands pass the policy regardless of exit code."""
+    (tmp_path / "note.txt").write_text("hello\n", encoding="utf-8")
+    tool = AutoBash(cwd=tmp_path)
+
+    assert await tool.check_read_only({"command": command})
+
+
+@pytest.mark.asyncio
+async def test_safe_redirection_is_allowed(tmp_path):
+    """safecmd allows output to /dev/null and input from workspace files."""
+    (tmp_path / "note.txt").write_text("hello\n", encoding="utf-8")
+    tool = AutoBash(cwd=tmp_path)
+
+    assert await tool.check_read_only({"command": "echo hello > /dev/null"})
+    assert await tool.check_read_only({"command": "cat < note.txt"})
+
+
+@pytest.mark.asyncio
+async def test_allow_and_denied_command_options_are_combined(tmp_path, monkeypatch):
+    """Allow constrains the first argument while denied still checks later flags."""
+    from reme_home import auto_bash
+
+    commands = auto_bash._READ_COMMANDS  # pylint: disable=protected-access
+    monkeypatch.setitem(commands, "echo", {"allow": ["hello"]})
+    tool = AutoBash(cwd=tmp_path)
+
+    assert await tool.check_read_only({"command": "echo hello"})
+    assert not await tool.check_read_only({"command": "echo goodbye"})
+
+    monkeypatch.setitem(commands, "echo", {"allow": ["hello"], "denied": ["-n"]})
+    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/config/benchmark.yaml

@@ -142,6 +142,10 @@ jobs:
         python_timeout:
           type: number
           description: "Execution timeout in seconds; defaults to 60."
+        memory_limit_bytes:
+          type: integer
+          minimum: 0
+          description: "Linux memory limit in bytes (default 4 GiB); macOS ignores nonzero values"
       required:
         - code
     steps:

reme/steps/common/python_execute.py

@@ -5,6 +5,8 @@ import sys
 from dataclasses import dataclass
 from typing import Any
 
+from ...utils.file_sandbox import create_file_sandbox_backend
+from ...utils.file_sandbox_backend import FileSandboxBackend
 from ..base_step import BaseStep
 from ...components import R
 
@@ -39,7 +41,31 @@ class PythonExecuteStep(BaseStep):
             self.context.response.answer = timeout_error
             return self.context.response
 
-        result = await self._run_python(code, timeout)
+        read_only_paths = self.context.get("read_only_paths", self.kwargs.get("read_only_paths"))
+        writable_paths = self.context.get("writable_paths", self.kwargs.get("writable_paths"))
+        memory_limit_bytes = self.context.get("memory_limit_bytes", self.kwargs.get("memory_limit_bytes"))
+        try:
+            backend = (
+                create_file_sandbox_backend(
+                    cwd=self.workspace_path,
+                    read_only_paths=[] if read_only_paths is None else read_only_paths,
+                    writable_paths=[] if writable_paths is None else writable_paths,
+                    memory_limit_bytes=memory_limit_bytes,
+                )
+                if read_only_paths is not None or writable_paths is not None
+                else None
+            )
+        except (TypeError, ValueError, RuntimeError) as exc:
+            self.context.response.success = False
+            self.context.response.answer = str(exc)
+            return self.context.response
+
+        try:
+            result = await self._run_python(code, timeout, backend)
+        except (OSError, RuntimeError) as exc:
+            self.context.response.success = False
+            self.context.response.answer = str(exc)
+            return self.context.response
         if result.timed_out:
             self.context.response.success = False
             self.context.response.answer = f"Python execution timed out after {timeout:g}s"
@@ -65,7 +91,19 @@ class PythonExecuteStep(BaseStep):
         )
         return self.context.response
 
-    async def _run_python(self, code: str, timeout: float) -> _PythonResult:
+    async def _run_python(self, code: str, timeout: float, backend: FileSandboxBackend | None = None) -> _PythonResult:
+        if backend is not None:
+            result = await backend.exec_shell(
+                [sys.executable, "-c", code],
+                cwd=str(self.workspace_path),
+                timeout=timeout,
+            )
+            return _PythonResult(
+                stdout=result.stdout.decode(errors="replace"),
+                stderr=result.stderr.decode(errors="replace"),
+                returncode=result.exit_code,
+                timed_out=result.exit_code == -1 and result.stderr == b"timed out",
+            )
         process = await asyncio.create_subprocess_exec(
             sys.executable,
             "-c",

reme/utils/file_sandbox.py

@@ -0,0 +1,18 @@
+"""Select a local file sandbox for supported operating systems."""
+
+import sys
+
+from .file_sandbox_backend import FileSandboxBackend
+from .path_bubble_backend import PathBubbleBackend
+from .path_seatbelt_backend import PathSeatbeltBackend
+
+
+def create_file_sandbox_backend(*, memory_limit_bytes: int | None = None, **kwargs) -> FileSandboxBackend:
+    """Construct the platform backend; never silently execute without isolation."""
+    if sys.platform.startswith("linux"):
+        limit = 4 * 1024**3 if memory_limit_bytes is None else memory_limit_bytes
+        return PathBubbleBackend(memory_limit_bytes=limit, **kwargs)
+    if sys.platform == "darwin":
+        limit = 0 if memory_limit_bytes is None else memory_limit_bytes
+        return PathSeatbeltBackend(memory_limit_bytes=limit, **kwargs)
+    raise RuntimeError(f"No file sandbox backend for {sys.platform}")

reme/utils/file_sandbox_backend.py

@@ -0,0 +1,132 @@
+"""Common contract and path validation for subprocess file sandboxes."""
+
+import asyncio
+import os
+import signal
+import sys
+from abc import abstractmethod
+
+from agentscope.tool import BackendBase, ExecResult
+
+
+def contains(root: str, path: str) -> bool:
+    """Whether a path is inside a configured root."""
+    return os.path.commonpath((root, path)) == root
+
+
+class FileSandboxBackend(BackendBase):
+    """Execute subprocesses with explicit host-path read and write access."""
+
+    def __init__(self, *, cwd: str | os.PathLike[str], read_only_paths: list[str], writable_paths: list[str]) -> None:
+        self._cwd = self._validate_path(os.fspath(cwd))
+        if not os.path.isdir(self._cwd):
+            raise ValueError("sandbox working directory must be a directory")
+        self._read_only_paths = self._validate_paths(read_only_paths, "read_only_paths")
+        self._writable_paths = self._validate_paths(writable_paths, "writable_paths")
+        if not any(contains(path, self._cwd) for path in self._read_only_paths + self._writable_paths):
+            self._read_only_paths.append(self._cwd)
+        runtime_prefix = os.path.realpath(sys.prefix)
+        if runtime_prefix not in ("/", "/usr", "/usr/local") and not runtime_prefix.startswith("/usr/"):
+            if not any(contains(path, runtime_prefix) for path in self._read_only_paths + self._writable_paths):
+                self._read_only_paths.append(self._validate_path(runtime_prefix))
+        paths = self._read_only_paths + self._writable_paths
+        if len(set(paths)) != len(paths):
+            raise ValueError("sandbox paths must not contain duplicates")
+        self._path_identities = {path: self._identity(path) for path in paths}
+
+    @staticmethod
+    def _identity(path: str) -> tuple[int, int]:
+        info = os.stat(path, follow_symlinks=False)
+        return info.st_dev, info.st_ino
+
+    @staticmethod
+    def _validate_path(raw: str) -> str:
+        if not isinstance(raw, str) or not raw or not os.path.isabs(raw):
+            raise ValueError("sandbox paths must be non-empty absolute paths")
+        path = os.path.normpath(raw)
+        if path == "/" or not os.path.exists(path) or os.path.realpath(path) != path:
+            raise ValueError(f"sandbox path must exist without symlinks and cannot be root: {raw}")
+        return path
+
+    @classmethod
+    def _validate_paths(cls, raw: list[str], name: str) -> list[str]:
+        if not isinstance(raw, list) or any(not isinstance(item, str) for item in raw):
+            raise ValueError(f"{name} must be a list of absolute paths")
+        return [cls._validate_path(path) for path in raw]
+
+    def _check_paths(self, cwd: str) -> None:
+        if not any(contains(path, cwd) for path in self._read_only_paths + self._writable_paths):
+            raise ValueError("command working directory must be inside a sandbox path")
+        for path, identity in self._path_identities.items():
+            if not os.path.exists(path) or os.path.realpath(path) != path or self._identity(path) != identity:
+                raise RuntimeError(f"sandbox path changed before execution: {path}")
+
+    async def getcwd(self) -> str:
+        """Return the host-identical sandbox working directory."""
+        return self._cwd
+
+    @abstractmethod
+    async def _start_process(self, command: list[str], cwd: str, *, stdin=asyncio.subprocess.DEVNULL):
+        """Start one sandboxed process, never falling back to host execution."""
+
+    def _release_process(self, process: asyncio.subprocess.Process) -> None:
+        """Release platform resources after the child has exited."""
+
+    async def exec_shell(
+        self,
+        command: list[str],
+        *,
+        cwd: str | None = None,
+        timeout: float | None = None,
+    ) -> ExecResult:
+        """Execute an argv list with a timeout inside the platform sandbox."""
+        process = await self._start_process(command, cwd or self._cwd)
+        try:
+            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
+        except TimeoutError:
+            self._kill_process(process)
+            await process.communicate()
+            return ExecResult(exit_code=-1, stdout=b"", stderr=b"timed out")
+        except BaseException:
+            self._kill_process(process)
+            await asyncio.shield(process.communicate())
+            raise
+        finally:
+            self._release_process(process)
+        return ExecResult(exit_code=process.returncode, stdout=stdout, stderr=stderr)
+
+    @staticmethod
+    def _kill_process(process: asyncio.subprocess.Process) -> None:
+        """Terminate the sandbox launcher and its process group."""
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+
+    async def read_file(self, path: str) -> bytes:
+        """Read through a sandboxed command so file rules still apply."""
+        result = await self.exec_shell(["cat", path])
+        if not result.ok():
+            raise OSError(result.stderr.decode(errors="replace"))
+        return result.stdout
+
+    async def write_file(self, path: str, data: bytes) -> None:
+        """Write through a sandboxed command so file rules still apply."""
+        if not isinstance(data, bytes):
+            raise TypeError("data must be bytes")
+        script = 'mkdir -p -- "$(dirname -- "$1")" && cat > "$1"'
+        process = await self._start_process(
+            ["/bin/sh", "-c", script, "sh", path],
+            self._cwd,
+            stdin=asyncio.subprocess.PIPE,
+        )
+        try:
+            _stdout, stderr = await process.communicate(data)
+        except BaseException:
+            self._kill_process(process)
+            await asyncio.shield(process.communicate())
+            raise
+        finally:
+            self._release_process(process)
+        if process.returncode != 0:
+            raise OSError(stderr.decode(errors="replace"))

reme/utils/path_bubble_backend.py

@@ -0,0 +1,129 @@
+"""Bubblewrap execution with host-identical, explicitly mounted paths."""
+
+import asyncio
+import os
+import shutil
+import uuid
+from pathlib import Path
+
+import sys
+
+from .file_sandbox_backend import FileSandboxBackend
+
+
+class PathBubbleBackend(FileSandboxBackend):
+    """Run commands with selected host paths at the same sandbox paths."""
+
+    def __init__(
+        self,
+        *,
+        cwd: str | os.PathLike[str],
+        read_only_paths: list[str],
+        writable_paths: list[str],
+        memory_limit_bytes: int = 4 * 1024**3,
+    ) -> None:
+        if isinstance(memory_limit_bytes, bool) or not isinstance(memory_limit_bytes, int) or memory_limit_bytes < 0:
+            raise ValueError("memory_limit_bytes must be a non-negative integer")
+        self._memory_limit_bytes = memory_limit_bytes
+        self._process_cgroups: dict[int, Path] = {}
+        super().__init__(cwd=cwd, read_only_paths=read_only_paths, writable_paths=writable_paths)
+
+    def _bwrap_argv(self, command: list[str], cwd: str) -> list[str]:
+        """Build one isolated execution with host-identical bind targets."""
+        if not command:
+            raise ValueError("command must not be empty")
+        self._check_paths(cwd)
+        args = ["bwrap", "--die-with-parent", "--new-session", "--unshare-all", "--proc", "/proc", "--dev", "/dev"]
+        for path in ("/usr", "/usr/local"):
+            if os.path.isdir(path):
+                args.extend(("--ro-bind", path, path))
+        for path in ("/bin", "/sbin", "/lib", "/lib64"):
+            if os.path.islink(path):
+                args.extend(("--symlink", os.readlink(path), path))
+            elif os.path.exists(path):
+                args.extend(("--ro-bind", path, path))
+        args.extend(("--dir", "/etc"))
+        for path in ("/etc/passwd", "/etc/group", "/etc/nsswitch.conf", "/etc/hosts", "/etc/resolv.conf", "/etc/ssl"):
+            if os.path.exists(path):
+                args.extend(("--ro-bind", path, path))
+        args.extend(("--tmpfs", "/tmp", "--tmpfs", "/run"))
+        mounts = [(path, "--ro-bind") for path in self._read_only_paths]
+        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))
+        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, "--"))
+        return args + command
+
+    async def _start_process(self, command: list[str], cwd: str, *, stdin=asyncio.subprocess.DEVNULL):
+        """Start bwrap without falling back to host execution."""
+        if not sys.platform.startswith("linux"):
+            raise RuntimeError("PathBubbleBackend requires Linux")
+        if shutil.which("bwrap") is None:
+            raise RuntimeError("PathBubbleBackend requires bwrap on PATH")
+        argv = self._bwrap_argv(command, cwd)
+        group = self._create_cgroup() if self._memory_limit_bytes else None
+        try:
+
+            def join_cgroup() -> None:
+                """Move the forked child before it executes bubblewrap."""
+                assert group is not None
+                descriptor = os.open(group / "cgroup.procs", os.O_WRONLY)
+                try:
+                    os.write(descriptor, str(os.getpid()).encode("ascii"))
+                finally:
+                    os.close(descriptor)
+
+            process = await asyncio.create_subprocess_exec(
+                *argv,
+                stdin=stdin,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                start_new_session=True,
+                preexec_fn=join_cgroup if group is not None else None,
+            )
+        except BaseException:
+            if group is not None:
+                group.rmdir()
+            raise
+        if group is not None:
+            self._process_cgroups[process.pid] = group
+        return process
+
+    def _create_cgroup(self) -> Path:
+        """Create a separate cgroup v2 for one execution, failing closed if unavailable."""
+        try:
+            entries = Path("/proc/self/cgroup").read_text(encoding="ascii").splitlines()
+            relative = next(line.split("::", 1)[1] for line in entries if line.startswith("0::"))
+            root = Path("/sys/fs/cgroup")
+            parent = root / relative.lstrip("/")
+            if ".." in Path(relative).parts or not parent.resolve().is_relative_to(root):
+                raise RuntimeError("invalid cgroup v2 path")
+            group = parent / f"reme-{uuid.uuid4().hex}"
+            group.mkdir()
+            try:
+                (group / "memory.max").write_text(str(self._memory_limit_bytes), encoding="ascii")
+            except BaseException:
+                group.rmdir()
+                raise
+            return group
+        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:
+        """Release the per-execution group after its process has been reaped."""
+        group = self._process_cgroups.pop(process.pid, None)
+        if group is None:
+            return
+        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
+
+    # Keep the existing helper name for callers that manage a process directly.
+    _release_cgroup = _release_process

reme/utils/path_seatbelt_backend.py

@@ -0,0 +1,83 @@
+"""macOS Seatbelt file sandbox with host-identical paths."""
+
+import asyncio
+import json
+import os
+import shutil
+import sys
+import tempfile
+import warnings
+
+from .file_sandbox_backend import FileSandboxBackend
+
+
+class PathSeatbeltBackend(FileSandboxBackend):
+    """Run each command under a deny-by-default Seatbelt profile."""
+
+    def __init__(
+        self,
+        *,
+        cwd: str | os.PathLike[str],
+        read_only_paths: list[str],
+        writable_paths: list[str],
+        memory_limit_bytes: int = 0,
+    ) -> None:
+        if isinstance(memory_limit_bytes, bool) or not isinstance(memory_limit_bytes, int) or memory_limit_bytes < 0:
+            raise ValueError("memory_limit_bytes must be a non-negative integer")
+        if memory_limit_bytes:
+            warnings.warn(
+                "PathSeatbeltBackend does not support memory_limit_bytes; the limit is ignored on macOS",
+                RuntimeWarning,
+                stacklevel=2,
+            )
+        super().__init__(cwd=cwd, read_only_paths=read_only_paths, writable_paths=writable_paths)
+        self._process_tempdirs: dict[int, str] = {}
+
+    @staticmethod
+    def _path_rule(path: str) -> str:
+        """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)})"
+
+    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")
+        read_paths = [path for path in system_roots if os.path.exists(path)]
+        read_paths += self._read_only_paths + self._writable_paths + [tempdir]
+        lines = ["(version 1)", "(deny default)", "(allow process*)", "(allow sysctl-read)", "(allow mach-lookup)"]
+        lines.extend(f"(allow file-read* {self._path_rule(path)})" for path in read_paths)
+        lines.extend(f"(allow file-write* {self._path_rule(path)})" for path in self._writable_paths + [tempdir])
+        return "\n".join(lines)
+
+    async def _start_process(self, command: list[str], cwd: str, *, stdin=asyncio.subprocess.DEVNULL):
+        if sys.platform != "darwin":
+            raise RuntimeError("PathSeatbeltBackend requires macOS")
+        if shutil.which("sandbox-exec") is None:
+            raise RuntimeError("PathSeatbeltBackend requires sandbox-exec on PATH")
+        if not command:
+            raise ValueError("command must not be empty")
+        self._check_paths(cwd)
+        tempdir = tempfile.mkdtemp(prefix="reme-seatbelt-", dir=os.path.realpath(tempfile.gettempdir()))
+        try:
+            env = {**os.environ, "HOME": self._cwd, "TMPDIR": tempdir}
+            process = await asyncio.create_subprocess_exec(
+                "/usr/bin/sandbox-exec",
+                "-p",
+                self._profile(tempdir),
+                *command,
+                cwd=cwd,
+                env=env,
+                stdin=stdin,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                start_new_session=True,
+            )
+        except BaseException:
+            shutil.rmtree(tempdir)
+            raise
+        self._process_tempdirs[process.pid] = tempdir
+        return process
+
+    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)

tests/unit/test_file_sandbox_backend.py

@@ -0,0 +1,98 @@
+"""Shared file sandbox selection and macOS policy construction."""
+
+import sys
+
+import pytest
+from agentscope.tool import ExecResult
+
+from reme.components.application_context import ApplicationContext
+from reme.steps.common.python_execute import PythonExecuteStep
+from reme.utils.file_sandbox import create_file_sandbox_backend
+from reme.utils.path_bubble_backend import PathBubbleBackend
+from reme.utils.path_seatbelt_backend import PathSeatbeltBackend
+
+
+def test_platform_selection_and_memory_defaults(tmp_path, monkeypatch):
+    """Linux retains cgroups while macOS defaults to no unsupported memory cap."""
+    monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "linux")
+    linux = create_file_sandbox_backend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    assert isinstance(linux, PathBubbleBackend)
+    assert linux._memory_limit_bytes == 4 * 1024**3  # pylint: disable=protected-access
+
+    monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "darwin")
+    mac = create_file_sandbox_backend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    assert isinstance(mac, PathSeatbeltBackend)
+    with pytest.warns(RuntimeWarning, match="limit is ignored on macOS"):
+        limited = create_file_sandbox_backend(
+            cwd=tmp_path,
+            read_only_paths=[],
+            writable_paths=[],
+            memory_limit_bytes=1024,
+        )
+    assert isinstance(limited, PathSeatbeltBackend)
+
+
+@pytest.mark.asyncio
+async def test_unsupported_platform_requires_backend_only_for_path_policy(tmp_path, monkeypatch):
+    """An unsupported platform runs ordinary Python but rejects a requested sandbox."""
+    monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "unsupported")
+    step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+
+    plain = await step(code="print('host')")
+    assert plain.success is True
+    assert plain.answer == "host\n"
+
+    restricted = await step(code="print('blocked')", read_only_paths=[])
+    assert restricted.success is False
+    assert restricted.answer == "No file sandbox backend for unsupported"
+
+
+def test_seatbelt_profile_limits_file_access(tmp_path):
+    """The profile grants configured paths and a separate per-run temporary directory."""
+    output = tmp_path / "output"
+    output.mkdir()
+    backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[str(tmp_path)], writable_paths=[str(output)])
+    profile = backend._profile(str(tmp_path / "temporary"))  # pylint: disable=protected-access
+    assert "(deny default)" in profile
+    assert f'(allow file-read* (subpath "{tmp_path}"))' in profile
+    assert f'(allow file-write* (subpath "{output}"))' in profile
+    assert f'(allow file-write* (subpath "{tmp_path}"))' not in profile
+    assert "network*" not in profile
+
+
+@pytest.mark.asyncio
+async def test_seatbelt_launch_failure_cleans_up_without_host_fallback(tmp_path, monkeypatch):
+    """A failed sandbox launcher never starts the requested command directly."""
+    monkeypatch.setattr(sys, "platform", "darwin")
+    monkeypatch.setattr("reme.utils.path_seatbelt_backend.shutil.which", lambda _: "/usr/bin/sandbox-exec")
+    calls = []
+
+    async def fail(*argv, **_kwargs):
+        calls.append(argv)
+        raise OSError("sandbox unavailable")
+
+    monkeypatch.setattr("reme.utils.path_seatbelt_backend.asyncio.create_subprocess_exec", fail)
+    backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    with pytest.raises(OSError, match="sandbox unavailable"):
+        await backend.exec_shell(["/bin/echo", "unsafe"])
+    assert calls[0][0] == "/usr/bin/sandbox-exec"
+    assert calls[0][-2:] == ("/bin/echo", "unsafe")
+    assert not backend._process_tempdirs  # pylint: disable=protected-access
+
+
+@pytest.mark.asyncio
+async def test_python_job_selects_seatbelt_on_macos(tmp_path, monkeypatch):
+    """Path injection reaches the macOS backend with no implicit memory cap."""
+    monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "darwin")
+    seen = {}
+
+    async def fake_exec(self, command, *, cwd=None, timeout=None):
+        seen.update(backend=self, command=command, cwd=cwd, timeout=timeout)
+        return ExecResult(exit_code=0, stdout=b"ok\n", stderr=b"")
+
+    monkeypatch.setattr(PathSeatbeltBackend, "exec_shell", fake_exec)
+    step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    response = await step(code="print('ok')", read_only_paths=[])
+    assert response.success is True
+    assert isinstance(seen["backend"], PathSeatbeltBackend)
+    assert seen["command"] == [sys.executable, "-c", "print('ok')"]

tests/unit/test_path_bubble_backend.py

@@ -0,0 +1,212 @@
+"""Path-preserving Bubblewrap mounts and job integration."""
+
+import sys
+from types import SimpleNamespace
+from pathlib import Path
+
+import pytest
+import yaml
+
+from agentscope.tool import ExecResult
+
+from reme.utils.path_bubble_backend import PathBubbleBackend
+from reme.components.application_context import ApplicationContext
+from reme.steps.common.python_execute import PythonExecuteStep
+
+
+@pytest.fixture(autouse=True)
+def linux_backend_selection(monkeypatch):
+    """Keep existing integration assertions on the Linux backend."""
+    monkeypatch.setattr("reme.utils.file_sandbox.sys.platform", "linux")
+
+
+def _mounts(argv):
+    return [
+        (argv[index], argv[index + 1], argv[index + 2])
+        for index in range(len(argv) - 2)
+        if argv[index] in ("--bind", "--ro-bind")
+    ]
+
+
+def test_path_bubble_backend_preserves_host_paths_and_mount_order(tmp_path):
+    """Nested writable mounts override a read-only parent at the same path."""
+    output = tmp_path / "output"
+    output.mkdir()
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[str(tmp_path)], writable_paths=[str(output)])
+
+    argv = backend._bwrap_argv(["/bin/sh", "-c", "pwd"], str(tmp_path))  # pylint: disable=protected-access
+    mounts = _mounts(argv)
+
+    assert ("--ro-bind", str(tmp_path), str(tmp_path)) in mounts
+    assert ("--bind", str(output), str(output)) in mounts
+    assert mounts.index(("--ro-bind", str(tmp_path), str(tmp_path))) < mounts.index(
+        ("--bind", str(output), str(output)),
+    )
+    assert argv[-3:] == ["/bin/sh", "-c", "pwd"]
+    assert argv[argv.index("--chdir") + 1] == str(tmp_path)
+    assert "--share-net" not in argv
+
+
+def test_path_bubble_backend_adds_workspace_read_only(tmp_path):
+    """The working directory is always accessible inside the sandbox."""
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    argv = backend._bwrap_argv(["true"], str(tmp_path))  # pylint: disable=protected-access
+    assert ("--ro-bind", str(tmp_path), str(tmp_path)) in _mounts(argv)
+
+
+@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 bytes."""
+    with pytest.raises(ValueError, match="memory_limit_bytes"):
+        PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_bytes=limit)
+
+
+@pytest.mark.asyncio
+async def test_each_execution_gets_separate_memory_cgroup(tmp_path, monkeypatch):
+    """Each process receives its own cgroup before bubblewrap is executed."""
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_bytes=1024)
+    groups = []
+    subprocess_options = []
+
+    def create_group():
+        group = tmp_path / f"group-{len(groups)}"
+        group.mkdir()
+        groups.append(group)
+        return group
+
+    async def create_process(*_argv, **kwargs):
+        subprocess_options.append(kwargs)
+        return SimpleNamespace(pid=100 + len(subprocess_options))
+
+    monkeypatch.setattr(sys, "platform", "linux")
+    monkeypatch.setattr("reme.utils.path_bubble_backend.shutil.which", lambda _: "/usr/bin/bwrap")
+    monkeypatch.setattr(backend, "_create_cgroup", create_group)
+    monkeypatch.setattr("reme.utils.path_bubble_backend.asyncio.create_subprocess_exec", create_process)
+    first = await backend._start_process(["true"], str(tmp_path))  # pylint: disable=protected-access
+    second = await backend._start_process(["true"], str(tmp_path))  # pylint: disable=protected-access
+
+    assert len(groups) == 2 and groups[0] != groups[1]
+    assert all(options["preexec_fn"] is not None for options in subprocess_options)
+    backend._release_cgroup(first)  # pylint: disable=protected-access
+    backend._release_cgroup(second)  # pylint: disable=protected-access
+    assert not any(group.exists() for group in groups)
+
+
+@pytest.mark.asyncio
+async def test_zero_memory_limit_skips_cgroup(tmp_path, monkeypatch):
+    """A zero limit bypasses cgroup creation and child enrollment."""
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_bytes=0)
+    options = {}
+
+    async def create_process(*_argv, **kwargs):
+        options.update(kwargs)
+        return SimpleNamespace(pid=123)
+
+    monkeypatch.setattr(sys, "platform", "linux")
+    monkeypatch.setattr("reme.utils.path_bubble_backend.shutil.which", lambda _: "/usr/bin/bwrap")
+    monkeypatch.setattr("reme.utils.path_bubble_backend.asyncio.create_subprocess_exec", create_process)
+    await backend._start_process(["true"], str(tmp_path))  # pylint: disable=protected-access
+    assert options["preexec_fn"] is None
+
+
+@pytest.mark.parametrize("path", ["relative", "/", "/does-not-exist"])
+def test_path_bubble_backend_rejects_invalid_mounts(tmp_path, path):
+    """Only existing absolute non-root paths can be mounted."""
+    with pytest.raises(ValueError):
+        PathBubbleBackend(cwd=tmp_path, read_only_paths=[path], writable_paths=[])
+
+
+def test_path_bubble_backend_rejects_symlink_and_duplicates(tmp_path):
+    """Ambiguous mount sources are rejected before a process starts."""
+    alias = tmp_path.parent / "alias"
+    alias.symlink_to(tmp_path)
+    try:
+        with pytest.raises(ValueError):
+            PathBubbleBackend(cwd=tmp_path, read_only_paths=[str(alias)], writable_paths=[])
+        with pytest.raises(ValueError):
+            PathBubbleBackend(cwd=tmp_path, read_only_paths=[str(tmp_path)], writable_paths=[str(tmp_path)])
+    finally:
+        alias.unlink()
+
+
+def test_path_bubble_backend_rejects_replaced_mount_source(tmp_path):
+    """A path swapped after validation cannot become a new mount source."""
+    source = tmp_path / "source"
+    source.mkdir()
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[str(source)], writable_paths=[])
+    source.rename(tmp_path / "old-source")
+    source.mkdir()
+
+    with pytest.raises(RuntimeError, match="sandbox path changed"):
+        backend._bwrap_argv(["true"], str(tmp_path))  # pylint: disable=protected-access
+
+
+@pytest.mark.asyncio
+async def test_python_execute_uses_path_backend_when_paths_supplied(tmp_path, monkeypatch):
+    """Python execution uses the same backend and timeout units."""
+    seen = {}
+
+    async def fake_exec(self, command, *, cwd=None, timeout=None):
+        seen.update(command=command, cwd=cwd, timeout=timeout, backend=self)
+        return ExecResult(exit_code=0, stdout=b"ok\n", stderr=b"")
+
+    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)])
+
+    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_bytes == 4 * 1024**3  # 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_bytes  # 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_bytes=0)
+    assert response.success is True
+    assert seen["limit"] == 0
+
+
+@pytest.mark.asyncio
+async def test_step_rejects_invalid_mount_parameters(tmp_path):
+    """Malformed path injection cannot silently disable sandboxing."""
+    step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    response = await step(code="print('unsafe')", writable_paths="/tmp")
+    assert response.success is False
+    assert "writable_paths must be a list" in response.answer
+
+
+@pytest.mark.asyncio
+async def test_sandbox_start_failure_does_not_run_on_host(tmp_path, monkeypatch):
+    """Python execution reports a sandbox failure instead of running on the host."""
+
+    async def unavailable(*_args, **_kwargs):
+        raise RuntimeError("bwrap unavailable")
+
+    monkeypatch.setattr(PathBubbleBackend, "_start_process", unavailable)
+    app_context = ApplicationContext(workspace_dir=str(tmp_path))
+    python_step = PythonExecuteStep(app_context=app_context)
+    python_result = await python_step(code="print('host')", read_only_paths=[])
+
+    assert python_result.success is False
+    assert "bwrap unavailable" in python_result.answer
+
+
+def test_job_hides_injected_path_policy():
+    """Path policy reaches the Step through context without becoming a tool argument."""
+    config_path = Path(__file__).parents[2] / "reme" / "config" / "benchmark.yaml"
+    parameters = yaml.safe_load(config_path.read_text(encoding="utf-8"))["jobs"]["python_execute"]["parameters"]
+    assert "read_only_paths" not in parameters["properties"]
+    assert "writable_paths" not in parameters["properties"]
+    assert "default" not in parameters["properties"]["memory_limit_bytes"]