Task Observation - AG2

Task Observation

The bridge between the Task lifecycle primitive and the network's per-agent track record. When an Agent runs an agent.task(..., capability="X") inside a network turn, a TaskMirror forwards the lifecycle events to the hub. On terminal events with a capability tag, the hub updates the worker's Resume.observed[capability].

In short: agents earn a track record on the network by completing capability-tagged tasks. Other agents (and operators) read that track record off the worker's Resume.

The Mechanism

TaskMirror is a stream subscriber that:

  1. Subscribes to TaskStarted, TaskProgress, TaskCompleted, TaskFailed, TaskExpired events on a stream.
  2. Forwards each as an ag2.task.* envelope to the hub via HubClient.
  3. On terminal events with spec.capability set, calls Hub.record_observation(...) so the hub's per-agent ObservedStat updates.

It's auto-attached by the default handler for the duration of every LLM turn — you don't need to wire it up manually. If you write a custom handler, attach it manually:

<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br> <br>from ag2.network import TaskMirror<br>from ag2.stream import MemoryStream<br>mirror = TaskMirror(<br> hub_client=client._hub_client,<br> owner_id=client.agent_id,<br> channel_id=metadata.channel_id,<br>)<br>stream = MemoryStream()<br>sub_ids = mirror.attach(stream)<br>try:<br> await client.agent.ask(text, stream=stream)<br>finally:<br> mirror.detach(stream, sub_ids)<br>

Capability Tagging

agent.task(...) accepts a capability keyword:

<br>1<br>2<br>3<br>4<br>5<br>6<br>7<br>8<br> <br>async with agent.task(<br> "survey: deployment patterns",<br> capability="research",<br> context=ctx,<br>) as task:<br> await task.progress({"step": "gather"})<br> # ... do work ...<br> await task.complete({"items_found": 7})<br>

ctx is the active Context (passed in by fast_depends to a tool body, or carried explicitly in scripts). It's important to pass context=ctx so the task fires its events on the LLM-turn's stream — that's the stream the mirror is attached to.

capability is a free-form string. Common values: "research", "summarisation", "review", "code_review". Whatever names your application uses internally for capability roles, use them here.

If capability is None (the default), the mirror still forwards lifecycle envelopes to the hub, but doesn't update Resume.observed. The track record is opt-in.

ObservedStat

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

Read it off the worker's resume:

<br>1<br>2<br>3<br>4<br> <br>resume = await hub.get_resume(bob.agent_id)<br>stat = resume.observed.get("research")<br>if stat:<br> print(f"completed={stat.completed}/{stat.n} median_latency={stat.p50_latency_ms}ms")<br>

The latency is computed from task_meta.started_at to the terminal event time, sourced from the hub's clock. With a MockClock you can construct deterministic latency values for testing.

What the Mirror Records

TaskMirror.record_observation(...) writes to:

Both updates happen inside the same hub transaction, so partial updates don't occur.

Where TaskMirror Fits in the Default Handler

Look at ag2.network.client.handlers._process_text if you want the exact wiring. Sketch:

<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br> <br>mirror = TaskMirror(<br> hub_client=client._hub_client,<br> owner_id=client.agent_id,<br> channel_id=metadata.channel_id,<br>)<br>sub_ids = mirror.attach(stream)<br>try:<br> reply = await client.agent.ask(<br> current_text,<br> stream=stream,<br> dependencies=dependencies,<br> )<br>finally:<br> mirror.detach(stream, sub_ids)<br>

Notably:

When to Skip Capability Tagging

Not every agent.task(...) deserves a capability tag. Tag only when:

Untagged tasks still get full lifecycle observation in the audit log — just no Resume.observed update. Use them for internal bookkeeping or sub-task delegation that doesn't represent an externally-visible capability.

Cross-Cutting Pattern

A common pattern: an agent has multiple capability roles. Tag each tool's task with the right capability and inspect the resume to see which capabilities are well-exercised.

<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br> <br>@worker.tool<br>async def research(topic: str, ctx: Context) -> str:<br> async with worker.task(f"research: {topic}", capability="research", context=ctx) as t:<br> # ...<br> return f"researched {topic}"<br>@worker.tool<br>async def summarise(text: str, ctx: Context) -> str:<br> async with worker.task("summarise", capability="summarisation", context=ctx) as t:<br> # ...<br> return f"summary: ..."<br>

After a few channels, worker.resume.observed will hold both "research" and "summarisation" ObservedStats, each tracking that capability independently.