Observation Events

#Attach an Observer

Pass a synchronous callback when you create or open an agent:

python

from daita import Agent, AgentEvent
 
events: list[AgentEvent] = []
 
agent = await Agent.open("atlas", observer=events.append)
try:
    result = await agent.run("Summarize revenue by region")
finally:
    await agent.close()
 
for event in events:
    print(event.kind.value, event.run_id, event.data.to_dict())

The observer runs on the foreground execution path and must return promptly. If it raises an exception, Daita swallows that observer failure so it cannot change the agent result.

#Event Kinds

Event	Emitted after
`run.started`	A run has been admitted
`model.completed`	A model request has completed
`tool.started`	A tool call is about to execute
`approval.requested`	A local write is awaiting a decision
`approval.decided`	The approval handler returned, failed, or was unavailable
`tool.completed`	Tool execution completed or failed
`run.completed`	The run reached a terminal outcome

Each immutable AgentEvent includes a timezone-aware timestamp, run_id, conversation_id, kind, and bounded JSON-compatible data.

#Delivery Contract

Observation is intentionally small:

events are best effort and non-durable;
callbacks are synchronous, not awaited background jobs;
event data is bounded and omits full transcript content;
there are no token-by-token text delta events; and
Daita does not ship an event store, trace backend, or telemetry exporter.

Persist, queue, or export events in caller-owned code when you need durable operational records.

#JSON Lines from the Terminal

For one-shot commands, write event records to standard error as JSON Lines:

bash

daita run atlas "Summarize revenue" \
  --model openai:gpt-5.6-terra \
  --events-jsonl

The final answer remains on standard output, which makes it possible to route answer text and operational events independently.

#Aggregate Measurements

The pure evaluation helper can turn caller-retained events into content-free counters:

python

from daita.evaluation import measure_observer_events
 
measurement = measure_observer_events(events)
print(measurement.model_calls, measurement.tool_calls)
print(measurement.total_tokens, measurement.estimated_cost_usd)

See Evaluation for baseline-versus-learned comparisons.