Day 5：Bash 工具 + 权限引擎

# bash 执行的环境变量最小集合——不让宿主机的敏感变量漏进子进程
_MINIMAL_ENV = {
    "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
    "HOME": os.environ.get("HOME", ""),
    "USER": os.environ.get("USER", ""),
    "SHELL": os.environ.get("SHELL", "/bin/bash"),
}


def run_sync(command: str, cwd: Path, timeout: int = 30) -> str:
    """同步执行 shell 命令。cwd 锁定在项目目录，超时后杀进程。"""
    try:
        proc = subprocess.run(
            command,
            shell=True,
            cwd=str(cwd),
            env=_MINIMAL_ENV,
            capture_output=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return f"error: command timed out after {timeout}s"
    ...
    truncated = truncate_output(output.strip(), max_chars=12000)
    if proc.returncode != 0:
        return f"exit code {proc.returncode}\n{truncated}"
    return truncated if truncated else "(no output)"

from __future__ import annotations

# Day 5 v1：render_diff 和 confirm_edit 的实现迁到了 prompt_ui.py。
# 这里只是个 re-export 垫片，让 v1 的 agent.py 不用改 import 也能跑。v2 会把它删掉。
from .prompt_ui import confirm_edit, render_diff

__all__ = ["confirm_edit", "render_diff"]

def confirm_command(command: str) -> bool:
    """让用户确认是否执行这条 bash 命令，默认不执行。"""
    return typer.confirm(f"Run this command?", default=False)

from .bash_runner import run_sync as _bash_run_sync

def bash(args: dict[str, Any], ctx: ToolContext) -> str:
    """执行 shell 命令。前置校验和用户确认在 agent.py 拦截块完成。"""
    command = args.get("command", "")
    if not command:
        return "error: missing required argument 'command'"
    timeout = int(args.get("timeout", 30))
    background = bool(args.get("background", False))

    # v1 只做同步；v4 接 background=True 分支
    if background:
        return "error: background mode not implemented yet (coming in v4)"

    return _bash_run_sync(command, ctx.cwd, timeout=timeout)

            # bash 拦截：打印命令预览，让用户确认后再执行
            elif call.name == "bash":
                command = call.arguments.get("command", "")
                timeout = call.arguments.get("timeout", 30)
                console.print(f"\n[bold yellow]Command:[/bold yellow] {command}")
                console.print(f"[dim]timeout: {timeout}s  cwd: {ctx.cwd}[/dim]")
                if not confirm_command(command):
                    result = ToolResult(call.id, "error: command rejected by user", is_error=True)
                    emit(f"observation: {result.content}")
                    tool_result_blocks.append(...)
                    continue

$ uv run agent-code "用 git_status 看一下当前仓库状态，然后用 git_diff 看改了什么"
Agent Code
cwd: /your/project
provider: anthropic  model: deepseek-v4-flash

tool_call: git_status {}
observation: On branch main
Changes not staged for commit:
  modified: ...

tool_call: git_diff {}
observation: diff --git ...
final: 当前仓库有一些未提交改动，我已经看到了 git status 和 git diff。

$ uv run agent-code "用 bash 跑一下 pytest --version"
...
tool_call: bash {'command': 'pytest --version', 'timeout': 30, 'background': False}

Command: pytest --version
timeout: 30s  cwd: /your/project
Run this command? [y/N]: y
observation: pytest 8.x.x
final: 当前环境中安装了 pytest 8.x.x。

$ uv run agent-code "用 bash 跑一下 python -c 'print(1/0)'"
...
observation: exit code 1
Traceback (most recent call last):
  File "<string>", line 1, in <module>
ZeroDivisionError: division by zero
final: 命令执行失败了——代码尝试用 0 做除数，Python 抛出了 ZeroDivisionError。

@dataclass
class PermissionRequest:
    """一次工具调用的权限请求。工具只描述意图，是否执行交给 harness 决定。"""
    tool_name: str
    args: dict
    mode: str
    cwd: Path


@dataclass
class PermissionDecision:
    """权限引擎的决策结果。behavior 是 allow / ask / deny 之一。"""
    behavior: str  # "allow" | "ask" | "deny"
    message: str | None = None


def decide_permission(request: PermissionRequest) -> PermissionDecision:
    """权限引擎入口：根据工具名、参数和当前模式决定 allow / ask / deny。"""
    tool_name = request.tool_name
    args = request.args
    mode = request.mode

    if tool_name in _ASK_TOOLS:
        return PermissionDecision("ask")

    # plan 模式：只允许只读工具。写类工具一律 deny。
    if mode == "plan":
        if tool_name in _READONLY_TOOLS:
            return PermissionDecision("allow")
        return PermissionDecision("deny", f"plan mode: {tool_name} is not allowed. ...")

    # 只读工具在所有模式下默认允许
    if tool_name in _READONLY_TOOLS:
        return PermissionDecision("allow")

    # bash 工具的危险命令检测——不管什么模式，危险命令先拦截
    if tool_name == "bash":
        command = args.get("command", "")
        danger_reason = _is_dangerous(command)
        if danger_reason:
            return PermissionDecision("deny", f"Dangerous command blocked: {danger_reason}")

    # acceptEdits 模式：文件编辑跳过确认 UI，但安全校验仍在 agent.py 做
    if mode == "acceptEdits":
        if tool_name in ("file_write", "file_edit"):
            return PermissionDecision("allow")

    # 默认模式：写文件和 bash 需要确认
    return PermissionDecision("ask")

            # 权限引擎统一入口：所有工具调用先包装成 PermissionRequest
            request = PermissionRequest(
                tool_name=call.name,
                args=call.arguments,
                mode=permission_mode,
                cwd=ctx.cwd,
            )
            decision = decide_permission(request)

            edit_preview: tuple[str, str, str] | None = None
            if call.name in ("file_write", "file_edit") and decision.behavior != "deny":
                # acceptEdits 只跳过确认 UI，不能跳过 Day 4 的安全校验
                # 路径解析 + read-before-edit + mtime 检查 + apply_single_replace 全在这里跑
                ...
                edit_preview = (path_str, old_content, new_content)

            if decision.behavior == "deny":
                # deny 路径：直接返回 error observation，不弹 UI
                result = ToolResult(call.id, f"error: {decision.message}", is_error=True)
                ...
                continue

            elif decision.behavior == "ask":
                if call.name in ("file_write", "file_edit"):
                    # ask 模式只负责 diff + confirm（校验已经做过）
                    ...
                elif call.name == "bash":
                    # 命令预览 + confirm
                    ...
                elif call.name in ("web_fetch", "web_search"):
                    # confirm_tool_use 让用户确认访问外部资源
                    ...
                elif call.name == "ask_user_question":
                    pass  # v3 接上

            # allow 路径 + ask 通过：执行工具
            result = tools.run(call, ctx)

    permission_mode: str = typer.Option("default", "--permission-mode", help="Permission mode: default, acceptEdits, plan"),

$ uv run agent-code "先读 hello.txt，再把里面的 hola 改成 hello"
...
tool_call: read_file {'path': 'hello.txt'}
observation: hola from agent
tool_call: file_edit {...}

Diff for hello.txt:
...
Apply this edit to hello.txt? [y/N]:

$ uv run agent-code --permission-mode acceptEdits "先读 hello.txt，再把里面的 hello 改成 hola"
...
tool_call: file_edit {...}
observation: Edited hello.txt: replaced 5 chars with 4 chars

$ uv run agent-code --permission-mode acceptEdits "用 bash 跑 echo hello"
...
Command: echo hello
Run this command? [y/N]:

$ uv run agent-code --permission-mode plan "创建 hello.txt 文件"
...
tool_call: file_write {'file_path': 'hello.txt', 'content': 'hello'}
observation: error: plan mode: file_write is not allowed. Only read-only tools can run in plan mode.
final: 当前处于 plan 模式，不允许执行写入操作。

$ uv run agent-code --permission-mode plan "读一下 hello.txt"
...
tool_call: read_file {'path': 'hello.txt'}
observation: hola from agent

$ uv run agent-code "用 bash 跑 git push --force origin main"
...
tool_call: bash {'command': 'git push --force origin main', 'timeout': 30, 'background': False}
observation: error: Dangerous command blocked: git push --force overwrites remote history

def prompt_single_choice(question: str, labels: list[str]) -> str | None:
    """展示一个 numbered menu 让用户单选，返回被选中的 label。"""
    import rich
    console = rich.console.Console()
    console.print(f"\n[bold yellow]? {question}[/bold yellow]")
    for i, label in enumerate(labels, 1):
        console.print(f"  {i}. {label}")
    console.print(f"  0. [dim]Skip / Other[/dim]")

    try:
        choice = typer.prompt("Choice", default="0")
        idx = int(choice)
        if 1 <= idx <= len(labels):
            return labels[idx - 1]
        return None
    except (ValueError, TypeError):
        return None

def _ask_user_question(args: dict[str, Any], ctx: ToolContext) -> str:
    """由 agent.py 拦截块处理——工具函数本身不读 stdin。"""
    prompt = args.get("prompt", "")
    options = args.get("options", [])
    if not prompt:
        return "error: missing required argument 'prompt'"
    if not options or not isinstance(options, list):
        return "error: options must be a non-empty list"
    return "error: ask_user_question must be handled by the harness, not executed directly"

                elif call.name == "ask_user_question":
                    question = call.arguments.get("prompt", "")
                    options = call.arguments.get("options", [])
                    if not isinstance(options, list):
                        options = []
                    labels = [str(o) for o in options]
                    selected = prompt_single_choice(question, labels)
                    if selected is None:
                        result = ToolResult(call.id, "User skipped the question.", is_error=False)
                    else:
                        result = ToolResult(call.id, f'User selected: "{selected}"', is_error=False)
                    emit(f"observation: {result.content}")
                    tool_result_blocks.append({
                        "type": "tool_result",
                        "tool_use_id": result.tool_call_id,
                        "content": result.content,
                        "is_error": result.is_error,
                    })
                    continue

$ uv run agent-code "我应该先修测试还是先修代码？用 ask_user_question 问我，选项三个：先修测试、先修代码、不确定"
...
tool_call: ask_user_question {'prompt': '修复这个 bug 应该从哪个方向入手？', 'options': ['先修测试', '先修代码', '不确定']}

? 修复这个 bug 应该从哪个方向入手？
  1. 先修测试
  2. 先修代码
  3. 不确定
  0. Skip / Other
Choice: 2
observation: User selected: "先修代码"
final: 你选择了"先修代码"。那我们从修改实现代码开始……

def start_background(command: str, cwd: Path) -> dict:
    """在后台启动 shell 命令，stdout/stderr 流式写入 .bg/<id>.out/.err。
    立即返回结构化信息，不等待命令结束。"""
    bg_id = f"bg-{uuid.uuid4().hex[:8]}"
    bg_dir = cwd / ".bg"
    bg_dir.mkdir(parents=True, exist_ok=True)
    out_path = bg_dir / f"{bg_id}.out"
    err_path = bg_dir / f"{bg_id}.err"

    out_f = open(str(out_path), "w")
    err_f = open(str(err_path), "w")

    proc = subprocess.Popen(
        command, shell=True, cwd=str(cwd), env=_MINIMAL_ENV,
        stdout=out_f, stderr=err_f,
    )

    def _wait_and_close() -> None:
        """等子进程结束，关闭文件描述符。在 daemon 线程里跑。"""
        proc.wait()
        out_f.close()
        err_f.close()

    t = threading.Thread(target=_wait_and_close, daemon=True)
    t.start()

    return {"background_id": bg_id, "output_file": ..., "stderr_file": ..., "pid": proc.pid, "message": ...}

    if background:
        return "error: background mode not implemented yet (coming in v4)"

    if background:
        # 后台执行：启动子进程后立即返回结构化信息，不阻塞 Agent Loop
        from .bg_manager import start_background
        result = start_background(command, ctx.cwd)
        return (
            f"Command running in background with ID: {result['background_id']}.\n"
            f"Output is being written to: {result['output_file']}\n"
            f"Stderr is being written to: {result['stderr_file']}\n"
            f"PID: {result['pid']}\n\n"
            f"{result['message']}"
        )

$ uv run agent-code "用 bash(background=True) 运行 sleep 5 && echo 'done from background'。拿到 background_id 后，用同步 bash 跑 sleep 6 && cat 对应的 .bg 输出文件"
...
tool_call: bash {'command': 'sleep 5 && echo done from background', 'timeout': 30, 'background': True}

Command: sleep 5 && echo done from background
Run this command? [y/N]: y
observation: Command running in background with ID: bg-a1b2c3d4.
Output is being written to: .bg/bg-a1b2c3d4.out
...
tool_call: bash {'command': 'sleep 6 && cat .bg/bg-a1b2c3d4.out', 'timeout': 10, 'background': False}

Command: sleep 6 && cat .bg/bg-a1b2c3d4.out
Run this command? [y/N]: y
observation: done from background
final: 后台任务执行完成，输出为 "done from background"。

$ uv run agent-code "用 bash(background=True) 运行 sleep 300，然后用 kill 杀掉它"
...
observation: Command running in background with ID: bg-e5f6g7h8.
PID: 12345
...
tool_call: bash {'command': 'kill 12345', 'timeout': 10, 'background': False}