feat: add managed local shell sessions (#9470)

2026-08-31 01:40:25 +08:00 · 2026-07-30 23:05:39 +08:00
parent ea1a80ab7f
commit d749d6fd41
11 changed files with 1121 additions and 17 deletions
@@ -41,8 +41,10 @@ from astrbot.core.tools.computer_tools import (
    FileUploadTool,
    FileWriteTool,
    GrepTool,
+    LocalExecuteShellTool,
    LocalPythonTool,
    PythonTool,
+    ShellSessionTool,
 )
 from astrbot.core.tools.message_tools import SendMessageToUserTool
 from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
@@ -224,7 +226,8 @@ class FunctionToolExecutor(BaseFunctionToolExecutor[AstrAgentContext]):
                )
            return tools
        if runtime == "local":
-            shell_tool = tool_mgr.get_builtin_tool(ExecuteShellTool)
+            shell_tool = LocalExecuteShellTool()
+            shell_session_tool = tool_mgr.get_builtin_tool(ShellSessionTool)
            python_tool = tool_mgr.get_builtin_tool(LocalPythonTool)
            read_tool = tool_mgr.get_builtin_tool(FileReadTool)
            write_tool = tool_mgr.get_builtin_tool(FileWriteTool)
@@ -232,6 +235,7 @@ class FunctionToolExecutor(BaseFunctionToolExecutor[AstrAgentContext]):
            grep_tool = tool_mgr.get_builtin_tool(GrepTool)
            return {
                shell_tool.name: shell_tool,
+                shell_session_tool.name: shell_session_tool,
                python_tool.name: python_tool,
                read_tool.name: read_tool,
                write_tool.name: write_tool,
@@ -70,11 +70,13 @@ from astrbot.core.tools.computer_tools import (
    GrepTool,
    ListSkillCandidatesTool,
    ListSkillReleasesTool,
+    LocalExecuteShellTool,
    LocalPythonTool,
    PromoteSkillCandidateTool,
    PythonTool,
    RollbackSkillReleaseTool,
    RunBrowserSkillTool,
+    ShellSessionTool,
    SyncSkillReleaseTool,
 )
 from astrbot.core.tools.cron_tools import FutureTaskTool
@@ -430,7 +432,8 @@ def _apply_local_env_tools(req: ProviderRequest, plugin_context: Context) -> Non
    if req.func_tool is None:
        req.func_tool = ToolSet()
    tool_mgr = plugin_context.get_llm_tool_manager()
-    req.func_tool.add_tool(tool_mgr.get_builtin_tool(ExecuteShellTool))
+    req.func_tool.add_tool(LocalExecuteShellTool())
+    req.func_tool.add_tool(tool_mgr.get_builtin_tool(ShellSessionTool))
    req.func_tool.add_tool(tool_mgr.get_builtin_tool(LocalPythonTool))
    req.func_tool.add_tool(tool_mgr.get_builtin_tool(FileReadTool))
    req.func_tool.add_tool(tool_mgr.get_builtin_tool(FileWriteTool))
@@ -450,7 +453,11 @@ def _build_local_mode_prompt() -> str:
    return (
        "You have access to the host local environment and can execute shell commands and Python code. "
        f"Current operating system: {system_name}. "
-        f"{shell_hint}"
+        f"{shell_hint} "
+        "Local shell commands automatically return a managed session when they "
+        "outlive the initial wait. Use `astrbot_shell_session` to list, poll, "
+        "write to, interrupt, or terminate those sessions. Do not add `&`, "
+        "`nohup`, or another detachment wrapper for ordinary long-running commands."
    )


@@ -1,12 +1,17 @@
 from __future__ import annotations

 import asyncio
+import hashlib
 import locale
 import os
 import shutil
+import signal
 import subprocess
 import sys
-from dataclasses import dataclass
+import time
+import uuid
+from dataclasses import dataclass, field
+from pathlib import Path
 from typing import Any

 if sys.version_info < (3, 14):
@@ -17,7 +22,10 @@ from astrbot.core.computer.file_read_utils import (
    detect_text_encoding,
    read_local_text_range_sync,
 )
-from astrbot.core.utils.astrbot_path import get_astrbot_root
+from astrbot.core.utils.astrbot_path import (
+    get_astrbot_root,
+    get_astrbot_system_tmp_path,
+)

 from ..olayer import FileSystemComponent, PythonComponent, ShellComponent
 from .base import ComputerBooter
@@ -84,8 +92,37 @@ def _decode_shell_output(output: bytes | None) -> str:
    return _decode_bytes_with_fallback(output, preferred_encoding="utf-8")


+@dataclass
+class _LocalShellSession:
+    """Runtime state for one managed local shell process."""
+
+    session_id: str
+    owner_id: str
+    process: asyncio.subprocess.Process
+    output_path: Path
+    started_at: float
+    output_event: asyncio.Event
+    reader_task: asyncio.Task[None]
+    wait_task: asyncio.Task[int]
+    timeout_task: asyncio.Task[None] | None = None
+    cursor: int = 0
+    timed_out: bool = False
+    terminated: bool = False
+
+
@dataclass
 class LocalShellComponent(ShellComponent):
+    _sessions: dict[str, _LocalShellSession] = field(
+        default_factory=dict,
+        init=False,
+        repr=False,
+    )
+    _sessions_lock: asyncio.Lock = field(
+        default_factory=asyncio.Lock,
+        init=False,
+        repr=False,
+    )
+
    async def exec(
        self,
        command: str,
@@ -160,6 +197,528 @@ class LocalShellComponent(ShellComponent):

        return await asyncio.to_thread(_run)

+    async def exec_managed(
+        self,
+        command: str,
+        *,
+        owner_id: str,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        timeout: int | None = None,
+        yield_time_ms: int = 10_000,
+        max_output_chars: int = 10_000,
+    ) -> dict[str, Any]:
+        """Start a locally managed shell process and briefly wait for it.
+
+        Args:
+            command: Shell command to execute.
+            owner_id: Unified message origin that owns the process.
+            cwd: Working directory for the process.
+            env: Additional environment variables.
+            timeout: Hard process lifetime in seconds. None disables it.
+            yield_time_ms: Maximum time to wait before returning a session ID.
+            max_output_chars: Maximum output bytes returned in this call.
+
+        Returns:
+            Process result with output, status, and session metadata.
+
+        Raises:
+            PermissionError: If the command matches a blocked pattern.
+            ValueError: If a timing or output limit is invalid.
+        """
+        if not _is_safe_command(command):
+            raise PermissionError("Blocked unsafe shell command.")
+        if yield_time_ms < 0 or yield_time_ms > 30_000:
+            raise ValueError("`yield_time_ms` must be between 0 and 30000.")
+        if timeout is not None and timeout <= 0:
+            raise ValueError("`timeout` must be greater than 0 when provided.")
+        if max_output_chars < 1:
+            raise ValueError("`max_output_chars` must be greater than 0.")
+
+        run_env = os.environ.copy()
+        if env:
+            run_env.update({str(k): str(v) for k, v in env.items()})
+        working_dir = Path(cwd).resolve() if cwd else Path(get_astrbot_root()).resolve()
+        session_id = f"sh_{uuid.uuid4().hex[:16]}"
+        owner_digest = hashlib.sha256(owner_id.encode("utf-8")).hexdigest()[:16]
+        output_dir = Path(get_astrbot_system_tmp_path()) / "shell" / owner_digest
+        output_dir.mkdir(parents=True, exist_ok=True)
+        output_path = output_dir / f"{session_id}.log"
+        output_path.touch()
+
+        process_kwargs: dict[str, Any] = {}
+        if os.name == "nt":
+            process_kwargs["creationflags"] = getattr(
+                subprocess,
+                "CREATE_NEW_PROCESS_GROUP",
+                0,
+            )
+        else:
+            process_kwargs["start_new_session"] = True
+
+        try:
+            process = await asyncio.create_subprocess_shell(
+                command,
+                cwd=working_dir,
+                env=run_env,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.STDOUT,
+                **process_kwargs,
+            )
+        except Exception:
+            output_path.unlink(missing_ok=True)
+            raise
+
+        output_event = asyncio.Event()
+
+        async def _capture_output() -> None:
+            if process.stdout is None:
+                return
+            with output_path.open("ab") as output_file:
+                while chunk := await process.stdout.read(8192):
+                    output_file.write(chunk)
+                    output_file.flush()
+                    output_event.set()
+
+        reader_task = asyncio.create_task(
+            _capture_output(),
+            name=f"local_shell_output_{session_id}",
+        )
+        wait_task = asyncio.create_task(
+            process.wait(),
+            name=f"local_shell_wait_{session_id}",
+        )
+        wait_task.add_done_callback(lambda _: output_event.set())
+        session = _LocalShellSession(
+            session_id=session_id,
+            owner_id=owner_id,
+            process=process,
+            output_path=output_path,
+            started_at=time.time(),
+            output_event=output_event,
+            reader_task=reader_task,
+            wait_task=wait_task,
+        )
+
+        if timeout is not None:
+
+            async def _enforce_timeout() -> None:
+                try:
+                    await asyncio.wait_for(
+                        asyncio.shield(wait_task),
+                        timeout=timeout,
+                    )
+                except asyncio.TimeoutError:
+                    session.timed_out = True
+                    logger.warning(
+                        "Managed local shell session timed out: session_id=%s pid=%s",
+                        session_id,
+                        process.pid,
+                    )
+                    await self._terminate_process(session)
+
+            session.timeout_task = asyncio.create_task(
+                _enforce_timeout(),
+                name=f"local_shell_timeout_{session_id}",
+            )
+
+        async with self._sessions_lock:
+            self._sessions[session_id] = session
+
+        if yield_time_ms > 0:
+            try:
+                await asyncio.wait_for(
+                    asyncio.shield(wait_task),
+                    timeout=yield_time_ms / 1000,
+                )
+            except asyncio.TimeoutError:
+                pass
+
+        return await self.poll_session(
+            owner_id=owner_id,
+            session_id=session_id,
+            cursor=0,
+            yield_time_ms=0,
+            max_output_chars=max_output_chars,
+        )
+
+    async def list_sessions(self, owner_id: str) -> dict[str, Any]:
+        """List managed shell sessions owned by one conversation.
+
+        Args:
+            owner_id: Unified message origin that owns the sessions.
+
+        Returns:
+            Session summaries scoped to the owner.
+        """
+        async with self._sessions_lock:
+            sessions = [
+                session
+                for session in self._sessions.values()
+                if session.owner_id == owner_id
+            ]
+
+        items = []
+        for session in sessions:
+            exit_code = session.process.returncode
+            status = (
+                "running"
+                if exit_code is None
+                else (
+                    "timed_out"
+                    if session.timed_out
+                    else (
+                        "terminated"
+                        if session.terminated
+                        else ("completed" if exit_code == 0 else "failed")
+                    )
+                )
+            )
+            try:
+                output_size = session.output_path.stat().st_size
+            except OSError:
+                output_size = session.cursor
+            items.append(
+                {
+                    "session_id": session.session_id,
+                    "pid": session.process.pid,
+                    "status": status,
+                    "exit_code": exit_code,
+                    "started_at": session.started_at,
+                    "unread_output_bytes": max(output_size - session.cursor, 0),
+                }
+            )
+        return {"sessions": items}
+
+    async def poll_session(
+        self,
+        *,
+        owner_id: str,
+        session_id: str,
+        cursor: int | None = None,
+        yield_time_ms: int = 0,
+        max_output_chars: int = 10_000,
+    ) -> dict[str, Any]:
+        """Read new output and status from a managed shell session.
+
+        Args:
+            owner_id: Unified message origin that owns the session.
+            session_id: Managed shell session identifier.
+            cursor: Byte offset to read from. Defaults to the last returned offset.
+            yield_time_ms: Maximum wait for new output or process completion.
+            max_output_chars: Maximum output bytes returned in this call.
+
+        Returns:
+            Incremental output, next cursor, process status, and exit code.
+
+        Raises:
+            ValueError: If the session is unavailable or an argument is invalid.
+        """
+        if yield_time_ms < 0 or yield_time_ms > 30_000:
+            raise ValueError("`yield_time_ms` must be between 0 and 30000.")
+        if max_output_chars < 1:
+            raise ValueError("`max_output_chars` must be greater than 0.")
+
+        session = await self._get_owned_session(owner_id, session_id)
+        read_cursor = session.cursor if cursor is None else cursor
+        if read_cursor < 0:
+            raise ValueError("`cursor` must be greater than or equal to 0.")
+
+        def _read_output() -> tuple[bytes, int, int]:
+            try:
+                output_size = session.output_path.stat().st_size
+            except FileNotFoundError:
+                return b"", read_cursor, read_cursor
+            normalized_cursor = min(read_cursor, output_size)
+            with session.output_path.open("rb") as output_file:
+                output_file.seek(normalized_cursor)
+                raw_output = output_file.read(max_output_chars)
+            return (
+                raw_output,
+                normalized_cursor + len(raw_output),
+                output_size,
+            )
+
+        if session.wait_task.done():
+            await session.reader_task
+        raw_output, next_cursor, output_size = await asyncio.to_thread(_read_output)
+
+        if not raw_output and session.process.returncode is None and yield_time_ms > 0:
+            session.output_event.clear()
+            raw_output, next_cursor, output_size = await asyncio.to_thread(_read_output)
+            if not raw_output and session.process.returncode is None:
+                output_waiter = asyncio.create_task(session.output_event.wait())
+                done, _ = await asyncio.wait(
+                    {output_waiter, session.wait_task},
+                    timeout=yield_time_ms / 1000,
+                    return_when=asyncio.FIRST_COMPLETED,
+                )
+                if output_waiter not in done:
+                    output_waiter.cancel()
+                    try:
+                        await output_waiter
+                    except asyncio.CancelledError:
+                        pass
+                if session.wait_task.done():
+                    await session.reader_task
+                raw_output, next_cursor, output_size = await asyncio.to_thread(
+                    _read_output
+                )
+
+        exit_code = session.process.returncode
+        if exit_code is not None:
+            await session.reader_task
+            raw_output, next_cursor, output_size = await asyncio.to_thread(_read_output)
+
+        exit_code = session.process.returncode
+        if exit_code is not None and not session.reader_task.done():
+            await session.reader_task
+            raw_output, next_cursor, output_size = await asyncio.to_thread(_read_output)
+
+        session.cursor = next_cursor
+        status = (
+            "running"
+            if exit_code is None
+            else (
+                "timed_out"
+                if session.timed_out
+                else (
+                    "terminated"
+                    if session.terminated
+                    else ("completed" if exit_code == 0 else "failed")
+                )
+            )
+        )
+        has_more = next_cursor < output_size
+        session_closed = exit_code is not None and not has_more
+        result = {
+            "session_id": session.session_id,
+            "pid": session.process.pid,
+            "status": status,
+            "stdout": _decode_shell_output(raw_output),
+            "stderr": "",
+            "exit_code": exit_code,
+            "cursor": next_cursor,
+            "has_more": has_more,
+            "session_closed": session_closed,
+        }
+        if session_closed:
+            await self._remove_session(session)
+        return result
+
+    async def write_session(
+        self,
+        *,
+        owner_id: str,
+        session_id: str,
+        chars: str,
+    ) -> dict[str, Any]:
+        """Write text to the stdin pipe of a managed shell session.
+
+        Args:
+            owner_id: Unified message origin that owns the session.
+            session_id: Managed shell session identifier.
+            chars: Text to write verbatim.
+
+        Returns:
+            Current process status after the write.
+
+        Raises:
+            ValueError: If the session is unavailable or no longer accepts input.
+        """
+        session = await self._get_owned_session(owner_id, session_id)
+        if session.process.returncode is not None or session.process.stdin is None:
+            raise ValueError(f"Shell session {session_id} is not accepting input.")
+        session.process.stdin.write(chars.encode("utf-8"))
+        await session.process.stdin.drain()
+        return {
+            "session_id": session_id,
+            "pid": session.process.pid,
+            "status": "running",
+            "written_chars": len(chars),
+        }
+
+    async def interrupt_session(
+        self,
+        *,
+        owner_id: str,
+        session_id: str,
+        yield_time_ms: int = 1_000,
+        max_output_chars: int = 10_000,
+    ) -> dict[str, Any]:
+        """Send an interrupt signal to a managed shell process group.
+
+        Args:
+            owner_id: Unified message origin that owns the session.
+            session_id: Managed shell session identifier.
+            yield_time_ms: Maximum wait for output or exit after the signal.
+            max_output_chars: Maximum output bytes returned after the signal.
+
+        Returns:
+            Incremental output and status after sending the interrupt.
+        """
+        session = await self._get_owned_session(owner_id, session_id)
+        if session.process.returncode is None:
+            if os.name == "nt":
+                session.process.send_signal(
+                    getattr(signal, "CTRL_BREAK_EVENT", signal.SIGTERM)
+                )
+            else:
+                try:
+                    os.killpg(session.process.pid, signal.SIGINT)
+                except ProcessLookupError:
+                    pass
+        return await self.poll_session(
+            owner_id=owner_id,
+            session_id=session_id,
+            yield_time_ms=yield_time_ms,
+            max_output_chars=max_output_chars,
+        )
+
+    async def terminate_session(
+        self,
+        *,
+        owner_id: str,
+        session_id: str,
+        max_output_chars: int = 10_000,
+    ) -> dict[str, Any]:
+        """Terminate a managed shell process group.
+
+        Args:
+            owner_id: Unified message origin that owns the session.
+            session_id: Managed shell session identifier.
+            max_output_chars: Maximum remaining output bytes to return.
+
+        Returns:
+            Remaining output and final process status.
+        """
+        session = await self._get_owned_session(owner_id, session_id)
+        session.terminated = True
+        await self._terminate_process(session)
+        return await self.poll_session(
+            owner_id=owner_id,
+            session_id=session_id,
+            yield_time_ms=0,
+            max_output_chars=max_output_chars,
+        )
+
+    async def shutdown_sessions(self) -> None:
+        """Terminate and remove every managed local shell session."""
+        async with self._sessions_lock:
+            sessions = list(self._sessions.values())
+        for session in sessions:
+            session.terminated = True
+        termination_results = await asyncio.gather(
+            *(self._terminate_process(session) for session in sessions),
+            return_exceptions=True,
+        )
+        for session, result in zip(sessions, termination_results, strict=True):
+            if isinstance(result, BaseException):
+                logger.warning(
+                    "Failed to terminate managed local shell session %s: %s",
+                    session.session_id,
+                    result,
+                )
+        await asyncio.gather(
+            *(session.reader_task for session in sessions),
+            return_exceptions=True,
+        )
+        for session in sessions:
+            await self._remove_session(session)
+
+    async def _get_owned_session(
+        self,
+        owner_id: str,
+        session_id: str,
+    ) -> _LocalShellSession:
+        """Resolve a shell session while enforcing conversation ownership.
+
+        Args:
+            owner_id: Unified message origin that must own the session.
+            session_id: Managed shell session identifier.
+
+        Returns:
+            Matching managed shell session.
+
+        Raises:
+            ValueError: If the session does not exist for this owner.
+        """
+        async with self._sessions_lock:
+            session = self._sessions.get(session_id)
+        if session is None or session.owner_id != owner_id:
+            raise ValueError(f"Shell session {session_id} was not found.")
+        return session
+
+    async def _terminate_process(self, session: _LocalShellSession) -> None:
+        """Gracefully terminate a process group, then force it if needed.
+
+        Args:
+            session: Managed shell session to terminate.
+        """
+        if session.process.returncode is not None:
+            return
+        if os.name == "nt":
+            try:
+                taskkill_result = await asyncio.to_thread(
+                    subprocess.run,
+                    ["taskkill", "/F", "/T", "/PID", str(session.process.pid)],
+                    stdout=subprocess.DEVNULL,
+                    stderr=subprocess.DEVNULL,
+                    timeout=5,
+                )
+            except Exception:
+                session.process.terminate()
+            else:
+                if taskkill_result.returncode != 0:
+                    session.process.terminate()
+        else:
+            try:
+                os.killpg(session.process.pid, signal.SIGTERM)
+            except ProcessLookupError:
+                pass
+
+        try:
+            await asyncio.wait_for(
+                asyncio.shield(session.wait_task),
+                timeout=5,
+            )
+        except asyncio.TimeoutError:
+            if os.name == "nt":
+                session.process.kill()
+            else:
+                try:
+                    os.killpg(session.process.pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+            await session.wait_task
+
+    async def _remove_session(self, session: _LocalShellSession) -> None:
+        """Remove a completed session and its temporary output file.
+
+        Args:
+            session: Managed shell session to remove.
+        """
+        async with self._sessions_lock:
+            if self._sessions.get(session.session_id) is session:
+                self._sessions.pop(session.session_id, None)
+        timeout_task = session.timeout_task
+        if (
+            timeout_task is not None
+            and timeout_task is not asyncio.current_task()
+            and not timeout_task.done()
+        ):
+            timeout_task.cancel()
+            try:
+                await timeout_task
+            except asyncio.CancelledError:
+                pass
+        session.output_path.unlink(missing_ok=True)
+        try:
+            session.output_path.parent.rmdir()
+        except OSError:
+            pass
+

@dataclass
 class LocalPythonComponent(PythonComponent):
@@ -411,6 +970,7 @@ class LocalBooter(ComputerBooter):
        logger.info(f"Local computer booter initialized for session: {session_id}")

    async def shutdown(self) -> None:
+        await self._shell.shutdown_sessions()
        logger.info("Local computer booter shutdown complete.")

    @property
@@ -680,3 +680,16 @@ def get_local_booter() -> ComputerBooter:
    if local_booter is None:
        local_booter = LocalBooter()
    return local_booter
+
+
+async def shutdown_local_booter() -> None:
+    """Shut down managed local computer resources without creating a booter."""
+    global local_booter
+    if local_booter is None:
+        return
+    booter = local_booter
+    local_booter = None
+    try:
+        await booter.shutdown()
+    except Exception as exc:
+        logger.warning("[Computer] Failed to shut down local booter: %s", exc)
@@ -19,6 +19,7 @@ from asyncio import Queue
 from astrbot.api import logger, sp
 from astrbot.core import LogBroker, LogManager
 from astrbot.core.astrbot_config_mgr import AstrBotConfigManager
+from astrbot.core.computer.computer_client import shutdown_local_booter
 from astrbot.core.config.default import VERSION
 from astrbot.core.conversation_mgr import ConversationManager
 from astrbot.core.cron import CronJobManager
@@ -387,6 +388,8 @@ class AstrBotCoreLifecycle:
        if self.cron_manager:
            await self.cron_manager.shutdown()

+        await shutdown_local_booter()
+
        for plugin in self.plugin_manager.context.get_all_stars():
            try:
                await self.plugin_manager._terminate_plugin(plugin)
@@ -418,6 +421,7 @@ class AstrBotCoreLifecycle:

    async def restart(self) -> None:
        """重启 AstrBot 核心生命周期管理类, 终止各个管理器并重新加载平台实例"""
+        await shutdown_local_booter()
        await self.provider_manager.terminate()
        await self.platform_manager.terminate()
        await self.kb_manager.terminate()
@@ -12,7 +12,7 @@ from .fs import (
    GrepTool,
 )
 from .python import LocalPythonTool, PythonTool
-from .shell import ExecuteShellTool
+from .shell import ExecuteShellTool, LocalExecuteShellTool, ShellSessionTool
 from .shipyard_neo import (
    AnnotateExecutionTool,
    BrowserBatchExecTool,
@@ -52,11 +52,13 @@ __all__ = [
    "GrepTool",
    "ListSkillCandidatesTool",
    "ListSkillReleasesTool",
+    "LocalExecuteShellTool",
    "LocalPythonTool",
    "PromoteSkillCandidateTool",
    "PythonTool",
    "RollbackSkillReleaseTool",
    "RunBrowserSkillTool",
+    "ShellSessionTool",
    "SyncSkillReleaseTool",
    "normalize_umo_for_workspace",
    "check_admin_permission",
@@ -10,6 +10,7 @@ from astrbot.api import FunctionTool
 from astrbot.core.agent.run_context import ContextWrapper
 from astrbot.core.agent.tool import ToolExecResult
 from astrbot.core.astr_agent_context import AstrAgentContext
+from astrbot.core.computer.booters.local import LocalShellComponent
 from astrbot.core.computer.computer_client import get_booter
 from astrbot.core.utils.astrbot_path import get_astrbot_system_tmp_path

@@ -23,6 +24,9 @@ from .util import (
 _COMPUTER_RUNTIME_TOOL_CONFIG = {
    "provider_settings.computer_use_runtime": ("local", "sandbox"),
 }
+_LOCAL_RUNTIME_TOOL_CONFIG = {
+    "provider_settings.computer_use_runtime": "local",
+}


 def _quote_redirect_path(path: str, *, local_runtime: bool) -> str:
@@ -90,8 +94,9 @@ class ExecuteShellTool(FunctionTool):
        context: ContextWrapper[AstrAgentContext],
        command: str,
        background: bool = False,
-        timeout: int | None = 300,
+        timeout: int | None = None,
        env: dict[str, Any] | None = None,
+        yield_time_ms: int = 10_000,
    ) -> ToolExecResult:
        if permission_error := check_admin_permission(context, "Shell execution"):
            return permission_error
@@ -102,17 +107,34 @@ class ExecuteShellTool(FunctionTool):
        )
        try:
            cwd: str | None = None
-            if is_local_runtime(context):
+            local_runtime = is_local_runtime(context)
+            if local_runtime:
                current_workspace_root = await workspace_root_for_context(context)
                current_workspace_root.mkdir(parents=True, exist_ok=True)
                cwd = str(current_workspace_root)

            env = dict(env or {})
+            if local_runtime:
+                if not isinstance(sb.shell, LocalShellComponent):
+                    return (
+                        "Error executing command: local shell component is unavailable."
+                    )
+                return json.dumps(
+                    await sb.shell.exec_managed(
+                        command,
+                        owner_id=context.context.event.unified_msg_origin,
+                        cwd=cwd,
+                        env=env,
+                        timeout=timeout,
+                        yield_time_ms=0 if background else yield_time_ms,
+                    ),
+                    ensure_ascii=False,
+                )
+
            effective_background = background and not _is_self_detached_command(command)

            stdout_file: str | None = None
            if effective_background:
-                local_runtime = is_local_runtime(context)
                stdout_file = _build_background_output_path(
                    local_runtime=local_runtime,
                )
@@ -140,6 +162,211 @@ class ExecuteShellTool(FunctionTool):
            return f"Error executing command: {detail}"


+@dataclass
+class LocalExecuteShellTool(ExecuteShellTool):
+    """Local shell tool that automatically yields long-running commands."""
+
+    description: str = (
+        "Execute a command in the shell. If it is still running after "
+        "yield_time_ms, the tool returns a managed shell session ID."
+    )
+    parameters: dict = field(
+        default_factory=lambda: {
+            "type": "object",
+            "properties": {
+                "command": {
+                    "type": "string",
+                    "description": "The shell command to execute in the current workspace.",
+                },
+                "yield_time_ms": {
+                    "type": "integer",
+                    "description": "Maximum time to wait for completion before returning a managed shell session. This does not stop the process.",
+                    "default": 10000,
+                    "minimum": 0,
+                    "maximum": 30000,
+                },
+                "timeout": {
+                    "type": "integer",
+                    "description": "Optional hard process lifetime in seconds. Omit it to allow the managed session to keep running.",
+                    "minimum": 1,
+                },
+                "env": {
+                    "type": "object",
+                    "description": "Optional environment variables to set.",
+                    "additionalProperties": {"type": "string"},
+                    "default": {},
+                },
+            },
+            "required": ["command"],
+        }
+    )
+
+    async def call(
+        self,
+        context: ContextWrapper[AstrAgentContext],
+        command: str,
+        yield_time_ms: int = 10_000,
+        timeout: int | None = None,
+        env: dict[str, Any] | None = None,
+    ) -> ToolExecResult:
+        """Execute a local command without a background-mode argument.
+
+        Args:
+            context: Current agent tool context.
+            command: Shell command to execute.
+            yield_time_ms: Maximum initial wait before returning a session.
+            timeout: Optional hard process lifetime.
+            env: Additional environment variables.
+
+        Returns:
+            JSON command result or a user-facing error.
+        """
+        return await super().call(
+            context,
+            command,
+            background=False,
+            timeout=timeout,
+            env=env,
+            yield_time_ms=yield_time_ms,
+        )
+
+
+@builtin_tool(config=_LOCAL_RUNTIME_TOOL_CONFIG)
+@dataclass
+class ShellSessionTool(FunctionTool):
+    """Manage shell sessions created by the local shell execution tool."""
+
+    name: str = "astrbot_shell_session"
+    description: str = (
+        "List, poll, write to, interrupt, or terminate managed shell sessions. "
+        "Sessions are isolated to the current conversation."
+    )
+    parameters: dict = field(
+        default_factory=lambda: {
+            "type": "object",
+            "properties": {
+                "action": {
+                    "type": "string",
+                    "enum": ["list", "poll", "write", "interrupt", "terminate"],
+                    "description": "Session operation to perform.",
+                },
+                "session_id": {
+                    "type": "string",
+                    "description": "Required for every action except list.",
+                },
+                "chars": {
+                    "type": "string",
+                    "description": "Text written verbatim when action is write.",
+                    "default": "",
+                },
+                "cursor": {
+                    "type": "integer",
+                    "description": "Optional byte cursor for poll. Omit to continue from the last returned output.",
+                    "minimum": 0,
+                },
+                "yield_time_ms": {
+                    "type": "integer",
+                    "description": "Maximum time poll or interrupt waits for output or exit.",
+                    "default": 5000,
+                    "minimum": 0,
+                    "maximum": 30000,
+                },
+                "max_output_chars": {
+                    "type": "integer",
+                    "description": "Maximum output bytes returned by poll, interrupt, or terminate.",
+                    "default": 10000,
+                    "minimum": 1,
+                    "maximum": 100000,
+                },
+            },
+            "required": ["action"],
+        }
+    )
+
+    async def call(
+        self,
+        context: ContextWrapper[AstrAgentContext],
+        action: str,
+        session_id: str | None = None,
+        chars: str = "",
+        cursor: int | None = None,
+        yield_time_ms: int = 5_000,
+        max_output_chars: int = 10_000,
+    ) -> ToolExecResult:
+        """Perform a conversation-scoped local shell session operation.
+
+        Args:
+            context: Current agent tool context.
+            action: Session operation to perform.
+            session_id: Managed session identifier, except for list.
+            chars: Text written for the write action.
+            cursor: Optional output byte cursor.
+            yield_time_ms: Maximum wait for output or process exit.
+            max_output_chars: Maximum output bytes to return.
+
+        Returns:
+            JSON session operation result or a user-facing error.
+        """
+        if permission_error := check_admin_permission(
+            context,
+            "Shell session management",
+        ):
+            return permission_error
+        if not is_local_runtime(context):
+            return "Error managing shell session: only local runtime is supported."
+
+        try:
+            sb = await get_booter(
+                context.context.context,
+                context.context.event.unified_msg_origin,
+            )
+            if not isinstance(sb.shell, LocalShellComponent):
+                return "Error managing shell session: local shell component is unavailable."
+
+            owner_id = context.context.event.unified_msg_origin
+            if action == "list":
+                result = await sb.shell.list_sessions(owner_id)
+            else:
+                if not session_id:
+                    return (
+                        "Error managing shell session: session_id is required "
+                        f"when action={action}."
+                    )
+                if action == "poll":
+                    result = await sb.shell.poll_session(
+                        owner_id=owner_id,
+                        session_id=session_id,
+                        cursor=cursor,
+                        yield_time_ms=yield_time_ms,
+                        max_output_chars=max_output_chars,
+                    )
+                elif action == "write":
+                    result = await sb.shell.write_session(
+                        owner_id=owner_id,
+                        session_id=session_id,
+                        chars=chars,
+                    )
+                elif action == "interrupt":
+                    result = await sb.shell.interrupt_session(
+                        owner_id=owner_id,
+                        session_id=session_id,
+                        yield_time_ms=yield_time_ms,
+                        max_output_chars=max_output_chars,
+                    )
+                elif action == "terminate":
+                    result = await sb.shell.terminate_session(
+                        owner_id=owner_id,
+                        session_id=session_id,
+                        max_output_chars=max_output_chars,
+                    )
+                else:
+                    return f"Error managing shell session: unsupported action {action}."
+            return json.dumps(result, ensure_ascii=False)
+        except Exception as exc:
+            detail = str(exc) or type(exc).__name__
+            return f"Error managing shell session: {detail}"
+
+
 def _is_self_detached_command(command: str) -> bool:
    lex = shlex.shlex(command, posix=False)
    lex.whitespace_split = True
@@ -1,7 +1,10 @@
 from __future__ import annotations

 import asyncio
+import os
+import shlex
 import subprocess
+import sys

 import pytest

@@ -28,6 +31,12 @@ class _FakeTaskkillResult:
        self.returncode = returncode


+def _python_command(code: str) -> str:
+    """Build a shell-safe Python command for the current operating system."""
+    args = [sys.executable, "-u", "-c", code]
+    return subprocess.list2cmdline(args) if os.name == "nt" else shlex.join(args)
+
+
 def test_local_shell_component_decodes_utf8_output(monkeypatch):
    def fake_run(*args, **kwargs):
        _ = args, kwargs
@@ -134,3 +143,140 @@ def test_local_shell_component_falls_back_when_windows_taskkill_fails(monkeypatc

    assert proc.killed
    assert proc.wait_timeout == 5
+
+
+@pytest.mark.asyncio
+async def test_managed_shell_returns_completed_output_without_open_session():
+    shell = LocalShellComponent()
+
+    result = await shell.exec_managed(
+        _python_command("print('hello')"),
+        owner_id="owner-a",
+        yield_time_ms=5_000,
+    )
+
+    assert result["status"] == "completed"
+    assert result["stdout"] == "hello\n"
+    assert result["exit_code"] == 0
+    assert result["session_closed"] is True
+    assert await shell.list_sessions("owner-a") == {"sessions": []}
+
+
+@pytest.mark.asyncio
+async def test_managed_shell_lists_and_terminates_running_session():
+    shell = LocalShellComponent()
+    result = await shell.exec_managed(
+        _python_command("import time; print('ready', flush=True); time.sleep(30)"),
+        owner_id="owner-a",
+        yield_time_ms=200,
+    )
+
+    try:
+        assert result["status"] == "running"
+        assert result["stdout"] == "ready\n"
+        session_id = result["session_id"]
+        assert (await shell.list_sessions("owner-b"))["sessions"] == []
+        sessions = (await shell.list_sessions("owner-a"))["sessions"]
+        assert [item["session_id"] for item in sessions] == [session_id]
+
+        stopped = await shell.terminate_session(
+            owner_id="owner-a",
+            session_id=session_id,
+        )
+
+        assert stopped["status"] == "terminated"
+        assert stopped["exit_code"] is not None
+        assert stopped["session_closed"] is True
+        assert await shell.list_sessions("owner-a") == {"sessions": []}
+    finally:
+        await shell.shutdown_sessions()
+
+
+@pytest.mark.asyncio
+async def test_managed_shell_accepts_stdin_and_polls_incremental_output():
+    shell = LocalShellComponent()
+    result = await shell.exec_managed(
+        _python_command("value = input(); print(f'got:{value}', flush=True)"),
+        owner_id="owner-a",
+        yield_time_ms=100,
+    )
+
+    try:
+        assert result["status"] == "running"
+        await shell.write_session(
+            owner_id="owner-a",
+            session_id=result["session_id"],
+            chars="hello\n",
+        )
+        completed = await shell.poll_session(
+            owner_id="owner-a",
+            session_id=result["session_id"],
+            yield_time_ms=5_000,
+        )
+        output = completed["stdout"]
+        if completed["status"] == "running":
+            completed = await shell.poll_session(
+                owner_id="owner-a",
+                session_id=result["session_id"],
+                yield_time_ms=5_000,
+            )
+            output += completed["stdout"]
+
+        assert completed["status"] == "completed"
+        assert output == "got:hello\n"
+        assert completed["session_closed"] is True
+    finally:
+        await shell.shutdown_sessions()
+
+
+@pytest.mark.asyncio
+async def test_managed_shell_hard_timeout_terminates_session():
+    shell = LocalShellComponent()
+    result = await shell.exec_managed(
+        _python_command("import time; time.sleep(30)"),
+        owner_id="owner-a",
+        timeout=1,
+        yield_time_ms=0,
+    )
+
+    try:
+        timed_out = await shell.poll_session(
+            owner_id="owner-a",
+            session_id=result["session_id"],
+            yield_time_ms=3_000,
+        )
+
+        assert timed_out["status"] == "timed_out"
+        assert timed_out["exit_code"] is not None
+        assert timed_out["session_closed"] is True
+    finally:
+        await shell.shutdown_sessions()
+
+
+@pytest.mark.asyncio
+async def test_managed_shell_keeps_completed_session_until_output_is_drained():
+    shell = LocalShellComponent()
+    result = await shell.exec_managed(
+        _python_command("print('x' * 25000)"),
+        owner_id="owner-a",
+        yield_time_ms=5_000,
+        max_output_chars=10_000,
+    )
+
+    try:
+        assert result["status"] == "completed"
+        assert result["has_more"] is True
+        output = result["stdout"]
+        while result["has_more"]:
+            result = await shell.poll_session(
+                owner_id="owner-a",
+                session_id=result["session_id"],
+                max_output_chars=10_000,
+            )
+            output += result["stdout"]
+
+        assert output == f"{'x' * 25000}\n"
+        assert result["session_closed"] is True
+        assert await shell.list_sessions("owner-a") == {"sessions": []}
+    finally:
+        await shell.shutdown_sessions()
@@ -801,9 +801,7 @@ class TestEnsurePersonaAndSkills:
        mock_context.persona_manager.resolve_selected_persona = AsyncMock(
            return_value=("conv-persona", persona, None, False)
        )
-        mock_event.get_extra.side_effect = (
-            lambda key: key == "enable_inline_genui"
-        )
+        mock_event.get_extra.side_effect = lambda key: key == "enable_inline_genui"
        req = ProviderRequest()
        req.conversation = MagicMock(persona_id="conv-persona")

@@ -818,9 +816,7 @@ class TestEnsurePersonaAndSkills:
    ):
        """Test inline GenUI instructions are added before conversation setup."""
        module = ama
-        mock_event.get_extra.side_effect = (
-            lambda key: key == "enable_inline_genui"
-        )
+        mock_event.get_extra.side_effect = lambda key: key == "enable_inline_genui"
        req = ProviderRequest()

        await module._ensure_persona_and_skills(req, {}, mock_context, mock_event)
@@ -1156,7 +1152,14 @@ class TestEnsurePersonaAndSkills:
            assert result.provider_request.func_tool is not None
            tool_names = result.provider_request.func_tool.names()
            assert "astrbot_execute_shell" in tool_names
+            assert "astrbot_shell_session" in tool_names
            assert "astrbot_execute_python" in tool_names
+            shell_tool = result.provider_request.func_tool.get_tool(
+                "astrbot_execute_shell"
+            )
+            assert shell_tool is not None
+            assert "background" not in shell_tool.parameters["properties"]
+            assert "yield_time_ms" in shell_tool.parameters["properties"]
        finally:
            if result.reset_coro:
                result.reset_coro.close()
@@ -54,9 +54,12 @@ class TestLocalBooterLifecycle:
    async def test_shutdown(self):
        """Test LocalBooter shutdown method."""
        booter = LocalBooter()
-        # Should not raise any exception
+        booter._shell.shutdown_sessions = AsyncMock()
+
        await booter.shutdown()

+        booter._shell.shutdown_sessions.assert_awaited_once()
+
    @pytest.mark.asyncio
    async def test_available(self):
        """Test LocalBooter available method returns True."""
@@ -504,6 +507,20 @@ class TestComputerClient:
        # Reset for other tests
        computer_client.local_booter = None

+    @pytest.mark.asyncio
+    async def test_shutdown_local_booter_clears_singleton(self):
+        """Test local managed resources are released during lifecycle shutdown."""
+        from astrbot.core.computer import computer_client
+
+        booter = MagicMock(spec=LocalBooter)
+        booter.shutdown = AsyncMock()
+        computer_client.local_booter = booter
+
+        await computer_client.shutdown_local_booter()
+
+        booter.shutdown.assert_awaited_once()
+        assert computer_client.local_booter is None
+
    @pytest.mark.asyncio
    async def test_get_booter_shipyard(self):
        """Test get_booter with shipyard type."""
@@ -1,12 +1,19 @@
 import asyncio
+import inspect
 import json
+from unittest.mock import AsyncMock

 import pytest

 from astrbot.core import sp
+from astrbot.core.computer.booters.local import LocalShellComponent
 from astrbot.core.provider import func_tool_manager as ftm
 from astrbot.core.provider.func_tool_manager import FunctionToolManager
-from astrbot.core.tools.computer_tools.shell import ExecuteShellTool
+from astrbot.core.tools.computer_tools.shell import (
+    ExecuteShellTool,
+    LocalExecuteShellTool,
+    ShellSessionTool,
+)
 from astrbot.core.tools.message_tools import SendMessageToUserTool
 from astrbot.core.tools.web_search_tools import (
    FirecrawlExtractWebPageTool,
@@ -49,6 +56,120 @@ def test_computer_tools_are_registered_as_builtin_tools():
    assert tool.name == "astrbot_execute_shell"
    assert tool.parameters["properties"]["background"]["default"] is False
    assert manager.is_builtin_tool("astrbot_execute_shell") is True
+    assert manager.is_builtin_tool("astrbot_shell_session") is True
+
+
+def test_local_execute_shell_schema_replaces_background_with_yield():
+    tool = LocalExecuteShellTool()
+
+    assert tool.name == "astrbot_execute_shell"
+    assert "background" not in tool.parameters["properties"]
+    assert tool.parameters["properties"]["yield_time_ms"]["default"] == 10_000
+    assert "background" not in inspect.signature(tool.call).parameters
+
+
+@pytest.mark.asyncio
+async def test_local_execute_shell_uses_managed_session(monkeypatch, tmp_path):
+    from astrbot.core.tools.computer_tools import shell as shell_tools
+
+    shell = LocalShellComponent()
+    shell.exec_managed = AsyncMock(
+        return_value={
+            "session_id": "sh_test",
+            "status": "running",
+            "stdout": "ready\n",
+            "stderr": "",
+            "exit_code": None,
+        }
+    )
+
+    class FakeBooter:
+        pass
+
+    booter = FakeBooter()
+    booter.shell = shell
+
+    class FakeConfig:
+        def get_config(self, umo):
+            return {"provider_settings": {"computer_use_runtime": "local"}}
+
+    class FakeEvent:
+        unified_msg_origin = "umo"
+        role = "admin"
+
+    class FakeAstrContext:
+        context = FakeConfig()
+        event = FakeEvent()
+
+    class FakeWrapper:
+        context = FakeAstrContext()
+
+    async def fake_get_booter(context, session_id):
+        return booter
+
+    monkeypatch.setattr(shell_tools, "get_booter", fake_get_booter)
+    monkeypatch.setattr(
+        shell_tools,
+        "workspace_root_for_context",
+        AsyncMock(return_value=tmp_path),
+    )
+
+    result = await LocalExecuteShellTool().call(
+        FakeWrapper(),
+        command="python server.py",
+        yield_time_ms=250,
+    )
+
+    assert json.loads(result)["session_id"] == "sh_test"
+    shell.exec_managed.assert_awaited_once_with(
+        "python server.py",
+        owner_id="umo",
+        cwd=str(tmp_path),
+        env={},
+        timeout=None,
+        yield_time_ms=250,
+    )
+
+
+@pytest.mark.asyncio
+async def test_shell_session_tool_lists_sessions_for_current_owner(monkeypatch):
+    from astrbot.core.tools.computer_tools import shell as shell_tools
+
+    shell = LocalShellComponent()
+    shell.list_sessions = AsyncMock(
+        return_value={"sessions": [{"session_id": "sh_test", "status": "running"}]}
+    )
+
+    class FakeBooter:
+        pass
+
+    booter = FakeBooter()
+    booter.shell = shell
+
+    class FakeConfig:
+        def get_config(self, umo):
+            return {"provider_settings": {"computer_use_runtime": "local"}}
+
+    class FakeEvent:
+        unified_msg_origin = "umo"
+        role = "admin"
+
+    class FakeAstrContext:
+        context = FakeConfig()
+        event = FakeEvent()
+
+    class FakeWrapper:
+        context = FakeAstrContext()
+
+    async def fake_get_booter(context, session_id):
+        return booter
+
+    monkeypatch.setattr(shell_tools, "get_booter", fake_get_booter)
+
+    result = await ShellSessionTool().call(FakeWrapper(), action="list")
+
+    assert json.loads(result)["sessions"][0]["session_id"] == "sh_test"
+    shell.list_sessions.assert_awaited_once_with("umo")


@pytest.mark.asyncio