4.7 嵌入式 Harness 合约 —— 你的稳定宿主 API

本节你会学到
为什么 agentao.host 要和 Transport / AgentEvent 并存
它暴露的三个表面（事件、策略快照、能力协议）以及各自解决的问题
agent.events() vs Transport(on_event=…) —— 何时用哪个（或都用）
agent.astream() —— 流式输出一轮 assistant 文本的稳定方式
端到端：~30 行写出一条租户审计流水线，跨 Agentao 版本升级也不会断

如果你看完 4.2 AgentEvent 顶部的 :::warning 知道"生产环境用 HostEvent"，本章就是这条建议背后的怎么做。

4.7.1 它解决什么问题

你在做多租户 SaaS。Agent 的每个动作都要落一行审计：租户、用户、工具名、参数、是否批准、是否成功。

你基于 AgentEvent（4.2 节）写好了，跑得很顺，上线。三个月后 Agentao 的下一个次版本发布，EventType.MEMORY_WRITE 改名了、两个 data 字段被挪了位置——你的审计流水线静默地漏行漏了一周，直到 ETL 计数对不上才被人发现。

这就是字段漂移问题。AgentEvent 是运行时的内部事件总线——驱动 CLI、调试 UI、replay 机制——这些消费方需要丰富细节，且能在每次发布时承受变更。生产宿主承受不起这种代价。

嵌入式 Harness 合约就是答案：agentao.host 下一组刻意的小表面，它的特征是：

以 Pydantic 模型冻结 —— 字段和类型是公开合约的一部分
schema 已快照到 docs/schema/host.events.v2.json —— CI 强制字节级一致
是内部事件的红线投影 —— 比如用户 prompt 文本不会出现在审计 body 里
有版本号 —— 加可选字段向后兼容；删字段或改名要 schema 版本号升级

只要你的代码只触到 agentao.host（加上有文档保证的 Agentao(...) 构造器和 chat() / arun() / astream() / events() / active_permissions() 方法），你就能保持向前兼容。

Harness 是运行时的边界，不是运行时本身。

合约把观测、策略、wire schema 三个表面，围绕一次 Agentao session 类型化。它不是一个 turnkey 聊天 runtime：驱动一轮还是用 agent.arun()（或 agent.chat()），或者用 agent.astream() 驱动并流式拿到 assistant 文本（见下文）。reasoning 文本和原始工具 I/O 仍不在合约内：要它们得从内部 Transport 或 ACP 选一个面。把 harness 想成发动机外面的"整车厂连接器"，不是发动机本身。

4.7.2 三个表面

agentao.host 暴露三个不同的表面。它们住在同一个 package 是因为共享"稳定宿主合约"的承诺，但解决的问题不一样：

表面	你怎么用	拿到什么	出处
事件	`agent.events()` 异步迭代器	一串 `HostEvent`（工具 / 子 agent / 权限三种生命周期）	审计、可观测、实时 UI
策略快照	`agent.active_permissions()`	JSON 安全的 `ActivePermissions`（mode + rules + sources）	设置 UI、审计富化、合规报告
能力协议	`from agentao.host.protocols import FileSystem, ShellExecutor, MCPRegistry, MemoryStore`	可注入 Docker / 虚拟 FS / 审计代理 / 程序化 MCP / 远程记忆后端的运行时 Protocol	见 2.2 第 3 档 · 能力协议与 6.4；端到端示例 `examples/protocol-injection/`

流式输出一轮的 assistant 文本也有稳定路径：agent.astream()——见 4.7.4 里的流式文本。

本章聚焦事件和策略快照——大多数读者最先用到的两块。能力协议在它们的构造时上下文里已经讲过；想看一次性替换全部四个槽位的可运行端到端形态，见 examples/protocol-injection/。

package 里还有第四件东西：ACP schema 面。

export_host_acp_json_schema() 暴露的是 Pydantic 化的 wire schema，给那些通过 ACP stdio 协议 进程外驱动 Agentao 的宿主用（IDE 插件、Node/Go/Rust 前端、微服务）。它不像上面三个那样是"消费 API"——而是给协议实现方的合约产物。进程内嵌入可以忽略；进程外嵌入应该参考这份 snapshot，而不是从运行时 trace 里反推 payload 形状。

