Merge 1193139aee into d6b02a20ba

plugins/home/README.md

@@ -0,0 +1,25 @@
+# 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. On Linux, the
+`memory_limit_mib` cgroup limit defaults to 4096 MiB; 0 disables it. Supplying
+the limit requires a path policy. macOS has no memory limit and warns when a
+nonzero value is supplied.
+
+Standard output and standard error each have a 1 MiB (1,048,576-byte) limit on
+all backends. Exceeding either limit terminates the command and returns a
+failure with bounded partial output and a diagnostic in standard error.

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.14",
+    "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,281 @@
+"""Run allowlisted shell commands without an approval prompt."""
+
+import asyncio
+import os
+import shlex
+from functools import cache
+
+from agentscope.message import TextBlock, ToolResultState
+from agentscope.permission import PermissionBehavior, PermissionContext, PermissionDecision
+from agentscope.tool import Bash, ToolChunk
+
+from reme.steps.base_step import BaseStep
+from reme.steps.common.file_sandbox import file_sandbox_for_step
+
+from reme.utils.local_backend import LocalProcessBackend
+
+_READ_COMMANDS = {
+    ":": {},
+    "cat": {},
+    "cd": {},
+    "cmp": {},
+    "diff": {"denied": ["--output"]},
+    "echo": {},
+    "false": {},
+    "file": {
+        "denied": ["-C", "--compile", "-p", "--preserve-date", "-z", "-Z", "--uncompress", "--uncompress-noreport"],
+    },
+    "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",
+            "--submodule",
+            "--ignore-submodules",
+            "--show-signature",
+            "--remerge-diff",
+            "--diff-merges",
+            "--pretty",
+            "--format",
+        ],
+    },
+    "grep": {},
+    "head": {},
+    "ls": {},
+    "pwd": {},
+    "sort": {"denied": ["-o", "-T", "--output", "--compress-program", "--temporary-directory"]},
+    "stat": {},
+    "tail": {},
+    "true": {},
+    "wc": {},
+    "which": {},
+}
+_SAFE_DESTINATIONS = {"/dev/null"}
+
+# Trusted shell setup, applied only after validating the user's command.
+# Git's configuration can launch helpers even for otherwise read-only commands.
+_READ_ONLY_SETUP = r"""
+unset GIT_EXTERNAL_DIFF
+export GIT_OPTIONAL_LOCKS=0 GIT_PAGER=/bin/cat GIT_NO_LAZY_FETCH=1 GIT_TERMINAL_PROMPT=0
+sort() {
+    # A large sort otherwise writes temporary files, even without -o.
+    command sort -T /dev/null "$@"
+}
+git() {
+    # These metadata queries do not read file contents through Git filters.
+    if [ "$1" != rev-parse ] && [ "$1" != ls-files ] && \
+        command git -c trace2.normalTarget=0 -c trace2.eventTarget=0 -c trace2.perfTarget=0 \
+        config --get-regexp '^filter\..*\.(clean|smudge|process)$' > /dev/null; then
+        echo 'Permission denied: Git repositories with external filters require unrestricted mode.' >&2
+        return 1
+    fi
+    case "$1" in
+        diff|log|show)
+            reme_git_subcommand="$1"
+            shift
+            set -- "$reme_git_subcommand" --no-ext-diff --no-textconv --ignore-submodules=all "$@"
+            ;;
+        status) shift; set -- status --ignore-submodules=all "$@" ;;
+    esac
+    command git --no-pager -c core.fsmonitor=false -c diff.submodule=short \
+        -c log.showSignature=false -c log.diffMerges=off -c format.pretty=medium \
+        -c trace2.normalTarget=0 -c trace2.eventTarget=0 -c trace2.perfTarget=0 "$@"
+}
+"""
+
+
+@cache
+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()}
+
+
+def _literal_shell(node: dict, command: str, quoted: bool = False) -> bool:
+    """Reject expansions and syntax whose effects safecmd does not fully check."""
+    from safecmd.bashxtract import word_text  # pylint: disable=import-outside-toplevel
+
+    kind = node.get("Type")
+    if kind in {"FuncDecl", "ForClause", "DeclClause", "ArithmExp", "ArithmCmd", "ProcSubst", "CmdSubst"}:
+        return False
+    if kind == "SglQuoted":
+        return True
+    quoted = quoted or kind == "DblQuoted"
+    if kind == "Lit":
+        value = node.get("Value", "")
+        if "\\" in value or (not quoted and any(char in value for char in "*?[]{}~")):
+            return False
+    for redirect in node.get("Redirs", []):
+        op = redirect.get("Op")
+        destination = word_text(redirect.get("Word", {}), command)
+        if op in {"<", "<<", "<<-"}:
+            continue
+        if op in {">&", "<&"} and (destination == "-" or destination.rstrip("-").isdigit()):
+            continue
+        if op not in {">", ">>", ">|", "<>", "&>", "&>>", ">&"} or destination != "/dev/null":
+            return False
+    for value in node.values():
+        children = value if isinstance(value, list) else [value]
+        if any(isinstance(child, dict) and not _literal_shell(child, command, quoted) for child in children):
+            return False
+    return True
+
+
+def _allowed_arguments(tokens: list[str]) -> bool:
+    """Supplement CmdSpec with subcommands, 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
+    # getopt and Git also accept unambiguous long-option abbreviations.
+    for token in tokens[1:]:
+        flag = token.split("=", 1)[0]
+        if flag.startswith("--") and len(flag) > 2:
+            if any(denied.startswith(flag) for denied in options.get("denied", []) if denied.startswith("--")):
+                return False
+    return True
+
+
+class AutoBash(Bash):
+    """AgentScope Bash with an optional conservative read-only command guard.
+
+    Shell expansions, function definitions and for loops are rejected because
+    their runtime arguments or environment can differ from the parsed tokens.
+    Git external filters require unrestricted mode; other Git helpers are
+    disabled. Use path injection for OS file isolation.
+    """
+
+    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):
+        if kwargs.get("backend") is None:
+            kwargs["backend"] = LocalProcessBackend()
+        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
+        try:
+            from safecmd.bashxtract import (  # pylint: disable=import-outside-toplevel
+                check_types,
+                collect_ops,
+                collect_redirects,
+                parse_bash,
+                visit_stmts,
+            )
+            from safecmd.core import normalize_dest, validate_cmd  # pylint: disable=import-outside-toplevel
+
+            ast = parse_bash(command)
+            check_types(ast)
+            if not _literal_shell(ast, command):
+                return False
+            commands = visit_stmts(ast.get("Stmts", []), command)
+            specs = _read_specs()
+            if any(not validate_cmd(tokens, specs) or not _allowed_arguments(tokens) for tokens in commands):
+                return False
+            return "=" not in collect_ops(ast) and all(
+                normalize_dest(destination) in _SAFE_DESTINATIONS for _, destination in collect_redirects(ast, command)
+            )
+        except Exception:  # pylint: disable=broad-exception-caught
+            # Parser failures and missing shfmt must never allow execution.
+            return False
+
+    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 await asyncio.to_thread(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 await asyncio.to_thread(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
+        if self.is_read_only:
+            # Trace destinations can write files even for metadata-only Git
+            # commands. Clear the entire inherited family on every invocation,
+            # including new Git trace variables, without changing os.environ.
+            trace_variables = " ".join(shlex.quote(name) for name in os.environ if name.startswith("GIT_TRACE"))
+            command = f"unset {trace_variables}\n" + _READ_ONLY_SETUP + 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):
+        """Run the command with the selected permission policy and backend."""
+        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))
+        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 = file_sandbox_for_step(self)
+        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)
+        success = True
+        texts = []
+        async for chunk in tool.call(command=command, timeout=timeout):
+            success = success and chunk.state != ToolResultState.ERROR
+            texts.extend(block.text for block in chunk.content if isinstance(block, TextBlock))
+        self.context.response.success = success
+        self.context.response.answer = "".join(texts)
+        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_mib:
+            type: integer
+            minimum: 0
+            description: "Linux cgroup memory limit in MiB (default 4096; 0 disables it); requires a path policy; macOS warns and ignores nonzero values"
+        required:
+          - command
+      steps:
+        - backend: auto_bash_step

plugins/home/tests/test_auto_bash.py

@@ -0,0 +1,606 @@
+"""Shell job and its optional read-only execution guard."""
+
+import asyncio
+import inspect
+import os
+import subprocess
+import threading
+from importlib.metadata import EntryPoint
+from pathlib import Path
+
+import pytest
+import psutil
+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.local_backend import LocalProcessBackend
+
+from reme.utils.file_sandbox.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.fixture(autouse=True)
+def isolated_git_config(tmp_path, monkeypatch):
+    """Keep Git tests independent of the developer's global and system config."""
+    monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(tmp_path / "global.gitconfig"))
+    monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
+
+
+@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.skipif(os.name != "posix", reason="POSIX shell pipelines")
+@pytest.mark.parametrize("background", [False, True])
+@pytest.mark.parametrize("cancel", [False, True])
+async def test_local_pipeline_timeout_and_cancellation(tmp_path, monkeypatch, background, cancel):
+    """Timeouts and cancellation clean up pipelines even after the shell exits."""
+    (tmp_path / "note.txt").write_text("hello\n", encoding="utf-8")
+    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    processes = []
+    descendants = []
+    original_start = asyncio.create_subprocess_exec
+    original_stop = LocalProcessBackend._stop_process  # pylint: disable=protected-access
+    started = asyncio.Event()
+
+    def group_children(pid):
+        children = []
+        for child in psutil.process_iter():
+            try:
+                if child.pid != pid and os.getpgid(child.pid) == pid:
+                    children.append(child)
+            except (ProcessLookupError, PermissionError):
+                pass
+        return children
+
+    async def start(*args, **kwargs):
+        process = await original_start(*args, **kwargs)
+        processes.append(process)
+        started.set()
+        return process
+
+    async def stop(self, process, communication):
+        descendants.extend(group_children(process.pid))
+        await original_stop(self, process, communication)
+
+    monkeypatch.setattr(asyncio, "create_subprocess_exec", start)
+    monkeypatch.setattr(LocalProcessBackend, "_stop_process", stop)
+    command = "tail -f note.txt | cat" + (" &" if background else "")
+    assert AutoBash._is_read_only(command)  # pylint: disable=protected-access
+    task = asyncio.create_task(step(command=command, timeout=60000 if cancel else 100))
+    try:
+        await asyncio.wait_for(started.wait(), timeout=2)
+        if cancel:
+            # Wait until pipeline children exist before cancelling the invocation.
+            async def pipeline_started():
+                while len(group_children(processes[0].pid)) < 2:
+                    await asyncio.sleep(0.01)
+
+            await asyncio.wait_for(pipeline_started(), timeout=2)
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(task, timeout=0.8)
+        else:
+            response = await asyncio.wait_for(task, timeout=0.8)
+            assert response.success is False
+            assert "timed out after 100ms" in response.answer
+        assert descendants
+        assert processes[0].returncode is not None
+        assert all(not child.is_running() or child.status() == psutil.STATUS_ZOMBIE for child in descendants)
+    finally:
+        if not task.done():
+            task.cancel()
+        for process in processes:
+            try:
+                os.killpg(process.pid, 9)
+            except ProcessLookupError:
+                pass
+        await asyncio.gather(task, return_exceptions=True)
+
+
+@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 -T denied.txt note.txt",
+        "sort -rTdenied.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 hello note.txt",
+        "file -C -m magic",
+        "file -bC -m magic",
+        "file --comp -m magic",
+        r"file -\C -m magic",
+        "file -z note.txt",
+        "file -p note.txt",
+        "echo hello >| denied.txt",
+        ": <> denied.txt",
+        "sort --out=denied.txt note.txt",
+        "git diff --out=denied.txt",
+        "git diff --ext-di",
+        "git show --textcon HEAD",
+        "git log --show-signature",
+        'git log --format="%G?"',
+        "git log --pretty=verify",
+        "git log --pre=verify",
+        "git show --format=verify",
+        "git show --remerge-diff",
+        "for PATH in .; do git status; done",
+        "git() { echo overridden; }; git status",
+        "file -* magic",
+    ],
+)
+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
+@pytest.mark.parametrize("limit", [0, 1, -1])
+async def test_step_rejects_memory_limit_without_path_policy(tmp_path, limit):
+    """An explicitly supplied limit cannot be ignored by host execution."""
+    step = AutoBashStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    response = await step(command="echo should-not-run", is_read_only=False, memory_limit_mib=limit)
+    assert response.success is False
+    assert response.answer == "memory_limit_mib requires read_only_paths or writable_paths to enable the sandbox"
+
+
+@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_read_only_parsing_does_not_block_event_loop(tmp_path, monkeypatch):
+    """A slow synchronous parser runs in a worker thread during permission checks."""
+    started = threading.Event()
+    release = threading.Event()
+
+    def slow_parse(_command):
+        started.set()
+        release.wait(timeout=2)
+        return True
+
+    tool = AutoBash(cwd=tmp_path)
+    monkeypatch.setattr(tool, "_is_read_only", slow_parse)
+    task = asyncio.create_task(tool.check_read_only({"command": "echo hello"}))
+    try:
+        assert await asyncio.wait_for(asyncio.to_thread(started.wait), timeout=1)
+        assert not task.done()
+    finally:
+        release.set()
+    assert await asyncio.wait_for(task, timeout=1)
+
+
+@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()
+
+
+@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 declares the job, Step, and tool defaults."""
+    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 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
+    assert job["parameters"]["properties"]["read_only_paths"]["type"] == "array"
+    assert job["parameters"]["properties"]["writable_paths"]["type"] == "array"
+    assert job["parameters"]["properties"]["memory_limit_mib"]["minimum"] == 0
+
+
+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",
+        "find . -name note.txt",
+        "diff note.txt note.txt",
+        "cmp note.txt note.txt",
+        "git status --short",
+        "git diff --stat",
+        "git rev-parse HEAD",
+        "git ls-files",
+        "echo hello > /dev/null",
+        "cat < note.txt",
+        "echo hello >| /dev/null",
+        "echo hello 2>&1",
+        "find . -name '*.txt'",
+        "(cat note.txt) | head -n 1",
+    ],
+)
+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_sort_does_not_create_temporary_files(tmp_path):
+    """Read-only sorting works in memory and fails closed when a spill is needed."""
+    (tmp_path / "small.txt").write_text("b\na\n", encoding="utf-8")
+    tool = AutoBash(cwd=tmp_path)
+
+    small = [chunk async for chunk in tool.call("sort small.txt")]
+    assert "".join(block.text for chunk in small for block in chunk.content) == "a\nb\n"
+
+    with (tmp_path / "large.txt").open("w", encoding="utf-8") as output:
+        for number in range(100000, 0, -1):
+            output.write(f"{number:012d} abcdefghijklmnopqrstuvwxyz\n")
+    large = [chunk async for chunk in tool.call("sort -S 1K large.txt > /dev/null")]
+    assert "sort:" in "".join(block.text for chunk in large for block in chunk.content)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("trace_variable", ["GIT_TRACE_REFS", "GIT_TRACE", "GIT_TRACE2_EVENT"])
+async def test_read_only_git_cannot_write_inherited_traces(tmp_path, monkeypatch, trace_variable):
+    """Allowed metadata queries must not write Git traces from the parent environment."""
+
+    def git(*args):
+        return subprocess.run(["git", *args], cwd=tmp_path, check=True, capture_output=True)
+
+    git("init", "-q")
+    git(
+        "-c",
+        "user.name=Test",
+        "-c",
+        "user.email=test@example.org",
+        "-c",
+        "commit.gpgsign=false",
+        "commit",
+        "--allow-empty",
+        "-qm",
+        "test",
+    )
+    expected = git("rev-parse", "HEAD").stdout.decode().strip()
+    marker = tmp_path / "trace output.txt"
+    tool = AutoBash(cwd=tmp_path)
+    # Set after construction to check that each invocation uses the current environment.
+    monkeypatch.setenv(trace_variable, str(marker))
+
+    chunks = [chunk async for chunk in tool.call("git rev-parse HEAD")]
+
+    assert chunks
+    assert all(chunk.state != ToolResultState.ERROR for chunk in chunks)
+    assert "".join(block.text for chunk in chunks for block in chunk.content).strip() == expected
+    assert not marker.exists()
+
+    # Control: the same environment still enables tracing for unrestricted execution.
+    unrestricted = AutoBash(cwd=tmp_path, is_read_only=False)
+    chunks = [chunk async for chunk in unrestricted.call("git rev-parse HEAD")]
+    assert all(chunk.state != ToolResultState.ERROR for chunk in chunks)
+    assert marker.stat().st_size > 0
+
+
+@pytest.mark.asyncio
+async def test_read_only_git_clears_future_trace_variables(tmp_path, monkeypatch):
+    """Trace isolation uses a prefix rule instead of an enumeration of known names."""
+    helper = tmp_path / "git"
+    helper.write_text('#!/bin/sh\nprintf "%s" "${GIT_TRACE_FUTURE-unset}"\n', encoding="utf-8")
+    helper.chmod(0o700)
+    monkeypatch.setenv("PATH", str(tmp_path) + os.pathsep + os.environ["PATH"])
+    monkeypatch.setenv("GIT_TRACE_FUTURE", "inherited")
+    tool = AutoBash(cwd=tmp_path)
+
+    chunks = [chunk async for chunk in tool.call("git rev-parse HEAD")]
+
+    assert all(chunk.state != ToolResultState.ERROR for chunk in chunks)
+    assert "".join(block.text for chunk in chunks for block in chunk.content) == "unset"
+    assert os.environ["GIT_TRACE_FUTURE"] == "inherited"
+
+
+@pytest.mark.asyncio
+async def test_git_pretty_config_cannot_launch_gpg(tmp_path):
+    """Git format aliases and defaults cannot smuggle signature verification."""
+    repo = tmp_path / "repo"
+    repo.mkdir()
+
+    def git(*args, input_data=None):
+        return subprocess.run(["git", *args], cwd=repo, input=input_data, check=True, capture_output=True)
+
+    git("init", "-q")
+    git("config", "user.name", "Test")
+    git("config", "user.email", "test@example.org")
+    (repo / "note.txt").write_text("hello\n", encoding="utf-8")
+    git("add", "note.txt")
+    git("commit", "-qm", "test")
+    commit = git("cat-file", "commit", "HEAD").stdout
+    signed = commit.replace(b"\n\n", b"\ngpgsig -----BEGIN PGP SIGNATURE-----\n \n -----END PGP SIGNATURE-----\n\n", 1)
+    signed_hash = git("hash-object", "-t", "commit", "-w", "--stdin", input_data=signed).stdout.strip().decode()
+    git("update-ref", "HEAD", signed_hash)
+
+    marker = tmp_path / "gpg-ran"
+    helper = tmp_path / "fake-gpg"
+    helper.write_text(f'#!/bin/sh\nprintf called > "{marker}"\nexit 1\n', encoding="utf-8")
+    helper.chmod(0o700)
+    git("config", "pretty.verify", "format:%G?")
+    git("config", "pretty.medium", "format:%G?")
+    git("config", "pretty.oneline", "format:%G?")
+    git("config", "format.pretty", "format:%G?")
+    git("config", "gpg.program", str(helper))
+    tool = AutoBash(cwd=repo)
+
+    for command in ("git log", "git show HEAD", "git log --oneline"):
+        chunks = [chunk async for chunk in tool.call(command)]
+        assert chunks
+        assert all(chunk.state != ToolResultState.ERROR for chunk in chunks)
+        assert not marker.exists()
+
+    for command in ("git log --pretty=verify", "git show --format=verify"):
+        chunks = [chunk async for chunk in tool.call(command)]
+        assert chunks[-1].state == ToolResultState.ERROR
+        assert not marker.exists()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("helper", ["fsmonitor", "diff", "textconv", "clean"])
+async def test_git_configured_helpers_cannot_write(tmp_path, helper):
+    """Repository configuration cannot turn an allowed read into helper execution."""
+
+    def git(*args):
+        return subprocess.run(["git", *args], cwd=tmp_path, check=True, capture_output=True)
+
+    git("init", "-q")
+    (tmp_path / "note.txt").write_text("before\n", encoding="utf-8")
+    (tmp_path / ".gitattributes").write_text("note.txt diff=test filter=test\n", encoding="utf-8")
+    git("add", "note.txt", ".gitattributes")
+    hook = tmp_path / "helper"
+    hook.write_text('#!/bin/sh\ntouch "' + str(tmp_path / "denied.txt") + '"\ncat\n', encoding="utf-8")
+    hook.chmod(0o700)
+    key = {
+        "fsmonitor": "core.fsmonitor",
+        "diff": "diff.external",
+        "textconv": "diff.test.textconv",
+        "clean": "filter.test.clean",
+    }[helper]
+    git("config", key, str(hook))
+    (tmp_path / "note.txt").write_text("after\n", encoding="utf-8")
+    command = "git status --short" if helper == "fsmonitor" else "git diff"
+    tool = AutoBash(cwd=tmp_path)
+
+    chunks = [chunk async for chunk in tool.call(command)]
+
+    assert not (tmp_path / "denied.txt").exists()
+    if helper == "clean":
+        assert chunks[-1].state == ToolResultState.ERROR
+        assert "external filters" in chunks[-1].content[0].text
+    else:
+        assert all(chunk.state != ToolResultState.ERROR for chunk in chunks)
+
+
+@pytest.mark.asyncio
+async def test_global_git_lfs_config_allows_metadata_queries(tmp_path, monkeypatch):
+    """A global filter alone must not block Git queries that never use it."""
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    subprocess.run(["git", "init", "-q", str(repo)], check=True, capture_output=True)
+    (repo / "note.txt").write_text("hello\n", encoding="utf-8")
+    subprocess.run(["git", "add", "note.txt"], cwd=repo, check=True, capture_output=True)
+
+    global_config = tmp_path / "global.gitconfig"
+    subprocess.run(
+        ["git", "config", "--file", str(global_config), "filter.lfs.process", "/bin/false"],
+        check=True,
+        capture_output=True,
+    )
+    monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(global_config))
+    tool = AutoBash(cwd=repo)
+
+    for command, expected in (("git rev-parse --git-dir", ".git"), ("git ls-files", "note.txt")):
+        chunks = [chunk async for chunk in tool.call(command)]
+        assert chunks
+        assert all(chunk.state != ToolResultState.ERROR for chunk in chunks)
+        assert expected in "".join(block.text for chunk in chunks for block in chunk.content)
+
+    chunks = [chunk async for chunk in tool.call("git diff")]
+    assert chunks[-1].state == ToolResultState.ERROR
+    assert "external filters" in chunks[-1].content[0].text
+
+
+@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"]})
+    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/config/benchmark.yaml

@@ -142,6 +142,10 @@ jobs:
         python_timeout:
           type: number
           description: "Execution timeout in seconds; defaults to 60."
+        memory_limit_mib:
+          type: integer
+          minimum: 0
+          description: "Linux memory limit in MiB (default 4096; 0 disables it); requires a path policy; macOS ignores nonzero values"
       required:
         - code
     steps:

reme/steps/common/file_sandbox.py

@@ -0,0 +1,23 @@
+"""Resolve an invocation's optional file sandbox from injected job parameters."""
+
+from ...utils.file_sandbox import create_file_sandbox_backend
+from ...utils.file_sandbox.file_sandbox_backend import FileSandboxBackend
+from ..base_step import BaseStep
+
+
+def file_sandbox_for_step(step: BaseStep) -> FileSandboxBackend | None:
+    """Keep path and memory policy selection identical across executor steps."""
+    assert step.context is not None
+    read_only_paths = step.context.get("read_only_paths", step.kwargs.get("read_only_paths"))
+    writable_paths = step.context.get("writable_paths", step.kwargs.get("writable_paths"))
+    memory_limit_mib = step.context.get("memory_limit_mib", step.kwargs.get("memory_limit_mib"))
+    if read_only_paths is None and writable_paths is None:
+        if memory_limit_mib is not None:
+            raise ValueError("memory_limit_mib requires read_only_paths or writable_paths to enable the sandbox")
+        return None
+    return create_file_sandbox_backend(
+        cwd=step.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_mib=memory_limit_mib,
+    )

