ag2-network-governance

AG2 Network — ガバナンス

ハブ側のすべてを扱います。ID、エージェントごとのルール、期待値、監査、タスクの観察が含まれます。ハブが信頼できる唯一の情報源です。すべての送信はハブを経由し、すべての観察はハブから読み取られ、すべてのポリシーはハブで確認されます。

前提条件：まず ag2-network-quickstart をお読みください。このスキルでは、Hub.open、Passport、Resume、チャネルのライフサイクル、agent_client.register(...) の流れについて理解していることを前提としています。

使用する場面

次のことが必要な場合に、このスキルを読み込んでください。

• 誰が誰と通信できるかを制限する（access / AccessBlock）
• トークンバケット方式のレート制限の意図を宣言する（limits.rate / RateBlock）。ルールには保存されますが、プロセス内ハブでは適用されません
• あふれを防ぐために受信トレイのサイズに上限を設ける（limits.inbox / InboxBlock）。上限に達する前に早期警告シグナルを受け取る（on_inbox_pressure / high_water）
• チャネルの TTL のデフォルト値、同時実行数の上限、委任の深さを設定する（limits / LimitsBlock）
• JWT スコープ、テナントごとのクォータ、フェデレーションなどのカスタムアクセス／ルーティングロジックを組み込む（HubArbiter / RuleBasedArbiter）
• 登録時にエージェントを認証する（AuthAdapter）
• チャネルを閉じるタイミングを調整する（acks_within、reply_within、max_silence、turn_within）
• コンプライアンスのために監査ログを読み取る、または検索する。あるいは、メトリクス／アラート用にライブの状態変更をストリーミングする（HubListener）
• ハブにカスタムの定期タスクを追加する（register_sweeper）
• 各エージェントの能力の実績を構築する（Resume.observed）
• タスクを実証的に完了したエージェントに基づいてルーティングする（例：「p50_latency_ms が最も低いリサーチャーに送信する」）

ID — すべてのエージェントが持つもの

ネットワーク上のエージェントは、3 つのデータクラスで表されます。ほとんどのフィールドはテナントが指定し、残りはハブが設定します。

from ag2.network import Passport, Resume, ResumeExample

passport = Passport(
    name="alice",                # required, unique within the hub
    owner="acme",                # optional, tenant id for multi-tenant deployments
    model="claude-sonnet-4-6",   # optional, surfaces on peer-lookup results
)

resume = Resume(
    claimed_capabilities=["analysis", "policy"],
    domains=["finance"],
    summary="Senior policy analyst — scenario synthesis and rebuttal review.",
    examples=[ResumeExample(title="Q3 risk brief", note="…")],
)

フィールド	設定元
Passport.name	テナント（必須、一意）
Passport.agent_id	登録時にハブが設定。ルーティングに使用
Passport.created_at	ハブが設定する ISO-Z タイムスタンプ
Resume.claimed_capabilities	テナント（自由形式の文字列："research"、"summarisation"、…）
Resume.summary	テナントが指定 — ピア検索用にインデックス化
Resume.observed	ハブが変更する、能力ごとの ObservedStat（n / completed / failed / expired / p50_latency_ms）
Resume.last_updated	ハブが設定する ISO-Z。変更時に更新

observed フィールドは、エージェントの実績です。エージェントが能力タグ付きタスクを実行すると、自動的に蓄積されます（下記の「タスクの観察」を参照）。

エージェントごとのルール

登録時に Rule を渡すと、ネットワーク上でのエージェントの動作を制御できます。

from ag2.network import (
    Rule, AccessBlock, LimitsBlock, RateBlock, InboxBlock, ChannelTypeAccess,
)

rule = Rule(
    access=AccessBlock(
        outbound_to=["bob", "carol"],           # whitelist of recipients (names or ids)
        channel_types=ChannelTypeAccess(
            initiate=["consulting", "discussion"],
            accept=["consulting", "discussion"],
        ),
    ),
    limits=LimitsBlock(                         # rate + inbox nest INSIDE limits
        channel_ttl_default="4h",               # default TTL for channels this agent creates
        delegation_depth=2,                     # max recursion through sub-task delegation
        rate=RateBlock(per_minute=60, burst=10),
        inbox=InboxBlock(max_pending=100),      # cap inbound queue depth
    ),
)

