Chapter 17

IDE 集成与 PTY 桥

同一个 AIAgent 怎么同时活在 ACP/Zed 编辑器、Ink 终端 UI、 浏览器 xterm.js dashboard 里。一个核心 + 三种界面 + JSON-RPC 统一协议—— 这是 LLM agent 多界面架构最干净的样本。

本章约 9,000 字 阅读 ~30 分钟 关键词:ACP · ContextVar · Ink · _SlashWorker · PTY · TIOCSWINSZ · xterm.js

ch11 我们看 CLI / Gateway / ACP 三个出口。但 ACP 那段只点了一下,TUI 完全没展开, Dashboard 也没讲怎么把 hermes --tui 嵌进浏览器。这一章拆这三个 IDE-class 表面的实现细节: ACP 用ContextVar绑定编辑器审批 callback;TUI 是 Ink/React 通过 stdio JSON-RPC 跟 Python 网关说话; Dashboard 用 WebSocket 把 PTY 字节双向转发给 xterm.js。3 种传输协议,1 个核心 agent, 零代码重复。

17.1三种 IDE-class 表面的需求差异

ACP(Zed/VS Code)TUI(Ink)Dashboard(浏览器)
传输stdio JSON-RPCstdio JSON-RPCWebSocket 字节流
UI 框架编辑器原生Ink (React for terminal)React + xterm.js
审批模型编辑器 diff 弹窗结构化 JSON event嵌入 hermes --tui 沿用 TUI
状态保持编辑器管Python 网关管PTY 进程管
典型用户软件工程师开发者(local CLI)团队 / 远程
核心差异需要 edit approval需要交互响应快需要远程可访问

所有三种都不应该各自实现一遍 Agent loop。Hermes 的解法: 核心 AIAgent 不变,只换 callback + 平台标签

17.2ACP Adapter:stdio JSON-RPC + ContextVar 审批

17.2.1 entry.py:严格 stdout 协议

acp_adapter/entry.py:212-266
def main(argv=None):
    args = _parse_args(argv)
    if args.version: _print_version(); return
    if args.check:   _run_check();   return
    if args.setup:   _run_setup();   return

    _setup_logging()       # 关键:logging 走 stderr,不污染 stdout
    _load_env()

    project_root = str(Path(__file__).resolve().parent.parent)
    if project_root not in sys.path:
        sys.path.insert(0, project_root)

    import acp
    from .server import HermesACPAgent

    # MCP discovery 在 asyncio 启动前(同步阶段安全)
    from tools.mcp_tool import discover_mcp_tools
    discover_mcp_tools()

    agent = HermesACPAgent()
    try:
        asyncio.run(acp.run_agent(agent, use_unstable_protocol=True))
    except KeyboardInterrupt:
        logger.info("Shutting down (KeyboardInterrupt)")

关键约定:stdout 仅 JSON-RPC,stderr 给人看_setup_logging() 强制 logging handler 写 stderr。任何 print() 都禁止——会污染协议。

17.2.2 HermesACPAgent 类:capability 协商

acp_adapter/server.py:445-527
class HermesACPAgent(acp.Agent):
    _SLASH_COMMANDS = {
        "help": "Show available commands",
        "model": "Show or change current model",
        "tools": "List available tools",
        "context": "Show conversation context info",
        "reset": "Clear conversation history",
        "compact": "Compress conversation context",
        "steer": "Inject guidance into running turn",
        "queue": "Queue prompt for after current turn",
    }

    def __init__(self, session_manager=None):
        super().__init__()
        self.session_manager = session_manager or SessionManager()
        self._conn: Optional[acp.Client] = None

    async def initialize(self, protocol_version=None, ...):
        return InitializeResponse(
            protocol_version=acp.PROTOCOL_VERSION,
            agent_info=Implementation(name="hermes-agent", version=HERMES_VERSION),
            agent_capabilities=AgentCapabilities(
                load_session=True,
                prompt_capabilities=PromptCapabilities(image=True),
                session_capabilities=SessionCapabilities(
                    fork=SessionForkCapabilities(),
                    list=SessionListCapabilities(),
                    resume=SessionResumeCapabilities(),
                ),
            ),
            auth_methods=build_auth_methods(),
        )