reme/steps/common/python_execute.py

@@ -1,27 +1,19 @@
 """Execute Python code and return printed stdout."""
 
-import asyncio
 import sys
-from dataclasses import dataclass
 from typing import Any
 
+from ...utils.local_backend import LocalProcessBackend, ProcessTimeout
 from ..base_step import BaseStep
 from ...components import R
+from .file_sandbox import file_sandbox_for_step
 
 DEFAULT_TIMEOUT = 60.0
 
 
-@dataclass(frozen=True)
-class _PythonResult:
-    stdout: str
-    stderr: str
-    returncode: int | None
-    timed_out: bool = False
-
-
 @R.register("python_execute_step")
 class PythonExecuteStep(BaseStep):
-    """Run Python code in a subprocess and return stdout as the response answer."""
+    """Run Python code and return stdout, limiting each output stream to 1 MiB."""
 
     async def execute(self):
         assert self.context is not None
@@ -39,58 +31,39 @@ class PythonExecuteStep(BaseStep):
             self.context.response.answer = timeout_error
             return self.context.response
 
-        result = await self._run_python(code, timeout)
-        if result.timed_out:
+        timed_out = False
+        try:
+            backend = file_sandbox_for_step(self)
+            result = await (backend or LocalProcessBackend()).exec_shell(
+                [sys.executable, "-c", code],
+                cwd=str(self.workspace_path),
+                timeout=timeout,
+                **({"raise_errors": True} if backend is None else {}),
+            )
+        except ProcessTimeout as exc:
+            result, timed_out = exc.result, True
+        except (TypeError, ValueError, OSError, RuntimeError) as exc:
+            self.context.response.success = False
+            self.context.response.answer = str(exc)
+            return self.context.response
+        timed_out = timed_out or (result.exit_code == -1 and result.stderr == b"timed out")
+        stdout = result.stdout.decode(errors="replace")
+        stderr = result.stderr.decode(errors="replace")
+        if timed_out:
             self.context.response.success = False
             self.context.response.answer = f"Python execution timed out after {timeout:g}s"