alice = await alice_hc.register(
    Agent("alice", config=config),
    Passport(name="alice"),
    Resume(),
    rule=rule,
)

Rule には2つのトップレベルブロックがあります。access（AccessBlock）と limits（LimitsBlock）です。RateBlock と InboxBlock は limits の内部にネストします（limits.rate および limits.inbox）。

ブロック	配置場所	制御内容	失敗時の動作
AccessBlock	access（トップレベル）	このエージェントが宛先にできる相手（outbound_to）／受け入れ可能な相手（inbound_from）、作成／参加できるチャネルの種類	AccessDeniedError
LimitsBlock	limits（トップレベル）	TTL のデフォルト値、delegation_depth、max_concurrent_channels／max_concurrent_tasks	AccessDeniedError
RateBlock	limits.rate（ネスト）	トークンバケットの値（per_minute、burst）	保存されますが、プロセス内では適用されません（デフォルトでは per_minute=0 で無効）
InboxBlock	limits.inbox（ネスト）	受信キューの深さ（max_pending）	送信者に InboxFull を返します

ルールチェックに失敗すると、ハブは channel.send(...) または hc.register(...) から該当するエラーを発生させます。エンベロープが WAL に書き込まれることはありません。ルールの変更は監査記録に残ります（set_rule を介した種別 AUDIT_KIND_RULE_SET）。一方、送信の拒否は組み込みの監査ログには記録されません。拒否は on_envelope_rejected リスナーへのファンアウトを通じて通知されます（組み込みの AuditLog は on_envelope_rejected ハンドラーを実装していません）。拒否を記録する必要がある場合は、HubListener を登録してください。これらのチェックを実行するコンポーネントと、ルールデータ以外の仕組みを組み込むための接点が、以下で説明する調停器です。

登録後のルールの更新

new_rule = Rule(access=AccessBlock(outbound_to=["bob"]))
await hub.set_rule(alice.agent_id, new_rule)  # emits AUDIT_KIND_RULE_SET

期間文字列の解析

LimitsBlock.channel_ttl_default には、parse_duration で解析される文字列を指定します。

from ag2.network import parse_duration

parse_duration("30s")  # 30
parse_duration("4h")   # 14400
parse_duration("2d")   # 172800

接尾辞には s、m、h、d を指定します。単位のない整数（または int）は秒として扱われます。空の文字列を指定すると 0 が返されます。戻り値は秒数を表す int です。

アービター — アクセスとルーティングの切り替え

ハブは Rule をインラインの if チェックで適用しなくなりました。アクセス／ルーティングに関するあらゆる判断を HubArbiter に委譲します。これは判断ポイントごとに 1 つのメソッドを持つ Protocol で、各メソッドは Allow() または Deny(reason, error=<NetworkError subclass>) を返します。

メソッド	…の前に呼び出される	デフォルトの Deny エラー
authorize_register(passport, resume, rule)	登録を確定する	AccessDeniedError
authorize_channel_open(manifest, creator, creator_rule, invitees, invitee_rules, active_creator_channels)	チャネルを作成する（招待者の inbound_from と作成者の max_concurrent_channels）	AccessDeniedError
authorize_send(envelope, sender, sender_rule, recipients)	エンベロープを WAL に追加する（送信アクセスと委譲の深さ）	AccessDeniedError
authorize_inbox(envelope, recipient, recipient_rule, current_pending)	受信者の受信トレイにエンキューする（容量）	InboxFull
authorize_dispatch(envelope, sender, recipient, recipient_rule)	1 件の配信をディスパッチする	AccessDeniedError
resolve_unknown_audience(envelope, unknown_ids)	ハブが認識していない ID にディスパッチする — None（何もせず破棄。単一ハブのデフォルト）または置換用の ID リスト（フェデレーション用フック）を返す	—

