Harden AutoBash policy and macOS sandbox supervision

plugins/home/src/reme_home/auto_bash.py

@@ -1,6 +1,7 @@
 """Run allowlisted shell commands without an approval prompt."""
 
 import asyncio
+from functools import cache
 
 from agentscope.message import TextBlock, ToolResultState
 from agentscope.permission import PermissionBehavior, PermissionContext, PermissionDecision
@@ -33,13 +34,15 @@ _READ_COMMANDS = {
             "--show-signature",
             "--remerge-diff",
             "--diff-merges",
+            "--pretty",
+            "--format",
         ],
     },
     "grep": {},
     "head": {},
     "ls": {},
     "pwd": {},
-    "sort": {"denied": ["-o", "--output", "--compress-program", "--temporary-directory"]},
+    "sort": {"denied": ["-o", "-T", "--output", "--compress-program", "--temporary-directory"]},
     "stat": {},
     "tail": {},
     "true": {},
@@ -54,6 +57,10 @@ _READ_ONLY_SETUP = r"""
 unset GIT_EXTERNAL_DIFF GIT_TRACE GIT_TRACE_SETUP GIT_TRACE_PERFORMANCE
 unset GIT_TRACE_PACK_ACCESS GIT_TRACE_PACKET GIT_TRACE_CURL GIT_TRACE2 GIT_TRACE2_EVENT GIT_TRACE2_PERF
 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 ] && \
@@ -71,12 +78,20 @@ git() {
         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 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
@@ -162,7 +177,6 @@ class AutoBash(Bash):
         if "$" in command or "`" in command:
             return False
         try:
-            from safecmd import CmdSpec  # pylint: disable=import-outside-toplevel
             from safecmd.bashxtract import (  # pylint: disable=import-outside-toplevel
                 check_types,
                 collect_ops,
@@ -176,8 +190,8 @@ class AutoBash(Bash):
             check_types(ast)
             if not _literal_shell(ast, command):
                 return False
-            specs = {CmdSpec(name, denied=options.get("denied")) for name, options in _READ_COMMANDS.items()}
             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(
@@ -224,6 +238,7 @@ 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))
@@ -249,9 +264,11 @@ class AutoBashStep(BaseStep):
             return self.context.response
 
         tool = AutoBash(cwd=self.workspace_path, is_read_only=is_read_only, backend=backend)
-        chunks = [chunk async for chunk in tool.call(command=command, timeout=timeout)]
-        self.context.response.success = all(chunk.state != ToolResultState.ERROR for chunk in chunks)
-        self.context.response.answer = "".join(
-            block.text for chunk in chunks for block in chunk.content if isinstance(block, TextBlock)
-        )
+        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/tests/test_auto_bash.py

@@ -28,6 +28,13 @@ def linux_backend_selection(monkeypatch):
     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."""
@@ -56,6 +63,8 @@ async def test_auto_bash_runs_in_workspace(tmp_path):
         "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",
@@ -82,6 +91,9 @@ async def test_auto_bash_runs_in_workspace(tmp_path):
         "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",
         "rg --hostname-bin=./hook --hyperlink-format=default hello .",
         "rg -z hello .",