这里向编辑器声明 Hermes 支持:加载历史会话、image 输入、session fork/list/resume。 编辑器收到后决定 UI 怎么显示(比如 Zed 会显示历史会话列表)。

17.2.3 Edit Approval:ContextVar 的精妙用法

每个 ACP session 都需要独立的编辑审批 callback。多 session 并发时, 怎么让某个工具 dispatch 找到对应 session 的 approval 函数?

Python ContextVar——同 asyncio 任务内传递的"隐式"变量。 不需要逐层传参,工具 dispatch 时直接 get()

acp_adapter/edit_approval.py:37-286
EditApprovalRequester = Callable[[EditProposal], bool]

_EDIT_APPROVAL_REQUESTER: ContextVar[EditApprovalRequester | None] = ContextVar(
    "ACP_EDIT_APPROVAL_REQUESTER",
    default=None,
)

def set_edit_approval_requester(requester) -> Token:
    """Bind an ACP edit approval requester for the current context."""
    return _EDIT_APPROVAL_REQUESTER.set(requester)

def maybe_require_edit_approval(tool_name: str, arguments: dict) -> str | None:
    """通用工具 dispatch 钩子。返回 None = 放行;返回 JSON 错误字符串 = 拦截。"""
    requester = get_edit_approval_requester()
    if requester is None:
        return None   # 不在 ACP 模式,放行

    try:
        proposal = build_edit_proposal(tool_name, arguments)
    except Exception as exc:
        return json.dumps({"error": f"Edit denied: cannot prepare diff ({exc})"})

    if proposal is None:
        return None    # 工具不属于"编辑类",放行

    try:
        approved = bool(requester(proposal))
    except Exception:
        approved = False    # 异常 = 拒

    if approved:
        return None
    return json.dumps({"error": "Edit approval denied by ACP client; file was not modified."})

这段 hook 在工具 dispatch 的每一次调用都会跑(model_tools.handle_function_call 调它)。 工具调用流程:

