4.7 The Embedded Harness Contract — Your Stable Host API

What you'll learn
Why there's a separate agentao.host package alongside Transport / AgentEvent
The three surfaces it exposes (events, policy snapshot, capability protocols) and which problem each solves
agent.events() vs. Transport(on_event=…) — when to use which (or both)
agent.astream() — the stable way to stream a turn's assistant text
End-to-end: a tenant-scoped audit pipeline in ~30 lines that won't break on the next Agentao release

If you've read 4.2 AgentEvent and the :::warning told you "use HostEvent instead for production", this chapter is the how behind that advice.

4.7.1 The problem this solves

You're shipping a multi-tenant SaaS. Every action the agent takes has to produce an audit row: tenant, user, tool name, args, was it approved, did it succeed.

You build it on top of AgentEvent (Part 4.2), it works, you ship. Three months later the next Agentao minor lands. EventType.MEMORY_WRITE got renamed. Two data fields moved. Your audit pipeline silently misses rows for a week before someone notices the ETL count diverging.

This is the churn problem. AgentEvent is the runtime's internal event bus — it powers the CLI, debug UI, and replay machinery, all of which need rich detail and can absorb churn release-by-release. A production host can't.

The embedded harness contract is the answer: a deliberately small surface at agentao.host that's:

Frozen as Pydantic models — fields and types are part of the public contract
Schema-snapshotted in docs/schema/host.events.v2.json — byte-equality enforced in CI
A redacted projection of internal events — e.g. user prompt text isn't in the audit body
Versioned — adding optional fields is backwards-compatible; removing or renaming requires a schema bump

If your code only touches agentao.host (plus the documented Agentao(...) constructor and chat() / arun() / astream() / events() / active_permissions() methods), you stay forward-compatible.

Harness wraps the runtime — it doesn't replace it.

The harness contract types the observability, policy, and wire-schema surfaces around an Agentao session. It is not a turnkey chat runtime: you still call agent.arun() (or agent.chat()) to drive a turn, or agent.astream() to drive it and stream its assistant text (below). Reasoning text and raw tool I/O are still outside the contract: for those you pick the internal Transport or ACP. Think of harness as the OEM connector around the engine, not the engine itself.

4.7.2 The three surfaces

agentao.host exposes three distinct surfaces. They live in one package because they share the "stable host contract" promise, but they solve different problems:

Surface	What you call	What it gives you	Used in
Events	`agent.events()` async iterator	A stream of `HostEvent` (tool / sub-agent / permission lifecycle)	Audit pipelines, observability, real-time UI
Policy snapshot	`agent.active_permissions()`	A JSON-safe `ActivePermissions` (mode + rules + sources)	Settings UI, audit-log enrichment, compliance reports
Capability protocols	`from agentao.host.protocols import FileSystem, ShellExecutor, MCPRegistry, MemoryStore`	Runtime-checkable Protocols you implement to inject Docker / virtual FS / audit proxies / programmatic MCP / remote memory backends	See 2.2 Tier 3 · capability protocols and 6.4; end-to-end demo `examples/protocol-injection/`

Streaming a turn's assistant text has a stable path too, agent.astream() — see Streaming text in 4.7.4.

This chapter focuses on events and policy snapshot — the parts most readers reach for first. Capability protocols are already covered in their construction-time context; for a runnable end-to-end shape that replaces all four slots at once, see examples/protocol-injection/.

There's also a fourth artifact in the package: the ACP schema surface.

export_host_acp_json_schema() exposes the Pydantic-typed wire schema for hosts that drive Agentao out-of-process (IDE plugin, Node/Go/Rust frontend, microservice) over the ACP stdio protocol. It isn't a consumption API like the three above — it's a contract artifact for protocol implementers. In-process embedders can ignore it; out-of-process embedders should reference the snapshot rather than reverse-engineering payloads from runtime traces.

Which names you need