@@ -222,18 +234,6 @@ async def test_unrestricted_tool_allows_commands_without_read_only_check(tmp_pat
     assert (tmp_path / "written.txt").exists()
 
 
-def test_auto_bash_defaults_to_sixty_seconds():
-    """Tool and default job expose the same 60-second timeout."""
-    manifest = Path(__file__).parents[1] / "src" / "reme_home" / "plugin.yaml"
-    with manifest.open(encoding="utf-8") as config_file:
-        job = yaml.safe_load(config_file)["application_defaults"]["jobs"]["auto_bash"]
-
-    assert inspect.signature(AutoBash.call).parameters["timeout"].default == 60000
-    assert AutoBash.input_schema["properties"]["timeout"]["default"] == 60000
-    assert job["parameters"]["properties"]["timeout"]["default"] == 60000
-    assert job["parameters"]["properties"]["is_read_only"]["default"] is True
-
-
 @pytest.mark.asyncio
 async def test_auto_bash_uses_path_backend_when_paths_supplied(tmp_path, monkeypatch):
     """AutoBash routes an injected path policy through the backend."""
@@ -270,12 +270,16 @@ async def test_sandbox_start_failure_does_not_run_on_host(tmp_path, monkeypatch)
 
 
 def test_home_manifest_declares_step_and_job():
-    """The installed manifest is the source for backend and job registration."""
+    """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
@@ -306,13 +310,10 @@ def test_home_plugin_registers_job_and_step(monkeypatch):
         "false",
         "grep hello note.txt",
         "find . -name note.txt",
-        "sort note.txt",
         "diff note.txt note.txt",
         "cmp note.txt note.txt",
         "git status --short",
         "git diff --stat",
-        "git log --oneline",
-        "git show HEAD",
         "git rev-parse HEAD",
         "git ls-files",
     ],
@@ -339,6 +340,65 @@ async def test_safe_redirection_is_allowed(tmp_path):
     assert await tool.check_read_only({"command": "(cat note.txt) | head -n 1"})
 
 
+@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
+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):
@@ -391,7 +451,6 @@ async def test_global_git_lfs_config_allows_metadata_queries(tmp_path, monkeypat
         capture_output=True,
     )
     monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(global_config))
-    monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
     tool = AutoBash(cwd=repo)
 
     for command, expected in (("git rev-parse --git-dir", ".git"), ("git ls-files", "note.txt")):
@@ -418,6 +477,11 @@ async def test_allow_and_denied_command_options_are_combined(tmp_path, monkeypat
     assert not await tool.check_read_only({"command": "echo goodbye"})
 
     monkeypatch.setitem(commands, "echo", {"allow": ["hello"], "denied": ["-n"]})
+    from safecmd import CmdSpec
+
+    specs = {spec for spec in auto_bash._read_specs() if "echo" not in spec.name}  # pylint: disable=protected-access
+    specs.add(CmdSpec("echo", denied=["-n"]))
+    monkeypatch.setattr(auto_bash, "_read_specs", lambda: specs)
     assert not await tool.check_read_only({"command": "echo hello -n"})
     assert await tool.check_read_only({"command": "find . -name note.txt"})
     assert not await tool.check_read_only({"command": "find . -delete"})

reme/steps/common/python_execute.py

@@ -43,33 +43,20 @@ class PythonExecuteStep(BaseStep):
 
         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
-
-        try:
             result = await self._run_python(code, timeout, backend)
-        except (OSError, RuntimeError) as exc:
+        except (TypeError, ValueError, OSError, RuntimeError) as exc:
             self.context.response.success = False
             self.context.response.answer = str(exc)
             return self.context.response
         if result.timed_out:
             self.context.response.success = False
             self.context.response.answer = f"Python execution timed out after {timeout:g}s"
-            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
+            stderr = result.stderr
+        else:
+            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
         self.context.response.metadata.update(
             {
                 "returncode": result.returncode,

reme/utils/file_sandbox/path_seatbelt_backend.py

@@ -3,10 +3,17 @@
 import asyncio
 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
 
@@ -92,23 +99,16 @@ class PathSeatbeltBackend(FileSandboxBackend):
         return "\n".join(lines)
 
     async def _start_process(self, command: list[str], cwd: str, *, stdin=asyncio.subprocess.DEVNULL):
-        if sys.platform != "darwin":
-            raise RuntimeError("PathSeatbeltBackend requires macOS")
-        if shutil.which("sandbox-exec") is None:
-            raise RuntimeError("PathSeatbeltBackend requires sandbox-exec on PATH")
-        if not command:
-            raise ValueError("command must not be empty")
-        self._check_paths(cwd)
+        """Start Seatbelt directly when the GUI launchd domain is unavailable."""
         tempdir = tempfile.mkdtemp(prefix="reme-seatbelt-", dir=os.path.realpath(tempfile.gettempdir()))
         try:
-            env = {**os.environ, "HOME": self._cwd, "TMPDIR": tempdir}
             process = await asyncio.create_subprocess_exec(
                 "/usr/bin/sandbox-exec",
                 "-p",
                 self._profile(tempdir),
                 *command,
                 cwd=cwd,
-                env=env,
+                env={**os.environ, "HOME": self._cwd, "TMPDIR": tempdir},
                 stdin=stdin,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
@@ -124,3 +124,144 @@ class PathSeatbeltBackend(FileSandboxBackend):
         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:
+        process = await asyncio.create_subprocess_exec(
+            "/bin/launchctl",
+            "print",
+            domain,
+            stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.DEVNULL,
+        )
+        return await process.wait() == 0
+
+    @staticmethod
+    async def _launchctl(*arguments: str) -> str:
+        process = await asyncio.create_subprocess_exec(
+            "/bin/launchctl",
+            *arguments,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+        if process.returncode:
+            raise RuntimeError(f"launchctl {' '.join(arguments[:1])} failed: {stderr.decode(errors='replace').strip()}")
+        return stdout.decode(errors="replace").strip()
+
+    @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)
+
+    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
+        timed_out = False
+        completed = False
+        pid = None
+        stdout = b""
+        stderr = b""
+        try:
+            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._launchctl("bootstrap", domain, str(plist_path))
+            registered = True
+            pid = int(await self._launchctl("kickstart", "-p", service))
+            try:
+                exit_code = await asyncio.wait_for(self._wait_status(status_path, pid), timeout=timeout)
+            except TimeoutError:
+                timed_out = True
+            else:
+                completed = True
+        finally:
+            try:
+                if registered:
+                    if not completed and pid is not None:
+                        self._kill_process(SimpleNamespace(pid=pid))
+                    await self._launchctl("bootout", service)
+            finally:
+                if not timed_out:
+                    stdout = stdout_path.read_bytes() if stdout_path.exists() else b""
+                    stderr = stderr_path.read_bytes() if stderr_path.exists() else b""
+                shutil.rmtree(control)
+        if timed_out:
+            return ExecResult(exit_code=-1, stdout=b"", stderr=b"timed out")
+        return ExecResult(exit_code=exit_code, stdout=stdout, stderr=stderr)
+
+    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"))

tests/integration/test_seatbelt_backend.py

@@ -61,6 +61,47 @@ async def test_seatbelt_allows_null_device_redirection(tmp_path):
     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."""
@@ -126,6 +167,26 @@ async def test_seatbelt_cannot_move_read_only_ancestor(tmp_path):
     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):