你需要哪些名字

agentao.host 导出的名字比一次集成会用到的多。它们稳定程度相同，分层只说明什么时候用到哪个。每个名字一行说明的完整清单见附录 A.10。

你在…	名字
观察运行中的 agent	`HostEvent`、`ToolLifecycleEvent`、`SubagentLifecycleEvent`、`PermissionDecisionEvent`、`ActivePermissions`，以及 `SubagentUsage`（子代理终态事件 `usage` 的类型）
流式输出文本，或读取一轮怎么结束的	`TextDelta`、`TurnOutcome`（下文）
从别处取消一轮	`CancellationToken`（2.6）
写宿主工具	继承 `Tool`（同步）或 `AsyncToolBase`（异步）。`RegistrableTool` 只用于类型标注，例如传给 `extra_tools=` 的列表（5.1）
挂第二个 `events()` 迭代器	`StreamSubscribeError`（4.7.5）
仅供参考	`EventStream`（运行时把流设为私有，通过 `agent.events()` 或 `agent.add_host_event_observer()` 访问）、`RFC3339UTCString`、`export_host_event_json_schema`、`export_host_acp_json_schema`（4.7.7）

4.7.3 三种事件类型

三个正交的生命周期事实。每个都是一个 Pydantic 模型，承载刚好够写一行审计的上下文，并通过 event_type 字段（discriminator）让你 isinstance 分派。

事件	阶段	何时触发
`ToolLifecycleEvent`	`started` · `completed` · `failed`	任何工具调用（内置或自定义）。取消会以 `phase="failed", outcome="cancelled"` 形式出现。
`PermissionDecisionEvent`	（单次决策，无阶段）	每次权限决策：`allow` / `deny` / `prompt`。消费方必须把 allow 也消化掉——审计行需要它。
`SubagentLifecycleEvent`	`spawned` · `completed` · `failed` · `cancelled`	子 agent 任务的生命周期。注意：这里 `cancelled` 是独立阶段（与工具事件不同）。`failed` 覆盖两种形态——见下。终态阶段带 `usage`（0.5.2）：子 agent 的请求上报的各项 token 计数，事件触发时已计入父级的会话累计。

failed 不只是"它崩了"

SubagentLifecycleEvent(phase="failed") 既会在子 agent 抛异常时触发，也会在它跑完却始终没给出答案时触发（空 turn、只有 reasoning、被长度截断、doom-loop 被拦停、LLM 调用失败、turn 预算耗尽）。把后者报成 completed 会让契约陈述一件不真实的事。

如果你在 phase == "failed" 上呼叫 on-call 或开事故单，这个告警现在也会被普通的"没答出来"触发。请用 error_type 区分：

"incomplete:<reason>" → 返回了但没答出来。<reason> ∈ no_output、reasoning_only、length_truncated、doom_loop、llm_error（即 TurnOutcome.incomplete_reason 的词表），外加 max_iterations（turn 预算——它是另一根轴，所以没被并进那个封闭集合）。还可能出现兜底值 cancelled / error / unknown；任何不认识的后缀都按"停在半路、原因未分类"处理。
其他任意值 → 异常类名，例如 "ValueError"。那才是真崩了。

取消不受影响：它仍然以 phase="cancelled" 到达。

HostEvent 是这三者的 discriminated union。用 isinstance 分支：

python

from agentao.host import (
    HostEvent,
    ToolLifecycleEvent,
    SubagentLifecycleEvent,
    PermissionDecisionEvent,
)

async for ev in agent.events():
    if isinstance(ev, ToolLifecycleEvent):
        ...
    elif isinstance(ev, PermissionDecisionEvent):
        ...
    elif isinstance(ev, SubagentLifecycleEvent):
        ...

完整字段表：附录 A.10。schema 文件：docs/schema/host.events.v2.json。

4.7.4 `agent.events()` vs `Transport(on_event=…)` —— 怎么选

