第 5 章：宿主桥——跨越边界的类型化请求

谁有权决定

上一章结束时留下了一个洞：kernel 里的 Python 几乎无所不能，但有一类操作它无权自行完成——创建子会话、记录目标、压缩上下文、给另一个 agent 发消息。这些操作的权威状态不在解释器里，而在 TypeScript 宿主里：会话转录归宿主写，provider 凭据归宿主拿，调度归宿主排。

官方文档对这条边界的表述是：

"Python skills use typed host requests for capabilities whose authoritative state belongs outside the kernel. … This keeps credentials, provider execution, transcript writes, worker routing, and scheduling out of Python while retaining a programmatic model interface."

「把权威状态留在宿主，同时保留程序化的模型接口」——这句话是整座桥的设计目标。它的实现叫 typed host request：Python 端调用 await rlm.host_request("类型", {负载})，TS 端按类型分发到处理器，结果原路返回。本章走完这座桥的全程。

Python 端：一次 host_request 的解剖

桥的 Python 端在 prime-agent-runtime/src/rlm/__init__.py，全文不到百行，值得完整读一遍：

async def host_request(request_type: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
    """Send a typed request to the Prime Agent host and await its reply. ..."""
    if not isinstance(request_type, str) or not request_type:
        raise TypeError("request_type must be a non-empty str")
    if payload is not None and not isinstance(payload, dict):
        raise TypeError(f"payload must be a dict or None, got {type(payload).__name__}")
    if Comm is None:
        raise RuntimeError("Jupyter comm support is unavailable in this kernel")
    _install_control_comm_handlers()

    loop = asyncio.get_running_loop()
    future: asyncio.Future[dict[str, Any]] = loop.create_future()
    comm = Comm(target_name=HOST_COMM_TARGET, primary=False)

    def _on_msg(msg: dict[str, Any]) -> None:
        ...  # status=="ok" → set_result；"error" → set_exception
        # 全部经 loop.call_soon_threadsafe(...) 回到 cell 的事件循环

    comm.on_msg(_on_msg)
    # request_type goes last so a payload "type" key cannot reroute the request.
    comm.open(data={**(payload or {}), "type": request_type})
    return await future

五个细节，每个都有理由：

1. 载体是 Jupyter comm，不是 stdin_request。comm 有 target_name（这里是常量 HOST_COMM_TARGET = "host.request"）和 comm_id，天然支持多路复用：并发的多个 host request 各用各的 comm，互不干扰；异步任务在 cell 结束后也能新开 comm 发请求。
2. 每次请求新建一个 comm（primary=False），收到回包即 comm.close()。请求与通道一一对应，归属清晰。
3. 回调用 call_soon_threadsafe resolve future。comm 回包在 kernel 的 IO 线程上到达，而 cell 的 await 挂在 asyncio 事件循环里——跨线程唤醒必须走这条安全通道。
4. request_type 放在 payload 合并的最后。注释一语道破：如果 payload 里混进一个 "type" 键，它不能劫持路由。一行代码的顺序就是一条安全规则。
5. 入口先做形状校验。类型非空字符串、payload 是 dict——在过桥之前就把畸形请求挡掉，宿主永远收到结构合法的数据。

还有那个下划线开头的 _install_control_comm_handlers()。它把 comm_msg/comm_close 注册进 kernel.control_handlers——为的是让回包能走 control 通道。为什么非 control 不可？这就到了整座桥最精彩的部分。

一条死锁推理

设想最直觉的设计：宿主处理完请求，把回包从 shell 通道发回去。会发生什么？

shell 通道是串行的——当前 cell 的 execute_request 独占着它。cell 里的 await rlm("subtask") 正在等回包，所以 execute_request 不会结束；而 kernel 不处理完 execute_request，就不会处理 shell 通道上的下一条消息——也就是那条回包。

回包等执行结束，执行等回包到达。死锁。

官方架构文档把这条推理写成了明文：

"IPython processes shell messages serially. Sending the admission response on the shell channel would deadlock: the active execute_request cannot finish until the response arrives, while the kernel will not process that shell response until the request finishes."

解法藏在 Jupyter 协议的通道划分里：control 通道与 shell 并行，专门用于中断、关机和这类「执行期间必须送达」的消息。于是 TS 端的回包发送写成了：

private async sendCommMessage(commId: string, data: Record<string, unknown>): Promise<void> {
	const channel = this.control ?? this.shell;
	...
	const msg = buildMessage("comm_msg", { comm_id: commId, data }, this.session, this.options.username);
	await channel.send(encode(msg, this.connection.key));
}

优先 control，shell 只是没连上 control 时的兜底。而 Python 端手工注册 control_handlers 的怪异举动，正是这个设计的镜像——Jupyter 默认只在 shell 通道分发 comm 消息，要让 control 通道来的 comm_msg 被认出来，必须显式接线。

TS 端：从 comm 到处理器

宿主侧的入口在 KernelManager 的 iopub 泵。comm 消息的分发先于普通输出的归属过滤——因为异步 Python 任务可能在调度它的 cell 已经 idle 之后才打开 comm。识别到 target 为 host.request 的 comm 后：

private startHostRequestFromComm(commId: string, data: unknown): void {
	if (this.handledHostRequestCommIds.has(commId)) {
		return;  // 同一请求的 comm_open 与 comm_msg 只处理一次
	}
	this.handledHostRequestCommIds.add(commId);

	const task = (async () => {
		try {
			const result = await this.handleHostRequest(data);
			await this.sendCommMessage(commId, { status: "ok", ...result });
		} catch (error) {
			await this.sendCommMessage(commId, { status: "error", error: errorMessage(error) });
		}
	})();
	this.inFlightHostRequests.add(task);
	void task.finally(() => { this.inFlightHostRequests.delete(task); });
}

每个 host request 是一个登记在案的 in-flight task；kernel dispose 时会等它们了结（上限 5 秒，HOST_REQUEST_DISPOSE_TIMEOUT_MS）。成功回 {status:"ok", ...result}，失败回 {status:"error", error}——与 Python 端 _on_msg 的三分支（ok/error/unexpected）严格对应。

分发本体只有几行，但有一处值得放大：

private async handleHostRequest(data: unknown): Promise<Record<string, unknown>> {
	...
	const handler = this.options.hostHandlers?.[data.type];
	if (!handler) {
		throw new Error(`host request type "${data.type}" is not available in this session`);
	}
	// Tag the request with the cell that triggered it. A blocking call is still
	// the in-flight execution; detached spawns (asyncio.create_task) fire after
	// the scheduling cell goes idle, so fall back to that last cell's source.
	const cellSourceCode = this.activeExecution?.code ?? this.lastCellCode;
	return handler({ ...data, cellSourceCode });
}

cellSourceCode 把每个请求归属到触发它的 cell：正在执行的取 activeExecution.code；cell 已 idle、由 detached asyncio 任务发起的，回退到 lastCellCode——这个「上一段程序」在执行结束后被特意保留，就是为这条回退链服务的。归属信息的用途第 6 章会看到：子代理的创建要记录「是哪段代码派的」。

处理器表：能力按会话配置开关

处理器在哪注册？AgentSession._createKernelHostHandlers()（core/agent-session.ts，约 L8761）。它的结构本身就是一份 host request 类型清单：

private _createKernelHostHandlers(): HostRequestHandlers {
	const handlers: HostRequestHandlers = {
		"rlm.run": createRlmRunHostHandler(async ({ prompt, kwargs, cellSourceCode }) => ({
			...(await this.runRlmChild(prompt, kwargs, cellSourceCode)),
		})),
		"rlm.find_models": createRlmFindModelsHostHandler(...),
		"rlm.list_subagents": createRlmListSubagentsHostHandler(...),
		"rlm.delete_subagent": createRlmDeleteSubagentHostHandler(...),
		"model.info": async () => ({ id: ..., provider: ..., input: ... }),
	};
	if (this._includeGoals) { /* goal.get / goal.create / goal.complete */ }
	if (this._includeCompactSkill) { /* compact.run / compact.status */ }
	if (this._autoRefineAllowedForSession()) { /* refine.run / refine.status */ }
	if (this._rlmHeartbeatController) { /* rlm_heartbeat.list/create/update/delete */ }
	if (this._agentMessageController && 技能对模型可见) { /* agent_message.list_agents / send */ }
	if (this._agentObserveController) { /* agent_observe.list / get / recent */ }
	// MCP 集成启用时，McpManager.hostHandlers() 再并入：
	// mcp.refresh / mcp.config /（交互式登录接线后）mcp.begin_login
	return handlers;
}

注意这张表的条件注册结构：kernel 里能调什么，精确反映这个会话装配了什么。goals 没启用，goal.create 就不在表里，调用会得到 host request type "goal.create" is not available in this session——一个准确的、可诊断的错误，而不是一次静默的失败或一段死代码。能力发现即错误信息。

验证在哪一侧

桥的分工纪律是：Python 端校验形状，TS 端校验语义。Python 端只管「payload 是 dict、type 是非空字符串」这类形状问题；真正的语义检查全在处理器里。以 rlm.run 为例，createRlmRunHostHandler 会断言 prompt 是字符串、kwargs 是对象，然后 runRlmChild 继续检查递归深度、名称可用性、模型选择——全部在 TS 侧，全部可以访问会话的真实状态。

这个分工的深意是：kernel 里跑的是模型生成的代码。宿主不能假设桥对面发来的任何东西是善意的或正确的；但宿主也不需要防御性地把业务逻辑复制到 Python 侧——因为权威处理器只有一份，在 TS 里。

实践应用

1. 权威状态决定处理器位置。划分「什么跑在嵌入环境、什么跑在宿主」的标准不是难易，而是状态的归属。凭据、转录、调度归宿主，所以它们只能以请求的形式被调用。
2. 先写死锁推理，再选通道。shell 串行 + 执行中等待回包 = 死锁，这个结论花三十秒就能推出来，但它决定了整座桥的形状。协议提供的并行通道（control）就是为这类场景准备的。
3. 能力开关优于功能开关。不是「编译进去、运行时判断」，而是「压根不注册」。缺席的处理器产生准确的错误信息，这本身就是文档。
4. 路由字段最后合并。任何「外部 payload + 内部元数据」合并的场景，内部关键字段都该放在覆盖位——一行代码的顺序就是一条注入防线。
5. 给异步归属留回退链。activeExecution?.code ?? lastCellCode 的模式——当前上下文缺失时退到「最近一次有意义的上下文」——适用于一切需要归属的审计与遥测。

总结

宿主桥用一个 comm target（host.request）、一条 control 回包通道、一张按会话配置组装的处理器表，把「Python 无所不能」和「宿主保留决策权」这两件矛盾的事合在了一起。它的每一处怪异——手工注册 control handler、request_type 最后合并、lastCellCode 回退——都是某个具体约束的投影。

表里最重要的那一项是 rlm.run。下一章看它被调用时发生什么：一次函数调用如何变成一个完整的子 Agent。