-            self.context.response.metadata.update(
-                {
-                    "returncode": result.returncode,
-                    "stderr": result.stderr,
-                    "python_timeout": timeout,
-                },
-            )
-            return self.context.response
-
-        stdout = result.stdout or ""
-        stderr = result.stderr or ""
-        self.context.response.success = result.returncode == 0
-        self.context.response.answer = stdout if stdout or result.returncode == 0 else stderr
+        else:
+            self.context.response.success = result.exit_code == 0
+            self.context.response.answer = stdout if stdout or result.exit_code == 0 else stderr
         self.context.response.metadata.update(
             {
-                "returncode": result.returncode,
+                "returncode": result.exit_code,
                 "stderr": stderr,
                 "python_timeout": timeout,
             },
         )
         return self.context.response
 
-    async def _run_python(self, code: str, timeout: float) -> _PythonResult:
-        process = await asyncio.create_subprocess_exec(
-            sys.executable,
-            "-c",
-            code,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-            cwd=self.workspace_path,
-        )
-        try:
-            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
-            return _PythonResult(
-                stdout=stdout.decode(),
-                stderr=stderr.decode(),
-                returncode=process.returncode,
-            )
-        except TimeoutError:
-            process.kill()
-            stdout, stderr = await process.communicate()
-            return _PythonResult(
-                stdout=stdout.decode(),
-                stderr=stderr.decode(),
-                returncode=process.returncode,
-                timed_out=True,
-            )
-
     @staticmethod
     def _parse_timeout(raw: Any) -> tuple[float, str]:
         try:

reme/utils/file_sandbox/__init__.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_mib: int | None = None, **kwargs) -> FileSandboxBackend:
+    """Construct the platform backend; never silently execute without isolation."""
+    if sys.platform.startswith("linux"):
+        limit = 4096 if memory_limit_mib is None else memory_limit_mib
+        return PathBubbleBackend(memory_limit_mib=limit, **kwargs)
+    if sys.platform == "darwin":
+        limit = 0 if memory_limit_mib is None else memory_limit_mib
+        return PathSeatbeltBackend(memory_limit_mib=limit, **kwargs)
+    raise RuntimeError(f"No file sandbox backend for {sys.platform}")

reme/utils/file_sandbox/file_sandbox_backend.py

@@ -0,0 +1,107 @@
+"""Common contract and path validation for subprocess file sandboxes."""
+
+import asyncio
+import os
+import sys
+from abc import abstractmethod
+
+from agentscope.tool import ExecResult
+
+from ..local_backend import LocalProcessBackend
+
+
+def contains(root: str, path: str) -> bool:
+    """Whether a path is inside a configured root."""
+    return os.path.commonpath((root, path)) == root
+
+
+class FileSandboxBackend(LocalProcessBackend):
+    """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)
+        # A virtual environment can point outside sys.prefix to its base
+        # interpreter and standard library (notably with Homebrew Python).
+        for runtime in (sys.prefix, sys.base_prefix, sys.executable):
+            path = os.path.realpath(runtime)
+            if path in ("/", "/usr", "/usr/local") or path.startswith("/usr/"):
+                continue
+            if not any(contains(root, path) for root in self._read_only_paths + self._writable_paths):
+                self._read_only_paths.append(self._validate_path(path))
+        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."""
+
+    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)
+        return await self._communicate(process, timeout=timeout)
+
+    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,
+        )
+        result = await self._communicate(process, data=data)
+        if not result.ok():
+            raise OSError(result.stderr.decode(errors="replace"))

reme/utils/file_sandbox/path_bubble_backend.py