两者都送事件，但服务的目的不一样。别把它们当对立选项——给每个消费者挑对的那个：

问题	用 `agent.events()`（harness）	用 `Transport(on_event=…)`
是给生产宿主用的，要前向兼容？	✅	❌ —— 字段会漂移
需要把 assistant 文本流式推给 UI？	❌ —— 投影时被剔掉了；用 `agent.astream()`（见下）	⚠️ 能用（`LLM_TEXT`），但属内部接口 —— 优先 `astream()`
需要 reasoning 文本（`THINKING`）或原始工具 I/O？	❌ —— 投影时被剔掉了	✅ —— 只有这里有
在做审计流水线 / SIEM 摄入 / 计费打表？	✅	❌
在做需要内部细节的 CLI / 调试工具？	❌ —— 投影太瘦	✅
需要异步 pull + 背压语义？	✅ —— `async for` + 有界队列	❌ —— 推送回调
需要多个并发消费者？	⚠️ 每个 `session_id` 过滤条件一个异步迭代器（MVP）；多个消费者用 `agent.add_host_event_observer`	✅ —— 自己加分发器扇出

大多数生产部署两者都用：events() 驱动审计 / 可观测流水线；UI 通过 astream() 流式显示回答，只有合约不带的东西（reasoning、原始工具 I/O）才额外挂 Transport 回调。它们零代码路径共享，互不干扰。

流式文本：`agent.astream()`

agent.astream(user_message, *, max_iterations=100, images=None, cancellation_token=None) 运行一轮，运行中产出 TextDelta，最后一项是本轮的 TurnOutcome：

python

from contextlib import aclosing  # Python 3.10+
from agentao.host import TextDelta, TurnOutcome

async with aclosing(agent.astream(prompt)) as stream:
    async for item in stream:
        if isinstance(item, TextDelta):
            send_to_ui(item.text)
        else:                         # TurnOutcome，总是最后一项
            store(item.text) if item.is_answer else report(item)

增量用于显示，结果以 outcome 为准。 所有增量拼起来不等于 TurnOutcome.text：本轮里每一次 LLM 调用都会流出文本，包括以工具调用结束的那次，所以"我先看一下文件"这类说明文字会流出来，却不在最终文本里。最终文本也可能是没有任何增量带过的字符串——[No response] 占位、中止说明或 [LLM API error: …]。增量拿来显示；要保存或据以行动的取 TurnOutcome.text，并先用 .is_answer 检查。
提前离开时要关闭它。 单用 break 不会关闭异步生成器：它要等被垃圾回收或事件循环关闭时才关闭，只要还有引用持有它，本轮就不会结束：它一直流到队列满（64 个增量），然后停在那里等待，同时一直占着 agent，之后的轮次会抛 TurnInProgressError。aclosing(...) 会关闭它。关闭流，或取消消费它的任务，都会取消本轮，并像被取消的 arun() 一样有界地等待清理。
结果属于本次调用。 只有本轮返回时才产出 TurnOutcome：有答案的、没有答案的（is_answer 为假），或被取消的（status="cancelled"）。抛出异常的一轮不产出结果：异常在已流出的文本之后从迭代器抛出，agent.last_turn 以 status="error" 记下它。agent.last_turn 是最近一轮的结果，可能属于另一个调用方。
一次只跑一轮。 同一个 agent 上并发的第二轮在 worker 开始执行时从迭代器抛出 TurnInProgressError；被拒绝的流不会收到正在运行那一轮的任何内容；它没有开始任何一轮，所以 agent.last_turn 不会为它更新。
cancellation_token= 会关联到流自己的 token——取消它就结束本轮。关联是单向的：关闭流而取消的一轮不会取消你的 token，所以要看结果的 status。队列和 events() 一样有界：消费方慢，本轮就变慢，内存不会增长。
不要为了抓文本而替换 agent.transport。 astream() 通过订阅 agent 的 transport 接入，从不替换它（NullTransport 和 SdkTransport 都支持 subscribe()；没有它的 transport 会让 astream() 在本轮开始前抛 TypeError）。按请求替换 transport 行不通：工具执行器持有自己的 transport 引用，工具事件和确认仍会发到旧 transport，replay 也会丢掉它的适配器。
流里没有的： reasoning 文本、工具和权限事件（仍在 agent.events() 上），以及子 agent 的文本（子 agent 运行在自己的 transport 上）。TextDelta 和 TurnOutcome 不是 HostEvent 成员，不投影进 replay，也不在 host.events.v2.json 里。