デフォルトは RuleBasedArbiter です。エージェントごとの Rule、つまり access（送信／受信の名前グロブ、チャネルタイプ）に加え、実際にチェックされる limits の上限（delegation_depth、max_concurrent_channels、inbox.max_pending）を適用します。limits.rate は適用されません。これは、この接続ポイントが導入される前にハブがインラインで行っていた処理とまったく同じです。Rule のみを使用する場合、アービターに触れる必要はありません。

独自のロジック（JWT スコープ、テナントごとのクォータ、フェデレーションルーティングなど）をルールデータの上に重ねる、またはルールデータの代わりに適用するには、アービターを差し替えます。BaseHubArbiter はすべてに対して Allow() を返すため、1 つのゲートをオーバーライドするサブクラスでは、残りも 許可 されます。ルールの適用を維持するには、RuleBasedArbiter() インスタンスに処理を委譲してください。

from ag2.network import HubArbiter, BaseHubArbiter, RuleBasedArbiter, Allow, Deny

class ScopedArbiter(BaseHubArbiter):
    def __init__(self, inner: HubArbiter) -> None:
        self._inner = inner                           # the rule checks
    async def authorize_send(self, envelope, sender, sender_rule, recipients):
        if not _token_has_scope(sender, "net.send"):
            return Deny("missing net.send scope")     # → AccessDeniedError back to the caller
        return await self._inner.authorize_send(envelope, sender, sender_rule, recipients)
    # authorize_register / _channel_open / _inbox / _dispatch / resolve_unknown_audience
    # all fall through to BaseHubArbiter's Allow() — explicitly re-delegate any you want enforced.

hub.register_arbiter(ScopedArbiter(RuleBasedArbiter()))   # one active arbiter; replaces the prior one
# hub.arbiter  → the active instance (read-only; handy in tests)

Deny.error は、ハブが送出する NetworkError のサブクラスを指定します（デフォルトは AccessDeniedError）。制御するには Deny(..., error=InboxFull) などを指定します。アービターは ゲートキーパー（状態が変更される 前 に呼び出されます）であり、後述する HubListener は オブザーバー（変更の 後 に通知されます）です。これは以下の AuthAdapter とは役割が異なります。AuthAdapter は登録時に一度だけ 認証情報 を認証しますが、アービターはチャネルの存続期間を通じて アクション を認可します。

認証

デフォルトでは、ハブは AuthRegistry.default() を使用します。これは NoAuth のみを含む（スキーム "none"）レジストリで、すべてのクレームを受け入れるため、認証情報がなくてもすべての登録が成功します。スキームは passport.auth.scheme（AuthBlock のフィールド。デフォルトは "none"）でパスポートごとに選択され、認証情報は passport.auth.claim（dict）に格納されます。本番環境では、次のようにします。

from ag2.network import AuthAdapter, AuthRegistry, NoAuth, ApiKeyAuth, AuthError, Hub
from ag2.network import AuthBlock, Passport
from ag2.knowledge import MemoryKnowledgeStore

import hmac
from typing import Any


class HMACAuth:
    scheme = "hmac"                                       # class-level scheme label

    async def validate(self, passport: Passport, claim: dict[str, Any]) -> None:
        expected = self._sign(passport.name)
        token = claim.get("token", "")
        if not hmac.compare_digest(expected, token):
            raise AuthError(f"bad hmac for {passport.name}")


# AuthRegistry takes a LIST of adapters at construction (keyed by .scheme).
registry = AuthRegistry([NoAuth(), HMACAuth()])

hub = await Hub.open(MemoryKnowledgeStore(), auth=registry)

AuthAdapter は scheme 属性と validate メソッドを持つ Protocol です。

class AuthAdapter(Protocol):
    scheme: str
    async def validate(self, passport: Passport, claim: dict[str, Any]) -> None: ...

