Function execute_bash

tutorials/14_bash_engine_deep_dive.py:454–501

执行 Bash 命令 — 源码 bash.rs:67-100 这是入口函数。它处理三种情况: 1. 后台执行 → spawn 后立即返回 PID 2. 有超时 → 用 asyncio.wait_for 包装 3. 普通执行 → 同步等待完成 在 Rust 源码中，它创建一个 tokio runtime 来支持异步超时。 Python 中我们用 asyncio 模拟。

(inp: BashCommandInput)

def execute_bash(inp: BashCommandInput) -> BashCommandOutput:
    """执行 Bash 命令 — 源码 bash.rs:67-100

    这是入口函数。它处理三种情况:
    1. 后台执行 → spawn 后立即返回 PID
    2. 有超时 → 用 asyncio.wait_for 包装
    3. 普通执行 → 同步等待完成

    在 Rust 源码中，它创建一个 tokio runtime 来支持异步超时。
    Python 中我们用 asyncio 模拟。
    """
    cwd = Path.cwd()

    # 第一步: 解析沙箱配置
    config = SandboxConfig()  # 实际从 ConfigLoader 加载
    request = resolve_request(
        config,
        enabled_override=(not inp.dangerously_disable_sandbox
                         if inp.dangerously_disable_sandbox is not None else None),
        namespace_override=inp.namespace_restrictions,
        network_override=inp.isolate_network,
        filesystem_mode_override=inp.filesystem_mode,
        allowed_mounts_override=inp.allowed_mounts,
    )
    sandbox_status = resolve_sandbox_status(request, cwd)

    # ========== 路径 1: 后台执行 ==========
    # 源码 bash.rs:71-96
    if inp.run_in_background:
        cmd_args, env = prepare_command(inp.command, cwd, sandbox_status, create_dirs=False)
        # 关键: stdin/stdout/stderr 全部设为 DEVNULL
        # 这让进程完全脱离——不占用终端、不阻塞管道
        process = subprocess.Popen(
            cmd_args,
            stdin=subprocess.DEVNULL,    # 不接受输入
            stdout=subprocess.DEVNULL,   # 不捕获输出
            stderr=subprocess.DEVNULL,   # 不捕获错误
            cwd=str(cwd),
            env=env,
        )
        return BashCommandOutput(
            background_task_id=str(process.pid),
            sandbox_status=sandbox_status,
            no_output_expected=True,
        )

    # ========== 路径 2 & 3: 前台执行（有/无超时）==========
    return _execute_bash_sync(inp, sandbox_status, cwd)


def _execute_bash_sync(