aimu.events

Run events: a frozen-dataclass event union plus one Callable[[RunEvent], None] sink. Attach a sink to a client, an Agent, an OrchestratorAgent (assemble(..., events=...)), or a workflow's from_client(...) to see what a run actually did. See how-to: observe a run.

Sink and dispatch

aimu.events.EventSink module-attribute

EventSink = Callable[[RunEvent], None]

One callable taking one :class:RunEvent. Its return value is ignored and an exception it raises is logged, not propagated (see :func:emit). It must be thread-safe: with concurrent_tool_calls=True tool events arrive from concurrent workers in nondeterministic order.

aimu.events.emit

emit(sink: Optional[EventSink], event: RunEvent) -> None

Deliver event to sink, logging rather than propagating its failure.

Observation must not change what it observes: a sink that raises would otherwise break a run that was working. Same contract as the SubagentObserver display hook.

A sink must be thread-safe. With concurrent_tool_calls=True the tool-loop engine dispatches a turn's tool calls from a ThreadPoolExecutor (sync) or an asyncio.TaskGroup (async), so :class:ToolCalled / :class:ToolDenied are emitted concurrently and their order is nondeterministic -- a sink that appends to a plain list, writes a file, or accumulates per-tool state needs its own lock. Turn and run events are emitted from the calling thread and stay ordered.

aimu.events.log_events

log_events(target: Logger, level: int = logging.INFO) -> EventSink

A sink that writes one line per event.

The shortest path to the comprehension payoff -- attach it and watch what actually happened -- and the sink the docs lead with.

Events

aimu.events.RunEvent dataclass

RunEvent(agent: Optional[str] = None, iteration: int = 0, run_id: Optional[str] = None, parent_run_id: Optional[str] = None)

Base for every event.

agent and iteration mirror :class:~aimu.models.StreamChunk's fields of the same name, so one sink can attribute events correctly inside a nested workflow.

