references/expectations-audit-listeners.md
# Expectations, audit log and hub listeners (verified against ag2 1.3.0)
## Expectations
The hub's expectation sweeper runs every `expectation_sweep_interval` seconds (default 10; `0` disables the loop). Each tick evaluates `metadata.manifest.expectations` on every active channel, dedupes per `(channel, expectation index, violator)`, fans `on_expectation_fired` out to listeners (the `AuditLog` writes `AUDIT_KIND_EXPECTATION_VIOLATED` here), then calls the handler named by `on_violation`. Unknown evaluator or handler names are silently skipped.
Built-in evaluators (`default_evaluators()`):
| Name | Class | `params["seconds"]` default | Fires when |
|---|---|---|---|
| `acks_within` | `AcksWithinEvaluator` | 30 | channel still `PENDING` and invitees haven't acked |
| `reply_within` | `ReplyWithinEvaluator` | 600 | `ACTIVE`; a participant addressed by `EV_TEXT` hasn't replied |
| `max_silence` | `MaxSilenceEvaluator` | 3600 | `ACTIVE`; no content envelope from anyone |
Built-in handlers (`default_handlers()`):
| `on_violation` | Class | Effect |
|---|---|---|
| `audit` | `AuditHandler` | no-op (the audit record is already written by the listener fan-out) |
| `notify_channel` | `NotifyChannelHandler` | posts `EV_EXPECTATION_VIOLATED` to the channel |
| `auto_close` | `AutoCloseHandler` | closes the channel, `reason="expectation_violated:<name>"` |
Adapter defaults: `consulting` = `acks_within(30, auto_close)` + `reply_within(600, auto_close)`; `conversation` = `max_silence(3600, audit)`; `discussion` = `turn_within(120, warn)` + `turn_within(600, hide)`; `workflow` = `turn_within(120, warn)` + `turn_within(600, auto_close)`. **No `turn_within` evaluator and no `warn` / `hide` handler ship**, so the discussion/workflow defaults are inert until you register them.
### Custom expectations (executed)
Expectations come **only** from the adapter manifest snapshotted at channel creation — `knobs={"expectations": ...}` is ignored. To change them, register an adapter instance whose manifest carries your list (it replaces the built-in adapter for that `(type, version)` for channels created afterwards):
```python
from ag2.network import ConversationAdapter, EV_TEXT, Expectation, ExpectationContext, Violation
class TooManyMessages:
name = "too_many_messages"
def evaluate(self, expectation: Expectation, context: ExpectationContext) -> Violation | None:
# context: metadata, state, wal, now_iso, now_seconds — keep evaluators pure
n = sum(1 for e in context.wal if e.event_type == EV_TEXT)
if n > int(expectation.params["max"]):
return Violation(expectation=expectation, violator_ids=[], detail={"text_count": n})
return None
conv = ConversationAdapter()
conv.manifest.expectations = [
Expectation(name="too_many_messages", on_violation="notify_channel", params={"max": 1}),
]
hub.register_adapter(conv)
hub.register_expectation_evaluator(TooManyMessages())
# hub.register_violation_handler(obj) # obj.name + async handle(self, hub, channel_id, violation)
```
`Violation(expectation, violator_ids=[], detail={})` — `expectation` is the `Expectation` object; there is no `channel_id` field. Empty `violator_ids` = channel-wide.
Deterministic tests: `Hub.open(store, expectation_sweep_interval=0, clock=fake_clock)` then `await hub._expectation_tick()` (test entry point, leading underscore by convention).
## Audit log
`hub.audit_log` (an `AuditLog`, itself the first `HubListener`) appends JSON records to `audit.jsonl` in the hub's `KnowledgeStore` — durable with `DiskKnowledgeStore`, process-lifetime with `MemoryKnowledgeStore`.
| Member | Purpose |
|---|---|
| `await hub.audit_log.read_all()` | every record (`list[dict]`, each has `kind` and `at`) |
| `await hub.audit_log.append(record)` | add your own kinds |
| `hub.audit_log.subscribe(async_cb)` / `unsubscribe(cb)` | live tail, `async def cb(record: dict)` |
| `hub.replace_audit_log(custom)` | swap in an `AuditLog` subclass |
| `hub.health()` | in-memory ops snapshot (includes `audit_log_bytes`) |
| Constant (`kind` string) | Written on | Fields besides `kind`, `at` |
|---|---|---|
| `AUDIT_KIND_AGENT_REGISTERED` (`agent_registered`) | register | `agent_id`, `name` |
| `AUDIT_KIND_AGENT_UNREGISTERED` | unregister | `agent_id`, `name` |
| `AUDIT_KIND_RESUME_SET` | `set_resume` / observation | `agent_id`, `version`, `source` (`RESUME_SOURCE_TENANT` / `RESUME_SOURCE_OBSERVED`; observed also has `capability`, `outcome`) |
| `AUDIT_KIND_SKILL_SET` | `set_skill` | `agent_id`, `removed` |
| `AUDIT_KIND_RULE_SET` | `set_rule` | `agent_id`, `version` |
| `AUDIT_KIND_CHANNEL_CREATED` | channel open | `channel_id`, `manifest_type`, `manifest_version`, `creator_id`, `participants` |
| `AUDIT_KIND_CHANNEL_CLOSED` / `_EXPIRED` | close / TTL | `channel_id`, `reason` |
| `AUDIT_KIND_EXPECTATION_VIOLATED` | violation | `channel_id`, `expectation`, `on_violation`, `params`, `violators`, `detail` |
| `AUDIT_KIND_TASK_TERMINATED` (`task_terminated`) | task reaches completed / failed / expired / cancelled | `task_id`, `owner_id`, `channel_id`, `outcome`, `capability`, `reason` (latency lives in `Resume.observed`) |
| `AUDIT_KIND_TURN_FAILED` | default handler's turn raised | `channel_id`, `agent_id`, `envelope_id`, `exc_type`, `exc_message` |
Records carry ids, not tenants: to filter by `Passport.owner`, join on `await hub.get_agent(agent_id)`.
```python
from ag2.network import AUDIT_KIND_EXPECTATION_VIOLATED
records = await hub.audit_log.read_all()
violations = [r for r in records if r["kind"] == AUDIT_KIND_EXPECTATION_VIOLATED]
on_channel = [r for r in records if r.get("channel_id") == channel_id]
acme_ids = {p.agent_id for p in await hub.list_agents() if p.owner == "acme"}
acme_events = [r for r in records if r.get("agent_id") in acme_ids]
```
## Hub listeners
`hub.register_listener(obj)` / `unregister_listener(obj)`. All methods `async`, awaited sequentially after the state change commits, each wrapped in try/except. Subclass `BaseHubListener` (all no-ops):
| Method | Fires when |
|---|---|
| `on_envelope_posted(envelope, metadata)` | envelope accepted, WAL-appended, dispatched |
| `on_envelope_rejected(envelope, reason)` | arbiter/validation denied a send (`reason` is the `NetworkError`) |
| `on_dispatch_failed(envelope, recipient_id, reason)` | delivery to one recipient raised |
| `on_channel_event(channel_id, kind, payload)` | `opened` / `closed` / `expired` / `participant_removed` / `participant_hidden` |
| `on_agent_event(agent_id, kind, payload)` | `registered` / `unregistered` / `resume_set` / `skill_set` / `rule_set` / `observation_recorded` |
| `on_expectation_fired(channel_id, expectation, violation)` | evaluator emitted a violation |
| `on_turn_failed(channel_id, agent_id, envelope_id, exc)` | a notify-handler turn raised |
| `on_task_event(task_id, kind, payload)` | task reached a terminal state (`completed` / `failed` / `expired` / `cancelled`), or `mirror_failed`; `started` / `progress` are not fanned out by the hub |
| `on_inbox_pressure(agent_id, pending, cap)` | recipient inbox first crosses `limits.inbox.high_water` |
```python
from ag2.network import BaseHubListener
class Metrics(BaseHubListener):
async def on_envelope_rejected(self, envelope, reason):
print("denied", envelope.event_type, reason)
async def on_task_event(self, task_id, kind, payload):
print("task", kind, payload.get("capability"))
hub.register_listener(Metrics())
```
A `Hub` subclass can override the same `on_*` methods directly instead of registering a listener.
## Custom sweepers
```python
async def refresh_tokens() -> None: ...
hub.register_sweeper("refresh_tokens", 60.0, refresh_tokens) # sync call; starts now if hub started
await hub.unregister_sweeper("refresh_tokens") # async!
```
Re-registering an existing name raises `ValueError`.
references/rules-arbiter-auth.md
# Rules, arbiter and authentication (verified against ag2 1.3.0)
## `Rule` — field reference
```python
from ag2.network import (
AccessBlock, ChannelTypeAccess, InboxBlock, LimitsBlock, RateBlock, Rule,
)
rule = Rule(
access=AccessBlock(
outbound_to=["bob", "research-*"], # fnmatch globs over Passport.name; default ["*"]
inbound_from=["*"], # who may address / invite this agent
channel_types=ChannelTypeAccess(initiate=["*"], accept=["*"]), # stored, NOT enforced (see below)
),
limits=LimitsBlock(
max_concurrent_channels=0, # 0 disables; checked at channel open (creator)
max_concurrent_tasks=0, # 0 disables; checked when the hub first observes a task
channel_ttl_default="2h", # default TTL for channels this agent creates
task_ttl_default="15m",
delegation_depth=5, # envelope.depth cap on sends; 0 disables
rate=RateBlock(per_minute=0, burst=0), # stored, NOT enforced
inbox=InboxBlock(max_pending=1000, overflow="reject", high_water=None),
),
)
```
| Check | Where (`RuleBasedArbiter`) | Error |
|---|---|---|
| `access.outbound_to` | `authorize_send` (every recipient, including the invite sent at channel open) | `AccessDeniedError` |
| `access.inbound_from` | `authorize_channel_open` (invitee pre-flight) + `authorize_dispatch` | `AccessDeniedError` |
| `limits.delegation_depth` | `authorize_send` (`envelope.depth > cap`) | `AccessDeniedError` |
| `limits.max_concurrent_channels` | `authorize_channel_open` | `AccessDeniedError` |
| `limits.max_concurrent_tasks` | `Hub.observe_task` (not the arbiter) | `AccessDeniedError` |
| `limits.inbox.max_pending` | `authorize_inbox` | `InboxFull` |
| `access.channel_types`, `limits.rate` | **nowhere** — data only. Enforce them in a custom arbiter. | — |
`InboxBlock.overflow` accepts `"drop_oldest"` / `"drop_newest"` but only `"reject"` is implemented.
`high_water=None` resolves to `int(max_pending * 0.8)`; `0` disables `on_inbox_pressure`.
`parse_duration("30s"|"15m"|"4h"|"2d")` → int seconds; a bare integer string or `int` is seconds; `""` → 0; unknown unit → `ValueError`.
`await hub.set_rule(agent_id, rule)` replaces the rule (audit `AUDIT_KIND_RULE_SET`, listener `on_agent_event(..., "rule_set", ...)`); `await hub.get_rule(agent_id)` reads it.
A denied send never reaches the WAL and is **not** written to the audit log; it fans out to `HubListener.on_envelope_rejected(envelope, reason)` (also for the invite of a denied channel open).
## `HubArbiter`
Protocol methods (all `async`), each returning `Allow()` or `Deny(reason, error=<NetworkError subclass>)` (default error `AccessDeniedError`):
| Method | Consulted before |
|---|---|
| `authorize_register(passport, resume, rule)` | committing a registration |
| `authorize_channel_open(manifest, creator, creator_rule, invitees, invitee_rules, active_creator_channels)` | creating a channel |
| `authorize_send(envelope, sender, sender_rule, recipients)` | WAL append |
| `authorize_inbox(envelope, recipient, recipient_rule, current_pending)` | enqueue into a recipient inbox |
| `authorize_dispatch(envelope, sender, recipient, recipient_rule)` | each delivery |
| `resolve_unknown_audience(envelope, unknown_ids)` | dispatch to unknown ids; `None` = drop (default) or a replacement id list (federation hook) |
`BaseHubArbiter` returns `Allow()` everywhere — a subclass that overrides one gate *allows* all others, so wrap a `RuleBasedArbiter()` to keep rule enforcement:
```python
from ag2.network import AccessDeniedError, Allow, BaseHubArbiter, Deny, HubArbiter, RuleBasedArbiter
class ChannelTypeArbiter(BaseHubArbiter):
"""Enforce AccessBlock.channel_types (the default arbiter doesn't) on top of the rule checks."""
def __init__(self, inner: HubArbiter) -> None:
self._inner = inner
async def authorize_channel_open(self, manifest, creator, creator_rule, invitees, invitee_rules,
active_creator_channels):
allowed = creator_rule.access.channel_types.initiate
if "*" not in allowed and manifest.type not in allowed:
return Deny(f"{creator.name} may not initiate {manifest.type}")
return await self._inner.authorize_channel_open(
manifest, creator, creator_rule, invitees, invitee_rules, active_creator_channels)
async def authorize_send(self, envelope, sender, sender_rule, recipients):
return await self._inner.authorize_send(envelope, sender, sender_rule, recipients)
async def authorize_inbox(self, envelope, recipient, recipient_rule, current_pending):
return await self._inner.authorize_inbox(envelope, recipient, recipient_rule, current_pending)
async def authorize_dispatch(self, envelope, sender, recipient, recipient_rule):
return await self._inner.authorize_dispatch(envelope, sender, recipient, recipient_rule)
hub.register_arbiter(ChannelTypeArbiter(RuleBasedArbiter())) # one active arbiter; replaces the previous
assert isinstance(hub.arbiter, ChannelTypeArbiter)
```
The arbiter is the gatekeeper (before the state change); `HubListener` is the observer (after). `AuthAdapter` authenticates credentials once at registration / re-attach; the arbiter authorizes actions throughout.
## Authentication
`AuthRegistry([...adapters])` keyed by each adapter's `scheme`. Default: `AuthRegistry.default()` = `NoAuth` only (scheme `"none"`, accept all). The passport selects the scheme with `Passport(auth=AuthBlock(scheme=..., claim={...}))`.
```python
import hmac
from ag2.knowledge import MemoryKnowledgeStore
from ag2.network import ApiKeyAuth, AuthBlock, AuthError, AuthRegistry, Hub, NoAuth, Passport
class HMACAuth:
scheme = "hmac"
def __init__(self, secret: bytes) -> None:
self._secret = secret
def sign(self, name: str) -> str:
return hmac.new(self._secret, name.encode(), "sha256").hexdigest()
async def validate(self, passport: Passport, claim: dict) -> None:
# Validate against YOUR secrets — the hub never stores claims (see below).
if not hmac.compare_digest(self.sign(passport.name), claim.get("token", "")):
raise AuthError(f"bad hmac for {passport.name}")
hm = HMACAuth(b"s3cret")
hub = await Hub.open(
MemoryKnowledgeStore(),
auth=AuthRegistry([NoAuth(), hm, ApiKeyAuth({"carol": "k1"})]), # ApiKeyAuth(keys=..., resolver=async fn)
)
hub.auth_schemes # ['none', 'hmac', 'api_key']
alice = await hc.register(agent, Passport(name="alice",
auth=AuthBlock(scheme="hmac", claim={"token": hm.sign("alice")})), Resume())
# bad claim -> AuthError from register()
```
1.3.0 behaviour worth knowing:
- The hub stores the passport **without** `auth.claim` (`(await hub.get_agent(id)).auth.claim == {}`), and wire responses never carry claims. A cross-process client re-attaching an identity registered under a credentialed scheme must pass its passport (with the credential) to `HubClient.attach(...)`.
- A `HelloFrame` re-attach must present the identity's registered scheme.
- Names shaped like an agent id (32 hex chars) or equal to an existing agent id are rejected (`ProtocolError`).
- `Hub.open(..., allow_remote_agent_registration=False)` (default): wire clients can't register `kind="remote_agent"` identities unless you opt in; when allowed, the registering connection's agents become its owners.
- Over the wire, every control-plane op is scoped to the identities bound to that connection (agent-, channel- and task-scoped checks); `fire_task_event` over the wire accepts only the mirror's `mirror_failed` report.
references/task-observation.md
# Task observation and capability track records (verified against ag2 1.3.0)
`Resume.observed[capability]` is an `ObservedStat(n, completed, failed, expired, p50_latency_ms)` the hub updates when a **capability-tagged** task reaches a terminal state on a stream that has a `TaskMirror` attached.
> Migration: `Agent.task(...)` is deprecated in 1.3.0 and removed in 1.4.0. Replace
> `async with agent.task(title, capability=..., context=ctx)` with
> `async with Task(owner_id=agent.name, spec=TaskSpec(title=..., capability=...), context=ctx)` (`from ag2.task import Task, TaskSpec`), or — for a whole model run — `agent.ask(..., capability=..., ttl_seconds=...)`.
## Three ways to produce an observation
| Path | When | How the mirror sees it |
|---|---|---|
| `Task(...)` inside a `@tool` | a tool's unit of work is the capability | pass `context=ctx`; the default handler's per-turn `TaskMirror` is on that stream |
| `agent.ask(..., capability=...)` | the whole model run is the capability | you call `ask` yourself on a stream with a mirror attached (custom handler or script) |
| peer `delegate(..., capability=...)` | — | only stored as `knobs["capability"]` on the consulting channel; it does **not** update anyone's `observed` by itself |
The mirror always records under **its own** `owner_id` (`client.agent_id`), regardless of `Task(owner_id=...)`.
### 1. `Task` in a tool (executed, default handler)
```python
from ag2 import Agent, Context
from ag2.task import Task, TaskSpec
worker_agent = Agent("worker", prompt="Always call the research tool, then answer.", config=config)
@worker_agent.tool # decorate the Agent before hc.register(...); AgentClient has no .tool
async def research(topic: str, ctx: Context) -> str:
"""Research a topic."""
async with Task(
owner_id=worker_agent.name,
spec=TaskSpec(title=f"research: {topic}", capability="research"),
context=ctx, # REQUIRED: emits on the turn's stream, where the TaskMirror listens
) as task:
await task.progress({"step": "gather"})
await task.complete({"items_found": 7})
return f"7 sources on {topic}"
worker = await hc.register(worker_agent, Passport(name="worker"), Resume(claimed_capabilities=["research"]))
```
Leaving the block without `complete()` completes it; an exception fails it; cancellation emits `TaskCancelled`. Inside the block the LLM's `tasks(action="progress"|"complete")` plugin tool resolves the active task via `TaskInject`.
### 2. `ask(capability=...)` with a manual mirror (executed)
```python
from ag2.network import TaskMirror
from ag2.stream import MemoryStream
mirror = TaskMirror(hub_client=worker._hub_client, owner_id=worker.agent_id, channel_id=None)
stream = MemoryStream()
subs = mirror.attach(stream)
try:
reply = await worker_agent.ask("Summarise X", stream=stream, capability="summarisation", ttl_seconds=60)
finally:
mirror.detach(stream, subs)
```
The run's task title is the agent name (never the prompt). `ttl_seconds` is an observation deadline enforced by the hub's TTL sweeper, not a local timeout; if hub expiry wins, `EXPIRED` is recorded once and a late completion is ignored. A tagged run waits for its background descendants before returning. To tag every network turn, use a custom notify handler that calls `client.agent.ask(*parts, stream=stream, dependencies=deps, capability="...")` — full handler in `ag2-network-tools-and-views/references/custom-handlers.md`.
Don't call `ask` on `ctx.stream` from inside a tool of the same running turn — the stream's turn lock is held.
## What gets written
On the accepted terminal:
- `Resume.observed[capability]` updated (deduped per `(owner_id, task_id)`), audit `AUDIT_KIND_RESUME_SET` with `source=RESUME_SOURCE_OBSERVED`, listener `on_agent_event(..., "observation_recorded", ...)`.
- `AUDIT_KIND_TASK_TERMINATED` record: `task_id`, `owner_id`, `channel_id`, `outcome`, `capability`, `reason`.
- `HubListener.on_task_event(task_id, "completed"|"failed"|"expired"|"cancelled", payload)`.
Untagged tasks are still observed (`hub.list_tasks()`, `task_terminated` audit) but don't touch `observed`.
```python
resume = await hub.get_resume(worker.agent_id)
stat = resume.observed.get("research")
if stat:
print(stat.completed, stat.n, stat.failed, stat.expired, stat.p50_latency_ms)
for t in await hub.list_tasks():
print(t.spec.title, t.state, t.spec.capability)
```
Latency is `started_at → terminal` on the hub's clock (deterministic with a fake `clock=`). Observed capabilities join the hub's capability index, so `peers(action="find", capability=..., sort_by="track_record")` can rank by them.
## Manual mirror in a custom handler
The default handler attaches a fresh `TaskMirror(hub_client=client._hub_client, owner_id=client.agent_id, channel_id=metadata.channel_id)` per inbound substantive envelope and detaches it after `ask`. Replicate that in any custom handler that runs the LLM, or tasks never reach the hub. The mirror swallows hub errors (reported as `mirror_failed`) so a flaky hub never crashes a turn. `Rule.limits.max_concurrent_tasks` is enforced when the hub first observes a task.
SKILL.md
---
name: ag2-network-governance
description: "Govern an AG2 multi-agent network hub-side: identity (`Passport`, `Resume`, `ObservedStat`), per-agent `Rule` (`AccessBlock`, `LimitsBlock` with nested `RateBlock` / `InboxBlock`), the swappable `HubArbiter` / `RuleBasedArbiter` / `BaseHubArbiter` (`Allow` / `Deny`) gate, `AuthRegistry` / `AuthAdapter` / `ApiKeyAuth` / `NoAuth` registration auth, channel `Expectation`s with `ExpectationEvaluator` / `Violation` and `audit` / `notify_channel` / `auto_close` handlers, the append-only `hub.audit_log` with `AUDIT_KIND_*` constants, live `HubListener` / `BaseHubListener` fan-out, `register_sweeper`, and capability track records via `TaskMirror` + `Task(owner_id=..., spec=TaskSpec(capability=...))` or `agent.ask(..., capability=...)` (replaces deprecated `agent.task`). Use for access policy, inbox caps, SLAs/timeouts, compliance trails, metrics/alerting, peer ranking by track record. Load after `ag2-network-quickstart`; for handlers, views, LLM network tools and envelopes see `ag2-network-tools-and-views`."
license: Apache-2.0
metadata:
ag2_version: "1.3.0"
---
# AG2 Network — Governance
Everything hub-side: identity, rules, the arbiter, auth, expectations, audit, listeners, and capability track records. Every send passes through the hub, so every policy is checked and every observation is read there.
> Prerequisite: `ag2-network-quickstart` (`Hub.open`, `HubClient.register`, channel lifecycle).
## When to use
- Restrict who talks to whom, cap inboxes / concurrent channels / delegation depth (`Rule`)
- Custom access logic — JWT scopes, tenant quotas, federation (`HubArbiter`)
- Authenticate agents at registration (`AuthRegistry`, `AuthAdapter`)
- Channel SLAs and auto-close timing (`Expectation`, evaluators, handlers)
- Compliance queries (`hub.audit_log`), live metrics/alerts (`HubListener`), periodic jobs (`register_sweeper`)
- Track which agent demonstrably does a capability well (`Resume.observed`)
## Rules
```python
from ag2.network import AccessBlock, InboxBlock, LimitsBlock, Rule
rule = Rule(
access=AccessBlock(outbound_to=["bob", "research-*"], inbound_from=["*"]), # globs over Passport.name
limits=LimitsBlock(
max_concurrent_channels=5, delegation_depth=2, channel_ttl_default="4h",
inbox=InboxBlock(max_pending=100), # high_water=None -> 80% early warning
),
)
alice = await hc.register(agent, Passport(name="alice", owner="acme"), Resume(), rule=rule)
await hub.set_rule(alice.agent_id, rule) # later change; audited as AUDIT_KIND_RULE_SET
```
Violations raise from `open(...)` / `send(...)`: `AccessDeniedError` (access, depth, concurrency) or `InboxFull`. Denied sends are **not** audited — they reach `HubListener.on_envelope_rejected`.
**Stored but not enforced** by the default `RuleBasedArbiter`: `access.channel_types` and `limits.rate`. Enforce them in a custom arbiter. Field reference: [references/rules-arbiter-auth.md](references/rules-arbiter-auth.md).
## Arbiter and auth
```python
from ag2.network import BaseHubArbiter, Deny, RuleBasedArbiter
class ScopedArbiter(BaseHubArbiter): # allows every gate you don't override -> re-delegate the others
def __init__(self, inner): # (authorize_register/_channel_open/_inbox/_dispatch) to keep rule checks
self._inner = inner
async def authorize_send(self, envelope, sender, sender_rule, recipients):
if sender.name in BLOCKED:
return Deny("missing net.send scope") # -> AccessDeniedError to the caller
return await self._inner.authorize_send(envelope, sender, sender_rule, recipients)
hub.register_arbiter(ScopedArbiter(RuleBasedArbiter()))
```
Auth: `Hub.open(store, auth=AuthRegistry([NoAuth(), ApiKeyAuth({"carol": "k1"}), MyAdapter()]))`; the passport selects one with `Passport(auth=AuthBlock(scheme="api_key", claim={"token": "k1"}))`; adapters raise `AuthError`. The hub stores passports **without** claims, so validate against your own secrets. Details, a full HMAC adapter and wire-security notes: [references/rules-arbiter-auth.md](references/rules-arbiter-auth.md).
## Expectations
Adapter manifests declare SLAs evaluated by the sweeper (`expectation_sweep_interval`, default 10s): `consulting` auto-closes on `acks_within(30s)` / `reply_within(600s)`; `conversation` audits `max_silence(3600s)`. Built-in handlers: `audit`, `notify_channel` (posts `EV_EXPECTATION_VIOLATED`), `auto_close`. The `turn_within` / `warn` / `hide` names on discussion/workflow manifests have **no** built-in implementation, and `knobs={"expectations": ...}` is ignored. Customise by registering an adapter with your own `manifest.expectations` plus `hub.register_expectation_evaluator(...)`. See [references/expectations-audit-listeners.md](references/expectations-audit-listeners.md).
## Audit and listeners
```python
from ag2.network import AUDIT_KIND_TASK_TERMINATED, BaseHubListener
records = await hub.audit_log.read_all() # dicts: kind, at, + kind-specific ids
ended = [r for r in records if r["kind"] == AUDIT_KIND_TASK_TERMINATED]
class Alerts(BaseHubListener): # override only what you need; all async
async def on_turn_failed(self, channel_id, agent_id, envelope_id, exc): ...
async def on_inbox_pressure(self, agent_id, pending, cap): ...
hub.register_listener(Alerts())
```
Audit kinds: `AGENT_REGISTERED` / `AGENT_UNREGISTERED` / `RESUME_SET` / `SKILL_SET` / `RULE_SET` / `CHANNEL_CREATED` / `CHANNEL_CLOSED` / `CHANNEL_EXPIRED` / `EXPECTATION_VIOLATED` / `TASK_TERMINATED` / `TURN_FAILED` (all `AUDIT_KIND_*`). Records carry ids (`agent_id`, `channel_id`, `task_id`), not tenant names. Field tables, `audit_log.subscribe`, and `register_sweeper` / `await unregister_sweeper` are in the reference.
## Capability track records
Capability-tagged task terminals update `Resume.observed[cap]` (`ObservedStat`: `n`, `completed`, `failed`, `expired`, `p50_latency_ms`) and write `AUDIT_KIND_TASK_TERMINATED`. `Agent.task(...)` is deprecated (removed in 1.4.0). Use `Task` from `ag2.task`:
```python
from ag2 import Context
from ag2.task import Task, TaskSpec
@worker_agent.tool
async def research(topic: str, ctx: Context) -> str:
"""Research a topic."""
async with Task(
owner_id=worker_agent.name,
spec=TaskSpec(title=f"research: {topic}", capability="research"),
context=ctx, # required: the default handler's TaskMirror listens on this stream
) as task:
await task.progress({"step": "gather"})
result = f"notes on {topic}"
return result
stat = (await hub.get_resume(worker.agent_id)).observed.get("research")
```
For a whole model run, call `agent.ask(..., capability="x", ttl_seconds=60)` on a stream with a `TaskMirror` attached, either in a custom handler or a script. The default handler does not pass `capability`. `delegate(capability=...)` only records a channel knob. More: [references/task-observation.md](references/task-observation.md).
## Reading hub state
`await hub.get_agent(id_or_name)` (Passport), `get_resume`, `get_rule`, `get_skill`, `list_agents(kind="agent"|"human"|"remote_agent")`, `get_channel`, `list_channels`, `read_wal`, `get_task`, `list_tasks`, plus `hub.arbiter`, `hub.auth_schemes`, `hub.health()`.
## Reference
- [references/rules-arbiter-auth.md](references/rules-arbiter-auth.md): `Rule` fields and their enforcement, the arbiter protocol, auth adapters, 1.3.0 wire security
- [references/expectations-audit-listeners.md](references/expectations-audit-listeners.md): evaluators, handlers, custom expectations, audit record fields, listener methods, sweepers
- [references/task-observation.md](references/task-observation.md): `Task` / `ask(capability=)` / `TaskMirror`, migration from `agent.task`
- Upstream: `website/docs/user-guide/network/expectations_and_audit.mdx`, `task_observation.mdx`, `hub_and_identity.mdx`; `website/docs/user-guide/tasks.mdx`