4.7.5 端到端：租户级审计流水线

两个可直接跑的入口

入门 —— examples/host_events.py：最小版，每条 HostEvent 打到 stdout。~50 行，OPENAI_API_KEY=sk-... uv run python examples/host_events.py 即跑。
生产模式 —— examples/host_audit_pipeline.py：下面这套审计循环的完整版，带 SQLite 持久化 + 跑完后 dump 审计表。

下面这套是 schema 稳定的代码骨架；任挑一个 example clone 下来 60 秒内就能看到真实输出。

下面是完整模式。每个工具调用、权限决策、子 agent 动作都对应一行审计——schema 跨 Agentao 版本稳定。下面字段名都是 agentao/host/models.py 真实定义的；完整类型签名见附录 A.10。

python

"""租户审计流水线。和 agent.arun() 并行运行。"""
import asyncio
import json
from agentao import Agentao
from agentao.host import (
    ToolLifecycleEvent,
    PermissionDecisionEvent,
    SubagentLifecycleEvent,
)

async def audit_loop(agent: Agentao, tenant_id: str, db):
    """消费 harness 事件，每条事实写一行审计。"""
    async for ev in agent.events():
        row = {
            "tenant_id":  tenant_id,
            "session_id": ev.session_id,
            "event_type": ev.event_type,  # discriminator
        }

        if isinstance(ev, ToolLifecycleEvent):
            # started_at 一定有；completed_at 在结束/失败时被填。
            ts = ev.completed_at or ev.started_at
            row.update({
                "ts":           ts,
                "tool_call_id": ev.tool_call_id,
                "tool_name":    ev.tool_name,
                "phase":        ev.phase,        # started | completed | failed
                "outcome":      ev.outcome,      # ok | error | cancelled
                "summary":      ev.summary,      # 已脱敏的宿主字符串
                "error_type":   ev.error_type,
            })
        elif isinstance(ev, PermissionDecisionEvent):
            row.update({
                "ts":             ev.decided_at,
                "tool_call_id":   ev.tool_call_id,
                "tool_name":      ev.tool_name,
                "decision_id":    ev.decision_id,
                "outcome":        ev.outcome,     # allow | deny | prompt
                "mode":           ev.mode,
                "matched_rule":   ev.matched_rule,    # dict 或 None
                "loaded_sources": ev.loaded_sources,  # list[str]
                "reason":         ev.reason,
            })
        elif isinstance(ev, SubagentLifecycleEvent):
            row.update({
                "ts":                ev.completed_at or ev.started_at,
                "child_session_id":  ev.child_session_id,
                "child_task_id":     ev.child_task_id,
                "phase":             ev.phase,    # spawned|completed|failed|cancelled
                # failed = 抛异常 或 跑完没答出来；后者带
                # "incomplete:<reason>" 前缀
                "error_type":        ev.error_type,
                "task_summary":      ev.task_summary,
            })

        await db.execute(
            "INSERT INTO agent_audit (tenant_id, session_id, ts, event_type, payload) "
            "VALUES ($1, $2, $3, $4, $5)",
            row["tenant_id"], row["session_id"], row["ts"],
            ev.event_type, json.dumps(row),
        )

# 与 arun() 一起接线
async def handle_request(tenant_id: str, message: str, db):
    agent = make_agent_for_session(tenant_id, ...)  # 你自己的工厂
    audit = asyncio.create_task(audit_loop(agent, tenant_id, db))
    try:
        reply = await agent.arun(message)
        return reply
    finally:
        audit.cancel()                # cancel 会释放队列和订阅
        await agent.aclose()