agentao.host exports more names than one integration meets. All of them are equally stable. The tiers say when you need each one. The full list with one line per name is Appendix A.10.

You are…	Names
Observing a running agent	`HostEvent`, `ToolLifecycleEvent`, `SubagentLifecycleEvent`, `PermissionDecisionEvent`, `ActivePermissions`, and `SubagentUsage` (the type of a terminal sub-agent event's `usage`)
Streaming text or reading how a turn ended	`TextDelta`, `TurnOutcome` (below)
Cancelling a turn from elsewhere	`CancellationToken` (2.6)
Writing a host tool	`Tool` (sync) or `AsyncToolBase` (async) to subclass. `RegistrableTool` only to annotate, for example a list passed as `extra_tools=` (5.1)
Attaching a second `events()` iterator	`StreamSubscribeError` (4.7.5)
Reference only	`EventStream` (the runtime keeps its stream private; reach it through `agent.events()` or `agent.add_host_event_observer()`), `RFC3339UTCString`, `export_host_event_json_schema`, `export_host_acp_json_schema` (4.7.7)

4.7.3 The three event types

Three orthogonal lifecycle facts. Each is a Pydantic model carrying just enough context for an audit row, with a discriminator field (event_type) so you can isinstance-dispatch.

Event	Phases	Fires on
`ToolLifecycleEvent`	`started` · `completed` · `failed`	Every tool call (built-in or custom). Cancellation surfaces as `phase="failed", outcome="cancelled"`.
`PermissionDecisionEvent`	(no phases — single decision per call)	Every permission decision: `allow` / `deny` / `prompt`. Consumers must drain even allow events — the audit row needs them.
`SubagentLifecycleEvent`	`spawned` · `completed` · `failed` · `cancelled`	Sub-agent task lifecycle. Note: `cancelled` is a distinct phase here (unlike tools). `failed` spans two shapes — see below. A terminal phase carries `usage` (0.5.2): the token counts the sub-agent's requests reported, already included in the parent's session totals when the event fires.

failed is not only "it crashed"

SubagentLifecycleEvent(phase="failed") fires both when the sub-agent raised and when it returned without ever answering (empty turn, reasoning only, length-truncated, doom-loop halted, LLM call failed, turn budget exhausted). Reporting the second as completed would make the contract state something untrue.

If you page on-call or open an incident on phase == "failed", that alert now fires on ordinary non-answers too. Discriminate on error_type:

"incomplete:<reason>" → returned without answering. <reason> ∈ no_output, reasoning_only, length_truncated, doom_loop, llm_error (the TurnOutcome.incomplete_reason vocabulary) plus max_iterations (turn budget — a separate axis, so it isn't folded into that closed set). Defensive values cancelled / error / unknown are possible; treat any unrecognized suffix as "stopped short, cause unclassified".
anything else → an exception class name, e.g. "ValueError". That's a real crash.

Cancellation is unaffected: it still arrives as phase="cancelled".

HostEvent is the discriminated union of these three. Use isinstance to branch:

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):
        ...

Full field list: Appendix A.10. The schemas live at docs/schema/host.events.v2.json.

4.7.4 `agent.events()` vs. `Transport(on_event=…)` — when to use which

Both deliver events, but they're for different jobs. Don't pick one instead of the other — pick the right one for each consumer:

Question	Use `agent.events()` (harness)	Use `Transport(on_event=…)`
Is this a production host that needs forward compatibility?	✅	❌ — fields will churn
Do you need the assistant's text streamed to a UI?	❌ — projected out; use `agent.astream()` (below)	⚠️ works (`LLM_TEXT`), but internal — prefer `astream()`
Do you need reasoning text (`THINKING`) or raw tool I/O?	❌ — projected out	✅ — only here
Building an audit pipeline / SIEM feed / billing meter?	✅	❌
Building CLI / debug tooling that wants every internal detail?	❌ — too redacted	✅
Need async pull semantics with backpressure?	✅ — `async for` with bounded queue	❌ — push callback
Need multiple concurrent consumers?	⚠️ One async iterator per `session_id` filter (MVP); several consumers via `agent.add_host_event_observer`	✅ — fan out via your own dispatcher

Most production deployments use both: events() drives the audit / observability pipeline; the UI streams the answer through astream(), and attaches a Transport callback only for what the contract doesn't carry (reasoning, raw tool I/O). They share zero code paths so they don't fight each other.

Streaming text: `agent.astream()`

agent.astream(user_message, *, max_iterations=100, images=None, cancellation_token=None) runs one turn and yields TextDelta items while it runs, then the turn's TurnOutcome as the last item:

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, always the last item
            store(item.text) if item.is_answer else report(item)

Deltas are for display; the outcome is the answer. Joined deltas are not TurnOutcome.text: every LLM call in the turn streams its text, including calls that end in tool calls, so narration like "Let me check the file" streams but is not in the final text. The final text can also be a string no delta carried — the [No response] placeholder, an abort notice, or an [LLM API error: …] string. Show the deltas; store or act on TurnOutcome.text, checked with .is_answer.
Close it when you leave early. break alone does not close an async generator: it is closed when it is garbage-collected or when the event loop shuts down, so while anything still references it, the turn stays open: it streams until the queue is full (64 deltas), then waits there, still holding the agent, and a later turn raises TurnInProgressError. aclosing(...) closes it. Closing, or cancelling the consuming task, cancels the turn and waits (bounded) for its cleanup, like a cancelled arun().
The outcome is this call's. TurnOutcome is yielded only when the turn returned — an answer, a turn with no answer (is_answer false), or a cancelled one (status="cancelled"). A turn that raised yields no outcome: its exception is raised from the iterator after the text already streamed, and agent.last_turn records it with status="error". agent.last_turn is the latest turn's outcome, which may be another caller's.
One turn at a time. A second concurrent turn on the same agent raises TurnInProgressError from the iterator when the worker starts it; a refused stream yields nothing from the running turn, and since it never started a turn, agent.last_turn is not set for it.
cancellation_token= is linked to the stream's own token — cancelling it ends the turn. The link is one way: a turn cancelled by closing the stream does not cancel your token, so read the outcome's status. The queue is bounded like events()'s: a slow consumer slows the turn instead of growing memory.
Never swap agent.transport to capture text. astream() subscribes to the agent's transport and never replaces it (NullTransport and SdkTransport both support subscribe(); a transport without it makes astream() raise TypeError before the turn starts). Replacing the transport per request does not work: the tool runner keeps its own reference, so tool events and confirmations would still go to the old transport, and replay would lose its adapter.
Not in the stream: reasoning text, tool and permission events (they stay on agent.events()), and sub-agent text (sub-agents run on their own transports). TextDelta and TurnOutcome are not HostEvent members, are not projected into replay, and are not in host.events.v2.json.

4.7.5 End-to-end: tenant-scoped audit pipeline

Two runnable starting points

First taste — examples/host_events.py: minimal, prints each HostEvent to stdout. ~50 lines, run with OPENAI_API_KEY=sk-... uv run python examples/host_events.py.
Production pattern — examples/host_audit_pipeline.py: the full audit-loop below, with SQLite persistence + after-turn table dump.

Read on for the schema-stable pattern; clone either example to get hands-on output in 60 seconds.

Here's a complete pattern. Every tool call, permission decision, and sub-agent action gets one row in your audit table — schema-stable across Agentao releases. Field names below are the actual ones in agentao/host/models.py; compare with Appendix A.10 for the full type signatures.

python

"""Tenant audit pipeline. Run alongside 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):
    """Drain harness events and write one audit row per fact."""
    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 always set; completed_at set on completion/failure.
            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,      # redacted host-facing string
                "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 or 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 = raised OR returned without answering; the
                # "incomplete:<reason>" prefix marks the latter.
                "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),
        )

# Wire it up alongside arun()
async def handle_request(tenant_id: str, message: str, db):
    agent = make_agent_for_session(tenant_id, ...)  # your factory
    audit = asyncio.create_task(audit_loop(agent, tenant_id, db))
    try:
        reply = await agent.arun(message)
        return reply
    finally:
        audit.cancel()                # cancellation releases queue/subscription
        await agent.aclose()

Why this pattern is robust:

Same-session ordering is guaranteed by the contract — your audit row order matches event order.
PermissionDecisionEvent precedes the matching ToolLifecycleEvent(phase="started") (same tool_call_id) so downstream views can stitch them.
Dropping a slow consumer doesn't drop events: backpressure is host-pulled via a bounded queue. Events block the producer rather than silently get dropped.
Start the iterator before the turn. There is no replay: events emitted before the first subscription are dropped, and a subscriber that starts mid-turn gets only later events. Cancelling the iterator releases its queue and subscription.
One async iterator per session_id filter. A second one on the same filter raises StreamSubscribeError on its first iteration while the first is open. For several consumers, use agent.add_host_event_observer() (rule).
When a later Agentao release adds a new internal event variant, your audit pipeline doesn't notice — that event isn't projected to harness, and any new HostEvent variant the projection does gain only adds optional fields.

4.7.6 `agent.active_permissions()` — policy snapshots

When your settings UI shows "this session can: read / write / fetch from these domains", you don't want to peek into the internal PermissionEngine. Use the public snapshot:

python

snap = agent.active_permissions()

snap.mode             # Literal: "read-only" | "workspace-write" | "full-access" | "plan"
snap.rules            # list[dict] — the resolved rules
snap.loaded_sources   # list[str] — provenance labels

loaded_sources carries stable string labels:

preset:<mode> — built-in preset (e.g. preset:workspace-write)
user:<path> — user-scoped JSON from ~/.agentao/permissions.json
injected:<name> — host-supplied policy via add_loaded_source()
default:no-engine — fallback when no engine is configured

default:no-engine means the agent has no engine. The snapshot then reports mode="workspace-write" with empty rules, but no rule is evaluated, not even that preset's, and every ask goes to the transport (5.4).

The snapshot is cached. set_mode() and add_loaded_source() with a new label invalidate the cache. Mutating engine.rules in place does not (5.4).

No project:<path> label. Project-scope <wd>/.agentao/permissions.json is intentionally not loaded — see 5.4. Hosts that need project-aware policy should inject it via add_loaded_source("injected:<name>") plus their own rule layer.

Pin this snapshot into your audit log on session start so you can later answer "what policy was active when this happened?" without replaying the whole engine.

4.7.7 Forward-compatibility guarantees

What agentao.host promises:

Adding a field to any model keeps code that reads the models working.
The JSON snapshots are closed. The models are extra="forbid", so a snapshot says additionalProperties: false, and a validator built from it rejects any field it does not list. So adding a field to an event model is a new snapshot version, as removing or renaming one is. A published snapshot is never rewritten: it keeps describing the releases that wrote it, and CI checks its digest.
Migrating a validator: host.events.v1.json describes events up to 0.5.14. Since 0.5.15 (#550) SubagentUsage has reasoning_tokens, always sent, so a v1 validator rejects every terminal SubagentLifecycleEvent that carries usage. Validate against host.events.v2.json instead. Code that reads the models, or ignores unknown keys, needs no change.
Internal types (agentao.transport.AgentEvent, agentao.tools.ToolExecutionResult, agentao.permissions.PermissionEngine) may change in any release. Don't import them directly into production code paths.
Schema snapshots are CI-enforced via tests/test_host_schema.py — a model change that shifts the wire form fails the build until both the model and the snapshot are updated together.

What this gives you operationally: the harness contract you build against today is the one you'll have at 0.9.x. That is a promise about this surface, not about the rest of the package — 0.5.0 kept every agentao.host model and both schema snapshots unchanged while removing agentao.harness, agentao.session and the eight callback kwargs on Agentao(...). So pin the package the ordinary 0.x way, agentao>=0.5.0,<0.6, and read the migration guide at each minor; what the contract buys is that the event-consuming half of your host does not move when you do.

4.7.8 What's not in the contract

The harness deliberately doesn't expose:

Public agent graph / descendants store API
Host-facing hooks list/disable API (rule-author guide: §5.7 Plugin Hooks)
Host-facing MCP reload / lifecycle events
Local plugin export/import; remote plugin share
External session import
Generated client SDKs

The CLI may build on the same events for its own UI, but its stores and commands are not promoted to the harness API.

If you find yourself wanting to reach past agentao.host for something missing, file an issue rather than relying on internal types.

4.7.9 Quick decision flow

Q: I need to react to agent events. Which surface?
│
├─ Driving a turn / getting the final reply?
│      → agent.arun() or agent.chat()  (Part 2)
│
├─ Streaming the assistant's text to a UI?
│      → agent.astream()               (§ 4.7.4)
│
├─ Reasoning text, raw tool output, LLM-call detail?
│      → Transport(on_event=…)         (Part 4.3, internal)
│
├─ Audit / SIEM / billing / compliance?
│      → agent.events()                (this chapter)
│
├─ Showing the active policy in a settings UI?
│      → agent.active_permissions()    (§ 4.7.6)
│
├─ Routing IO through Docker / virtual FS / audit proxy?
│      → from agentao.host.protocols import FileSystem, ShellExecutor
│        (Part 2.2 / Part 6.4)
│
├─ Driving Agentao from a non-Python host (IDE, Node, Go, Rust)?
│      → ACP stdio protocol            (Part 3.1)
│        (use export_host_acp_json_schema() for the wire types)
│
└─ Anything else? → check Appendix A.10, then file an issue.

TL;DR

agentao.host is the stable, schema-snapshotted, forward-compatible host surface. Pin to it for production code.
Three surfaces: events() for streams, active_permissions() for policy snapshots, agentao.host.protocols for capability injection — plus astream() to stream a turn's text.
astream() for the answer text, events() for audit / observability, Transport only for what neither carries (reasoning, raw tool I/O). Deltas are for display; the answer is the final TurnOutcome.
isinstance-dispatch on HostEvent to route to the right handler. The three event types are orthogonal lifecycle facts, not a hierarchy.
30 lines + a database is all it takes to ship a tenant audit pipeline that survives release upgrades.

→ Reference dive: Appendix A.10 · Embedded Harness Contract → Schemas: docs/schema/host.events.v2.json → Design rationale: docs/design/embedded-host-contract.md

→ Next: Part 5 · Extend Agent Behavior

4.7 The Embedded Harness Contract — Your Stable Host API ​

4.7.1 The problem this solves ​

4.7.2 The three surfaces ​

Which names you need ​

4.7.3 The three event types ​

4.7.4 agent.events() vs. Transport(on_event=…) — when to use which ​

Streaming text: agent.astream() ​

4.7.5 End-to-end: tenant-scoped audit pipeline ​

4.7.6 agent.active_permissions() — policy snapshots ​

4.7.7 Forward-compatibility guarantees ​

4.7.8 What's not in the contract ​

4.7.9 Quick decision flow ​

TL;DR ​

4.7 The Embedded Harness Contract — Your Stable Host API

4.7.1 The problem this solves

4.7.2 The three surfaces

Which names you need

4.7.3 The three event types

4.7.4 `agent.events()` vs. `Transport(on_event=…)` — when to use which

Streaming text: `agent.astream()`

4.7.5 End-to-end: tenant-scoped audit pipeline

4.7.6 `agent.active_permissions()` — policy snapshots

4.7.7 Forward-compatibility guarantees

4.7.8 What's not in the contract

4.7.9 Quick decision flow

TL;DR