@@ -0,0 +1,124 @@
+"""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
+
+# Run after exec of a fresh interpreter: Python preexec_fn can deadlock when
+# the parent has threads. The requested command starts only after enrollment.
+_CGROUP_EXEC = """
+import os
+import sys
+with open(sys.argv[1], 'w', encoding='ascii') as group:
+    group.write(str(os.getpid()))
+os.execvp(sys.argv[2], sys.argv[2:])
+"""
+
+
+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_mib: int = 4096,
+    ) -> None:
+        if isinstance(memory_limit_mib, bool) or not isinstance(memory_limit_mib, int) or memory_limit_mib < 0:
+            raise ValueError("memory_limit_mib must be a non-negative integer")
+        self._memory_limit_mib = memory_limit_mib
+        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_mib else None
+        try:
+            if group is not None:
+                argv = [sys.executable, "-I", "-S", "-c", _CGROUP_EXEC, str(group / "cgroup.procs"), *argv]
+            process = await self._spawn_process(
+                argv,
+                stdin=stdin,
+            )
+        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_mib * 1024**2), 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

reme/utils/file_sandbox/path_seatbelt_backend.py

@@ -0,0 +1,329 @@
+"""macOS Seatbelt file sandbox with host-identical paths."""
+
+import asyncio
+import errno
+import json
+import os
+import plistlib
+import shutil
+import sys
+import tempfile
+import uuid
+import warnings
+from pathlib import Path
+from types import SimpleNamespace
+
+import psutil
+from agentscope.tool import ExecResult
+
+from .file_sandbox_backend import FileSandboxBackend, contains
+from ..local_backend import (
+    LocalProcessBackend,
+    OutputLimitExceeded,
+    capture_output,
+    finish_cleanup,
+    output_limit_result,
+)
+
+
+class LaunchctlError(RuntimeError):
+    """Preserve launchctl's error code so missing services can be handled narrowly."""
+
+    def __init__(self, message: str, exit_code: int):
+        super().__init__(message)
+        self.exit_code = exit_code
+
+
+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_mib: int = 0,
+    ) -> None:
+        if isinstance(memory_limit_mib, bool) or not isinstance(memory_limit_mib, int) or memory_limit_mib < 0:
+            raise ValueError("memory_limit_mib must be a non-negative integer")
+        if memory_limit_mib:
+            warnings.warn(
+                "PathSeatbeltBackend does not support memory_limit_mib; 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)",
+            # dyld opens the root directory; /bin/sh reads the system shell selector.
+            '(allow file-read* (literal "/") (literal "/private/var/select/sh"))',
+        ]
+        ancestors = set()
+        for path in read_paths:
+            parent = os.path.dirname(path)
+            while parent != "/":
+                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(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.
+        lines.append('(allow file-write* (literal "/dev/null"))')
+        for path in self._writable_paths + [tempdir]:
+            # Seatbelt combines allow rules with OR, unlike ordered bwrap mounts.
+            # Exclude read-only descendants from each writable parent's grant.
+            exclusions = [
+                f"(require-not {self._path_rule(read_only)})"
+                for read_only in self._read_only_paths
+                if contains(path, read_only)
+            ]
+            rule = self._path_rule(path)
+            if exclusions:
+                rule = f"(require-all {rule} {' '.join(exclusions)})"
+            lines.append(f"(allow file-write* {rule})")
+        # A writable ancestor can otherwise be renamed, moving a protected
+        # subtree out of the path-based exclusion above. Pin the protected
+        # path and every ancestor that falls within a writable root.
+        pinned = set()
+        for read_only in self._read_only_paths:
+            current = read_only
+            while current != "/":
+                if any(contains(writable, current) for writable in self._writable_paths):
+                    pinned.add(current)
+                current = os.path.dirname(current)
+        for path in sorted(pinned):
+            rule = f"(literal {json.dumps(path)})"
+            lines.append(f"(deny file-write-unlink {rule})")
+            lines.append(f"(deny file-write-create {rule})")
+        return "\n".join(lines)
+
+    async def _start_process(self, command: list[str], cwd: str, *, stdin=asyncio.subprocess.DEVNULL):
+        """Start Seatbelt directly when the GUI launchd domain is unavailable."""
+        tempdir = tempfile.mkdtemp(prefix="reme-seatbelt-", dir=os.path.realpath(tempfile.gettempdir()))
+        try:
+            process = await self._spawn_process(
+                ["/usr/bin/sandbox-exec", "-p", self._profile(tempdir), *command],
+                cwd=cwd,
+                env={**os.environ, "HOME": self._cwd, "TMPDIR": tempdir},
+                stdin=stdin,
+            )
+        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)
+
+    @staticmethod
+    async def _has_gui_launchd(domain: str) -> bool:
+        result = await LocalProcessBackend().exec_shell(["/bin/launchctl", "print", domain], timeout=5)
+        return result.ok()
+
+    @staticmethod
+    async def _launchctl(*arguments: str) -> str:
+        result = await LocalProcessBackend().exec_shell(["/bin/launchctl", *arguments], timeout=5)
+        if not result.ok():
+            raise LaunchctlError(
+                f"launchctl {arguments[0]} failed: {result.stderr.decode(errors='replace').strip()}",
+                result.exit_code,
+            )
+        return result.stdout.decode(errors="replace").strip()
+
+    async def _bootstrap(self, domain: str, plist: Path, service: str) -> None:
+        """Resolve registration before rolling it back, even when its reply is interrupted."""
+        registration = asyncio.create_task(self._launchctl("bootstrap", domain, str(plist)))
+        try:
+            await asyncio.shield(registration)
+        except BaseException:
+
+            async def rollback():
+                # Killing the client cannot undo a registration accepted by launchd.
+                # Wait for the bounded launchctl operation before issuing bootout.
+                await asyncio.gather(registration, return_exceptions=True)
+                try:
+                    await self._launchctl("bootout", service)
+                except LaunchctlError as exc:
+                    if exc.exit_code != errno.ESRCH:
+                        raise
+
+            await finish_cleanup(rollback())
+            raise
+
+    @staticmethod
+    async def _wait_status(status: Path, pid: int) -> int:
+        while True:
+            if status.exists():
+                return int(status.read_text(encoding="ascii"))
+            if not psutil.pid_exists(pid):
+                raise RuntimeError("launchd sandbox job exited without reporting a status")
+            await asyncio.sleep(0.02)
+
+    @staticmethod
+    async def _read_fifo(fd: int, buffer: bytearray, name: str, exceeded: asyncio.Future, finished: asyncio.Event):
+        """Drain launchd output through a bounded pipe instead of an unbounded disk file."""
+        while True:
+            try:
+                chunk = os.read(fd, 65536)
+            except BlockingIOError:
+                if finished.is_set():
+                    return
+                await asyncio.sleep(0.005)
+                continue
+            capture_output(buffer, chunk, name, exceeded)
+            await asyncio.sleep(0)
+
+    async def _run_job(self, command: list[str], cwd: str, *, data: bytes | None, timeout: float | None) -> ExecResult:
+        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)
+        domain = f"gui/{os.getuid()}"
+        if not await self._has_gui_launchd(domain):
+            warnings.warn(
+                "No macOS GUI launchd domain; process-group cleanup is unavailable, running directly under Seatbelt",
+                RuntimeWarning,
+                stacklevel=2,
+            )
+            process = await self._start_process(
+                command,
+                cwd,
+                stdin=asyncio.subprocess.PIPE if data is not None else asyncio.subprocess.DEVNULL,
+            )
+            return await self._communicate(process, data=data, timeout=timeout)
+        control = Path(tempfile.mkdtemp(prefix="reme-seatbelt-", dir=os.path.realpath(tempfile.gettempdir())))
+        sandbox_tmp = control / "sandbox-tmp"
+        sandbox_tmp.mkdir()
+        stdout_path = control / "stdout"
+        stderr_path = control / "stderr"
+        status_path = control / "status"
+        stdin_path = control / "stdin"
+        label = f"ai.agentscope.reme.sandbox.{uuid.uuid4().hex}"
+        service = f"{domain}/{label}"
+        registered = False
+        pid = None
+        stdout, stderr = bytearray(), bytearray()
+        exceeded = asyncio.get_running_loop().create_future()
+        finished = asyncio.Event()
+        output_fds = []
+        readers = []
+
+        async def cleanup(completed):
+            try:
+                if registered:
+                    if not completed and pid is not None:
+                        self._kill_process(SimpleNamespace(pid=pid))
+                    await self._launchctl("bootout", service)
+            finally:
+                finished.set()
+                try:
+                    await asyncio.wait_for(asyncio.gather(*readers), timeout=0.25)
+                except TimeoutError:
+                    pass
+
+        try:
+            for path, buffer, name in ((stdout_path, stdout, "stdout"), (stderr_path, stderr, "stderr")):
+                os.mkfifo(path, 0o600)
+                # Holding both ends open prevents a transient EOF before launchd
+                # opens its writer. Kernel pipe capacity bounds pending output.
+                fd = os.open(path, os.O_RDWR | os.O_NONBLOCK)
+                output_fds.append(fd)
+                readers.append(asyncio.create_task(self._read_fifo(fd, buffer, name, exceeded, finished)))
+            if data is not None:
+                stdin_path.write_bytes(data)
+            # The shell reports the command's status in a control file. Its
+            # child inherits the launchd job's process group, so launchd can
+            # reap same-group background processes when the shell exits.
+            script = '"$@"; result=$?; printf "%s\\n" "$result" > "$0.tmp"; mv "$0.tmp" "$0"'
+            job = {
+                "Label": label,
+                "ProgramArguments": [
+                    "/bin/sh",
+                    "-c",
+                    script,
+                    str(status_path),
+                    "/usr/bin/sandbox-exec",
+                    "-p",
+                    self._profile(str(sandbox_tmp)),
+                    *command,
+                ],
+                "WorkingDirectory": cwd,
+                "EnvironmentVariables": {**os.environ, "HOME": self._cwd, "TMPDIR": str(sandbox_tmp)},
+                "StandardInPath": str(stdin_path) if data is not None else "/dev/null",
+                "StandardOutPath": str(stdout_path),
+                "StandardErrorPath": str(stderr_path),
+                "RunAtLoad": False,
+                "KeepAlive": False,
+                "AbandonProcessGroup": False,
+            }
+            plist_path = control / "job.plist"
+            with os.fdopen(os.open(plist_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb") as stream:
+                plistlib.dump(job, stream)
+            await self._bootstrap(domain, plist_path, service)
+            registered = True
+
+            async def run():
+                nonlocal pid
+                pid = int(await self._launchctl("kickstart", "-p", service))
+                return await self._watch_output(self._wait_status(status_path, pid), exceeded)
+
+            try:
+                exit_code = await self._execute(run(), timeout=timeout, cleanup=cleanup)
+            except TimeoutError:
+                return ExecResult(exit_code=-1, stdout=b"", stderr=b"timed out")
+            except OutputLimitExceeded as exc:
+                return output_limit_result(bytes(stdout), bytes(stderr), str(exc))
+            if exceeded.done():
+                return output_limit_result(bytes(stdout), bytes(stderr), exceeded.result())
+            return ExecResult(
+                exit_code=exit_code,
+                stdout=bytes(stdout),
+                stderr=bytes(stderr),
+            )
+        finally:
+            for reader in readers:
+                reader.cancel()
+            await asyncio.gather(*readers, return_exceptions=True)
+            for fd in output_fds:
+                os.close(fd)
+            shutil.rmtree(control)
+
+    async def exec_shell(
+        self,
+        command: list[str],
+        *,
+        cwd: str | None = None,
+        timeout: float | None = None,
+    ) -> ExecResult:
+        return await self._run_job(command, cwd or self._cwd, data=None, timeout=timeout)
+
+    async def write_file(self, path: str, data: bytes) -> None:
+        if not isinstance(data, bytes):
+            raise TypeError("data must be bytes")
+        script = 'mkdir -p -- "$(dirname -- "$1")" && cat > "$1"'
+        result = await self._run_job(["/bin/sh", "-c", script, "sh", path], self._cwd, data=data, timeout=None)
+        if not result.ok():
+            raise OSError(result.stderr.decode(errors="replace"))

reme/utils/local_backend.py

@@ -0,0 +1,247 @@
+"""Shared subprocess execution with bounded timeout and cancellation cleanup."""
+
+import asyncio
+import os
+import signal
+import sys
+from collections.abc import Awaitable, Callable
+
+import psutil
+from agentscope.tool import ExecResult, LocalBackend
+
+_CLEANUP_TIMEOUT = 0.25
+OUTPUT_LIMIT_BYTES = 1024 * 1024
+
+
+async def finish_cleanup(operation: Awaitable):
+    """Finish rollback after cancellation, deferring additional cancellation requests."""
+    task = asyncio.ensure_future(operation)
+    while True:
+        try:
+            return await asyncio.shield(task)
+        except asyncio.CancelledError:
+            if task.done():
+                return task.result()
+
+
+class OutputLimitExceeded(RuntimeError):
+    """Identify the output stream that exceeded its byte budget."""
+
+
+def capture_output(buffer: bytearray, chunk: bytes, stream: str, exceeded: asyncio.Future) -> None:
+    """Retain at most one stream's budget, signalling overflow without blocking draining."""
+    remaining = OUTPUT_LIMIT_BYTES - len(buffer)
+    buffer.extend(chunk[:remaining])
+    if len(chunk) > remaining and not exceeded.done():
+        exceeded.set_result(stream)
+
+
+def output_limit_result(stdout: bytes, stderr: bytes, stream: str) -> ExecResult:
+    """Return bounded partial output and an explicit failure diagnostic."""
+    message = f"\nOutput limit exceeded: {stream} exceeds {OUTPUT_LIMIT_BYTES} bytes; command terminated.\n".encode()
+    diagnostic = (stderr[: max(0, OUTPUT_LIMIT_BYTES - len(message))] + message)[:OUTPUT_LIMIT_BYTES]
+    return ExecResult(exit_code=1, stdout=stdout[:OUTPUT_LIMIT_BYTES], stderr=diagnostic)
+
+
+class ProcessTimeout(TimeoutError):
+    """Carry the reaped process output for callers with their own timeout protocol."""
+
+    def __init__(self, result: ExecResult):
+        super().__init__("timed out")
+        self.result = result
+
+
+class LocalProcessBackend(LocalBackend):
+    """Keep local file helpers; terminate commands exceeding 1 MiB on either output stream."""
+
+    async def exec_shell(
+        self,
+        command: list[str],
+        *,
+        cwd: str | None = None,
+        timeout: float | None = None,
+        raise_errors: bool = False,
+    ) -> ExecResult:
+        """Run an argv in its own session and collect output within the deadline."""
+        try:
+            process = await self._start_process(command, cwd)
+        except OSError as exc:
+            if raise_errors:
+                raise
+            return ExecResult(exit_code=127, stdout=b"", stderr=str(exc).encode("utf-8"))
+        return await self._communicate(process, timeout=timeout, raise_errors=raise_errors)
+
+    async def _start_process(self, command: list[str], cwd: str | None, *, stdin=None):
+        """Start a host process; sandboxes override platform-specific setup."""
+        return await self._spawn_process(command, cwd=cwd, stdin=stdin)
+
+    async def _spawn_process(self, command: list[str], **kwargs):
+        """Capture output in a separate process group for every execution backend."""
+        spawning = asyncio.create_task(
+            asyncio.create_subprocess_exec(
+                *command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                start_new_session=os.name == "posix",
+                **kwargs,
+            ),
+        )
+        try:
+            return await asyncio.shield(spawning)
+        except asyncio.CancelledError:
+
+            async def rollback():
+                outcome = (await asyncio.gather(spawning, return_exceptions=True))[0]
+                if isinstance(outcome, BaseException):
+                    return
+                exceeded = asyncio.get_running_loop().create_future()
+                communication = asyncio.create_task(
+                    self._collect_output(outcome, None, bytearray(), bytearray(), exceeded),
+                )
+                await self._stop_process(outcome, communication)
+
+            await finish_cleanup(rollback())
+            raise
+
+    def _release_process(self, process: asyncio.subprocess.Process) -> None:
+        """Release any platform resources owned by a completed execution."""
+
+    @staticmethod
+    async def _watch_output(operation: Awaitable, exceeded: asyncio.Future):
+        """Stop waiting as soon as either captured stream exceeds its budget."""
+        task = asyncio.ensure_future(operation)
+        try:
+            await asyncio.wait((task, exceeded), return_when=asyncio.FIRST_COMPLETED)
+            if exceeded.done():
+                raise OutputLimitExceeded(exceeded.result())
+            return await task
+        finally:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+
+    @staticmethod
+    async def _collect_output(process, data, stdout, stderr, exceeded):
+        """Drain both pipes concurrently while retaining bounded byte buffers."""
+
+        async def read(reader, buffer, name):
+            while chunk := await reader.read(65536):
+                capture_output(buffer, chunk, name, exceeded)
+
+        async def write():
+            if process.stdin is not None:
+                try:
+                    if data is not None:
+                        process.stdin.write(data)
+                        await process.stdin.drain()
+                except (BrokenPipeError, ConnectionResetError):
+                    pass
+                finally:
+                    process.stdin.close()
+
+        tasks = [
+            asyncio.create_task(read(process.stdout, stdout, "stdout")),
+            asyncio.create_task(read(process.stderr, stderr, "stderr")),
+            asyncio.create_task(write()),
+        ]
+        try:
+            await asyncio.gather(*tasks)
+            await process.wait()
+            return bytes(stdout), bytes(stderr)
+        finally:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+    @staticmethod
+    async def _execute(operation: Awaitable, *, timeout: float | None, cleanup: Callable[[bool], Awaitable[None]]):
+        """Apply one deadline and always clean up success, failure, or cancellation."""
+        task = asyncio.ensure_future(operation)
+        completed = False
+        try:
+            result = await asyncio.wait_for(asyncio.shield(task), timeout=timeout)
+            completed = True
+            return result
+        finally:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+            await cleanup(completed)
+
+    async def _communicate(
+        self,
+        process: asyncio.subprocess.Process,
+        *,
+        data: bytes | None = None,
+        timeout: float | None = None,
+        raise_errors: bool = False,
+    ) -> ExecResult:
+        """Collect pipes and release the process, preserving each caller's timeout protocol."""
+        stdout_buffer, stderr_buffer = bytearray(), bytearray()
+        exceeded = asyncio.get_running_loop().create_future()
+        communication = asyncio.create_task(
+            self._collect_output(process, data, stdout_buffer, stderr_buffer, exceeded),
+        )
+
+        async def cleanup(completed):
+            try:
+                if not completed:
+                    await self._stop_process(process, communication)
+            finally:
+                self._release_process(process)
+
+        try:
+            stdout, stderr = await self._execute(
+                self._watch_output(asyncio.shield(communication), exceeded),
+                timeout=timeout,
+                cleanup=cleanup,
+            )
+        except OutputLimitExceeded as exc:
+            return output_limit_result(bytes(stdout_buffer), bytes(stderr_buffer), str(exc))
+        except TimeoutError:
+            if raise_errors:
+                stdout, stderr = communication.result() if not communication.cancelled() else (b"", b"")
+                raise ProcessTimeout(ExecResult(exit_code=process.returncode, stdout=stdout, stderr=stderr)) from None
+            return ExecResult(exit_code=-1, stdout=b"", stderr=b"timed out")
+        return ExecResult(exit_code=process.returncode, stdout=stdout, stderr=stderr)
+
+    @staticmethod
+    def _kill_process(process: asyncio.subprocess.Process) -> None:
+        """Kill the group and known detached children on macOS and Windows."""
+        if os.name != "posix" or sys.platform == "darwin":
+            if os.name == "posix":
+                try:
+                    os.killpg(process.pid, signal.SIGSTOP)
+                except ProcessLookupError:
+                    pass
+            try:
+                parent = psutil.Process(process.pid)
+                for child in reversed(parent.children(recursive=True)):
+                    try:
+                        child.kill()
+                    except (psutil.Error, OSError):
+                        pass
+                if os.name != "posix":
+                    parent.kill()
+            except (psutil.Error, OSError):
+                pass
+        if os.name == "posix":
+            # The launcher may have exited while a pipeline still holds stdout.
+            try:
+                os.killpg(process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+
+    async def _stop_process(self, process: asyncio.subprocess.Process, communication: asyncio.Task) -> None:
+        """Kill descendants and bound draining even if a child escaped the group."""
+        self._kill_process(process)
+        try:
+            await asyncio.wait_for(asyncio.shield(communication), timeout=_CLEANUP_TIMEOUT)
+        except TimeoutError:
+            communication.cancel()
+            await asyncio.gather(communication, return_exceptions=True)
+            # asyncio has no public API to close a Process's captured pipes.
+            # An escaped descendant must not keep these transports alive.
+            process._transport.close()  # pylint: disable=protected-access
+            try:
+                await asyncio.wait_for(process.wait(), timeout=_CLEANUP_TIMEOUT)
+            except TimeoutError:
+                pass

tests/integration/test_seatbelt_backend.py

@@ -0,0 +1,231 @@
+"""Exercise macOS sandbox startup and file isolation using real subprocesses."""
+
+import asyncio
+import sys
+import venv
+
+import psutil
+
+import pytest
+
+from reme.utils.file_sandbox.path_seatbelt_backend import PathSeatbeltBackend
+
+pytestmark = pytest.mark.skipif(sys.platform != "darwin", reason="Seatbelt requires macOS")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "command",
+    [
+        ["/bin/sh", "-c", "pwd"],
+        [sys.executable, "-c", "import os; print(os.getcwd())"],
+    ],
+)
+async def test_seatbelt_starts_shell_and_python(tmp_path, command):
+    """System runtime access permits startup and resolution of the working directory."""
+    workspace = tmp_path.resolve()
+    backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[])
+
+    result = await backend.exec_shell(command, timeout=10)
+
+    assert result.exit_code == 0, result.stderr.decode(errors="replace")
+    assert result.stdout.decode().strip() == str(workspace)
+    assert result.stderr == b""
+
+
+@pytest.mark.asyncio
+async def test_seatbelt_starts_virtual_environment_python(tmp_path):
+    """A symlinked virtual environment interpreter can load its base runtime."""
+    workspace = tmp_path.resolve()
+    environment = workspace / "venv"
+    venv.EnvBuilder(with_pip=False, symlinks=True).create(environment)
+    backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[])
+
+    result = await backend.exec_shell(
+        [str(environment / "bin" / "python"), "-c", "import os; print(os.getcwd())"],
+        timeout=10,
+    )
+
+    assert result.exit_code == 0, result.stderr.decode(errors="replace")
+    assert result.stdout.decode().strip() == str(workspace)
+
+
+@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."""
+    backend = PathSeatbeltBackend(cwd=tmp_path.resolve(), read_only_paths=[], writable_paths=[])
+
+    result = await backend.exec_shell(["/bin/sh", "-c", "echo hi > /dev/null"], timeout=5)
+
+    assert result.exit_code == 0, result.stderr.decode(errors="replace")
+    assert result.stdout == b""
+
+
+@pytest.mark.asyncio
+async def test_seatbelt_write_file_with_launchd_stdin(tmp_path):
+    """The launchd job delivers input bytes through its configured stdin file."""
+    workspace = tmp_path.resolve()
+    backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)])
+    target = workspace / "nested" / "note.txt"
+
+    await backend.write_file(str(target), b"hello\n")
+
+    assert target.read_bytes() == b"hello\n"
+
+
+@pytest.mark.asyncio
+async def test_seatbelt_without_gui_launchd_uses_direct_sandbox(tmp_path, monkeypatch):
+    """A missing GUI domain warns while Seatbelt still enforces the path policy."""
+    root = tmp_path.resolve()
+    workspace = root / "workspace"
+    workspace.mkdir()
+    outside = root / "outside.txt"
+    backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)])
+
+    async def missing(_domain):
+        return False
+
+    monkeypatch.setattr(backend, "_has_gui_launchd", missing)
+    code = (
+        "from pathlib import Path;"
+        "Path('inside.txt').write_text('allowed');"
+        f"Path({str(outside)!r}).write_text('denied')"
+    )
+    with pytest.warns(RuntimeWarning, match="No macOS GUI launchd domain"):
+        result = await backend.exec_shell([sys.executable, "-c", code], timeout=5)
+    assert result.exit_code != 0
+    assert (workspace / "inside.txt").read_text() == "allowed"
+    assert not outside.exists()
+
+    with pytest.warns(RuntimeWarning, match="No macOS GUI launchd domain"):
+        await backend.write_file(str(workspace / "input.txt"), b"hello")
+    assert (workspace / "input.txt").read_bytes() == b"hello"
+
+
+@pytest.mark.asyncio
+async def test_seatbelt_runtime_permissions_preserve_file_isolation(tmp_path):
+    """Startup grants do not expose sibling files or writable access to protected children."""
+    root = tmp_path.resolve()
+    workspace = root / "workspace"
+    protected = workspace / "protected"
+    output = protected / "output"
+    output.mkdir(parents=True)
+    (protected / "note.txt").write_text("readable", encoding="utf-8")
+    (root / "outside.txt").write_text("private", encoding="utf-8")
+    backend = PathSeatbeltBackend(
+        cwd=workspace,
+        read_only_paths=[str(protected)],
+        writable_paths=[str(workspace), str(output)],
+    )
+    code = """
+from pathlib import Path
+assert Path('protected/note.txt').read_text() == 'readable'
+Path('allowed.txt').write_text('ok')
+Path('protected/output/allowed.txt').write_text('ok')
+for path, operation in [
+    (Path('protected/denied.txt'), 'write'),
+    (Path('../outside.txt'), 'read'),
+    (Path('../outside.txt'), 'write'),
+]:
+    try:
+        path.read_text() if operation == 'read' else path.write_text('denied')
+    except PermissionError:
+        pass
+    else:
+        raise AssertionError(f'{operation} unexpectedly allowed: {path}')
+"""
+
+    result = await backend.exec_shell([sys.executable, "-c", code], timeout=10)
+
+    assert result.exit_code == 0, result.stderr.decode(errors="replace")
+    assert (workspace / "allowed.txt").read_text() == "ok"
+    assert (output / "allowed.txt").read_text() == "ok"
+    assert not (protected / "denied.txt").exists()
+    assert (root / "outside.txt").read_text() == "private"
+
+
+@pytest.mark.asyncio
+async def test_seatbelt_cannot_move_read_only_ancestor(tmp_path):
+    """Moving a writable ancestor must not change a protected file's effective path."""
+    workspace = tmp_path.resolve()
+    protected = workspace / "parent" / "protected"
+    protected.mkdir(parents=True)
+    note = protected / "note.txt"
+    note.write_text("original", encoding="utf-8")
+    backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[str(protected)], writable_paths=[str(workspace)])
+
+    result = await backend.exec_shell(
+        ["/bin/sh", "-c", "mv parent renamed && echo changed > renamed/protected/note.txt"],
+        timeout=5,
+    )
+    assert result.exit_code != 0
+    assert note.read_text(encoding="utf-8") == "original"
+    assert not (workspace / "renamed").exists()
+
+    sibling = await backend.exec_shell(["/bin/sh", "-c", "echo allowed > sibling.txt"], timeout=5)
+    assert sibling.exit_code == 0, sibling.stderr.decode(errors="replace")
+    assert (workspace / "sibling.txt").read_text(encoding="utf-8").strip() == "allowed"
+
+
+@pytest.mark.asyncio
+async def test_seatbelt_reaps_same_group_child_after_parent_exits(tmp_path):
+    """launchd clears a background child after its foreground command returns."""
+    workspace = tmp_path.resolve()
+    backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)])
+    marker = workspace / "late.txt"
+    child_code = f"import time;from pathlib import Path;time.sleep(1);Path({str(marker)!r}).write_text('late')"
+    parent_code = (
+        "import subprocess,sys;"
+        f"subprocess.Popen([sys.executable,'-c',{child_code!r}],"
+        "stdin=subprocess.DEVNULL,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)"
+    )
+
+    result = await backend.exec_shell([sys.executable, "-c", parent_code], timeout=5)
+
+    assert result.exit_code == 0, result.stderr.decode(errors="replace")
+    await asyncio.sleep(1.2)
+    assert not marker.exists()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("cancel", [False, True], ids=["timeout", "cancel"])
+async def test_seatbelt_stops_detached_child(tmp_path, cancel):
+    """A child that left the launcher's process group should be stopped when observed."""
+    workspace = tmp_path.resolve()
+    backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)])
+    pidfile = workspace / "child.pid"
+    marker = workspace / "late.txt"
+    output = workspace / "child.out"
+    child_code = (
+        f"import os,time;os.setsid();open({str(pidfile)!r},'w').write(str(os.getpid()));"
+        f"time.sleep(1.2);open({str(marker)!r},'w').write('late')"
+    )
+    parent_code = (
+        f"import subprocess,sys,time;f=open({str(output)!r},'w');"
+        f"subprocess.Popen([sys.executable,'-c',{child_code!r}],stdin=f,stdout=f,stderr=f);"
+        "f.close();time.sleep(10)"
+    )
+    task = asyncio.create_task(backend.exec_shell([sys.executable, "-c", parent_code], timeout=10 if cancel else 0.5))
+    try:
+        for _ in range(100):
+            if pidfile.exists():
+                break
+            await asyncio.sleep(0.02)
+        assert pidfile.exists(), output.read_text() if output.exists() else "child did not start"
+        if cancel:
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(task, timeout=4)
+        else:
+            result = await asyncio.wait_for(task, timeout=4)
+            assert result.exit_code == -1 and result.stderr == b"timed out"
+        await asyncio.sleep(1.2)
+        assert not marker.exists()
+    finally:
+        if not task.done():
+            task.cancel()
+        if pidfile.exists():
+            try:
+                psutil.Process(int(pidfile.read_text(encoding="utf-8"))).kill()
+            except psutil.Error:
+                pass