tests/unit/test_file_sandbox_backend.py

@@ -1,7 +1,9 @@
 """Shared file sandbox selection and macOS policy construction."""
 
 import asyncio
+import plistlib
 import sys
+from pathlib import Path
 
 import pytest
 from agentscope.tool import ExecResult
@@ -109,7 +111,7 @@ def test_seatbelt_excludes_read_only_child_from_writable_parent(tmp_path):
 @pytest.mark.asyncio
 async def test_sandbox_timeout_does_not_wait_forever_for_inherited_pipe(tmp_path, monkeypatch):
     """A surviving child with an output pipe must not hold the timeout response open."""
-    backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
+    backend = PathBubbleBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[], memory_limit_mib=0)
 
     class StuckProcess:
         """Model a child that keeps its output pipe open after termination."""
@@ -140,19 +142,27 @@ async def test_seatbelt_launch_failure_cleans_up_without_host_fallback(tmp_path,
     """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")
-    calls = []
+    seen = {}
 
-    async def fail(*argv, **_kwargs):
-        calls.append(argv)
-        raise OSError("sandbox unavailable")
+    async def fail(*arguments):
+        seen["arguments"] = arguments
+        with open(arguments[-1], "rb") as stream:
+            seen["job"] = plistlib.load(stream)
+        raise RuntimeError("launchd unavailable")
 
-    monkeypatch.setattr("reme.utils.file_sandbox.path_seatbelt_backend.asyncio.create_subprocess_exec", fail)
     backend = PathSeatbeltBackend(cwd=tmp_path, read_only_paths=[], writable_paths=[])
-    with pytest.raises(OSError, match="sandbox unavailable"):
+    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 calls[0][0] == "/usr/bin/sandbox-exec"
-    assert calls[0][-2:] == ("/bin/echo", "unsafe")
-    assert not backend._process_tempdirs  # pylint: disable=protected-access
+    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