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).
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.