> ## Documentation Index
> Fetch the complete documentation index at: https://codexceed.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# avatar.bus

EventBus — bounded, non-blocking fan-out + one privileged journal (ADR-0001, Phase 3.1).

Stamps the global order (`event_id`) on every event, appends it losslessly to the
privileged `JsonlEventJournal` (if wired), then fans it out to each subscriber's
**bounded** queue. The bound is *soft*: only droppable `*_update` events are shed when a
subscriber is at its cap; lifecycle/control events are always enqueued (they may exceed
the cap). A slow or broken subscriber can therefore never stall publishing or its peers,
and a subscriber's drops show up as `event_id` gaps — never silent reordering. The
journal stays lossless regardless of what any subscriber sheds.

This was the foundation's simple unbounded fan-out (lived in `session.py`); Lane 1 grew
it into its own module behind the same `events()` / `EventSink` API.

## Classes

### `EventBus`

Stamps, journals, and **fans events to every subscriber's bounded queue** (§13).

Each `subscribe()` gets its own queue, so independent observers — a TUI, a telemetry
exporter, a benchmark collector — each see the same stream rather than competing for
one. Fan-out is non-blocking: a full subscriber sheds droppable updates instead of
blocking the publisher. A late subscriber sees only events published after it
subscribed; the `journal` (and the in-memory `history`) is the lossless record.

Args:
session\_id: Stamped on every event so a stream/journal groups back to its run.
journal: The privileged lossless sink; every published event is appended to it
in order before fan-out. `None` disables on-disk journaling (the foundation
behavior — in-memory `history` is still kept).

```python theme={null}
EventBus(session_id: str, *, journal: JsonlEventJournal | None = None) -> None
```

#### `EventBus.close(self) -> None`

Signal end-of-stream to every subscriber and close the journal (idempotent).

#### `EventBus.emit(self, draft: Annotated[AgentStart | AgentEnd | TurnStart | TurnEnd | PhaseChanged | DeclarationRequired | ModelDecisionEvent | ModelUpdate | ToolStart | ToolEnd | ApprovalRequested | ApprovalResolved | DecisionError | ModelUsage | VerificationPlanFrozen | VerificationStart | VerificationEnd | CancellationObserved | TaskEscalated, FieldInfo(annotation=NoneType, required=True, discriminator='type')]) -> Annotated[AgentStart | AgentEnd | TurnStart | TurnEnd | PhaseChanged | DeclarationRequired | ModelDecisionEvent | ModelUpdate | ToolStart | ToolEnd | ApprovalRequested | ApprovalResolved | DecisionError | ModelUsage | VerificationPlanFrozen | VerificationStart | VerificationEnd | CancellationObserved | TaskEscalated, FieldInfo(annotation=NoneType, required=True, discriminator='type')]`

Awaitable publish — the frozen async interface (delegates to `publish_nowait`).

Args:
draft: The event to publish.

Returns:
The stamped event.

#### `EventBus.publish_nowait(self, draft: Annotated[AgentStart | AgentEnd | TurnStart | TurnEnd | PhaseChanged | DeclarationRequired | ModelDecisionEvent | ModelUpdate | ToolStart | ToolEnd | ApprovalRequested | ApprovalResolved | DecisionError | ModelUsage | VerificationPlanFrozen | VerificationStart | VerificationEnd | CancellationObserved | TaskEscalated, FieldInfo(annotation=NoneType, required=True, discriminator='type')]) -> Annotated[AgentStart | AgentEnd | TurnStart | TurnEnd | PhaseChanged | DeclarationRequired | ModelDecisionEvent | ModelUpdate | ToolStart | ToolEnd | ApprovalRequested | ApprovalResolved | DecisionError | ModelUsage | VerificationPlanFrozen | VerificationStart | VerificationEnd | CancellationObserved | TaskEscalated, FieldInfo(annotation=NoneType, required=True, discriminator='type')]`

Stamp `draft`, journal it losslessly, and fan it out (never blocks, §13).

Args:
draft: The event to publish; mutated in place with the ordering keys.

Returns:
The stamped event.

#### `EventBus.subscribe(self, *, max_queue: int | None = None) -> 'Queue'`

Register an independent consumer and return its bounded queue.

Args:
max\_queue: The soft cap for this consumer (droppable updates are shed above
it). Defaults to a generous value that never drops a normal-size stream.

Returns:
A queue that will receive every event published from now on (subject to the
drop policy for `*_update` events), then a `None` close sentinel. An
already-closed bus returns a queue holding only the sentinel.