拒否するには AuthError を送出してください。登録時にハブは passport.auth.scheme でアダプターを検索し、adapter.validate(passport, passport.auth.claim) を呼び出して、成功すると AUDIT_KIND_AGENT_REGISTERED を記録します。リモートエージェントのパスポートでは、ローカル認証チェックはスキップされます。このライブラリには、すべてを受け入れる NoAuth（デフォルト）と、claim["token"] とトークンを定数時間で比較する ApiKeyAuth(keys=..., resolver=...) が付属しています。

期待値 — チャネルレベルの SLA

各アダプターはマニフェストにデフォルト値を設定します。期待値のスイーパータスクは expectation_sweep_interval（デフォルトは 10s）ごとにこれらを評価し、違反をハンドラーにディスパッチします。

組み込み評価器

default_evaluators() には、ちょうど 3 つの評価器が含まれています。

名前	クラス	seconds のデフォルト値	しきい値
"acks_within"	AcksWithinEvaluator	30	まだ保留中のすべての招待者が、チャンネル作成から params["seconds"] 以内に ACK する必要があります（チャンネルが PENDING の間のみ）。
"reply_within"	ReplyWithinEvaluator	600	EV_TEXT の宛先となった参加者は、params["seconds"] 以内に返信する必要があります（ACTIVE の間のみ）。
"max_silence"	MaxSilenceEvaluator	3600	params["seconds"] の間、誰からもチャンネルにコンテンツエンベロープが届いていません（チャンネル全体が対象）。

discussion と workflow のアダプターはマニフェストで turn_within の期待値を宣言していますが、組み込みの turn_within 評価器は含まれていません。また、"warn" と "hide" は組み込みハンドラーではありません。スイーパーは、name に対応する評価器が登録されていない期待値、または on_violation に対応するハンドラーが登録されていない期待値を、何も通知せずスキップします（_expectation_tick は .get(...) を実行し、None の場合は continue します）。turn_within／warn／hide を有効にするには、独自の評価器（register_expectation_evaluator）とハンドラー（register_violation_handler）を登録してください。

アダプターごとのデフォルトの期待値

アダプター	デフォルト値
consulting	acks_within(30s, auto_close)、reply_within(600s, auto_close)
conversation	max_silence(3600s, audit)
discussion	turn_within(120s, warn)、turn_within(600s, hide)
workflow	turn_within(120s, warn)、turn_within(600s, auto_close)

違反ハンドラー

from ag2.network import Expectation

Expectation(name="acks_within", on_violation="auto_close", params={"seconds": 30})

default_handlers() には、on_violation をキーとする、ちょうど 3 つのハンドラーが含まれています。

on_violation	ハンドラークラス	効果
"audit"	AuditHandler	何もしないハンドラーです。 実際の監査記録（AUDIT_KIND_EXPECTATION_VIOLATED）は、ハブがハンドラーを呼び出す前にファンアウトする on_expectation_fired 経由で AuditLog リスナーが書き込みます。そのため、AuditHandler 自体は何もしません。チャンネルは継続します。
"notify_channel"	NotifyChannelHandler	すべてのチャンネル参加者に EV_EXPECTATION_VIOLATED を投稿します。チャンネルは継続します。（監査記録は引き続き AuditLog リスナーが書き込みます。）
"auto_close"	AutoCloseHandler	reason="expectation_violated:<name>" でチャンネルを閉じます。（監査記録は引き続き AuditLog リスナーが書き込みます。）

組み込みの "warn"／"hide" ハンドラーはありません。これらの名前は discussion／workflow のマニフェストにのみ記載されており、対応するハンドラーを登録するまでは何もしません。

アダプターのデフォルト値の上書き

チャンネルのノブに expectations を渡すと、アダプターのデフォルト値を置き換えられます。

channel = await alice.open(
    type="conversation",
    target=bob.agent_id,
    knobs={
        "expectations": [
            {"name": "max_silence", "on_violation": "auto_close",
             "params": {"seconds": 600}},
        ],
    },
)