这套模式为什么稳：

同会话顺序由合约保证——你的审计行顺序就是事件顺序。
同一个 tool_call_id 的 PermissionDecisionEvent 永远在 ToolLifecycleEvent(phase="started") 之前——下游可以拼起来。
慢消费者不会丢事件：背压走有界队列，生产者会被阻塞而不是默默丢事件。
先启动迭代器，再开始这一轮。 事件不重放：第一次订阅之前发出的事件直接丢弃，中途开始的订阅者只收到之后的事件。取消迭代器会释放它的队列和订阅。
每个 session_id 过滤条件只支持一个异步迭代器。第一个还开着时，同一过滤条件上的第二个会在首次迭代时抛 StreamSubscribeError。要多个消费者，用 agent.add_host_event_observer()（规则）。
之后某个 Agentao 版本新增了内部事件变体时，你的审计流水线根本不会注意到——那种事件不会被投影到 harness；harness 自己如果增加新变体，也只会加可选字段。

4.7.6 `agent.active_permissions()` —— 策略快照

设置页要展示"本会话可以：读 / 写 / 访问这些域名"时，你不想偷看内部 PermissionEngine。用公开快照：

python

snap = agent.active_permissions()

snap.mode             # Literal: "read-only" | "workspace-write" | "full-access" | "plan"
snap.rules            # list[dict] —— 解析后的规则
snap.loaded_sources   # list[str] —— 来源标签

loaded_sources 用稳定字符串标签：

preset:<mode> —— 内置预设（如 preset:workspace-write）
user:<path> —— ~/.agentao/permissions.json 的用户级
injected:<name> —— 宿主用 add_loaded_source() 注入的策略
default:no-engine —— 没配引擎时的兜底

default:no-engine 表示这个 agent 没有引擎。此时快照报告 mode="workspace-write"、rules 为空，但不评估任何规则，连那个预设的规则也不评估，所有询问都交给 transport（5.4）。

快照有缓存。set_mode() 和带新标签的 add_loaded_source() 会使缓存失效，直接原地修改 engine.rules 不会（5.4）。

没有 project:<path> 标签。 项目级 <wd>/.agentao/permissions.json 故意不加载 —— 见 5.4。需要项目感知策略的 host 应通过 add_loaded_source("injected:<name>") + 自己的规则层注入。

会话开始时把这个快照钉进审计日志，事后查"那时候到底是什么策略生效"就不用回放整个引擎。

4.7.7 前向兼容承诺

agentao.host 承诺什么：

加字段：读取这些模型的代码继续工作。
JSON 快照是封闭的。 模型是 extra="forbid"，所以快照写的是 additionalProperties: false，用它构建的校验器会拒绝它没有列出的字段。因此给事件模型加字段，和删字段、改名一样，都是一个新的快照版本。已发布的快照永远不会重写：它继续描述写出它的那些版本，CI 会检查它的摘要。
迁移校验器： host.events.v1.json 描述到 0.5.14 为止的事件。自 0.5.15（#550）起 SubagentUsage 有 reasoning_tokens，并且总会发送，所以 v1 校验器会拒绝每个带 usage 的终态 SubagentLifecycleEvent。请改用 host.events.v2.json 校验。读取模型、或忽略未知键的代码不用改。
内部类型（agentao.transport.AgentEvent / agentao.tools.ToolExecutionResult / agentao.permissions.PermissionEngine）任何版本都可能变。不要直接 import 进生产代码路径。
schema 快照由 CI 强制：tests/test_host_schema.py 会从 Pydantic 模型重生成 schema，做字节级断言——一个改动同时改了模型和 wire 形状但忘了更新 schema 时，CI 会失败。

运营上这给你什么：你今天对接的 harness 合约，到 0.9.x 还是同一份合约。这是对这个表面的承诺，不是对整个包的承诺 —— 0.5.0 在移除 agentao.harness、agentao.session 和 Agentao(...) 上八个回调参数的同时，agentao.host 的每个模型和两份 schema 快照都原样未动。所以包本身仍按 0.x 的常规方式锁：agentao>=0.5.0,<0.6，每个次版本读一遍迁移指南；合约带给你的是：你升级的时候，宿主里消费事件的那一半不用跟着动。

