Day 8：交互式 Shell + Plan Mode 闭环

typer.prompt(">") -> input_queue -> run_agent(...) -> console.print trace

uv add prompt_toolkit

main thread: PromptSession + key bindings + slash dispatch
worker thread: run_agent + provider.complete + tools.run

@dataclass
class RuntimeState:
    permission_mode: str = "default"
    model: str = "deepseek-v4-pro"
    provider: str = "anthropic"
    abort_event: threading.Event = field(default_factory=threading.Event)
    input_queue: "queue.Queue[str]" = field(default_factory=queue.Queue)

async def _run() -> None:
    loop = asyncio.get_running_loop()

    def terminal_asker(func: Callable[[], Any]) -> Any:
        async def ask_in_terminal() -> Any:
            return await run_in_terminal(func)
        return asyncio.run_coroutine_threadsafe(ask_in_terminal(), loop).result()

    prompt_ui.set_terminal_asker(terminal_asker)

    with patch_stdout():
        while True:
            text = (await session.prompt_async("> ")).strip()
            if text.startswith("/"):
                result = dispatch_slash(text, make_slash_context())
                ...
            job_queue.put(text)

$ uv run agent-code
Agent Code
cwd: /your/project
provider: anthropic  model: deepseek-v4-pro

输入 /help 查看命令，输入 /exit 退出。
> 读一下 agent_code/cli.py 然后一句话说说它做什么
tool_call: read_file {'path': 'agent_code/cli.py'}
final: cli.py 是 CLI 入口，负责解析参数、分派 slash、跑 Agent Loop。
 default · deepseek-v4-pro
>

def cycle_permission_mode(self) -> str:
    """shift+tab 循环：default → acceptEdits → plan → default。只主线程调，无需锁。"""
    order = ["default", "acceptEdits", "plan"]
    idx = order.index(self.permission_mode) if self.permission_mode in order else 0
    self.permission_mode = order[(idx + 1) % len(order)]
    return self.permission_mode

@kb.add("s-tab")
def _(event: Any) -> None:
    new_mode = state.cycle_permission_mode()
    print(f"[mode → {new_mode}]")

def _cmd_model(args: list[str], ctx: SlashContext) -> SlashResult:
    if not args:
        return SlashResult(handled=True, message=f"provider: {ctx.provider}  model: {ctx.model}")
    target = args[0]
    if ctx.state is not None:
        ctx.state.model = target
    return SlashResult(handled=True, message=f"model → {target}（下一轮生效，当前轮不变）")

$ uv run agent-code
 default · deepseek-v4-pro
> （按 shift+tab）
[mode → acceptEdits]
 accept edits · deepseek-v4-pro
> /model deepseek-chat
model → deepseek-chat（下一轮生效，当前轮不变）
 accept edits · deepseek-chat

if busy.is_set():
    state.input_queue.put(text)
    print("[queued] turn 结束后自动处理")
else:
    job_queue.put(text)

finally:
    busy.clear()
while not state.input_queue.empty():
    job_queue.put(state.input_queue.get())

if state.abort_event.is_set():
    emit("interrupted by user")
    if response.tool_calls:
        blocks = [
            {"type": "tool_result", "tool_use_id": c.id,
             "content": "Interrupted by user", "is_error": True}
            for c in response.tool_calls
        ]
        messages.append({"role": "user", "content": blocks})
    return AgentResult(final="interrupted", trace=trace, messages=messages)

if forced is not None:
    continuation_count += 1
    emit(f"continue: {forced}")
    messages.append({"role": "user", "content": f"continue: {forced}"})
    continue

> 读 agent_code/ 下所有 .py 文件，分别说一句它们做什么
tool_call: read_file ...
> 顺便数一下一共多少个 .py 文件
[queued] turn 结束后自动处理
...（上一轮结束后自动接着跑）

> 把 agent_code 每个文件都读一遍再逐个总结
tool_call: read_file {'path': 'agent_code/agent.py'}
（按 ESC）
interrupted by user
 default · deepseek-v4-pro
>

连续只读工具 -> 并行组
写工具 / 未知工具 -> 单独串行组
tool_result 回填顺序 -> 必须保持 tool_use 顺序

@dataclass
class Tool:
    name: str
    description: str
    run: ToolFunc
    parameters: dict[str, Any] = field(...)
    is_read_only: bool = False

def partition_tool_calls(calls, tools) -> list[list]:
    batches: list[list] = []
    current: list = []
    for call in calls:
        tool = tools.get(call.name)
        if tool is not None and tool.is_read_only:
            current.append(call)
        else:
            if current:
                batches.append(current)
                current = []
            batches.append([call])
    if current:
        batches.append(current)
    return batches

with ThreadPoolExecutor(max_workers=4) as ex:
    results = list(
        ex.map(lambda c: execute_one_tool_call(c, ctx, state, tools, emit), batch)
    )