tests/unit/test_file_sandbox_backend.py

@@ -0,0 +1,297 @@
+"""Shared file sandbox selection and macOS policy construction."""
+
+import asyncio
+import os
+import plistlib
+import sys
+from pathlib import Path
+
+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.local_backend import LocalProcessBackend
+from reme.utils.file_sandbox import create_file_sandbox_backend
+from reme.utils.file_sandbox.path_bubble_backend import PathBubbleBackend
+from reme.utils.file_sandbox.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_mib == 4096  # 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_mib=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 '(allow file-write* (literal "/dev/null"))' in profile
+    assert '(allow file-write* (subpath "/dev"))' not in profile
+    assert "network*" not in profile
+
+
+def test_sandbox_reads_virtual_environment_base_runtime(tmp_path, monkeypatch):
+    """A virtual environment grants its real base interpreter without widening writes."""
+    workspace = tmp_path / "workspace"
+    venv = tmp_path / "venv"
+    runtime = tmp_path / "runtime"
+    interpreter = tmp_path / "interpreter" / "python"
+    workspace.mkdir()
+    (venv / "bin").mkdir(parents=True)
+    runtime.mkdir()
+    interpreter.parent.mkdir(parents=True)
+    interpreter.touch()
+    (venv / "bin" / "python").symlink_to(interpreter)
+    monkeypatch.setattr(sys, "prefix", str(venv))
+    monkeypatch.setattr(sys, "base_prefix", str(runtime))
+    monkeypatch.setattr(sys, "executable", str(venv / "bin" / "python"))
+
+    backend = PathSeatbeltBackend(cwd=workspace, read_only_paths=[], writable_paths=[str(workspace)])
+    profile = backend._profile(str(workspace / "temporary"))  # pylint: disable=protected-access
+
+    assert f'(allow file-read* (subpath "{runtime}"))' in profile
+    assert f'(allow file-read* (literal "{interpreter}"))' in profile
+    assert f'(allow file-write* (subpath "{runtime}"))' not in profile
+    assert f'(allow file-write* (literal "{interpreter}"))' not in profile
+
+
+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"
+    protected.mkdir()
+    output = protected / "output"
+    output.mkdir()
+    backend = PathSeatbeltBackend(
+        cwd=tmp_path,
+        read_only_paths=[str(protected)],
+        writable_paths=[str(tmp_path), str(output)],
+    )
+    profile = backend._profile(str(tmp_path / "temporary"))  # pylint: disable=protected-access
+    assert (
+        f'(allow file-write* (require-all (subpath "{tmp_path}") ' f'(require-not (subpath "{protected}"))))'
+    ) in profile
+    assert f'(allow file-write* (subpath "{output}"))' in profile
+
+
+@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
+
+
+@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.file_sandbox.path_seatbelt_backend.shutil.which", lambda _: "/usr/bin/sandbox-exec")
+    seen = {}
+
+    async def fail(*arguments):
+        if arguments[0] == "bootout":
+            import errno
+            from reme.utils.file_sandbox.path_seatbelt_backend import LaunchctlError
+
+            raise LaunchctlError("service not found", errno.ESRCH)
+        seen["arguments"] = arguments
+        with open(arguments[-1], "rb") as stream:
+            seen["job"] = plistlib.load(stream)
+        raise RuntimeError("launchd unavailable")
+
+    backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    monkeypatch.setattr(backend, "_launchctl", fail)
+
+    async def available(_domain):
+        return True
+
+    monkeypatch.setattr(backend, "_has_gui_launchd", available)
+    with pytest.raises(RuntimeError, match="launchd unavailable"):
+        await backend.exec_shell(["/bin/echo", "unsafe"])
+    assert seen["arguments"][0] == "bootstrap"
+    assert seen["job"]["AbandonProcessGroup"] is False
+    assert seen["job"]["ProgramArguments"][-2:] == ["/bin/echo", "unsafe"]
+    assert not Path(seen["arguments"][-1]).exists()
+
+
+@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')"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["success", "failure", "timeout", "cancel", "error", "kickstart_error"])
+async def test_seatbelt_launchd_lifecycle(tmp_path, monkeypatch, mode):
+    """Every launchd outcome unregisters the job and removes its control files."""
+    from unittest.mock import AsyncMock, Mock
+
+    monkeypatch.setattr(sys, "platform", "darwin")
+    monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.shutil.which", lambda _: "sandbox-exec")
+    backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    monkeypatch.setattr(backend, "_has_gui_launchd", AsyncMock(return_value=True))
+    killed = Mock()
+    monkeypatch.setattr(backend, "_kill_process", killed)
+    calls = []
+    control = None
+    started = asyncio.Event()
+
+    async def launchctl(*args):
+        nonlocal control
+        calls.append(args[0])
+        if args[0] == "bootstrap":
+            control = Path(args[-1]).parent
+        if args[0] == "kickstart" and mode == "kickstart_error":
+            raise RuntimeError("kickstart failed")
+        if args[0] == "bootout":
+            # Output is collected after launchd has reaped the process group.
+            (control / "stdout").write_bytes(b"output")
+            (control / "stderr").write_bytes(b"diagnostic")
+        return "12345"
+
+    async def wait_status(*_args):
+        started.set()
+        if mode == "error":
+            raise RuntimeError("status failed")
+        if mode in ("timeout", "cancel"):
+            await asyncio.Event().wait()
+        return 7 if mode == "failure" else 0
+
+    monkeypatch.setattr(backend, "_launchctl", launchctl)
+    monkeypatch.setattr(backend, "_wait_status", wait_status)
+    task = asyncio.create_task(backend.exec_shell(["true"], timeout=0.01 if mode == "timeout" else None))
+    if mode == "cancel":
+        await asyncio.wait_for(started.wait(), timeout=1)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=1)
+    elif mode in ("error", "kickstart_error"):
+        with pytest.raises(RuntimeError, match="failed"):
+            await asyncio.wait_for(task, timeout=1)
+    else:
+        result = await asyncio.wait_for(task, timeout=1)
+        if mode == "timeout":
+            assert result == ExecResult(exit_code=-1, stdout=b"", stderr=b"timed out")
+        else:
+            assert result.exit_code == (7 if mode == "failure" else 0)
+            assert result.stdout == b"output" and result.stderr == b"diagnostic"
+    assert calls == ["bootstrap", "kickstart", "bootout"]
+    assert not control.exists()
+    assert killed.call_count == int(mode in ("timeout", "cancel", "error"))
+
+
+@pytest.mark.asyncio
+async def test_python_timeout_preserves_exit_status_and_stderr(tmp_path):
+    """The shared executor retains the existing unsandboxed Python metadata."""
+    step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    response = await step(
+        code="import sys, time; print('diagnostic', file=sys.stderr, flush=True); time.sleep(60)",
+        python_timeout=0.3,
+    )
+    assert response.success is False
+    assert response.answer == "Python execution timed out after 0.3s"
+    assert response.metadata == {
+        "returncode": -9 if os.name == "posix" else 1,
+        "stderr": "diagnostic\n",
+        "python_timeout": 0.3,
+    }