カスタム評価器

from ag2.network import EV_TEXT, Expectation
from ag2.network import ExpectationContext, Violation


class TooManyMessagesEvaluator:
    name = "too_many_messages"

    # Signature: evaluate(self, expectation, context) -> Violation | None
    def evaluate(self, expectation: Expectation, context: ExpectationContext) -> Violation | None:
        threshold = int(expectation.params["max"])
        text_count = sum(1 for e in context.wal if e.event_type == EV_TEXT)
        if text_count > threshold:
            return Violation(
                expectation=expectation,                 # the Expectation object, not a string
                violator_ids=[],                          # channel-wide
                detail={"text_count": text_count, "threshold": threshold},
            )
        return None

Violation は Violation(expectation: Expectation, violator_ids: list[str] = [], detail: dict = {}) です。expectation は Expectation オブジェクトで、name、on_violation、params を保持します。channel_id はフィールドではありません（ハブがすでに把握しています）。評価器はチャンネルの状態を入力とする純粋関数で、I/O も変更も行わないため、簡単にテストできます。hub.register_expectation_evaluator(TooManyMessagesEvaluator()) で登録してください。

決定論的なテスト

hub = await Hub.open(MemoryKnowledgeStore(), expectation_sweep_interval=0)

# Manually advance state and tick:
clock.advance(45)
await hub._expectation_tick()  # operator API (leading underscore by convention)

監査ログ

ハブは追記専用の監査ログ（AuditLog インスタンス）を保持しており、パブリックな hub.audit_log プロパティから参照できます（内部属性は hub._audit_log です。hub.replace_audit_log(...) でインスタンスを差し替えられます）。

records = await hub.audit_log.read_all()
for r in records:
    print(r["kind"], r["at"], r)

各レコードは通常の dict で、少なくとも kind と at（ISO-Z 形式のタイムスタンプ）を持ちます。種別固有のフィールドも併記されます。

監査の種類

from ag2.network import (
    AUDIT_KIND_AGENT_REGISTERED,
    AUDIT_KIND_AGENT_UNREGISTERED,
    AUDIT_KIND_RESUME_SET,
    AUDIT_KIND_SKILL_SET,
    AUDIT_KIND_RULE_SET,
    AUDIT_KIND_CHANNEL_CREATED,
    AUDIT_KIND_CHANNEL_CLOSED,
    AUDIT_KIND_CHANNEL_EXPIRED,
    AUDIT_KIND_TASK_TERMINATED,
    AUDIT_KIND_EXPECTATION_VIOLATED,
)

種類	発生時	主なフィールド
AUDIT_KIND_AGENT_REGISTERED	hc.register(...)	agent_id、name、owner
AUDIT_KIND_AGENT_UNREGISTERED	hc.unregister(agent_id)	agent_id
AUDIT_KIND_RESUME_SET	hub.set_resume(...)	ソース：RESUME_SOURCE_TENANT または RESUME_SOURCE_OBSERVED
AUDIT_KIND_SKILL_SET	hub.set_skill(...)	更新されたスキルの Markdown
AUDIT_KIND_RULE_SET	hub.set_rule(...)	新しい Rule
AUDIT_KIND_CHANNEL_CREATED	alice.open(...)	creator_id、マニフェストの種類／バージョン、参加者
AUDIT_KIND_CHANNEL_CLOSED	すべてのクローズ経路	reason
AUDIT_KIND_CHANNEL_EXPIRED	TTL スイーパー	TTL の詳細
AUDIT_KIND_TASK_TERMINATED	TaskMirror を介して agent.task(...) が終端状態に到達したとき	owner_id、capability、outcome、latency_ms
AUDIT_KIND_EXPECTATION_VIOLATED	期待値評価器のしきい値が経過したとき	expectation、channel_id、評価器の詳細

一般的なクエリ

# All violations on the system.
violations = [r for r in await hub.audit_log.read_all()
              if r["kind"] == AUDIT_KIND_EXPECTATION_VIOLATED]

