fix(sandbox): stop owned command processes when a run is cancelled (#7772)

* fix(sandbox): stop owned command processes when a run is cancelled

* fix(sandbox): handle incomplete AIO output responses

---------

Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -19,6 +19,11 @@ This section accumulates work toward the **2.3.0** milestone
 
 ### Fixed
 
+- **sandbox:** Cancelled native Runs now stop their owned AIO and POSIX local
+  bash processes, including ordinary `nohup` children, before releasing the
+  execution lease. Concurrent work and background services from successful
+  earlier turns remain running. ([#7772])
+  
 - **agents:** A clarification no longer overwrites an earlier question when the
   provider reuses a tool-call ID across turns (for example `call_0` on every
   response). Its message ID, which is also the human-input `request_id`, named
@@ -15966,3 +15971,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7729]: https://github.com/bytedance/deer-flow/pull/7729
 [#7732]: https://github.com/bytedance/deer-flow/pull/7732
 [#7762]: https://github.com/bytedance/deer-flow/pull/7762
+[#7772]: https://github.com/bytedance/deer-flow/pull/7772

CHANGELOG_zh.md

@@ -12725,3 +12725,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#7729]: https://github.com/bytedance/deer-flow/pull/7729
 [#7732]: https://github.com/bytedance/deer-flow/pull/7732
 [#7762]: https://github.com/bytedance/deer-flow/pull/7762
+[#7772]: https://github.com/bytedance/deer-flow/pull/7772

README.md

@@ -2834,6 +2834,8 @@ async with runtime:
 
 Gateway run finalization also retains ownership through caller cancellation: it finishes terminal persistence, stream completion, and resource cleanup before propagating the interruption. Backend I/O must terminate or enforce its own timeout. Gateway shutdown includes workers still finalizing after their terminal status commits; an expired shutdown drain budget is logged and does not cancel their outstanding writes.
 
+Cancelled, replaced, failed or hard-stopped executions stop their owned bash processes before releasing the sandbox lease on AIO and POSIX LocalSandbox, including ordinary `nohup` children. Background services from successful earlier turns remain running. Cleanup tracks inherited execution markers and process groups; it is not a security boundary against scripts deliberately escaping that ownership. Windows LocalSandbox and third-party providers retain their existing behavior.
+
 Terminal cleanup releases the run journal and its message-sequence callback even when diagnostic logs retain a checkpoint-inspection traceback.
 
 The factory still does not load YAML or create SQL infrastructure: the caller supplies the config snapshot, repository, and lifecycle. Because it accepts a caller-owned `system_prompt`, direct integrations also own any model-visible wording about those limits; the default middleware enforces the runtime limits regardless. The factory does not mount the Gateway owner-scoped HTTP routes or Web UI, so direct applications must expose their own result API/UI if they need those surfaces. For ordinary delegation only, `SubagentRuntime(...)` needs no asynchronous startup.

backend/packages/harness/deerflow/agents/middlewares/run_input_middleware.py

@@ -32,6 +32,9 @@ async def manage_current_work(action: Literal["continue", "replace", "stop"], ru
         clear_goal = runtime.state.get("goal") is not None
         await control.agent_runs.authorize_control(thread_id=control.thread_id, run_id=control.inputs.run_id, clear_goal=clear_goal)
         await control.stop_children()
+        from deerflow.sandbox.lease import cancel_sandbox_execution_lease_async
+
+        await cancel_sandbox_execution_lease_async(runtime.context, reset=True)
         await control.agent_runs.authorize_control(thread_id=control.thread_id, run_id=control.inputs.run_id, clear_goal=clear_goal)
         control.abandoned = True
         if clear_goal:

backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md

@@ -4,7 +4,7 @@
 
 AIO has two distinct stdin contracts: the persistent `/v1/shell` transport is a
 PTY, while `/v1/bash` uses a subprocess pipe that stays open for writes. Keep
-persistent shell commands unchanged: the Lark broker shim already ignores TTY
+persistent shell stdin unchanged: the Lark broker shim already ignores TTY
 input. Only fresh `_run_bash_exec` calls prefix the original command with
 `exec < /dev/null` so default stdin gets immediate EOF; explicit
 pipes/heredocs/files still override fd0. Those sessions are created and
@@ -20,6 +20,16 @@ INFO logs contain argc/exit/elapsed only, never argument values.
 
 Regressions: `tests/test_aio_sandbox.py`, `tests/test_lark_broker.py`.
 
+## Command cancellation
+
+Managed bash uses existing async exec/poll APIs without retrying execution.
+Read only the current console entry; keep exit codes and persistent-shell stdin.
+Legacy Shell needs a same-line owner prefix. Poll by returned session/command identity; stop inherited owner processes through
+a separate fixed control command with pidfds. Missing command state or cleanup
+receipts fence reuse through existing quarantine. New output from a running
+command needs its next offset; empty polls retain the previous offsets.
+Markers support cooperative cleanup, not hostile-code isolation.
+
 ## Reuse admission
 
 An implicit-shell fence is sticky for the container's lifetime, even after a

backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py

@@ -1,7 +1,9 @@
 import base64
+import json
 import logging
 import math
 import threading
+import time
 import uuid
 from dataclasses import dataclass, field
 
@@ -11,6 +13,7 @@ from agent_sandbox.core.api_error import ApiError
 from pydantic import ValidationError
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
+from deerflow.sandbox.command_scope import COMMAND_OWNER_ENV, CommandScopeCancelled, command_lock, current_command_scope
 from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT as _MAX_DOWNLOAD_SIZE
 from deerflow.sandbox.download import download_limit, read_download_stream
 from deerflow.sandbox.remote_list_dir import parse_remote_list_dir_output, remote_list_dir_command
@@ -18,6 +21,7 @@ from deerflow.sandbox.sandbox import Sandbox, _validate_extra_env
 from deerflow.sandbox.search import GrepMatch, path_matches, should_ignore_path_under_root, truncate_line
 
 from .backend import sandbox_http_trust_env
+from .command_cleanup import cleanup_command
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +72,8 @@ class AioSandbox(Sandbox):
     #: the next — recorded bash evidence cannot prove a clean environment.
     persistent_shell_sessions = True
 
+    supports_command_cancellation = True
+
     def __init__(
         self,
         id: str,
@@ -124,6 +130,8 @@ class AioSandbox(Sandbox):
         self._recovery_session_id: str | None = None
         self._default_shell_corrupted = False
         self._closed = False
+        self._default_shell_stopped = False
+        self._command_cleanup_failed = False
         # Set to True after bash.exec answers 404 (image predates /v1/bash/*),
         # so later env-bearing calls fail fast instead of re-hitting HTTP (#3921).
         self._bash_exec_unsupported = False
@@ -339,7 +347,7 @@ class AioSandbox(Sandbox):
             # before its execution lease ends. Fencing on pending let a
             # concurrent acquire recycle a healthy container that another run
             # was actively using.
-            return bool(self._default_shell_corrupted or shell.ambiguous or bash.ambiguous)
+            return bool(self._default_shell_corrupted or getattr(self, "_command_cleanup_failed", False) or shell.ambiguous or bash.ambiguous)
 
     def _create_shell_session(self, client) -> str:
         session_id = str(uuid.uuid4())
@@ -406,8 +414,9 @@ class AioSandbox(Sandbox):
 
         Caller must hold ``self._lock``.
         """
-        if self._default_shell_corrupted and self._recovery_session_id is None:
+        if (self._default_shell_corrupted or self._default_shell_stopped) and self._recovery_session_id is None:
             self._recovery_session_id = self._create_shell_session(client)
+            self._default_shell_stopped = False
         return self._recovery_session_id
 
     def _exec_shell(
@@ -428,8 +437,92 @@ class AioSandbox(Sandbox):
         }
         if session_id is not None:
             kwargs["id"] = session_id
+        scope = current_command_scope(self)
+        if scope is not None:
+            return self._exec_owned_shell(client, command, session_id=session_id, timeout=timeout, kwargs=kwargs, scope=scope)
         return self._format_shell_result(client.shell.exec_command(**kwargs))
 
+    def _exec_owned_shell(self, client, command, *, session_id, timeout, kwargs, scope):
+        scope.check()
+        # Keep the original command at top level; legacy AIO rejects a
+        # newline prefix. The last matching console entry belongs to this call.
+        prepared = f"export {COMMAND_OWNER_ENV}={scope.token}; {command}"
+        started = time.monotonic()
+        result = client.shell.exec_command(**{**kwargs, "command": prepared, "async_mode": True, "request_options": self._bounded_cleanup_request_options()})
+        data = result.data if result else None
+        active_id = getattr(data, "session_id", None) or session_id
+        if not active_id:
+            self._command_cleanup_failed = True
+            raise RuntimeError("AIO did not return the command session identity")
+        while True:
+            if scope.cancelled.is_set():
+                self._stop_owned_processes(scope)
+                raise CommandScopeCancelled("Sandbox execution was cancelled")
+            remaining = timeout - (time.monotonic() - started)
+            if remaining <= 0:
+                client.shell.kill_process(id=active_id, request_options=self._bounded_cleanup_request_options())
+                if session_id is None or self._recovery_session_id == active_id:
+                    self._recovery_session_id = None
+                    self._default_shell_stopped = True
+                return "", 124, "hard_timeout"
+            wait = client.shell.wait_for_process(id=active_id, seconds=1, max_wait_seconds=1, request_options=self._bounded_cleanup_request_options())
+            status = getattr(getattr(wait, "data", None), "status", None)
+            if scope.cancelled.is_set():
+                continue
+            if status == "running":
+                continue
+            view = client.shell.view(id=active_id, request_options=self._bounded_cleanup_request_options())
+            data = view.data if view else None
+            if data is None:
+                self._command_cleanup_failed = True
+                raise RuntimeError("AIO command result is unavailable")
+            output = next((entry.output or "" for entry in reversed(data.console or []) if entry.command == prepared), None)
+            if output is None:
+                self._command_cleanup_failed = True
+                raise RuntimeError("AIO did not return this command's console record")
+            return output, getattr(data, "exit_code", None), getattr(data, "status", status)
+
+    def cancel_command_scope(self, scope_id: str) -> None:
+        scope = self._command_scopes.find(scope_id)
+        if scope is None:
+            return
+        self.request_command_cancel(scope_id)
+        self._stop_owned_processes(scope)
+
+    def _stop_owned_processes(self, scope) -> None:
+        with scope.lock:
+            if scope.cleaned.is_set():
+                return
+            self._cleanup_owned_processes(scope)
+
+    def _cleanup_owned_processes(self, scope) -> None:
+        client = self._client
+        if client is None:
+            raise RuntimeError("Sandbox client closed before command cleanup")
+        session_id = None
+        try:
+            session_id = self._create_shell_session(client)
+            client.shell.exec_command(command=cleanup_command(scope.token), id=session_id, async_mode=True, request_options=self._bounded_cleanup_request_options())
+            deadline = time.monotonic() + 5
+            while True:
+                result = client.shell.view(id=session_id, request_options=self._bounded_cleanup_request_options())
+                output, code, status = self._format_shell_result(result)
+                if status != "running" and output:
+                    break
+                if time.monotonic() >= deadline:
+                    raise RuntimeError("Sandbox process cleanup did not return a result")
+                time.sleep(0.1)
+            receipt = json.loads(output)
+            if code not in (0, None) or status not in (None, "completed") or receipt.get("remaining") != 0:
+                raise RuntimeError("Owned sandbox processes did not stop")
+            scope.cleaned.set()
+        except Exception:
+            self._command_cleanup_failed = True
+            raise
+        finally:
+            if session_id is not None:
+                self._cleanup_session_best_effort(client, session_id, context="cancelled execution cleanup", request_options=self._bounded_cleanup_request_options())
+
     def _rotate_and_retry_shell(
         self,
         client,
@@ -513,7 +606,7 @@ class AioSandbox(Sandbox):
                     scope_id,
                     _ScopedShellSession(),
                 )
-            with scoped.lock:
+            with command_lock(scoped.lock, current_command_scope(self)):
                 # Registration and command execution are separated by the
                 # per-scope wait.  Revalidate identity after that wait so a
                 # release/close which removed this exact scope is a hard
@@ -602,7 +695,8 @@ class AioSandbox(Sandbox):
             return f"Error: {e}"
 
     def release_command_scope(self, scope_id: str) -> None:
-        """Clean up one subagent's explicit server-side shell session."""
+        """Release execution metadata and its explicit shell session."""
+        self._command_scopes.release(scope_id)
         with self._scope_registry_lock:
             scoped = self._scoped_shell_sessions.pop(scope_id, None)
         if scoped is None:
@@ -791,7 +885,7 @@ class AioSandbox(Sandbox):
         effective_timeout = self._effective_command_timeout(timeout)
         if env:
             return self._execute_with_env(command, env, effective_timeout)
-        with self._lock:
+        with command_lock(self._lock, current_command_scope(self)):
             try:
                 client = self._client
                 if getattr(self, "_closed", False) or client is None:
@@ -925,6 +1019,80 @@ class AioSandbox(Sandbox):
                 return retried
         return output
 
+    def _render_bash_output(self, stdout, stderr, exit_code, status, timeout) -> tuple[str, str | None]:
+        output = stdout
+        if stderr:
+            output += f"\nStd Error:\n{stderr}" if output else stderr
+
+        if status == "timed_out":
+            notice = self._format_timeout_notice(timeout)
+            output = f"{output}\n{notice}" if output else notice
+            return f"{output}\nExit Code: 124", status
+
+        if status == "killed":
+            notice = "Command was killed before completion."
+            output = f"{output}\n{notice}" if output else notice
+            return output, status
+
+        if status == "running":
+            notice = "Error: Sandbox command returned a non-terminal running status; command outcome is unknown and was not retried."
+            output = f"{output}\n{notice}" if output else notice
+            return output, status
+
+        if status not in (None, "completed"):
+            notice = self._unexpected_status_notice(status)
+            output = f"{output}\n{notice}" if output else notice
+            return output, status
+
+        if exit_code not in (0, None):
+            # Mirror LocalSandbox: keep the actual shell status in the
+            # output text (acceptance-checklist evidence).
+            output = f"{output}\nExit Code: {exit_code}" if output else f"Command exited with code {exit_code}"
+        return output if output else "(no output)", status
+
+    def _run_owned_bash(self, command, env, timeout, scope):
+        scope.check()
+        session_id = self._create_bash_session(self._client)
+        started = time.monotonic()
+        try:
+            prepared = f"exec < /dev/null\nexport {COMMAND_OWNER_ENV}={scope.token}\n{command}"
+            response = self._client.bash.exec(command=prepared, session_id=session_id, env=env, async_mode=True, hard_timeout=timeout, request_options=self._bounded_cleanup_request_options())
+            initial = response.data if response else None
+            command_id = getattr(initial, "command_id", None)
+            if not command_id:
+                self._command_cleanup_failed = True
+                raise RuntimeError("AIO did not return the bash command identity")
+            stdout, stderr = [], []
+            offset = stderr_offset = 0
+            while True:
+                if scope.cancelled.is_set():
+                    self._stop_owned_processes(scope)
+                    raise CommandScopeCancelled("Sandbox execution was cancelled")
+                if time.monotonic() - started >= timeout:
+                    self._client.bash.kill(session_id=session_id, signal="SIGKILL", request_options=self._bounded_cleanup_request_options())
+                    return self._render_bash_output("".join(stdout), "".join(stderr), 124, "timed_out", timeout)
+                response = self._client.bash.output(session_id=session_id, command_id=command_id, offset=offset, stderr_offset=stderr_offset, wait=True, wait_timeout=1, request_options=self._bounded_cleanup_request_options())
+                data = response.data if response else None
+                command_info = getattr(data, "command", None)
+                if command_info is None:
+                    self._command_cleanup_failed = True
+                    raise RuntimeError("AIO bash command result is unavailable")
+                stdout.append(data.stdout or "")
+                stderr.append(data.stderr or "")
+                status = command_info.status
+                if scope.cancelled.is_set():
+                    continue
+                if status != "running":
+                    return self._render_bash_output("".join(stdout), "".join(stderr), command_info.exit_code, status, timeout)
+                if (data.stdout and data.offset is None) or (data.stderr and data.stderr_offset is None):
+                    self._command_cleanup_failed = True
+                    raise RuntimeError("AIO bash output position is unavailable")
+                # Empty polls may omit offsets; retain the last known positions.
+                offset = data.offset if data.offset is not None else offset
+                stderr_offset = data.stderr_offset if data.stderr_offset is not None else stderr_offset
+        finally:
+            self._cleanup_bash_session_best_effort(self._client, session_id)
+
     def _run_bash_exec(
         self,
         command: str,
@@ -932,10 +1100,13 @@ class AioSandbox(Sandbox):
         timeout: float,
     ) -> tuple[str, str | None]:
         """Single bash.exec invocation in an explicitly released fresh session."""
-        with self._lock:
+        scope = current_command_scope(self)
+        with command_lock(self._lock, scope):
             for attempt in range(2):
                 session_id: str | None = None
                 try:
+                    if scope is not None:
+                        return self._run_owned_bash(command, env, timeout, scope)
                     session_id = self._create_bash_session(self._client)
                     result = self._client.bash.exec(
                         # /v1/bash keeps a subprocess stdin pipe open for writes.
@@ -952,39 +1123,7 @@ class AioSandbox(Sandbox):
                         request_options=self._command_request_options(timeout),
                     )
                     data = result.data if result else None
-                    stdout = (data.stdout or "") if data else ""
-                    stderr = (data.stderr or "") if data else ""
-                    exit_code = getattr(data, "exit_code", None) if data else None
-                    status = getattr(data, "status", None) if data else None
-                    output = stdout
-                    if stderr:
-                        output += f"\nStd Error:\n{stderr}" if output else stderr
-
-                    if status == "timed_out":
-                        notice = self._format_timeout_notice(timeout)
-                        output = f"{output}\n{notice}" if output else notice
-                        return f"{output}\nExit Code: 124", status
-
-                    if status == "killed":
-                        notice = "Command was killed before completion."
-                        output = f"{output}\n{notice}" if output else notice
-                        return output, status
-
-                    if status == "running":
-                        notice = "Error: Sandbox command returned a non-terminal running status; command outcome is unknown and was not retried."
-                        output = f"{output}\n{notice}" if output else notice
-                        return output, status
-
-                    if status not in (None, "completed"):
-                        notice = self._unexpected_status_notice(status)
-                        output = f"{output}\n{notice}" if output else notice
-                        return output, status
-
-                    if exit_code not in (0, None):
-                        # Mirror LocalSandbox: keep the actual shell status in the
-                        # output text (acceptance-checklist evidence).
-                        output = f"{output}\nExit Code: {exit_code}" if output else f"Command exited with code {exit_code}"
-                    return output if output else "(no output)", status
+                    return self._render_bash_output(getattr(data, "stdout", None) or "", getattr(data, "stderr", None) or "", getattr(data, "exit_code", None), getattr(data, "status", None), timeout)
                 except httpx.TimeoutException:
                     return self._transport_timeout_error(timeout), "transport_timeout"
                 except ApiError as e:

backend/packages/harness/deerflow/community/aio_sandbox/command_cleanup.py

@@ -0,0 +1,63 @@
+"""Fixed, model-independent cleanup for Linux AIO commands carrying a run marker."""
+
+import re
+import shlex
+
+# pidfds keep a recycled PID from receiving a late cancellation signal. Only
+# exact inherited execution markers are selected; no command-name matching.
+_SCRIPT = r"""import json, os, select, signal, sys, time
+probe = os.pidfd_open(os.getpid())
+os.close(probe)
+needle = b'DEERFLOW_COMMAND_OWNER=' + sys.argv[1].encode('ascii')
+deadline = time.monotonic() + 4
+stopped = set()
+empty = 0
+while time.monotonic() < deadline:
+    owned = []
+    blocked = False
+    for name in os.listdir('/proc'):
+        if not name.isdecimal() or int(name) == os.getpid():
+            continue
+        fd = None
+        matched = False
+        try:
+            fd = os.pidfd_open(int(name))
+            with open('/proc/' + name + '/environ', 'rb') as stream:
+                environment = stream.read()
+            if needle not in environment.split(b'\0'):
+                os.close(fd)
+                continue
+            matched = True
+            signal.pidfd_send_signal(fd, signal.SIGTERM)
+            owned.append((int(name), fd))
+        except OSError as error:
+            if matched and not isinstance(error, ProcessLookupError):
+                blocked = True
+            if fd is not None:
+                os.close(fd)
+    if not owned and not blocked:
+        empty += 1
+        if empty >= 2:
+            break
+        time.sleep(0.05)
+        continue
+    empty = 0
+    time.sleep(0.15)
+    for pid, fd in owned:
+        try:
+            if not select.select([fd], [], [], 0)[0]:
+                signal.pidfd_send_signal(fd, signal.SIGKILL)
+            stopped.add(pid)
+        except ProcessLookupError:
+            pass
+        finally:
+            os.close(fd)
+    time.sleep(0.05)
+print(json.dumps({'stopped': len(stopped), 'remaining': 0 if empty >= 2 else 1}))
+"""
+
+
+def cleanup_command(token: str) -> str:
+    if re.fullmatch(r"[0-9a-f]{32}", token) is None:
+        raise ValueError("Invalid execution marker")
+    return "python3 -c " + shlex.quote("exec(" + repr(_SCRIPT) + ")") + " " + shlex.quote(token)

backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -94,6 +94,7 @@ from deerflow.runtime.goal import (
 )
 from deerflow.runtime.goal_progress import observe_goal_operations
 from deerflow.runtime.keyed_lock import AsyncKeyedLockTable
+from deerflow.runtime.runs.schemas import GOAL_RUN_HARD_STOP_REASONS
 from deerflow.runtime.runs.stream_cleanup import close_agent_stream
 from deerflow.runtime.serialization import serialize
 from deerflow.runtime.stream_bridge import StreamBridge
@@ -1113,6 +1114,14 @@ async def run_agent(
         nonlocal checkpoint_rollback_completed
         if record.ownership_lost:
             return
+        from deerflow.sandbox.lease import cancel_sandbox_execution_lease_async
+
+        cleanup_error = None
+        try:
+            await cancel_sandbox_execution_lease_async(runtime_ctx)
+        except Exception:
+            cleanup_error = "Sandbox process cleanup could not be confirmed"
+            logger.warning("Run %s: %s", run_id, cleanup_error, exc_info=True)
         await _ensure_stream_producer_lease()
         await run_manager.set_finalizing(run_id, True)
         if record.ownership_lost:
@@ -1122,7 +1131,7 @@ async def run_agent(
             await run_manager.set_status(
                 run_id,
                 RunStatus.error,
-                error="Rolled back by user",
+                error=f"Rolled back by user; {cleanup_error}" if cleanup_error else "Rolled back by user",
                 **({"persist": False} if record.rollback_finalization_pending else terminal_status_kwargs),
             )
             if record.ownership_lost:
@@ -1153,6 +1162,7 @@ async def run_agent(
             await run_manager.set_status(
                 run_id,
                 RunStatus.interrupted,
+                error=cleanup_error,
                 **terminal_status_kwargs,
             )
             logger.info("Run %s was cancelled", run_id)
@@ -1776,6 +1786,10 @@ async def run_agent(
             # collects the most severe / first / all reasons) instead of each
             # guard writing directly to the same key.
             stop_reason = runtime_context.get("stop_reason") if runtime_context is not None else None
+            if stop_reason in GOAL_RUN_HARD_STOP_REASONS:
+                from deerflow.sandbox.lease import cancel_sandbox_execution_lease_async
+
+                await cancel_sandbox_execution_lease_async(runtime_ctx)
             produced_output_paths = await _produced_output_paths(
                 pre_run_workspace_snapshot,
                 thread_id=thread_id,
@@ -1803,6 +1817,12 @@ async def run_agent(
         await _finish_cancellation(record.abort_action)
 
     except Exception as exc:
+        try:
+            from deerflow.sandbox.lease import cancel_sandbox_execution_lease_async
+
+            await cancel_sandbox_execution_lease_async(runtime_ctx)
+        except Exception:
+            logger.warning("Failed to stop commands after run failure", exc_info=True)
         checkpoint_resume_rejected = isinstance(exc, CheckpointResumeRejected)
         error_msg = f"{exc}"
         stop_status_kwargs: dict[str, Any] = {}

backend/packages/harness/deerflow/sandbox/AGENTS.md

@@ -2,6 +2,10 @@
 
 Sandbox restore requires a thread ID, even forks.
 
+Signal managed bash before draining cancellation. Stop only the lease owner’s
+processes before outer release; normal completion keeps previews. Replacement
+resets its command generation. Regression: `test_sandbox_command_cancellation.py`.
+
 On startup failure, close reads and kill/reap the child, preserving the error.
 Only `finally` closes write ends.
 

backend/packages/harness/deerflow/sandbox/command_scope.py

@@ -0,0 +1,95 @@
+"""Process ownership for cancellable bash calls; retained only for one execution lease."""
+
+from __future__ import annotations
+
+import threading
+import uuid
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+from typing import Any
+
+COMMAND_OWNER_ENV = "DEERFLOW_COMMAND_OWNER"
+
+
+class CommandScopeCancelled(RuntimeError):
+    """An explicitly cancelled execution must not start or retry commands."""
+
+
+@dataclass
+class CommandScope:
+    token: str = field(default_factory=lambda: uuid.uuid4().hex)
+    cancelled: threading.Event = field(default_factory=threading.Event)
+    cleaned: threading.Event = field(default_factory=threading.Event)
+    lock: threading.Lock = field(default_factory=threading.Lock)
+    processes: list[Any] = field(default_factory=list)
+
+    def check(self) -> None:
+        if self.cancelled.is_set():
+            raise CommandScopeCancelled("Sandbox execution was cancelled")
+
+
+class CommandScopes:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._scopes: dict[str, CommandScope] = {}
+
+    def get(self, owner: str) -> CommandScope:
+        with self._lock:
+            scope = self._scopes.get(owner)
+            if scope is None:
+                scope = self._scopes[owner] = CommandScope()
+            return scope
+
+    def find(self, owner: str) -> CommandScope | None:
+        with self._lock:
+            return self._scopes.get(owner)
+
+    def request_cancel(self, owner: str) -> None:
+        scope = self.get(owner)
+        scope.cancelled.set()
+
+    def release(self, owner: str) -> None:
+        with self._lock:
+            self._scopes.pop(owner, None)
+
+
+_active: ContextVar[tuple[object, CommandScope] | None] = ContextVar("sandbox_command_owner", default=None)
+
+
+@contextmanager
+def command_owner(sandbox: object, owner: str | None):
+    """Bind server-owned execution identity across the synchronous tool bridge."""
+    if not owner or not getattr(sandbox, "supports_command_cancellation", False):
+        yield
+        return
+    scope = sandbox._command_scopes.get(owner)
+    scope.check()
+    token = _active.set((sandbox, scope))
+    try:
+        yield
+    finally:
+        _active.reset(token)
+
+
+def current_command_scope(sandbox: object | None = None) -> CommandScope | None:
+    active = _active.get()
+    if active is None or (sandbox is not None and active[0] is not sandbox):
+        return None
+    return active[1]
+
+
+@contextmanager
+def command_lock(lock, scope: CommandScope | None):
+    """A cancelled queued command must not wait for another execution's I/O."""
+    if scope is None:
+        with lock:
+            yield
+        return
+    while not lock.acquire(timeout=0.1):
+        scope.check()
+    try:
+        scope.check()
+        yield
+    finally:
+        lock.release()

backend/packages/harness/deerflow/sandbox/lease.py

@@ -57,7 +57,7 @@ async def _drain_task_after_cancellation[T](task: asyncio.Task[T]) -> T:
                 return task.result()
 
 
-async def run_sync_lifecycle_operation[T](func: Callable[..., T], /, *args: Any, **kwargs: Any) -> T:
+async def run_sync_lifecycle_operation[T](func: Callable[..., T], /, *args: Any, _on_cancel: Callable[[], None] | None = None, **kwargs: Any) -> T:
     """Run blocking client work without letting cancellation outlive it.
 
     ``asyncio.to_thread`` cannot stop its worker when the awaiting task is
@@ -71,6 +71,11 @@ async def run_sync_lifecycle_operation[T](func: Callable[..., T], /, *args: Any,
     try:
         return await asyncio.shield(operation_task)
     except asyncio.CancelledError as cancellation:
+        if _on_cancel is not None:
+            try:
+                _on_cancel()
+            except Exception:
+                logger.warning("Could not signal cancelled sandbox command", exc_info=True)
         try:
             await _drain_task_after_cancellation(operation_task)
         except Exception:
@@ -648,6 +653,28 @@ class SandboxLeaseManager:
                 release_on_last=acquire_release_on_last,
             )
 
+    def cancel(self, owner_id: str, *, reset: bool = False) -> None:
+        """Stop one execution while retaining its existing sandbox holder."""
+        with self._metadata_lock:
+            binding = self._bindings_by_owner.get(owner_id)
+        if binding is None:
+            return
+        with self._serializer.hold(binding.thread_key):
+            with self._metadata_lock:
+                if self._bindings_by_owner.get(owner_id) is not binding:
+                    return
+            sandbox = self._provider.get(binding.sandbox_id)
+            if sandbox is None:
+                raise SandboxNotFoundError("Sandbox unavailable during cancellation", sandbox_id=binding.sandbox_id)
+            cancel = getattr(sandbox, "cancel_command_scope", None)
+            if callable(cancel):
+                cancel(owner_id)
+            if reset:
+                sandbox.release_command_scope(owner_id)
+
+    async def cancel_async(self, owner_id: str, *, reset: bool = False) -> None:
+        await run_sync_lifecycle_operation(self.cancel, owner_id, reset=reset)
+
     def release(self, owner_id: str) -> None:
         """Release one execution and park the sandbox only after the last user."""
         with self._metadata_lock:
@@ -742,6 +769,7 @@ def discard_sandbox_lease_manager(provider: SandboxProvider) -> None:
 def register_sandbox_prewarm_cleanup(context: dict[str, Any]) -> None:
     """Admit prewarm only inside an unconditional async lease-release finally."""
     context[SANDBOX_PREWARM_CLEANUP_CONTEXT_KEY] = _PREWARM_CLEANUP_MARKER
+    ensure_sandbox_lease_owner(context)
 
 
 def sandbox_prewarm_cleanup_registered(context: Any) -> bool:
@@ -788,6 +816,17 @@ def _sandbox_execution_lease(context: Any) -> tuple[str, str] | None:
     return owner_id, sandbox_id
 
 
+async def cancel_sandbox_execution_lease_async(context: Any, *, reset: bool = False) -> None:
+    """Cancel only the currently bound execution; never initialize a sandbox."""
+    lease = _sandbox_execution_lease(context)
+    if lease is None:
+        return
+    from deerflow.sandbox.sandbox_provider import get_sandbox_provider
+
+    owner_id, _sandbox_id = lease
+    await get_sandbox_lease_manager(get_sandbox_provider()).cancel_async(owner_id, reset=reset)
+
+
 def release_sandbox_execution_lease(context: Any) -> None:
     """Release a lead/embedded execution lease at its outer lifecycle fence."""
     lease = _sandbox_execution_lease(context)

backend/packages/harness/deerflow/sandbox/local/local_sandbox.py

@@ -11,13 +11,15 @@ import stat
 import subprocess
 import sys
 import threading
-from contextlib import suppress
+import time
+from contextlib import nullcontext, suppress
 from dataclasses import dataclass
 from functools import cached_property
 from pathlib import Path, PurePosixPath
 from typing import NamedTuple
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
+from deerflow.sandbox.command_scope import COMMAND_OWNER_ENV, current_command_scope
 from deerflow.sandbox.download import DEFAULT_DOWNLOAD_LIMIT, download_limit
 from deerflow.sandbox.env_policy import build_sandbox_env
 from deerflow.sandbox.local.list_dir import list_dir
@@ -327,6 +329,30 @@ def escape_powershell_path(match: re.Match[str], path: str) -> str:
 
 
 class LocalSandbox(Sandbox):
+    supports_command_cancellation = os.name == "posix"
+
+    def cancel_command_scope(self, scope_id: str) -> None:
+        scope = self._command_scopes.find(scope_id)
+        if scope is None:
+            return
+        self.request_command_cancel(scope_id)
+        with scope.lock:
+            processes = list(scope.processes)
+        for process in processes:
+            if process.poll() is None:
+                self._terminate_process_group(process, process.pid)
+                continue
+            # A returned shell may leave nohup children. Verify the execution
+            # marker in this recorded group before signalling a recycled PGID.
+            members = subprocess.run(["pgrep", "-g", str(process.pid)], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=2)
+            pids = ",".join(x for x in members.stdout.split() if x.isdecimal())
+            if not pids:
+                continue
+            result = subprocess.run(["ps", "eww", "-o", "command=", "-p", pids], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=2)
+            if f"{COMMAND_OWNER_ENV}={scope.token}" in result.stdout:
+                with suppress(ProcessLookupError):
+                    os.killpg(process.pid, signal.SIGKILL)
+
     #: Every call is a fresh ``subprocess.run([shell, "-c", ...])`` process —
     #: no shell state survives into the next command.
     persistent_shell_sessions = False
@@ -1082,20 +1108,29 @@ class LocalSandbox(Sandbox):
 
         Returns ``(stdout, stderr, returncode, timed_out)``.
         """
+        scope = current_command_scope()
+        if scope is not None:
+            scope.check()
+            env = {**(os.environ if env is None else env), COMMAND_OWNER_ENV: scope.token}
         timed_out = False
         stdout_read_fd, stdout_write_fd = os.pipe()
         stderr_read_fd, stderr_write_fd = os.pipe()
         process = None
         try:
-            process = subprocess.Popen(
-                args,
-                shell=False,
-                stdin=subprocess.DEVNULL,
-                stdout=stdout_write_fd,
-                stderr=stderr_write_fd,
-                start_new_session=True,
-                env=env,
-            )
+            with scope.lock if scope is not None else nullcontext():
+                if scope is not None:
+                    scope.check()
+                process = subprocess.Popen(
+                    args,
+                    shell=False,
+                    stdin=subprocess.DEVNULL,
+                    stdout=stdout_write_fd,
+                    stderr=stderr_write_fd,
+                    start_new_session=True,
+                    env=env,
+                )
+                if scope is not None:
+                    scope.processes.append(process)
             # With start_new_session=True, the new process is also the
             # process-group leader, so its PID is the stable group identifier.
             process_group_id = process.pid
@@ -1129,7 +1164,19 @@ class LocalSandbox(Sandbox):
         stderr_capture, stderr_thread = LocalSandbox._start_pipe_drain(stderr_read_fd, "deerflow-bash-stderr-drain")
         try:
             try:
-                process.wait(timeout=timeout)
+                if scope is None:
+                    process.wait(timeout=timeout)
+                else:
+                    deadline = time.monotonic() + timeout
+                    while process.poll() is None:
+                        if scope.cancelled.is_set():
+                            LocalSandbox._terminate_process_group(process, process_group_id)
+                            break
+                        remaining = deadline - time.monotonic()
+                        if remaining <= 0:
+                            raise subprocess.TimeoutExpired(args, timeout)
+                        with suppress(subprocess.TimeoutExpired):
+                            process.wait(timeout=min(0.1, remaining))
             except subprocess.TimeoutExpired:
                 timed_out = True
                 LocalSandbox._terminate_process_group(process, process_group_id)

backend/packages/harness/deerflow/sandbox/middleware.py

@@ -29,6 +29,7 @@ from deerflow.sandbox.lease import (
     get_sandbox_lease_manager,
     run_sync_lifecycle_operation,
     sandbox_lease_owner,
+    sandbox_prewarm_cleanup_registered,
 )
 from deerflow.sandbox.overwrite import unwrap_sandbox
 from deerflow.sandbox.prewarm import close_sandbox_prewarm, start_sandbox_prewarm
@@ -481,6 +482,10 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
     @override
     async def aafter_agent(self, state: SandboxMiddlewareState, runtime: Runtime) -> dict | None:
         await close_sandbox_prewarm(runtime)
+        if sandbox_prewarm_cleanup_registered(runtime.context):
+            # The existing outer execution fence retains this holder through
+            # cancellation and finalization, then performs the ordinary release.
+            return await super().aafter_agent(state, runtime)
         sandbox, fork_restored = unwrap_sandbox(state.get("sandbox"))
         if sandbox is not None:
             sandbox_id = sandbox["sandbox_id"]

backend/packages/harness/deerflow/sandbox/sandbox.py

@@ -1,6 +1,7 @@
 import re
 from abc import ABC, abstractmethod
 
+from deerflow.sandbox.command_scope import CommandScopes
 from deerflow.sandbox.search import GrepMatch
 
 # POSIX env-var name rule: letter or underscore, then letters/digits/underscores.
@@ -77,8 +78,20 @@ class Sandbox(ABC):
     #: silently mounting credentials.
     lark_cli_broker: bool | None = False
 
+    supports_command_cancellation = False
+
     def __init__(self, id: str):
         self._id = id
+        self._command_scopes = CommandScopes()
+
+    def request_command_cancel(self, scope_id: str) -> None:
+        """Signal a cancelled bash execution without blocking the event loop."""
+        if self.supports_command_cancellation:
+            self._command_scopes.request_cancel(scope_id)
+
+    def cancel_command_scope(self, scope_id: str) -> None:
+        """Stop commands owned by this execution; legacy providers keep their behavior."""
+        self.request_command_cancel(scope_id)
 
     @property
     def id(self) -> str:
@@ -139,8 +152,8 @@ class Sandbox(ABC):
         return self.execute_command(command, env=env, timeout=timeout)
 
     def release_command_scope(self, scope_id: str) -> None:
-        """Release provider-specific command state for one execution scope."""
-        del scope_id
+        """Forget completed execution ownership without stopping background services."""
+        self._command_scopes.release(scope_id)
 
     @abstractmethod
     def read_file(

backend/packages/harness/deerflow/sandbox/tools.py

@@ -10,7 +10,7 @@ import string
 from collections.abc import AsyncIterator, Callable, Iterator
 from contextlib import asynccontextmanager, contextmanager
 from contextvars import ContextVar
-from functools import lru_cache
+from functools import lru_cache, partial
 from pathlib import Path
 
 from langchain.tools import tool
@@ -27,6 +27,7 @@ from deerflow.config.paths import VIRTUAL_PATH_PREFIX
 from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH, WRITE_ELISION_PLACEHOLDER_TEMPLATE
 from deerflow.runtime.secret_context import read_active_secrets
 from deerflow.runtime.user_context import resolve_runtime_user_id
+from deerflow.sandbox.command_scope import command_owner
 from deerflow.sandbox.env_policy import is_blocked_env_name
 from deerflow.sandbox.exceptions import (
     SandboxError,
@@ -2325,12 +2326,16 @@ async def _run_sync_tool_after_async_sandbox_init(
     """Initialize lazily via async provider, then run sync tool body off-thread."""
     try:
         async with sandbox_authorization_scope_async(runtime):
-            await ensure_sandbox_initialized_async(runtime)
+            sandbox = await ensure_sandbox_initialized_async(runtime)
 
             if func is None:
                 return "Error: Tool implementation not available"
 
-            return await run_sync_lifecycle_operation(func, runtime, *args)
+            owner = sandbox_lease_owner(runtime.context)
+            signal_cancel = None
+            if func is bash_tool.func and owner and getattr(sandbox, "supports_command_cancellation", False):
+                signal_cancel = partial(sandbox.request_command_cancel, owner)
+            return await run_sync_lifecycle_operation(func, runtime, *args, _on_cancel=signal_cancel)
     except SandboxError as e:
         return f"Error: {e}"
     except Exception as e:
@@ -2346,19 +2351,20 @@ def _execute_bash_command(
     timeout: float | None = None,
 ) -> str:
     """Route subagent bash calls through their isolated shell-session scope."""
-    scope_id = sandbox_command_scope(runtime.context)
-    scoped_execute = getattr(sandbox, "execute_command_in_scope", None)
-    if scope_id is not None and callable(scoped_execute):
-        return scoped_execute(
-            command,
-            env=env,
-            timeout=timeout,
-            scope_id=scope_id,
-        )
-    # Keep duck-typed custom providers and test doubles compatible: the scoped
-    # method is an additive Sandbox API, and ordinary/lead executions retain
-    # the original execute_command path.
-    return sandbox.execute_command(command, env=env, timeout=timeout)
+    with command_owner(sandbox, sandbox_lease_owner(runtime.context)):
+        scope_id = sandbox_command_scope(runtime.context)
+        scoped_execute = getattr(sandbox, "execute_command_in_scope", None)
+        if scope_id is not None and callable(scoped_execute):
+            return scoped_execute(
+                command,
+                env=env,
+                timeout=timeout,
+                scope_id=scope_id,
+            )
+        # Keep duck-typed custom providers and test doubles compatible: the scoped
+        # method is an additive Sandbox API, and ordinary/lead executions retain
+        # the original execute_command path.
+        return sandbox.execute_command(command, env=env, timeout=timeout)
 
 
 def ensure_thread_directories_exist(runtime: Runtime | None) -> None:

backend/packages/harness/deerflow/subagents/executor.py

@@ -2130,9 +2130,13 @@ class SubagentExecutor:
             async def cleanup_execution() -> None:
                 if execution_context is not None:
                     try:
-                        from deerflow.sandbox.lease import release_sandbox_execution_lease_async
+                        from deerflow.sandbox.lease import cancel_sandbox_execution_lease_async, release_sandbox_execution_lease_async
 
-                        await release_sandbox_execution_lease_async(execution_context)
+                        try:
+                            if result.status in (SubagentStatus.CANCELLED, SubagentStatus.FAILED, SubagentStatus.TIMED_OUT):
+                                await cancel_sandbox_execution_lease_async(execution_context)
+                        finally:
+                            await release_sandbox_execution_lease_async(execution_context)
                     except Exception:
                         logger.warning(
                             "[trace=%s] Failed to release sandbox execution lease for subagent %s",

backend/tests/test_github_token_plumbing.py

@@ -42,6 +42,7 @@ def _new_aio_sandbox_with_session_state():
     sbx._session_creation_state_lock = threading.Lock()
     sbx._shell_session_creation_state = _SessionCreationState()
     sbx._bash_session_creation_state = _SessionCreationState()
+    sbx._default_shell_stopped = False
     return sbx
 
 

backend/tests/test_sandbox_command_cancellation.py

@@ -0,0 +1,420 @@
+from __future__ import annotations
+
+import asyncio
+import os
+import shlex
+import signal
+import subprocess
+import sys
+import threading
+from pathlib import Path
+
+import pytest
+
+from deerflow.sandbox.command_scope import command_owner
+from deerflow.sandbox.lease import run_sync_lifecycle_operation
+from deerflow.sandbox.local.local_sandbox import LocalSandbox
+
+
+def _alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+        state = subprocess.run(["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True, check=False).stdout.strip()
+        return bool(state) and not state.startswith("Z")
+    except ProcessLookupError:
+        return False
+
+
+async def _pid(path: Path) -> int:
+    for _ in range(200):
+        if path.exists():
+            return int(path.read_text())
+        await asyncio.sleep(0.01)
+    raise AssertionError("child did not start")
+
+
+def _command(path: Path, *, background: bool) -> str:
+    code = f"import os,time;from pathlib import Path;Path({str(path)!r}).write_text(str(os.getpid()));time.sleep(30)"
+    command = f"{shlex.quote(sys.executable)} -u -c {shlex.quote(code)}"
+    return f"{command} > /dev/null 2>&1 &" if background else command
+
+
+@pytest.mark.skipif(os.name != "posix", reason="POSIX process groups")
+@pytest.mark.asyncio
+async def test_cancel_scope_stops_foreground_and_nohup_without_stopping_previous_preview(tmp_path):
+    sandbox = LocalSandbox("owned-commands")
+    foreground = None
+    preview_pid = None
+    preview_group = None
+    try:
+        with command_owner(sandbox, "previous-run"):
+            sandbox.execute_command(_command(tmp_path / "preview.pid", background=True))
+        preview_pid = await _pid(tmp_path / "preview.pid")
+        preview_group = os.getpgid(preview_pid)
+        sandbox.release_command_scope("previous-run")
+        assert _alive(preview_pid), "ordinary completion must retain preview services"
+
+        with command_owner(sandbox, "current-run"):
+            sandbox.execute_command("nohup " + _command(tmp_path / "background.pid", background=True))
+        background_pid = await _pid(tmp_path / "background.pid")
+
+        async def execute_foreground():
+            with command_owner(sandbox, "current-run"):
+                return await run_sync_lifecycle_operation(
+                    sandbox.execute_command,
+                    _command(tmp_path / "foreground.pid", background=False),
+                    _on_cancel=lambda: sandbox.request_command_cancel("current-run"),
+                )
+
+        foreground = asyncio.create_task(execute_foreground())
+        foreground_pid = await _pid(tmp_path / "foreground.pid")
+        foreground.cancel()
+        foreground.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(foreground, 2)
+        await asyncio.to_thread(sandbox.cancel_command_scope, "current-run")
+        for _ in range(100):
+            if not _alive(foreground_pid) and not _alive(background_pid):
+                break
+            await asyncio.sleep(0.01)
+        assert not _alive(foreground_pid)
+        assert not _alive(background_pid)
+        assert _alive(preview_pid)
+
+        sandbox.release_command_scope("current-run")
+        with command_owner(sandbox, "current-run"):
+            assert sandbox.execute_command("printf resumed") == "resumed"
+        assert _alive(preview_pid)
+    finally:
+        if foreground is not None and not foreground.done():
+            foreground.cancel()
+            await asyncio.gather(foreground, return_exceptions=True)
+        sandbox.cancel_command_scope("current-run")
+        if preview_group is not None:
+            try:
+                os.killpg(preview_group, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+
+
+@pytest.mark.asyncio
+async def test_cancellation_signal_runs_before_worker_drain_and_survives_repeated_cancel():
+    started = threading.Event()
+    stop = threading.Event()
+    finished = threading.Event()
+
+    def work(**_):
+        started.set()
+        stop.wait(2)
+        finished.set()
+
+    task = asyncio.create_task(run_sync_lifecycle_operation(work, _on_cancel=stop.set))
+    try:
+        assert await asyncio.to_thread(started.wait, 1)
+        task.cancel()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert stop.is_set(), "cancel must be delivered before waiting for the blocking operation"
+        assert finished.is_set(), "the worker must still drain before cancellation escapes"
+    finally:
+        stop.set()
+        await asyncio.gather(task, return_exceptions=True)
+
+
+class _Shell:
+    def __init__(self):
+        self.command = None
+        self.killed = []
+        self.calls = 0
+        self.cancel = None
+        self.controls = []
+
+    def exec_command(self, **kwargs):
+        from types import SimpleNamespace as N
+
+        if kwargs["command"].startswith("python3 -c "):
+            self.controls.append(kwargs["id"])
+            return N(data=N(session_id=kwargs["id"], status="running"))
+        self.calls += 1
+        self.command = kwargs["command"]
+        assert kwargs.get("async_mode") is True
+        if self.cancel:
+            self.cancel()
+        return N(data=N(session_id="returned-session", status="running"))
+
+    def wait_for_process(self, **kwargs):
+        from types import SimpleNamespace as N
+
+        return N(data=N(status="completed"))
+
+    def create_session(self, **kwargs):
+        pass
+
+    def cleanup_session(self, *args, **kwargs):
+        pass
+
+    def view(self, **kwargs):
+        from types import SimpleNamespace as N
+
+        if kwargs["id"] in self.controls:
+            return N(data=N(output='{"remaining":0}', exit_code=0, status="completed"))
+        return N(data=N(status="completed", exit_code=7, console=[N(command="previous command", output="OLD_OUTPUT"), N(command=self.command, output="NEW_OUTPUT")]))
+
+    def kill_process(self, **kwargs):
+        from types import SimpleNamespace as N
+
+        self.killed.append(kwargs["id"])
+        return N(success=True)
+
+
+def _aio(monkeypatch):
+    from types import SimpleNamespace as N
+
+    from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox
+
+    shell = _Shell()
+    monkeypatch.setattr("deerflow.community.aio_sandbox.aio_sandbox.AioSandboxClient", lambda **_: N(shell=shell))
+    return AioSandbox("aio", "http://127.0.0.1:9", home_dir="/tmp"), shell
+
+
+def test_aio_owned_command_keeps_only_its_output_and_exit_code(monkeypatch):
+    sandbox, shell = _aio(monkeypatch)
+    with command_owner(sandbox, "run"):
+        result = sandbox.execute_command("same command")
+    assert "NEW_OUTPUT" in result
+    assert "OLD_OUTPUT" not in result
+    assert "Exit Code: 7" in result
+    assert shell.calls == 1
+
+
+def test_aio_cancel_uses_separate_control_without_replaying_or_killing_shared_shell(monkeypatch):
+    sandbox, shell = _aio(monkeypatch)
+    shell.cancel = lambda: sandbox.request_command_cancel("run")
+    with command_owner(sandbox, "run"):
+        result = sandbox.execute_command("long command")
+    assert "cancelled" in result
+    assert shell.killed == []
+    assert len(shell.controls) == 1
+    assert shell.calls == 1
+    assert not sandbox.requires_container_recycle
+
+
+@pytest.mark.asyncio
+async def test_cancelled_aio_waiter_never_starts_or_interrupts_another_scope(monkeypatch):
+    sandbox, shell = _aio(monkeypatch)
+    sandbox._lock.acquire()
+    try:
+
+        async def waiter():
+            with command_owner(sandbox, "queued-run"):
+                return await run_sync_lifecycle_operation(sandbox.execute_command, "queued", _on_cancel=lambda: sandbox.request_command_cancel("queued-run"))
+
+        task = asyncio.create_task(waiter())
+        await asyncio.sleep(0.05)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, 1)
+        assert shell.calls == 0
+        assert shell.killed == []
+    finally:
+        sandbox._lock.release()
+
+
+@pytest.mark.skipif(os.name != "posix", reason="POSIX process groups")
+@pytest.mark.asyncio
+@pytest.mark.parametrize("outcome", ["interrupt", "failure", "loop_capped"])
+async def test_native_bash_to_worker_stops_owned_processes_before_release(monkeypatch, tmp_path, outcome):
+    from types import SimpleNamespace as N
+    from unittest.mock import AsyncMock, MagicMock
+
+    from deerflow.runtime.runs.manager import RunManager
+    from deerflow.runtime.runs.worker import RunContext, run_agent
+    from deerflow.sandbox.lease import get_sandbox_lease_manager
+    from deerflow.sandbox.sandbox_provider import reset_sandbox_provider, set_sandbox_provider
+    from deerflow.sandbox.tools import bash_tool
+
+    sandbox = LocalSandbox("local:worker-cancel")
+    provider = MagicMock()
+    provider.get.return_value = sandbox
+    provider.get_scoped.return_value = sandbox
+    provider.acquire.return_value = sandbox.id
+    provider.acquire_async = AsyncMock(return_value=sandbox.id)
+    monkeypatch.setattr("deerflow.sandbox.tools.is_host_bash_allowed", lambda: True)
+    run_manager = RunManager()
+    record = await run_manager.create("native-bash-cancellation")
+    background = tmp_path / "background.pid"
+    foreground = tmp_path / "foreground.pid"
+    release_seen = []
+
+    def release(_sandbox_id):
+        release_seen.append(True)
+        assert not _alive(int(background.read_text()))
+        if foreground.exists():
+            assert not _alive(int(foreground.read_text()))
+
+    provider.release.side_effect = release
+
+    class Agent:
+        async def astream(self, graph_input, config, **kwargs):
+            context = config["configurable"]["__pregel_runtime"].context
+            state = {"sandbox": {"sandbox_id": sandbox.id}, "thread_data": {key: str(tmp_path) for key in ("workspace_path", "outputs_path", "uploads_path")}}
+            runtime = N(context=context, state=state, config=config)
+            command = "nohup " + _command(background, background=True).replace(shlex.quote(sys.executable), "python3").replace(str(tmp_path), "/mnt/user-data/workspace")
+            result = await bash_tool.coroutine(runtime=runtime, command=command)
+            assert not result.startswith("Error:"), result
+            await _pid(background)
+            if outcome == "interrupt":
+                command = _command(foreground, background=False).replace(shlex.quote(sys.executable), "python3").replace(str(tmp_path), "/mnt/user-data/workspace")
+                await bash_tool.coroutine(runtime=runtime, command=command)
+            elif outcome == "failure":
+                raise RuntimeError("scripted model failure after starting a process")
+            else:
+                context["stop_reason"] = outcome
+            yield {"messages": []}
+
+    set_sandbox_provider(provider)
+    task = asyncio.create_task(run_agent(N(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock()), run_manager, record, ctx=RunContext(checkpointer=None), agent_factory=lambda **_: Agent(), graph_input={}, config={}))
+    record.task = task
+    try:
+        await _pid(background)
+        if outcome == "interrupt":
+            await _pid(foreground)
+            await run_manager.cancel(record.run_id)
+        await asyncio.wait_for(task, 5)
+        assert release_seen == [True]
+        assert record.status.value == ("interrupted" if outcome == "interrupt" else "error" if outcome == "failure" else "success")
+    finally:
+        if not task.done():
+            task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+        for path in (background, foreground):
+            if path.exists() and _alive(int(path.read_text())):
+                os.killpg(os.getpgid(int(path.read_text())), signal.SIGKILL)
+        get_sandbox_lease_manager(provider).close()
+        reset_sandbox_provider()
+
+
+def test_aio_failed_cleanup_never_claims_clean_or_allows_warm_reuse(monkeypatch):
+    from types import SimpleNamespace as N
+
+    sandbox, shell = _aio(monkeypatch)
+    with command_owner(sandbox, "run"):
+        assert "NEW_OUTPUT" in sandbox.execute_command("echo done")
+    monkeypatch.setattr(sandbox, "_create_shell_session", lambda _: "control")
+    monkeypatch.setattr(sandbox, "_cleanup_session_best_effort", lambda *_, **__: None)
+    monkeypatch.setattr(shell, "exec_command", lambda **_: N(data=N(session_id="control", status="running")))
+    monkeypatch.setattr(shell, "view", lambda **_: N(data=N(output='{"remaining":1}', exit_code=0, status="completed")))
+    with pytest.raises(RuntimeError, match="did not stop"):
+        sandbox.cancel_command_scope("run")
+    assert sandbox.requires_container_recycle
+
+
+@pytest.mark.parametrize("cancel", [False, True])
+def test_aio_env_command_keeps_output_exit_and_cleans_its_session(monkeypatch, cancel):
+    from types import SimpleNamespace as N
+
+    sandbox, shell = _aio(monkeypatch)
+    closed = []
+
+    class Bash:
+        calls = 0
+
+        def create_session(self, **kwargs):
+            pass
+
+        def exec(self, **kwargs):
+            assert kwargs["async_mode"]
+            assert kwargs["env"] == {"FAKE_KEY": "test-only"}
+            self.calls += 1
+            if cancel:
+                sandbox.request_command_cancel("run")
+            return N(data=N(command_id="actual-command"))
+
+        def output(self, **kwargs):
+            assert kwargs["command_id"] == "actual-command"
+            return N(data=N(stdout="中文 output", stderr="diagnostic", offset=12, stderr_offset=10, command=N(status="completed", exit_code=7)))
+
+        def close_session(self, session_id, **kwargs):
+            closed.append(session_id)
+
+    bash = Bash()
+    sandbox._client.bash = bash
+    with command_owner(sandbox, "run"):
+        result = sandbox.execute_command("echo result", env={"FAKE_KEY": "test-only"})
+    assert bash.calls == 1
+    assert len(closed) == 1
+    if cancel:
+        assert "cancelled" in result
+        assert len(shell.controls) == 1
+    else:
+        assert "中文 output" in result and "diagnostic" in result and "Exit Code: 7" in result
+        assert shell.controls == []
+    assert not sandbox.requires_container_recycle
+
+
+def test_aio_cleanup_does_not_report_success_without_pidfd_support():
+    from deerflow.community.aio_sandbox.command_cleanup import cleanup_command
+
+    command = shlex.split(cleanup_command("a" * 32))
+    # Run the actual controller with an unsupported-kernel response; no real
+    # process is inspected or signalled on any platform.
+    bootstrap = "import json,os,select,signal,sys,time,errno; os.listdir=lambda _:['123']; os.pidfd_open=lambda _:(_ for _ in ()).throw(OSError(errno.ENOSYS,'pidfd unavailable')); script=sys.argv.pop(1); exec(script)"
+    result = subprocess.run([sys.executable, "-c", bootstrap, command[2], command[3]], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=2)
+    assert result.returncode != 0
+    assert "pidfd unavailable" in result.stderr
+    assert '"remaining": 0' not in result.stdout
+
+
+@pytest.mark.parametrize("case", ["missing-command", "missing-stdout-cursor", "missing-stderr-cursor", "completed-without-cursors", "empty-poll"])
+def test_aio_owned_bash_handles_optional_output_fields_without_unsafe_reuse(monkeypatch, case):
+    from types import SimpleNamespace as N
+    from unittest.mock import MagicMock
+
+    from agent_sandbox.types.bash_command_info import BashCommandInfo
+    from agent_sandbox.types.bash_output_result import BashOutputResult
+
+    sandbox, _ = _aio(monkeypatch)
+    bash = MagicMock()
+    bash.exec.return_value = N(data=N(command_id="owned-command"))
+    completed = {"stdout": "done", "stderr": "diagnostic", "status": "completed"}
+    frames = {
+        "missing-command": [{"stdout": "partial", "command_missing": True}],
+        "missing-stdout-cursor": [{"stdout": "partial", "stderr_offset": 0}],
+        "missing-stderr-cursor": [{"stderr": "partial", "offset": 0}],
+        "completed-without-cursors": [completed],
+        "empty-poll": [{"stdout": "one", "stderr": "two", "offset": 3, "stderr_offset": 3}, {}, completed],
+    }[case]
+    # A second response makes an erroneous continuation deterministic instead
+    # of leaving this regression in a polling loop until its timeout.
+    frames = [*frames, completed]
+    observed = []
+
+    def output(**kwargs):
+        observed.append(kwargs)
+        frame = dict(frames[len(observed) - 1])
+        missing = frame.pop("command_missing", False)
+        status = frame.pop("status", "running")
+        command = None if missing else BashCommandInfo(command_id="owned-command", command="echo result", status=status, exit_code=7 if status == "completed" else None)
+        return N(data=BashOutputResult(session_id=kwargs["session_id"], command=command, **frame))
+
+    bash.output.side_effect = output
+    sandbox._client.bash = bash
+    with command_owner(sandbox, "run"):
+        result = sandbox.execute_command("echo result", env={"FAKE_KEY": "test-only"})
+
+    bash.exec.assert_called_once()
+    bash.close_session.assert_called_once()
+    if case.startswith("missing-"):
+        assert result.startswith("Error: AIO bash")
+        assert sandbox.requires_container_recycle
+        assert len(observed) == 1
+    else:
+        assert "done" in result and "diagnostic" in result and "Exit Code: 7" in result
+        assert not sandbox.requires_container_recycle
+        if case == "empty-poll":
+            assert len(observed) == 3
+            assert [(row["offset"], row["stderr_offset"]) for row in observed] == [(0, 0), (3, 3), (3, 3)]
+            assert result.startswith("onedone\n") and "twodiagnostic" in result
+        else:
+            assert len(observed) == 1