flowchart TB
  Tool["LLM 调 write_file('foo.py', '...')"] --> Disp["handle_function_call"]
  Disp --> Hook["maybe_require_edit_approval()"]
  Hook --> CV["ContextVar.get('ACP_EDIT_APPROVAL_REQUESTER')"]
  CV --> Q1{"requester is None?"}
  Q1 -->|是(不是 ACP 模式)| Pass["放行,正常执行"]:::good
  Q1 -->|否| Build["build_edit_proposal()
(diff against current)"] Build --> Ask["requester(proposal)
→ Zed 弹 diff 给用户"] Ask --> Decide{"用户点了?"} Decide -->|Allow Once| Pass Decide -->|Deny / Timeout / Exception| Block["返回 JSON 错误,工具不执行"]:::warn classDef good fill:#ecf3eb,stroke:#2f5d3a,color:#2f5d3a classDef warn fill:#f5ede0,stroke:#a86420,color:#a86420

requester 怎么"问"编辑器? Zed/VS Code 收到 request_permission JSON-RPC, 显示 diff,等用户点。Python 这边是阻塞同步等的(future.result(timeout=60))。 60 秒不答自动拒。

acp_adapter/edit_approval.py (make_acp_edit_approval_requester)
def make_acp_edit_approval_requester(request_permission_fn, loop, session_id,
                                       timeout=60.0, auto_approve_getter=None):
    """Return a sync requester that bridges to ACP permissions."""

    def _requester(proposal: EditProposal) -> bool:
        from acp.schema import PermissionOption
        from agent.async_utils import safe_schedule_threadsafe

        # 1. 自动 approve 策略(.cursorignore 类似规则)
        if auto_approve_getter is not None:
            policy, cwd = auto_approve_getter()
            if should_auto_approve_edit(proposal, policy, cwd):
                logger.info("Auto-approved ACP edit under policy %s: %s", policy, proposal.path)
                return True

        # 2. 否则发请求让用户决定
        options = [
            PermissionOption(option_id="allow_once", kind="allow_once", name="Allow edit"),
            PermissionOption(option_id="deny", kind="reject_once", name="Deny"),
        ]
        tool_call = build_acp_edit_tool_call(proposal)
        coro = request_permission_fn(
            session_id=session_id,
            tool_call=tool_call,
            options=options,
        )
        future = safe_schedule_threadsafe(coro, loop, logger=logger, ...)
        if future is None:
            return False
        try:
            response = future.result(timeout=timeout)   # 阻塞 60s
        except (FutureTimeout, Exception):
            future.cancel()
            return False

        outcome = getattr(response, "outcome", None)
        return (
            getattr(outcome, "outcome", None) == "selected"
            and getattr(outcome, "option_id", None) == "allow_once"
        )

    return _requester
ContextVar 是这里的 Hero 没有 ContextVar,你得逐层传 approval_cb 参数,污染所有工具签名。 ContextVar 让任意嵌套工具都能透明拿到当前 session 的审批函数。 Python 标准库 3.7+ 的杀手级特性。Async-aware,任务间隔离。

17.3TUI:Ink (React) 通过 stdio 跟 Python 网关说话

17.3.1 进程模型

flowchart LR
  User["用户运行
hermes --tui"] --> Node["Node.js
(Ink + React)"] Node -.stdio JSON-RPC.-> Py["Python
tui_gateway"] Py --> Agent["AIAgent
(platform='tui')"] Py --> Slash["_SlashWorker
(persistent subprocess)"] Slash --> CLI["HermesCLI 实例
(处理 / 命令)"] Agent --> LLM["LLM API"] classDef ts fill:#e8eff5,stroke:#2c5282,color:#2c5282 classDef py fill:#f5ede0,stroke:#8b1538,color:#8b1538 class Node ts class Py,Agent,Slash,CLI py

两层进程:

  1. Ink 进程:TypeScript/React,渲染终端 UI。它不知道 Agent 怎么跑——只知道收 JSON 事件、显示出来。
  2. Python tui_gateway:跑 AIAgent,通过 stdio 给 Ink 发事件。同时 spawn 一个常驻 SlashWorker 处理 / 命令(避免每次 / 都重启 CLI)。

17.3.2 entry.py:tight stdio loop

tui_gateway/entry.py:187-251
def main():
    _install_sidecar_publisher()

    # MCP discovery 提前(同步阶段安全)
    cfg = read_raw_config() or {}
    if isinstance(cfg.get("mcp_servers"), dict):
        from tools.mcp_tool import discover_mcp_tools
        discover_mcp_tools()

    # 首事件:让 Ink 知道 gateway 起来了 + 当前 skin
    if not write_json({
        "jsonrpc": "2.0",
        "method": "event",
        "params": {"type": "gateway.ready", "payload": {"skin": resolve_skin()}},
    }):
        sys.exit(0)   # 一开始就 broken pipe 直接退

    # 主循环:读一行,dispatch,写响应
    for raw in sys.stdin:
        line = raw.strip()
        if not line: continue
        try:
            req = json.loads(line)
        except json.JSONDecodeError:
            write_json({"jsonrpc": "2.0", "error": {"code": -32700, "message": "parse error"}})
            continue

        resp = dispatch(req)
        if resp is not None:
            write_json(resp)

17.3.3 dispatch + thread pool 解决"长任务卡住主循环"

问题:某些 RPC 跑几秒到几分钟(session.resume 加载几百条历史;slash.exec 调 LLM 改 model)。如果在主线程里同步跑,这段时间里新来的事件(比如用户按打断) 也没法处理。

Hermes 的解法:"长 handler 走 thread pool,短 handler 走主线程":

tui_gateway/server.py:137-253
_LONG_HANDLERS = frozenset({
    "browser.manage",
    "cli.exec",
    "session.branch",
    "session.compress",
    "session.resume",
    "shell.exec",
    "skills.manage",
    "slash.exec",
})

_pool = concurrent.futures.ThreadPoolExecutor(
    max_workers=_rpc_pool_workers,
    thread_name_prefix="tui-rpc",
)

def dispatch(req, transport=None):
    t = transport or _stdio_transport
    token = bind_transport(t)
    try:
        normalized = _normalize_request(req)
        if isinstance(normalized, dict):
            return normalized
        _rid, method, _params = normalized
        if method not in _LONG_HANDLERS:
            return handle_request(req)    # 短 → 主线程同步返回

        # 长 → 拷贝 context,扔进 pool,主线程立刻返回 None
        ctx = contextvars.copy_context()
        def run():
            try:
                resp = handle_request(req)
            except Exception as exc:
                resp = _err(req.get("id"), -32000, f"handler error: {exc}")
            if resp is not None:
                t.write(resp)         # write_json 是锁保护的,并发安全

        _pool.submit(lambda: ctx.run(run))
        return None    # 主线程立刻让出,接下一条
    finally:
        reset_transport(token)

关键"copy_context":thread 里的 handler 需要看到 transport ContextVar 才能写响应。 contextvars.copy_context() 把当前 context 一并送进 worker thread。

17.3.4 _SlashWorker:持久 HermesCLI 实例

/ 命令(/model/tools/skills...)由 HermesCLI 处理。 但每次都 spawn 一个新 CLI 进程会花 1-2 秒——太慢。Hermes 的解法:一个常驻 subprocess, 通过 JSON 来回通信:

tui_gateway/server.py:_SlashWorker
class _SlashWorker:
    """Persistent HermesCLI subprocess for slash commands."""

    def __init__(self, session_key, model):
        self._lock = threading.Lock()
        self._seq = 0
        self.stderr_tail = []
        self.stdout_queue = queue.Queue()

        argv = [
            sys.executable, "-m", "tui_gateway.slash_worker",
            "--session-key", session_key,
        ]
        if model:
            argv += ["--model", model]

        self.proc = subprocess.Popen(
            argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
            stderr=subprocess.PIPE, text=True, bufsize=1,
            cwd=os.getcwd(), env=os.environ.copy(),
        )
        threading.Thread(target=self._drain_stdout, daemon=True).start()
        threading.Thread(target=self._drain_stderr, daemon=True).start()

    def run(self, command):
        if self.proc.poll() is not None:
            raise RuntimeError("slash worker exited")

        with self._lock:
            self._seq += 1
            rid = self._seq
            self.proc.stdin.write(json.dumps({"id": rid, "command": command}) + "\n")
            self.proc.stdin.flush()
            while True:
                msg = self.stdout_queue.get(timeout=_SLASH_WORKER_TIMEOUT_S)
                if msg is None: break
                if msg.get("id") != rid: continue
                if not msg.get("ok"):
                    raise RuntimeError(msg.get("error", "failed"))
                return str(msg.get("output", "")).rstrip()

对应的 worker 端(单独脚本):

tui_gateway/slash_worker.py (~76 行,核心)
def main():
    args = p.parse_args()
    os.environ["HERMES_SESSION_KEY"] = args.session_key
    os.environ["HERMES_INTERACTIVE"] = "1"

    # 静默 init,redirect stdout 防污染
    with contextlib.redirect_stdout(io.StringIO()), \
         contextlib.redirect_stderr(io.StringIO()):
        cli = HermesCLI(model=args.model or None,
                         compact=True,
                         resume=args.session_key)

    # 主循环:读 JSON,跑命令,回 JSON
    for raw in sys.stdin:
        req = json.loads(raw)
        rid = req.get("id")
        try:
            out = _run(cli, req.get("command", ""))
            sys.stdout.write(json.dumps({"id": rid, "ok": True, "output": out}) + "\n")
        except Exception as e:
            sys.stdout.write(json.dumps({"id": rid, "ok": False, "error": str(e)}) + "\n")
        sys.stdout.flush()

def _run(cli, command):
    # 巧妙:HermesCLI 用 rich.Console 输出 — 重定向控制台底层文件到 StringIO
    buf = io.StringIO()
    cli.console = Console(file=buf, force_terminal=True, width=120)
    with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf):
        cli.process_command(command)
    return buf.getvalue().rstrip()
为啥 rich.Console 要 patch 文件 rich.Console 在构造时抓取当前 stdout 文件描述符,后续 contextlib.redirect_stdout 无效。所以要直接换 cli.console.file = buf。这种"library 不配合 stdlib pattern"是常见小坑。 Hermes 用 patch 解决——你写 LLM 集成时遇到类似坑要记得这招。

17.4Web Dashboard:把 `hermes --tui` 嵌进浏览器

17.4.1 架构

Dashboard 不是重写 TUI——它是把 TUI 嵌进浏览器。通过 PTY + WebSocket + xterm.js。

flowchart LR
  Browser["浏览器
xterm.js"] -.WebSocket /api/pty.-> Server["FastAPI
web_server.py"] Server --> PTY["PtyBridge
(POSIX PTY)"] PTY --> Hermes["子进程:
hermes --tui
(Ink + Python gateway)"] Hermes -.ANSI bytes.-> PTY PTY -.bytes.-> Server Server -.WebSocket.-> Browser classDef browser fill:#e8eff5,stroke:#2c5282,color:#2c5282 classDef server fill:#f5ede0,stroke:#8b1538,color:#8b1538 classDef hermes fill:#ecf3eb,stroke:#2f5d3a,color:#2f5d3a class Browser browser class Server,PTY server class Hermes hermes

17.4.2 PtyBridge:POSIX-only 优雅降级

hermes_cli/pty_bridge.py:1-100 (摘要)
"""PTY bridge for `hermes dashboard` chat tab.

Design constraints:
* **POSIX-only.** 依赖 ``fcntl``, ``termios``, ``ptyprocess``.
  原生 Windows 没这些。导入时抛 ImportError,Dashboard 的 /chat tab
  显示"用 WSL2"提示而不是 crash。其他功能(sessions、jobs、metrics)Windows 正常。
"""

class PtyBridge:
    @classmethod
    def spawn(cls, argv, *, cwd=None, env=None, cols=80, rows=24):
        if not _PTY_AVAILABLE:
            if sys.platform.startswith("win"):
                raise PtyUnavailableError(
                    "Pseudo-terminals unavailable. Hermes Windows = WSL only.")
            if ptyprocess is None:
                raise PtyUnavailableError(
                    "pip install ptyprocess (or pip install -e '.[pty]')")

        spawn_env = (os.environ.copy() if env is None else env.copy())
        if not spawn_env.get("TERM"):
            spawn_env["TERM"] = "xterm-256color"   # 一定要 set TERM

        proc = ptyprocess.PtyProcess.spawn(
            list(argv),
            cwd=cwd, env=spawn_env,
            dimensions=(rows, cols),
        )
        return cls(proc)

    def resize(self, cols, rows):
        """Forward terminal resize via TIOCSWINSZ ioctl."""
        if self._closed:
            return
        # struct winsize: rows, cols, xpixel, ypixel
        winsize = struct.pack("HHHH", max(1, rows), max(1, cols), 0, 0)
        try:
            fcntl.ioctl(self._fd, termios.TIOCSWINSZ, winsize)
        except OSError:
            pass

关键设计:

17.4.3 WebSocket /api/pty:resize 协议

hermes_cli/web_server.py:3455-3562 (摘要)
_RESIZE_RE = re.compile(rb"\x1b\[RESIZE:(\d+);(\d+)\]")

@app.websocket("/api/pty")
async def pty_ws(ws: WebSocket):
    # --- auth 检查 ---
    if not _DASHBOARD_EMBEDDED_CHAT_ENABLED:
        await ws.close(code=4403); return
    token = ws.query_params.get("token", "")
    if not hmac.compare_digest(token.encode(), _SESSION_TOKEN.encode()):
        await ws.close(code=4401); return
    if not _ws_request_is_allowed(ws):
        await ws.close(code=4403); return
    await ws.accept()

    # --- 平台兜底 ---
    if not _PTY_BRIDGE_AVAILABLE:
        await ws.send_text(
            "\r\n\x1b[31mChat unavailable: 需要 POSIX PTY,Windows 用 WSL2.\x1b[0m\r\n"
        )
        await ws.close(code=1011); return

    # --- spawn `hermes --tui` 子进程 ---
    argv, cwd, env = _resolve_chat_argv(resume=resume, sidecar_url=sidecar_url)
    bridge = PtyBridge.spawn(argv, cwd=cwd, env=env)
    loop = asyncio.get_running_loop()

    # --- 读 PTY 写 WS ---
    async def pump_pty_to_ws():
        while True:
            chunk = await loop.run_in_executor(
                None, bridge.read, _PTY_READ_CHUNK_TIMEOUT)
            if chunk is None: return     # EOF
            if not chunk:                  # 本 tick 无数据
                await asyncio.sleep(0); continue
            await ws.send_bytes(chunk)
    reader_task = asyncio.create_task(pump_pty_to_ws())

    # --- 读 WS 写 PTY,识别 resize escape ---
    try:
        while True:
            msg = await ws.receive()
            if msg["type"] == "websocket.disconnect": break
            raw = msg.get("bytes")
            if raw is None:
                text = msg.get("text")
                raw = text.encode("utf-8") if isinstance(text, str) else b""
            if not raw: continue

            # Resize escape 本地消费,不写 PTY
            match = _RESIZE_RE.match(raw)
            if match and match.end() == len(raw):
                cols = int(match.group(1))
                rows = int(match.group(2))
                bridge.resize(cols=cols, rows=rows)
                continue

            bridge.write(raw)
    except WebSocketDisconnect:
        pass
    finally:
        reader_task.cancel()
        bridge.close()

这里有几个非常巧妙的设计:

  1. 自定义 resize escape \x1b[RESIZE:cols;rows]:不是标准 ANSI(标准没法表达 resize), 但浏览器 xterm.js fit addon 检测窗口大小后合成一个,服务器端识别它,转 TIOCSWINSZ。 标准协议外的 in-band 信号。
  2. run_in_executor 把同步 PTY read 包装成 async: PTY 读是阻塞的,丢给默认 thread pool,asyncio loop 不卡。
  3. Ephemeral SESSION_TOKEN:每次启动随机生成 32 字节 token,只通过同源 SPA HTML 注入。 防止任意源 WebSocket 连接。
  4. WebSocket 没法在 upgrade 时设 Authorization header,所以 token 走 query param。 用 hmac.compare_digest 防 timing attack。

17.4.4 Windows 优雅降级

Native Windows 没 POSIX PTY。Hermes 在"功能消费点"而非启动时检测:

if not _PTY_BRIDGE_AVAILABLE:
    await ws.send_text(
        "\r\n\x1b[31mChat unavailable: the embedded terminal requires a "
        "POSIX PTY, which native Windows Python doesn't provide.\x1b[0m\r\n"
        "\x1b[33mInstall Hermes inside WSL2 to use the dashboard's /chat "
        "tab — the rest of the dashboard works here.\x1b[0m\r\n"
    )
    await ws.close(code=1011)

用户体验:

不崩、不烦、信息精确。这种"局部 degrade"是好的 cross-platform 工程范式。

17.5三种界面共用一个 AIAgent

看三段 AIAgent 构造代码,差异令人惊讶地小:

CLI(cli.py:4819-4865 摘要)

self.agent = AIAgent(
    ...
    platform="cli",
    quiet_mode=not self.verbose,
    clarify_callback=self._clarify_callback,
    reasoning_callback=self._current_reasoning_callback(),
    thinking_callback=self._on_thinking,
    tool_progress_callback=self._on_tool_progress,
    tool_start_callback=self._on_tool_start if self._inline_diffs_enabled else None,
    tool_complete_callback=self._on_tool_complete,
    stream_delta_callback=self._stream_delta if self.streaming_enabled else None,
)

TUI(tui_gateway/server.py:2000-2053 摘要)

return AIAgent(
    ...
    platform="tui",
    quiet_mode=True,
    ephemeral_system_prompt=system_prompt or None,
    **_agent_cbs(sid),    # 所有 callback 走 JSON-RPC event 发给 Ink
)

ACP(acp_adapter/session.py:596-628 摘要)

kwargs = {
    "platform": "acp",
    "enabled_toolsets": _expand_acp_enabled_toolsets(["hermes-acp"], ...),
    "quiet_mode": True,
    ...
}
agent = AIAgent(**kwargs)
# ACP stdio: stdout 必须 JSON-RPC,把人可读输出转 stderr
agent._print_fn = _acp_stderr_print

差异表:

方面CLITUIACP
platform"cli""tui""acp"
quiet_modeFalse(verbose)TrueTrue
工具 callbackrich progress + inline diffJSON-RPC eventACP framework
edit approval交互式 promptJSON-RPC approval.requestContextVar + permissions
输出去向stdout/stderrJSON-RPC transportACP notifications
执行模型主线程阻塞thread poolasyncio loop

所有不变的都是工具发现、MCP registry 加载、provider credential 解析、agent 自身逻辑。 所有的都是callback + 平台标签——非常薄的接口。这是"Single Core, N Surfaces" 设计模式的教科书示例。

17.65 条设计原则提炼

  1. 核心 Agent 是不可知。AIAgent 不知道是 CLI/TUI/ACP 用它。 所有差异通过 callback 注入,通过 platform 字符串标记。
  2. 所有 IDE-class 表面用 JSON-RPC(stdio 或 WebSocket)。一份消息格式覆盖所有 transport。
  3. Capability 抽象超越具体 UI。Edit approval 在 ACP 用 ContextVar, 在 TUI 用 JSON-RPC event,在 CLI 用同步 prompt。三种实现,同一个"capability"。
  4. 长 handler 必须异步。短 RPC 主线程同步;长 RPC(session 加载、slash 执行、shell) 丢 thread pool。让短 RPC(打断、approval)随时能跑。
  5. Platform-specific feature 在消费点降级。 不在启动时检测 PTY,在用户打开 /chat tab 那一刻才检测+提示。其他功能继续可用。

17.7本章带走的

这章对你做产品的意义 下次你要给一个 LLM agent 加一个新 UI 表面(比如 Slack DM、移动 app、VR 终端), 这一章告诉你:不要重写 agent 逻辑。新建一个 callback bundle、一个 transport 实现、 一个 platform 字符串。Core 不动。Hermes 用 3 种表面验证了这个模式的可扩展性。

章末练习

  1. Easy ACP 为什么用 ContextVar 而不是逐层传 approval_cb 参数?用 50 字回答。
  2. Easy Dashboard 的 resize escape \x1b[RESIZE:cols;rows] 不是标准 ANSI。 为什么用 in-band(夹在字节流里)而不是 out-of-band(单独 JSON 消息)?
  3. Medium _SlashWorker 是每 session 一个常驻 subprocess。如果有 100 个并发 TUI session, 就有 100 个 subprocess。Hermes 怎么管理这些?(提示:idle timeout、session GC)
  4. Medium TUI dispatch 把 8 个 method 标为 _LONG_HANDLERS 走 thread pool。 如果未来要加一个 cron.tick 方法,应不应该加进 long handlers?讨论判断标准。
  5. Hard 设计一个新 IDE-class 表面:VR/AR Agent UI(Apple Vision Pro)。 它需要什么 callback、什么 transport? Capability "vision-based file approval"(用户看着 diff 点头 = 同意)怎么落地?
  6. Hard Dashboard 嵌入整个 hermes --tui 进程,而不是直接调 AIAgent。 讨论这个决策的 trade-off:为什么不写一个 React TUI 直接复用 tui_gateway? (提示:Ink + xterm.js 显示差异、心智模型一致性)