# Everything that happened on one channel.
channel_records = [r for r in await hub.audit_log.read_all()
                   if r.get("channel_id") == channel_id]

# All registrations for one tenant.
acme_agents = [r for r in await hub.audit_log.read_all()
               if r["kind"] == AUDIT_KIND_AGENT_REGISTERED
               and r.get("owner") == "acme"]

監査ログは、DiskKnowledgeStore をバックエンドにする場合は永続化されます。MemoryKnowledgeStore の場合は、ハブが存続している間だけ保持されます。

Hub リスナー — ライブなプログラムによる可観測性

監査ログは永続的な記録です。ハブの状態変化に対するライブな反応（メトリクスバックエンドへのプッシュ、ダッシュボードへのストリーム配信、オンコール担当者へのアラート）には、**HubListener**を登録します。これは読み取り専用の Protocol で、状態遷移がコミットされるたびにハブから通知されます。（組み込みの監査ログ自体も、こうしたリスナーの1つです。）

メソッド（正確なシグネチャ）	呼び出されるタイミング
on_envelope_posted(envelope, metadata)	エンベロープが受理され、WAL に追記され、状態に反映され、ディスパッチされたとき
on_envelope_rejected(envelope, reason)	アービター／検証によって送信が拒否されたとき（reason は型付きの NetworkError です）
on_dispatch_failed(envelope, recipient_id, reason)	1人の受信者への配信で例外が発生したとき（reason は BaseException です）
on_channel_event(channel_id, kind, payload)	kind ∈ opened / closed / expired / participant_removed / participant_hidden のとき
on_agent_event(agent_id, kind, payload)	kind ∈ registered / unregistered / resume_set / skill_set / rule_set / observation_recorded のとき
on_expectation_fired(channel_id, expectation, violation)	expectation evaluator が Violation を出力したとき
on_turn_failed(channel_id, agent_id, envelope_id, exc)	エージェントの notify-handler ターンで例外が発生したとき（デフォルトのハンドラーはここにエラーを送ります）
on_task_event(task_id, kind, payload)	ag2.task.* のライフサイクルイベントが観測されたとき（kind ∈ started / progress / completed / failed / expired / cancelled / mirror_failed）
on_inbox_pressure(agent_id, pending, cap)	受信者の受信箱が初めて limits.inbox.high_water を超えたとき（エンベロープごとではなく、しきい値を超えるたびに1回呼び出されます）

すべてのメソッドは async です。ハブは登録順に各メソッドを順次呼び出して待機し、それぞれを try/except で囲みます。バグのあるリスナーがディスパッチを停滞させることはありません。処理は高速に保ってください（I/O は独自のタスクにキューイングしてください）。BaseHubListener（すべてのメソッドが pass）をサブクラス化し、必要なメソッドだけをオーバーライドします。

from ag2.network import BaseHubListener

class MetricsListener(BaseHubListener):
    async def on_envelope_posted(self, envelope, metadata):
        statsd.incr(f"net.envelope.{envelope.event_type}")
    async def on_inbox_pressure(self, agent_id, pending, cap):
        statsd.gauge(f"net.inbox.{agent_id}", pending / cap)
    async def on_turn_failed(self, channel_id, agent_id, envelope_id, exc):
        sentry.capture_exception(exc)

hub.register_listener(MetricsListener())     # hub.unregister_listener(inst) to detach

ハブのサブクラス化に関連する仕組みが2つあります。

• Hub 自体の on_* フック — 同じメソッド群が Hub の空メソッドとして存在します。Hub のサブクラスでは、これらを直接オーバーライドできます（ファンアウト時に、登録済みリスナーとともにバインドされたメソッドも呼び出されます）。提供するハブの派生版そのものに可観測性を組み込む場合はサブクラスを使い、プラグイン可能なアドオンには register_listener を使います。
• hub.register_sweeper(name, interval_seconds, fn) / unregister_sweeper(name) — 組み込みの TTL および expectation スイーパーと並行して、独自の定期コルーチンをハブの interval-sweeper 機構に追加します。Hub.start() の実行後にサブクラスから登録したスイーパーはすぐに開始され、まだ実行されていない場合は開始までキューに入ります。