tests/unit/test_path_bubble_backend.py

@@ -0,0 +1,248 @@
+"""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.file_sandbox.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 MiB."""
+    with pytest.raises(ValueError, match="memory_limit_mib"):
+        PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=limit)
+
+
+def test_one_mib_is_written_as_bytes_to_cgroup(tmp_path, monkeypatch):
+    """The public MiB limit is converted only at the Linux cgroup boundary."""
+    proc_file = tmp_path / "proc-self-cgroup"
+    proc_file.write_text("0::/\n", encoding="ascii")
+    cgroup_root = tmp_path / "cgroup"
+    cgroup_root.mkdir()
+    original_path = Path
+
+    def sandbox_path(value):
+        if value == "/proc/self/cgroup":
+            return proc_file
+        if value == "/sys/fs/cgroup":
+            return cgroup_root
+        return original_path(value)
+
+    monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.Path", sandbox_path)
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=1)
+    group = backend._create_cgroup()  # pylint: disable=protected-access
+    assert (group / "memory.max").read_text(encoding="ascii") == str(1024**2)
+
+
+@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_mib=1024)
+    groups = []
+    subprocess_options = []
+    subprocess_commands = []
+
+    def create_group():
+        group = tmp_path / f"group-{len(groups)}"
+        group.mkdir()
+        groups.append(group)
+        return group
+
+    async def create_process(*argv, **kwargs):
+        subprocess_commands.append(argv)
+        subprocess_options.append(kwargs)
+        return SimpleNamespace(pid=100 + len(subprocess_options))
+
+    monkeypatch.setattr(sys, "platform", "linux")
+    monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.shutil.which", lambda _: "/usr/bin/bwrap")
+    monkeypatch.setattr(backend, "_create_cgroup", create_group)
+    monkeypatch.setattr("reme.utils.file_sandbox.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("preexec_fn" not in options for options in subprocess_options)
+    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
+    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_mib=0)
+    options = {}
+
+    async def create_process(*_argv, **kwargs):
+        options.update(kwargs)
+        return SimpleNamespace(pid=123)
+
+    monkeypatch.setattr(sys, "platform", "linux")
+    monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.shutil.which", lambda _: "/usr/bin/bwrap")
+    monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.asyncio.create_subprocess_exec", create_process)
+    await backend._start_process(["true"], str(tmp_path))  # pylint: disable=protected-access
+    assert "preexec_fn" not in options
+
+
+@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_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
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("limit", [0, 1, -1])
+async def test_python_execute_rejects_memory_limit_without_path_policy(tmp_path, limit):
+    """A requested memory policy must never be silently ignored."""
+    step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    response = await step(code="print('should not run')", memory_limit_mib=limit)
+    assert response.success is False
+    assert response.answer == "memory_limit_mib requires read_only_paths or writable_paths to enable the sandbox"
+
+
+@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_mib"]

