花一天读完 OpenHarness：一个 11,733 行的 Agent Harness 到底长什么样

src/openharness/
│
├── cli.py                 ← 入口：Typer CLI
│
├── engine/                ← 🧠 核心中的核心：Agent Loop
│   ├── query_engine.py    ← while True: stream → tool_use → execute → loop
│   ├── query.py           ← 真正的循环实现
│   ├── messages.py        ← 消息格式
│   ├── cost_tracker.py    ← token 计费
│   └── stream_events.py   ← 流式事件类型
│
├── tools/                 ← 🔧 43 个 Tool（Bash, Read, Write, Glob...）
│   ├── base.py            ← BaseTool + ToolRegistry
│   └── *_tool.py          ← 每个文件一个 Tool 实现
│
├── permissions/           ← 🛡️ 权限检查（default/plan/full_auto）
├── hooks/                 ← ⚡ 生命周期钩子（PreToolUse/PostToolUse）
│
├── prompts/               ← 📝 System Prompt 组装工厂
├── skills/                ← 📚 按需加载的 .md 知识文件
├── memory/                ← 🧠 跨会话持久记忆
├── plugins/               ← 🔌 插件系统
├── commands/              ← 💬 斜杠命令注册
├── mcp/                   ← 🌐 Model Context Protocol Client
├── tasks/                 ← 📋 后台任务管理
├── coordinator/           ← 🤝 多 Agent 编排
│
├── config/                ← ⚙️ 配置管理
├── state/                 ← 状态存储
├── services/              ← 辅助服务
├── bridge/                ← Python ↔ React TUI 通信桥
├── ui/                    ← UI 层入口
└── keybindings/           ← 快捷键配置

# cli.py:12-21
app = typer.Typer(
    name="openharness",
    help="Oh my Harness! An AI-powered coding assistant.",
    invoke_without_command=True,
)

# cli.py:346-377
if print_mode is not None:
    # 非交互模式
    asyncio.run(run_print_mode(...))
    return

# 交互模式
asyncio.run(run_repl(...))

async def run_repl(...) -> None:
    if backend_only:
        await run_backend_host(...)
        return

    exit_code = await launch_react_tui(...)

# react_launcher.py:78-102
env["OPENHARNESS_FRONTEND_CONFIG"] = json.dumps({
    "backend_command": build_backend_command(...),  # ← python -m openharness --backend-only
    "initial_prompt": prompt,
})

process = await asyncio.create_subprocess_exec(
    npm, "exec", "--", "tsx", "src/index.tsx", ...
)

第 1 步：你敲 `oh`
         → Python 进程 A 启动
         → 它的工作只有一件事：启动 Node.js

第 2 步：Node.js 启动了
         → 用 React/Ink 渲染你看到的 TUI 界面
         → 但 Node.js 不会 AI 逻辑啊
         → 于是它反过来 spawn 了 Python 进程 B（--backend-only 模式）

第 3 步：Python 进程 B 启动
         → 这才是真正干活的后端
         → 进程 A 使命完成，退出

class FrontendRequest(BaseModel):
    type: Literal[
        "submit_line",           # 用户输入一行
        "permission_response",   # 权限弹窗的回答
        "question_response",     # 问题弹窗的回答
        "list_sessions",
        "shutdown",
    ]
    line: str | None = None
    allowed: bool | None = None
    answer: str | None = None

# runtime.py:89-176 (简化版)
async def build_runtime(...) -> RuntimeBundle:
    settings = load_settings().merge_cli_overrides(...)
    plugins = load_plugins(settings, cwd)

    resolved_api_client = AnthropicApiClient(
        api_key=settings.resolve_api_key(),
        base_url=settings.base_url,
    )
    mcp_manager = McpClientManager(load_mcp_server_configs(settings, plugins))
    await mcp_manager.connect_all()

    tool_registry = create_default_tool_registry(mcp_manager)
    hook_executor = HookExecutor(...)

    engine = QueryEngine(
        api_client=resolved_api_client,
        tool_registry=tool_registry,
        permission_checker=PermissionChecker(settings.permission),
        system_prompt=build_runtime_system_prompt(...),
        hook_executor=hook_executor,
        ...
    )

    return RuntimeBundle(
        api_client=resolved_api_client,
        tool_registry=tool_registry,
        hook_executor=hook_executor,
        engine=engine,
        ...
    )

@dataclass
class RuntimeBundle:
    api_client: SupportsStreamingMessages   # LLM API 客户端
    cwd: str                                 # 工作目录
    mcp_manager: McpClientManager           # MCP 外部工具
    tool_registry: ToolRegistry             # 43 个 Tool
    app_state: AppStateStore                # UI 状态
    hook_executor: HookExecutor             # 生命周期 Hook
    engine: QueryEngine                     # Agent Loop 引擎
    commands: object                        # 斜杠命令
    external_api_client: bool
    session_id: str = ""