on_inbox_pressure は limits.inbox.high_water（InboxBlock フィールド）によって制御されます。これは絶対的な保留件数のしきい値（int | None）です。None（デフォルト）は int(limits.inbox.max_pending * 0.8) に自動設定され、0 は通知を無効にします。これはハード上限である InboxFull（上限は limits.inbox.max_pending で、アービターの authorize_inbox によって適用されます）に対する早期警告として機能します。

タスクの観測 — 実績の構築

能力タグ付きタスクは、エージェントの Resume.observed[capability] を自動的に更新します。これにより、ネットワークは「bob は調査タスクを47件完了し、レイテンシの中央値は4.2秒」と把握できます。

タスクにタグを付ける

agent.task(..., capability="X") には、自由形式の能力文字列を指定できます。

# `.tool` and `.task(...)` live on `Agent`, not on the `AgentClient` returned
# by `hc.register(...)`. So decorate the Agent before registering.
@worker_agent.tool
async def research(topic: str, ctx: Context) -> str:
    async with worker_agent.task(
        f"research: {topic}",
        capability="research",
        context=ctx,
    ) as task:
        await task.progress({"step": "gather"})
        # ... do work ...
        await task.complete({"items_found": 7})
    return "researched"


worker = await worker_hc.register(worker_agent, Passport(name="worker"), Resume())

context=ctx を渡すと、タスクのイベントは LLM ターンのストリーム上で発生します。TaskMirror がアタッチされているのはこのストリームです。指定しない場合、イベントはハブに届きません。

capability=None（デフォルト）の場合、ライフサイクルイベントは引き続きハブの監査ログに記録されますが、Resume.observed は更新されません。実績の追跡はオプトイン方式です。

実績を読み取る

resume = await hub.get_resume(bob.agent_id)
stat = resume.observed.get("research")
if stat:
    print(f"completed={stat.completed}/{stat.n}  "
          f"failed={stat.failed}  "
          f"p50_latency={stat.p50_latency_ms}ms")

@dataclass(slots=True)
class ObservedStat:
    n: int = 0                        # total terminal events
    completed: int = 0
    failed: int = 0
    expired: int = 0
    p50_latency_ms: int | None = None  # rolling median of started_at → completed_at

レイテンシは、ハブのクロックを使い、task_meta.started_at から終端イベントの時刻までを計算します。テストで MockClock を使うと、決定論的なレイテンシを構築できます。

TaskMirror の役割

デフォルトのハンドラーは、アクティブなチャネルにスコープされた TaskMirror をターンごとに自動でアタッチします。ミラーは LLM ターンのストリーム上で TaskStarted / TaskProgress / TaskCompleted / TaskFailed / TaskExpired イベントをサブスクライブし、それぞれを ag2.task.* エンベロープとしてハブに転送します。また、終端イベントに capability タグが付いている場合は、Hub.record_observation(...) を呼び出して Resume.observed を更新します。

カスタムハンドラーを作成した場合に限り、TaskMirror を手動でアタッチします。

from ag2.network import TaskMirror
from ag2.stream import MemoryStream

mirror = TaskMirror(
    hub_client=client._hub_client,
    owner_id=client.agent_id,
    channel_id=metadata.channel_id,
)
stream = MemoryStream()
sub_ids = mirror.attach(stream)
try:
    await client.agent.ask(text, stream=stream)
finally:
    mirror.detach(stream, sub_ids)

ミラーはエージェントごとではなく、ターンごとにアタッチされます。つまり、受信したエンベロープごとに新しいミラーが作成されます。また、ハブへの転送エラーは暗黙的に無視されるため、不安定なハブ接続によって LLM ターンがクラッシュすることはありません。

capability タグを付けない場合

次の場合にのみタグを付けます。