tests/unit/test_process_output_limits.py

@@ -0,0 +1,156 @@
+"""Execution output stays bounded across pipe and launchd backends."""
+
+# pylint: disable=protected-access
+
+import asyncio
+import os
+import plistlib
+import stat
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock
+
+import pytest
+
+from reme.components.application_context import ApplicationContext
+from reme.steps.common.python_execute import PythonExecuteStep
+from reme.utils.file_sandbox.path_bubble_backend import PathBubbleBackend
+from reme.utils.file_sandbox.path_seatbelt_backend import PathSeatbeltBackend
+from reme.utils.local_backend import LocalProcessBackend, OUTPUT_LIMIT_BYTES
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("backend_name", ["local", "bubble", "seatbelt"])
+@pytest.mark.parametrize("stream", ["stdout", "stderr", "both"])
+@pytest.mark.parametrize("extra", [0, 1])
+async def test_pipe_output_boundary(tmp_path, monkeypatch, backend_name, stream, extra):
+    """Each stream has its own budget; crossing either budget fails execution."""
+    local = LocalProcessBackend()
+    backend = local
+    if backend_name != "local":
+        backend_class = PathBubbleBackend if backend_name == "bubble" else PathSeatbeltBackend
+        backend = backend_class(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+        # Exercise the shared collector with a real process, without requiring
+        # kernel sandbox support or changing the host launchd domain.
+        monkeypatch.setattr(backend, "_start_process", local._start_process)
+        if backend_name == "seatbelt":
+            monkeypatch.setattr(backend, "_has_gui_launchd", AsyncMock(return_value=False))
+    size = OUTPUT_LIMIT_BYTES + extra
+    code = "import os;" + ";".join(
+        f"os.write({fd}, b'x' * {size})" for fd in (1, 2) if stream in ("both", "stdout" if fd == 1 else "stderr")
+    )
+    if backend_name == "seatbelt":
+        with pytest.warns(RuntimeWarning, match="No macOS GUI launchd domain"):
+            result = await backend.exec_shell([sys.executable, "-c", code], timeout=5)
+    else:
+        result = await backend.exec_shell([sys.executable, "-c", code], timeout=5)
+    assert len(result.stdout) <= OUTPUT_LIMIT_BYTES
+    assert len(result.stderr) <= OUTPUT_LIMIT_BYTES
+    if extra:
+        assert not result.ok()
+        assert b"Output limit exceeded:" in result.stderr
+    else:
+        assert result.ok(), result.stderr
+        assert result.stdout == (b"x" * size if stream in ("stdout", "both") else b"")
+        assert result.stderr == (b"x" * size if stream in ("stderr", "both") else b"")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("fd", [1, 2])
+async def test_continuous_output_stops_process(tmp_path, fd):
+    """Overflow terminates a live writer before it can perform a later write."""
+    marker = tmp_path / "late.txt"
+    code = (
+        "import os,time;from pathlib import Path;"
+        f"os.write({fd}, b'x' * {OUTPUT_LIMIT_BYTES + 1});"
+        f"time.sleep(1);Path({str(marker)!r}).write_text('late')"
+    )
+    backend = LocalProcessBackend()
+    process = await backend._start_process([sys.executable, "-c", code], str(tmp_path))
+    result = await asyncio.wait_for(backend._communicate(process), timeout=3)
+    assert not result.ok()
+    assert b"Output limit exceeded:" in result.stderr
+    assert process.returncode is not None
+    assert not marker.exists()
+
+
+@pytest.mark.asyncio
+async def test_python_step_reports_output_overflow(tmp_path):
+    """Python callers receive bounded partial output and failure metadata."""
+    step = PythonExecuteStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    response = await step(code=f"import os;os.write(1, b'x' * {OUTPUT_LIMIT_BYTES + 1})")
+    assert response.success is False
+    assert len(response.answer) == OUTPUT_LIMIT_BYTES
+    assert "Output limit exceeded: stdout" in response.metadata["stderr"]
+    assert response.metadata["returncode"] != 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(os.name != "posix", reason="launchd uses POSIX FIFOs")
+@pytest.mark.parametrize("stream", ["stdout", "stderr", "both"])
+@pytest.mark.parametrize("extra", [0, 1])
+async def test_launchd_output_boundary(tmp_path, monkeypatch, stream, extra):
+    """Real FIFO writers obey each stream's budget and are reaped on overflow."""
+    backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.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=True))
+    size = OUTPUT_LIMIT_BYTES + extra
+    code = "import os,time;" + ";".join(
+        f"os.write({fd}, b'x' * {size})" for fd in (1, 2) if stream in ("both", "stdout" if fd == 1 else "stderr")
+    )
+    if extra:
+        code += ";time.sleep(60)"
+    job = {}
+    process = None
+    control = None
+    calls = []
+
+    async def launchctl(*args):
+        nonlocal job, process, control
+        calls.append(args[0])
+        if args[0] == "bootstrap":
+            control = Path(args[-1]).parent
+            with open(args[-1], "rb") as source:
+                job = plistlib.load(source)
+            for key in ("StandardOutPath", "StandardErrorPath"):
+                assert stat.S_ISFIFO(os.stat(job[key]).st_mode)
+        elif args[0] == "kickstart":
+            with open(job["StandardOutPath"], "wb") as stdout, open(job["StandardErrorPath"], "wb") as stderr:
+                process = await asyncio.create_subprocess_exec(
+                    sys.executable,
+                    "-c",
+                    code,
+                    stdout=stdout,
+                    stderr=stderr,
+                    start_new_session=True,
+                )
+            return str(process.pid)
+        elif args[0] == "bootout":
+            if process.returncode is None:
+                process.kill()
+            await process.wait()
+        return ""
+
+    async def wait_status(*_args):
+        return await process.wait()
+
+    monkeypatch.setattr(backend, "_launchctl", launchctl)
+    monkeypatch.setattr(backend, "_wait_status", wait_status)
+    try:
+        result = await asyncio.wait_for(backend.exec_shell(["test-command"], timeout=5), timeout=7)
+        assert len(result.stdout) <= OUTPUT_LIMIT_BYTES
+        assert len(result.stderr) <= OUTPUT_LIMIT_BYTES
+        assert result.ok() is (extra == 0)
+        if extra:
+            assert b"Output limit exceeded:" in result.stderr
+        else:
+            assert result.stdout == (b"x" * size if stream in ("stdout", "both") else b"")
+            assert result.stderr == (b"x" * size if stream in ("stderr", "both") else b"")
+        assert calls == ["bootstrap", "kickstart", "bootout"]
+        assert process.returncode is not None
+        assert not control.exists()
+    finally:
+        if process is not None and process.returncode is None:
+            process.kill()
+            await process.wait()