4.7.8 不在合约里的东西

Harness 故意不暴露：

公开 agent graph / descendants store API
宿主侧的 hooks list / disable API（规则作者视角见 §5.7 插件 Hooks）
宿主侧的 MCP reload / lifecycle 事件
本地插件 export/import；远程插件分享
外部会话 import
生成的客户端 SDK

CLI 可能基于同一套事件构建自己的 UI，但它的 stores 和命令不会被提升到 harness API 表面。

如果你发现非伸手到 agentao.host 之外才能拿到的东西——先开 issue，不要依赖内部类型。

4.7.9 决策流速查

Q: 我要消费 Agent 事件，用哪个表面？
│
├─ 驱动一轮 / 拿最终回答？
│      → agent.arun() 或 agent.chat()  (Part 2)
│
├─ 把 assistant 文本流式推给 UI？
│      → agent.astream()               (§ 4.7.4)
│
├─ reasoning 文本、原始工具输出、LLM 调用细节？
│      → Transport(on_event=…)         (Part 4.3，内部接口)
│
├─ 审计 / SIEM / 计费 / 合规？
│      → agent.events()                (本章)
│
├─ 设置 UI 里展示当前策略？
│      → agent.active_permissions()    (§ 4.7.6)
│
├─ 把 IO 路由到 Docker / 虚拟 FS / 审计代理？
│      → from agentao.host.protocols import FileSystem, ShellExecutor
│        (Part 2.2 / Part 6.4)
│
├─ 从非 Python 宿主（IDE、Node、Go、Rust）驱动 Agentao？
│      → ACP stdio 协议                (Part 3.1)
│        （wire 类型用 export_host_acp_json_schema()）
│
└─ 其他？ → 先看附录 A.10，再考虑提 issue。

TL;DR

agentao.host 是稳定的、schema 快照的、前向兼容的宿主表面。 生产代码就 pin 这个。
三个表面：events() 接事件流、active_permissions() 取策略快照、agentao.host.protocols 注入能力——另有 astream() 流式输出一轮的文本。
回答文本用 astream()，审计 / 可观测用 events()，Transport 只用于两者都不带的东西（reasoning、原始工具 I/O）。增量用于显示，结果以最后的 TurnOutcome 为准。
isinstance 分派 HostEvent 把事件路由到对应 handler。三种事件是正交的生命周期事实，不是层级关系。
30 行 + 一张数据库表就能落出能扛住版本升级的租户审计流水线。

→ 参考速查：附录 A.10 · 嵌入 Harness 合约 → Schema：docs/schema/host.events.v2.json → 设计文档：docs/design/embedded-host-contract.md

→ 下一节：第 5 部分 · 扩展 Agent 行为

4.7 嵌入式 Harness 合约 —— 你的稳定宿主 API ​

4.7.1 它解决什么问题 ​

4.7.2 三个表面 ​

你需要哪些名字 ​

4.7.3 三种事件类型 ​

4.7.4 agent.events() vs Transport(on_event=…) —— 怎么选 ​

流式文本：agent.astream() ​

4.7.5 端到端：租户级审计流水线 ​

4.7.6 agent.active_permissions() —— 策略快照 ​

4.7.7 前向兼容承诺 ​

4.7.8 不在合约里的东西 ​

4.7.9 决策流速查 ​

TL;DR ​

4.7 嵌入式 Harness 合约 —— 你的稳定宿主 API

4.7.1 它解决什么问题

4.7.2 三个表面

你需要哪些名字

4.7.3 三种事件类型

4.7.4 `agent.events()` vs `Transport(on_event=…)` —— 怎么选

流式文本：`agent.astream()`

4.7.5 端到端：租户级审计流水线

4.7.6 `agent.active_permissions()` —— 策略快照

4.7.7 前向兼容承诺

4.7.8 不在合约里的东西

4.7.9 决策流速查

TL;DR