run_id identifies one run of an agent, so two concurrent runs of the same agent stay apart; parent_run_id is the run that started it, when the caller said so (AIMU's spawn tools always do). Both are None outside a run, e.g. a bare client.chat(), and on turn events that reach a durable client.events sink with no agent run behind them (the same limit as agent and iteration attribution).

aimu.events.RunStarted dataclass

RunStarted(agent: Optional[str] = None, iteration: int = 0, run_id: Optional[str] = None, parent_run_id: Optional[str] = None, task: str = '')

Bases: RunEvent

A Runner (an agent or a workflow) began a run.

Fires once at the top of run(), before the first model turn, and is paired with exactly one :class:RunFinished. task is the task string as the caller passed it.

aimu.events.ModelTurnStarted dataclass

ModelTurnStarted(agent: Optional[str] = None, iteration: int = 0, run_id: Optional[str] = None, parent_run_id: Optional[str] = None, model: str = '', message_count: int = 0, tool_names: tuple[str, ...] = ())

Bases: RunEvent

One model request is about to be issued.

Fires inside chat() / generate(), before the provider is called, and is paired with exactly one :class:ModelTurnFinished even when the request raises.

message_count is how many messages this request carries (including the user turn it is about to append, and the system message when this turn seeds one); it is 1 for a stateless generate(), which sends only the prompt. tool_names are the tools advertised on this turn, not the ones called.

aimu.events.RequestPrepared dataclass

RequestPrepared(agent: Optional[str] = None, iteration: int = 0, run_id: Optional[str] = None, parent_run_id: Optional[str] = None, provider: str = '', model: str = '', endpoint: Optional[str] = None, payload: Any = None)

Bases: RunEvent

The payload as it goes to the provider, after every adaptation AIMU applies.

This is the event that makes "when a model surprises you, the surprise should be the model's" checkable: between a caller's chat() and the wire sit the four-tier generate_kwargs merge, the GENERATE_KWARG_SUPPORT renames and drops, thinking resolution, strip_inert_keys, and provider format adaptation. Without this, none of it is visible at runtime.

The payload is the request, unredacted -- it contains whatever the caller put in the conversation. Redacting here would reintroduce the hiding this exists to remove; a sink that ships events off the machine is the right place to filter.

It is plain JSON data (see aimu.models._internal.json_safe): NOT_GIVEN keys are dropped, since the SDK omits them from the wire, and anything with no JSON form becomes a visible marker. client.last_request keeps the raw payload.

endpoint is the server the request went to (base URL, credentials and query stripped), so two OpenAI-compatible servers are distinguishable and a recorded request can be re-sent. None means there was no wire request: HuggingFace and LlamaCpp run in-process and record the nearest equivalent, which is inspectable but not replayable.

aimu.events.ModelTurnFinished dataclass

ModelTurnFinished(agent: Optional[str] = None, iteration: int = 0, run_id: Optional[str] = None, parent_run_id: Optional[str] = None, model: str = '', text: Optional[str] = None, usage: Optional[dict] = None, duration_s: float = 0.0, error: Optional[BaseException] = None, stop_reason: Optional[str] = None, tool_calls: list = list(), thinking: Optional[str] = None)

Bases: RunEvent

The model request that a :class:ModelTurnStarted announced has ended.

Fires for every started turn, including one that failed: error is then the exception (ContextOverflowError, ModelConnectionError, a provider 4xx, ...) and text / usage are None. A sink pairing started/finished can therefore close its span unconditionally.

On a streamed turn this fires when the stream is actually drained (or abandoned), not when it was created, so usage -- which only populates at the end -- is real and duration_s covers the whole stream. text is the concatenated GENERATING content on a streamed turn, and the returned string on a non-streamed one.

Not emitted at all on the schema= structured-output path: chat() / generate() return before :class:ModelTurnStarted is emitted when schema is set, since that path makes exactly one call and reports itself a different way (see Agent._structured_run_events / :class:RunFinished). There is therefore no started turn to leave unpaired there either.

stop_reason is the provider's own word for how the turn ended ("end_turn", "tool_use", "length", "refusal", ...), the same value as client.last_stop_reason. It is here because a sink often never holds the client: one handed a spawned sub-agent's events cannot read that sub-agent's client at all. None means the provider said nothing.

tool_calls are the tool calls the model requested on this turn, parsed and before any dispatch, each {"name", "arguments", "id"}: a tools-only round's text is empty, so without them a recorded turn would show no response at all. thinking is the turn's reasoning (client.last_thinking), None when there was none. Both are read when the event fires, so on a streamed turn they are complete.

aimu.events.ToolCalled dataclass

ToolCalled(agent: Optional[str] = None, iteration: int = 0, run_id: Optional[str] = None, parent_run_id: Optional[str] = None, name: str = '', arguments: dict = dict(), result: Optional[str] = None, error: Optional[str] = None, duration_s: float = 0.0)

Bases: RunEvent

A tool the model asked for was dispatched.

Fires after the call returns, for every outcome the model gets to see: a normal result, a tool that raised (error set, result carrying the message handed back to the model), invalid arguments, and a name the model invented that matches no tool. It does not fire for a call a tool_approval policy refused -- that is :class:ToolDenied.

arguments is what the model passed, before AIMU's coercion. With concurrent_tool_calls=True these are emitted from worker threads (sync) or concurrent tasks (async): see :func:emit on sink thread-safety and ordering.

aimu.events.ToolDenied dataclass

ToolDenied(agent: Optional[str] = None, iteration: int = 0, run_id: Optional[str] = None, parent_run_id: Optional[str] = None, name: str = '', arguments: dict = dict(), reason: str = '')

Bases: RunEvent

A tool_approval policy refused this call.

Fires instead of :class:ToolCalled, before the tool runs: nothing was executed. The model still sees a tool message saying the call was not approved.

reason is what the policy gave when it returned Denied(reason), and "" when it returned a bare False. It is the same text the model was shown, so a sink recording denials records why without having to reproduce the policy's logic.

aimu.events.ContextCompacted dataclass

ContextCompacted(agent: Optional[str] = None, iteration: int = 0, run_id: Optional[str] = None, parent_run_id: Optional[str] = None, dropped: list = list(), before_tokens: int = 0, after_tokens: int = 0)

Bases: RunEvent

Conversation history was rewritten by an Agent's compaction= callable.

Fires once per compaction, after the rewrite, before the next model turn.

dropped is the removed messages, not a count, so a caller who wants the discarded turns still has them. They are copies, so a later in-place edit of client.messages (the provenance tag an agent loop adds, say) cannot mutate an already-emitted event.

before_tokens / after_tokens are AIMU's own default token estimate (see :func:aimu.context.count_tokens), not a measurement of whatever the compaction callable itself counted to decide what to drop. A callable that used a real tokenizer, a word count, or any other budget will disagree with these numbers -- that is stated rather than hidden, since AIMU cannot see inside an opaque callable to know what it actually counted. Treat them as rough orientation, not a claim about the number that drove the decision.

aimu.events.RunFinished dataclass

RunFinished(agent: Optional[str] = None, iteration: int = 0, run_id: Optional[str] = None, parent_run_id: Optional[str] = None, result: Optional[str] = None, error: Optional[BaseException] = None, ended_by: Optional[RunEnding] = None, max_iterations: Optional[int] = None)

Bases: RunEvent

A Runner finished, successfully or not.

Fires in a finally, so it is emitted even when the run raised: error is the exception in that case and result is None.

result is also None on a streamed run (run(stream=True)): the chunks go to the caller as they arrive and the runner never assembles a final string, so there is no result to report. Read the text off the GENERATING chunks instead.

ended_by names the decision that ended the run, so a host can say why without re-deriving the loop's control flow from the events before it:

• "answered": the model's last turn was a real answer and no inbox message was waiting.
• "wrap_up": the round budget ran out while the model still wanted tools (or had produced nothing), so the loop made one tools-disabled call asking for a final answer, and got one.
• "error": the run raised; error is the exception (DegenerateTurnError when even the wrap-up produced nothing, TruncatedTurnError, ModelRefusalError, CancelledError, ...).

It is None from a Runner that has no such decision to report (a workflow).

max_iterations is the round budget the loop ran under, for reading iteration against: a wrap-up at iteration == max_iterations is the uncounted call after the cap. A delivered inbox message restarts the count, so a messaged run can legitimately pass it. None where there was no loop (a schema= run makes one call).