tests/unit/test_sandbox_startup_cancellation.py

@@ -0,0 +1,164 @@
+"""Cancellation during startup must not abandon OS-owned sandbox resources."""
+
+# pylint: disable=protected-access
+
+import asyncio
+import errno
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock
+
+import pytest
+
+from reme.utils.file_sandbox.path_bubble_backend import PathBubbleBackend
+from reme.utils.file_sandbox.path_seatbelt_backend import LaunchctlError, PathSeatbeltBackend
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("late_registration", [False, True])
+@pytest.mark.parametrize("cancel_twice", [False, True])
+async def test_cancelled_bootstrap_waits_and_unregisters(tmp_path, monkeypatch, late_registration, cancel_twice):
+    """Unregister after a pending bootstrap settles, including a late successful reply."""
+    backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.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=True))
+    entered = asyncio.Event()
+    reply = asyncio.Event()
+    bootout_entered = asyncio.Event()
+    bootout_reply = asyncio.Event()
+    services = set()
+    calls = []
+    control = None
+
+    async def launchctl(*args):
+        nonlocal control
+        calls.append(args[0])
+        if args[0] == "bootstrap":
+            control = Path(args[-1]).parent
+            if not late_registration:
+                services.add("sandbox")
+            entered.set()
+            await reply.wait()
+            services.add("sandbox")
+        elif args[0] == "bootout":
+            assert reply.is_set(), "bootout must not race an unfinished registration"
+            assert control.exists(), "keep control files until the service is removed"
+            bootout_entered.set()
+            await bootout_reply.wait()
+            services.clear()
+        else:
+            pytest.fail(f"Unexpected launchctl call: {args[0]}")
+        return ""
+
+    monkeypatch.setattr(backend, "_launchctl", launchctl)
+    task = asyncio.create_task(backend.exec_shell(["true"]))
+    await asyncio.wait_for(entered.wait(), timeout=1)
+    task.cancel()
+    await asyncio.sleep(0)
+    assert not task.done()
+    reply.set()
+    if cancel_twice:
+        await asyncio.wait_for(bootout_entered.wait(), timeout=1)
+        task.cancel()
+        await asyncio.sleep(0)
+        assert not task.done()
+    bootout_reply.set()
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(task, timeout=1)
+    assert calls == ["bootstrap", "bootout"]
+    assert not services
+    assert not control.exists()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("accepted", [False, True])
+async def test_failed_bootstrap_attempts_unregister(tmp_path, monkeypatch, accepted):
+    """A failed reply may still hide an accepted registration; absence alone is benign."""
+    backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    plist = tmp_path / "job.plist"
+    calls = []
+    services = set()
+
+    async def launchctl(*args):
+        calls.append(args[0])
+        if args[0] == "bootstrap":
+            if accepted:
+                services.add("sandbox")
+            raise LaunchctlError("registration reply failed", 5)
+        if not services:
+            raise LaunchctlError("service not found", errno.ESRCH)
+        services.clear()
+        return ""
+
+    monkeypatch.setattr(backend, "_launchctl", launchctl)
+    with pytest.raises(LaunchctlError, match="registration reply failed"):
+        await backend._bootstrap("gui/501", plist, "gui/501/sandbox")
+    assert calls == ["bootstrap", "bootout"]
+    assert not services
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_rollback_reports_unregistration_failure(tmp_path, monkeypatch):
+    """Permission and other cleanup failures must not be mistaken for a missing service."""
+    backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    launchctl = AsyncMock(side_effect=[LaunchctlError("bootstrap failed", 5), LaunchctlError("bootout denied", 1)])
+    monkeypatch.setattr(backend, "_launchctl", launchctl)
+    with pytest.raises(LaunchctlError, match="bootout denied"):
+        await backend._bootstrap("gui/501", tmp_path / "job.plist", "gui/501/sandbox")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("memory_limit", [0, 1])
+@pytest.mark.parametrize("spawn_fails", [False, True])
+async def test_cancelled_bubble_start_reaps_before_removing_cgroup(tmp_path, monkeypatch, memory_limit, spawn_fails):
+    """A delayed process handle is reclaimed before its cgroup is removed."""
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=memory_limit)
+    monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.sys.platform", "linux")
+    monkeypatch.setattr("reme.utils.file_sandbox.path_bubble_backend.shutil.which", lambda _: "bwrap")
+    group = tmp_path / "cgroup"
+
+    def create_group():
+        group.mkdir()
+        return group
+
+    monkeypatch.setattr(backend, "_create_cgroup", create_group)
+    entered = asyncio.Event()
+    reply = asyncio.Event()
+    processes = []
+    create_process = asyncio.create_subprocess_exec
+
+    async def delayed_spawn(*_args, **kwargs):
+        if not spawn_fails:
+            processes.append(await create_process(sys.executable, "-c", "import time;time.sleep(60)", **kwargs))
+        entered.set()
+        await reply.wait()
+        if spawn_fails:
+            raise FileNotFoundError("spawn failed after cancellation")
+        return processes[0]
+
+    monkeypatch.setattr("reme.utils.local_backend.asyncio.create_subprocess_exec", delayed_spawn)
+    task = asyncio.create_task(backend._start_process(["true"], str(tmp_path)))
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=2)
+        task.cancel()
+        await asyncio.sleep(0)
+        task.cancel()
+        await asyncio.sleep(0)
+        assert not task.done()
+        assert group.exists() is bool(memory_limit)
+        reply.set()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=2)
+        assert all(process.returncode is not None for process in processes)
+        assert not group.exists()
+        assert not backend._process_cgroups
+    finally:
+        reply.set()
+        if not task.done():
+            task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+        for process in processes:
+            if process.returncode is None:
+                process.kill()
+                await process.communicate()