• このタスクが、エージェントのレジュメで追跡したい能力を表している。
• 失敗やレイテンシのシグナルが、（ルーティング、アラート、ピアのランキングに影響するなど）運用上重要である。

タグなしのタスクも、ライフサイクル監査記録にはすべて記録されますが、Resume.observed は更新されません。外部から見える能力を表さない内部の記録管理には、タグなしのタスクを使用します。

横断的なパターン：複数の能力を持つワーカー

@worker_agent.tool
async def research(topic: str, ctx: Context) -> str:
    async with worker_agent.task(f"research: {topic}", capability="research", context=ctx) as t:
        # ...
    return "..."


@worker_agent.tool
async def summarise(text: str, ctx: Context) -> str:
    async with worker_agent.task("summarise", capability="summarisation", context=ctx) as t:
        # ...
    return "..."

いくつかのチャネルで処理すると、worker.resume.observed には "research" と "summarisation" の両方の ObservedStat が格納され、それぞれ個別に追跡されます。これにより、ピア探索クエリ（peers(action="find", capability="research")。ag2-network-tools-and-views を参照）で、レイテンシや完了率に基づいてランキングできます。

ハブの状態を読み取る

呼び出し	戻り値
await hub.get_channel(channel_id)	ChannelMetadata のスナップショット（状態、参加者、close_reason）
await hub.get_resume(agent_id)	現在の Resume（observed を含む）
await hub.get_passport(agent_id)	現在の Passport
await hub.list_agents(kind=None)	登録済みのパスポート。kind="agent" / "human" / "remote_agent" で Passport.kind によるフィルタリングが可能
await hub.read_wal(channel_id)	そのチャネル内の Envelope の順序付きリスト
await hub.audit_log.read_all()	すべての監査記録
hub.arbiter	アクティブな HubArbiter（読み取り専用）

ハブは変更のたびに Resume.last_updated を設定するため、タイムスタンプを比較して古いビューを検出できます。これらの pull 呼び出しではなく push を利用するには、HubListener を登録します。

クイックリファレンス — インポート

from ag2.network import (
    # Identity
    Passport, Resume, ResumeExample, ObservedStat,
    # Rules
    Rule, AccessBlock, LimitsBlock, RateBlock, InboxBlock,
    ChannelTypeAccess, parse_duration,
    # Arbiter (swappable access / routing seam)
    HubArbiter, BaseHubArbiter, RuleBasedArbiter, Allow, Deny,
    # Listeners (live observability)  — hub.register_listener(...)
    HubListener, BaseHubListener,
    # Auth
    AuthAdapter, AuthRegistry, AuthBlock, NoAuth, ApiKeyAuth,
    # Expectations
    Expectation,
    ExpectationEvaluator, ExpectationContext,
    AcksWithinEvaluator, ReplyWithinEvaluator, MaxSilenceEvaluator,
    AuditHandler, NotifyChannelHandler, AutoCloseHandler,
    Violation, ViolationHandler,
    default_evaluators, default_handlers,
    # Audit kinds
    AUDIT_KIND_AGENT_REGISTERED,
    AUDIT_KIND_AGENT_UNREGISTERED,
    AUDIT_KIND_RESUME_SET,
    AUDIT_KIND_SKILL_SET,
    AUDIT_KIND_RULE_SET,
    AUDIT_KIND_CHANNEL_CREATED,
    AUDIT_KIND_CHANNEL_CLOSED,
    AUDIT_KIND_CHANNEL_EXPIRED,
    AUDIT_KIND_TASK_TERMINATED,
    AUDIT_KIND_EXPECTATION_VIOLATED,
    RESUME_SOURCE_OBSERVED, RESUME_SOURCE_TENANT,
    # Task observation
    TaskMirror,
    # Errors — the full family (no `LimitsExceeded`; limit/access denials raise AccessDeniedError, inbox-full raises InboxFull)
    NetworkError, AccessDeniedError, AuthError, InboxFull, NotFoundError, ProtocolError,
)