Observation Events

#Attach an Observer

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

from pathlib import Path
 
from daita import Agent, AgentEvent, LocalWorkspace
 
events: list[AgentEvent] = []
 
agent = await Agent.open(
    "atlas",
    workspace=LocalWorkspace(Path("/absolute/path/project")),
    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

Each immutable AgentEvent includes a timezone-aware timestamp, run_id, conversation_id, run_origin, kind, and bounded JSON-compatible data. model.text_delta is emitted only when the admitted profile and provider support streaming; each fragment is capped at 1,024 characters and the terminal transcript still stores only the canonical completed response.

#Delivery Contract

Observation is intentionally small:

• events are best effort and non-durable;
• callbacks are synchronous, not awaited background jobs;
• lifecycle event data is bounded and omits full transcript content;
• streamed text fragments are presentation events, not durable transcript records;
• 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:

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:

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.