tool_result_blocks.extend(results)

with self._lock:
    self.entries[path] = (mtime_ns, len(content))

> 同时读 agent_code/agent.py、agent_code/cli.py、agent_code/tools.py，各一句话说职责
tool_call: read_file {'path': 'agent_code/cli.py'}
tool_call: read_file {'path': 'agent_code/agent.py'}
tool_call: read_file {'path': 'agent_code/tools.py'}
final: agent.py 跑 loop，cli.py 是入口，tools.py 是工具表。

@dataclass
class TodoItem:
    content: str
    status: str
    active_form: str

def todo_write(args: dict[str, Any], ctx: ToolContext) -> str:
    state = ctx.runtime_state
    if state is None:
        return "error: no runtime state"
    items = [
        TodoItem(
            content=t.get("content", ""),
            status=t.get("status", "pending"),
            active_form=t.get("activeForm", ""),
        )
        for t in args.get("todos", [])
    ]
    state.todo_store = items

    lines = [_render_todos(items), "", "Todos updated."]
    completed = sum(1 for t in items if t.status == "completed")
    kws = ("test", "pytest", "verify", "lint", "check")
    has_verify = any(any(k in t.content.lower() for k in kws) for t in items)
    if completed >= 3 and not has_verify:
        lines.append("提示：关掉了 3+ 个任务但没有验证步骤，建议先加一个测试/验证项再收尾。")
    return "\n".join(lines)

def _cmd_todo(_args: list[str], ctx: SlashContext) -> SlashResult:
    items = ctx.state.todo_store if ctx.state else []
    icon = {"pending": "○", "in_progress": "◉", "completed": "✓"}
    body = "\n".join(f"  {icon.get(t.status, '?')} {t.content}" for t in items) or "(no todos)"
    return SlashResult(handled=True, message=body)

active = next((t.active_form for t in state.todo_store if t.status == "in_progress"), "")
todo = f" · {active}" if active else ""
return f" {mode} · {state.model}{todo} "

> 用 todo_write 列个计划再执行：1 读 cli.py 2 给 cli.py 顶部加注释 3 跑 git_status 验证。然后照着做
tool_call: todo_write {...}
  ◉ 读 cli.py
  ○ 给 cli.py 顶部加注释
  ○ 跑 git_status 验证
Todos updated.
> /todo
  ✓ 读 cli.py
  ✓ 给 cli.py 顶部加注释
  ◉ 跑 git_status 验证

def enter_plan_mode(args: dict[str, Any], ctx: ToolContext) -> str:
    state = ctx.runtime_state
    if state is None:
        return "error: no runtime state"
    state.permission_mode = "plan"
    return (
        "Plan mode on. Write tools are denied. Draft a clear plan, then present it "
        "(or call exit_plan_mode(plan_summary)). The harness will ask the user to "
        "approve before writes unlock."
    )


def exit_plan_mode(args: dict[str, Any], ctx: ToolContext) -> str:
    return "Plan approved. Write tools are now enabled."

if call.name == "exit_plan_mode":
    plan_summary = call.arguments.get("plan_summary", "")
    if not confirm_plan(plan_summary):
        obs = "Plan not approved. Revise the plan and call exit_plan_mode again."
        return {"type": "tool_result", "tool_use_id": call.id, "content": obs, "is_error": True}
    state.permission_mode = "acceptEdits"

if state.permission_mode == "plan" and final.strip():
    if confirm_plan(final):
        state.permission_mode = "acceptEdits"
        messages.append({"role": "user", "content": "Plan approved. Implement it now."})
    else:
        messages.append({"role": "user", "content": "Plan not approved. Revise the plan and present it again."})
    continue

def confirm_plan(plan_summary: str) -> bool:
    def _do() -> bool:
        buffer = StringIO()
        Console(file=buffer, no_color=True).print(
            Panel(plan_summary or "(empty plan)", title="Plan", border_style="blue")
        )
        panel = buffer.getvalue()
        if _terminal_asker is not None:
            _write_real_terminal(panel)
        else:
            typer.echo(panel, nl=False)
        return typer.confirm("Approve this plan and exit plan mode?", default=False)

    return _ask(_do)

$ uv run agent-code --permission-mode plan
 plan · deepseek-v4-pro
> 写一个 day8_demo.py，放一个 fibonacci 函数和对应的 pytest 测试
tool_call: read_file {'path': 'pyproject.toml'}
...
╭─ Plan ─────────────────────────────────────╮
│ 1. 建 day8_demo.py                          │
│ 2. 写 fibonacci                             │
│ 3. 写 pytest 测试                           │
╰─────────────────────────────────────────────╯
Approve this plan and exit plan mode? [y/N]: y
 accept edits · deepseek-v4-pro
tool_call: file_write {'file_path': 'day8_demo.py', ...}