用户发消息 → 调 API → AI 回复 → 结束

# query.py:53-86
async def run_query(
    context: QueryContext,
    messages: list[ConversationMessage],
) -> AsyncIterator[tuple[StreamEvent, UsageSnapshot | None]]:
    """Run the conversation loop until the model stops requesting tools."""
    for _ in range(context.max_turns):
        final_message: ConversationMessage | None = None
        usage = UsageSnapshot()

        async for event in context.api_client.stream_message(
            ApiMessageRequest(
                model=context.model,
                messages=messages,
                system_prompt=context.system_prompt,
                max_tokens=context.max_tokens,
                tools=context.tool_registry.to_api_schema(),
            )
        ):
            if isinstance(event, ApiTextDeltaEvent):
                yield AssistantTextDelta(text=event.text), None
                continue

            if isinstance(event, ApiMessageCompleteEvent):
                final_message = event.message
                usage = event.usage

        if final_message is None:
            raise RuntimeError("Model stream finished without a final message")

        messages.append(final_message)
        yield AssistantTurnComplete(message=final_message, usage=usage), usage

        if not final_message.tool_uses:
            return

for _ in range(context.max_turns):   # 默认 8 轮

tools=context.tool_registry.to_api_schema()

if isinstance(event, ApiTextDeltaEvent):
    yield AssistantTextDelta(text=event.text), None  # 打字增量
    continue

if isinstance(event, ApiMessageCompleteEvent):
    final_message = event.message  # 完整消息
    usage = event.usage

if not final_message.tool_uses:
    return

tool_calls = final_message.tool_uses

if len(tool_calls) == 1:
    # Single tool: sequential (stream events immediately)
    tc = tool_calls[0]
    yield ToolExecutionStarted(tool_name=tc.name, tool_input=tc.input), None
    result = await _execute_tool_call(context, tc.name, tc.id, tc.input)
    yield ToolExecutionCompleted(
        tool_name=tc.name,
        output=result.content,
        is_error=result.is_error,
    ), None
    tool_results = [result]
else:
    # Multiple tools: execute concurrently, emit events after
    for tc in tool_calls:
        yield ToolExecutionStarted(tool_name=tc.name, tool_input=tc.input), None

    async def _run(tc):
        return await _execute_tool_call(context, tc.name, tc.id, tc.input)

    results = await asyncio.gather(*[_run(tc) for tc in tool_calls])
    tool_results = list(results)

    for tc, result in zip(tool_calls, tool_results):
        yield ToolExecutionCompleted(
            tool_name=tc.name,
            output=result.content,
            is_error=result.is_error,
        ), None

AI 请求执行 tool
    │
    ▼
① PreToolUse Hook
   → 例如 security-guidance 插件会检查危险命令
   → Hook 可以直接 block 掉这次执行
    │
    ▼
② 找到 tool 实现
   → tool_registry.get(tool_name)
    │
    ▼
③ 校验输入参数（用 Pydantic）
   → tool.input_model.model_validate(tool_input)
   → 类型不对直接报错
    │
    ▼
④ 权限检查
   → permission_checker.evaluate(...)
   → 检查 mode（default/plan/full_auto）
   → 检查 path_rules（某些路径不允许）
   → 检查 denied_commands（某些命令不允许）
   → 如果需要确认 → 调 permission_prompt 弹窗
    │
    ▼
⑤ 真正执行 tool
    │
    ▼
⑥ PostToolUse Hook（记日志等）

decision = context.permission_checker.evaluate(
    tool_name,
    is_read_only=tool.is_read_only(parsed_input),
    file_path=_file_path,
    command=_command,
)
if not decision.allowed:
    if decision.requires_confirmation and context.permission_prompt is not None:
        confirmed = await context.permission_prompt(tool_name, decision.reason)
        if not confirmed:
            return ToolResultBlock(
                tool_use_id=tool_use_id,
                content=f"Permission denied for {tool_name}",
                is_error=True,
            )

第 1 轮：
  messages = [{ role: "user", text: "读一下 README.md 然后总结" }]
  → 调 API，传入所有 tool schema
  → AI 回复: "我来读一下" + tool_use: Read({ file_path: "README.md" })
  → tool_uses 非空，继续循环
  → 执行 Read tool：
    ① PreToolUse Hook 通过
    ② tool_registry.get("Read") 找到工具
    ③ Pydantic 校验 file_path
    ④ 权限检查（只读操作，通过）
    ⑤ 读文件
    ⑥ PostToolUse Hook
  → 把 tool 结果作为 user 消息追加
  → messages 现在有 3 条

第 2 轮：
  → 调 API
  → AI 回复: "这个 README 主要讲了三点：1... 2... 3..."
  → tool_uses 为空
  → return，循环结束