EventBus and SessionController API Reference¶

EventBus¶

PubSub event bus for cross-turn event streaming.

Decouples event producers (agents) from consumers (protocol handlers). Events are broadcast to all subscribers for a given session via _send() with no publish-side buffering or coalescing.

Event coalescing/merging is the subscriber's responsibility. Subscribers should drain their queue using drain_and_merge() to batch-merge consecutive same-type events (e.g., PartDeltaEvent text chunks) for efficient processing.

Safety features: - Bounded asyncio.Queue with configurable overflow policies - Automatic cleanup of dead subscribers - Queue.shutdown()-based shutdown (no sentinel None)

init ¶

__init__(
    max_queue_size: int = DEFAULT_QUEUE_MAXSIZE,
    replay_buffer_size: int = 100,
    session_controller: SessionController | None = None,
    overflow_policy: OverflowPolicy = "drop_oldest",
) -> None

Initialize the event bus.

Parameters:

Name	Type	Description	Default
`max_queue_size`	`int`	Maximum buffer size for subscriber queues.	`DEFAULT_QUEUE_MAXSIZE`
`replay_buffer_size`	`int`	Maximum number of events retained per session for replay.	`100`
`session_controller`	`SessionController \| None`	Optional session controller for hierarchy queries.	`None`
`overflow_policy`	`OverflowPolicy`	Policy for handling full subscriber queues. One of `drop_oldest`, `drop_newest`, `drop_subscriber`. `block` is NOT supported (would deadlock the run loop).	`'drop_oldest'`

Raises:

Type	Description
`ValueError`	If `overflow_policy` is `"block"` or not a valid policy.

clear_replay_buffer ¶

clear_replay_buffer(session_id: str) -> None

Clear the replay buffer for a session.

Removes all historical events from the replay buffer so that new subscribers only receive events from this point forward. This should be called at the start of each turn to prevent stale events (including terminal events like StreamCompleteEvent) from previous turns being replayed to new subscribers.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session whose replay buffer to clear.	required

close_session `async` ¶

close_session(session_id: str) -> None

Close all subscriptions for a session.

Shuts down all subscriber queues to signal QueueShutDown to consumers, and clears the replay buffer.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session to close subscriptions for.	required

get_subscriber_counts `async` ¶

get_subscriber_counts() -> dict[str, int]

Get subscriber counts per session.

Returns:

Type	Description
`dict[str, int]`	A snapshot mapping session IDs to subscriber counts.

publish `async` ¶

publish(session_id: str, event: Any) -> None

Publish an event to all subscribers for a session.

Wraps the event in an EventEnvelope and sends it directly via _send(). PartDeltaEvent with delta=None is dropped (no content to deliver). Coalescing is handled subscriber-side by drain_and_merge().

Each published event is assigned a monotonically increasing event_id so that subscribers can request conditional replay via last_event_id.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session that produced the event.	required
`event`	`Any`	The event to broadcast.	required

subscribe `async` ¶

subscribe(
    session_id: str,
    scope: str = "session",
    *,
    replay: bool = True,
    last_event_id: int | None = None
) -> Queue[EventEnvelope]

Subscribe to events for a session.

New subscribers receive replayed historical events from the replay buffer before live events. Events published during the replay phase are drained and re-inserted after historical events to preserve ordering and avoid loss.

Conditional replay is supported via replay and last_event_id:

replay=False: Skip the replay buffer entirely (zero historical events). Only live events published after subscription are delivered.
replay=True, last_event_id=None: Replay all buffered events (default — backward compatible).
replay=True, last_event_id=N: Replay only events with event_id > N. If the buffer's oldest event has event_id > N + 1 (gap detected, meaning events N+1..oldest-1 have been evicted), fall back to replaying all events from that buffer to avoid silent data loss.

For scope="all", the filtering applies per-buffer after collecting from all buffers.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session to subscribe to.	required
`scope`	`str`	Subscription scope - "session" (exact match), "descendants" (self + children), or "subtree" (self + parent + siblings). Deprecated: descendants scope The "descendants" scope is deprecated for protocol server use. It has known issues with replay buffer data loss, O(N) recursive traversal, and duplicate deliveries. Protocol servers should use "session" scope with explicit child consumers via `ProtocolEventConsumerMixin._on_spawn_session_start()` instead. The "descendants" enum value is retained for backward compatibility.	`'session'`
`replay`	`bool`	If False, skip replay buffer (no historical events).	`True`
`last_event_id`	`int \| None`	If provided with `replay=True`, only replay events with `event_id > last_event_id`. Gap detection falls back to full replay when the buffer is missing contiguous events.	`None`

Returns:

Type	Description
`Queue[EventEnvelope]`	An `asyncio.Queue` to consume events from.

unsubscribe `async` ¶

unsubscribe(session_id: str, queue: Queue[EventEnvelope]) -> None

Unsubscribe from events.

Shuts down the subscriber's queue so the consumer receives QueueShutDown. Cleans up empty subscriber lists to prevent memory leaks.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session to unsubscribe from.	required
`queue`	`Queue[EventEnvelope]`	The queue returned by subscribe().	required

SessionController¶

Bases: SessionControllerAgentMixin, SessionControllerRunsMixin, SessionControllerCloseMixin

Manages per-session agent lifecycle.

Extracted from ACP's AgentPoolACPAgent._session_agents and OpenCode's ServerState._session_agents.

Safety features: - Single global lock for session creation (no DCL) - Per-session turn lock for serialization - Explicit cleanup of all resources - Support for all agent types (with per-session agents for NativeAgentConfig only)

runtime_registry `property` ¶

runtime_registry: RuntimeAgentRegistry

Runtime agent registry for programmatically-created agents.

init ¶

__init__(
    pool: AgentPool[Any],
    store: SessionPersistence | None = None,
    cleanup_callback: Callable[[str], Awaitable[None]] | None = None,
    max_concurrent_runs: int | None = None,
    session_ttl_seconds: float = DEFAULT_SESSION_TTL_SECONDS,
    cleanup_interval_seconds: float | None = None,
    deferred_cleanup_interval_seconds: float = 60.0,
) -> None

Initialize the session controller.

Parameters:

Name	Type	Description	Default
`pool`	`AgentPool[Any]`	The agent pool to resolve agents from.	required
`store`	`SessionPersistence \| None`	Optional session store for persistence.	`None`
`cleanup_callback`	`Callable[[str], Awaitable[None]] \| None`	Optional callback invoked when a session is cleaned up.	`None`
`max_concurrent_runs`	`int \| None`	Maximum number of concurrent runs across all sessions.	`None`
`session_ttl_seconds`	`float`	TTL for idle sessions in seconds.	`DEFAULT_SESSION_TTL_SECONDS`
`cleanup_interval_seconds`	`float \| None`	Interval for the session TTL cleanup loop. Defaults to `session_ttl_seconds / 2`.	`None`
`deferred_cleanup_interval_seconds`	`float`	Interval for the deferred call expiry cleanup loop in seconds.	`60.0`

cancel_all_pending_questions ¶

cancel_all_pending_questions() -> list[str]

Cancel all pending questions across all sessions.

Iterates over every session, cancels each pending question's future, and returns the IDs of all cancelled questions.

Returns:

Type	Description
`list[str]`	List of cancelled question IDs.

cancel_run_for_session ¶

cancel_run_for_session(session_id: str) -> bool

Cancel the active run for a session.

Only cancels the run if the RunHandle is still active (complete_event is not set). If the handle has already completed, the cancel is skipped.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session whose run should be cancelled.	required

Returns:

Type	Description
`bool`	`True` if cancellation was initiated, `False` if the
`bool`	session/run was not found or already completed.

cancel_session_pending_questions ¶

cancel_session_pending_questions(session_id: str) -> list[str]

Cancel pending questions for a specific session.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session whose pending questions should be cancelled.	required

Returns:

Type	Description
`list[str]`	List of cancelled question IDs.

close_session `async` ¶

close_session(session_id: str) -> None

Close a session and clean up resources.

Delegates to :meth:_close_session_run_turn which performs checkpoint-on-close (if needed) and then the standardized 7-step cleanup ordering via :meth:_close_session_unlocked.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session to close.	required

find_sessions_by_agent_name ¶

find_sessions_by_agent_name(agent_name: str) -> list[SessionState]

Find all active sessions associated with a given agent name.

Parameters:

Name	Type	Description	Default
`agent_name`	`str`	The agent name to search for.	required

Returns:

Type	Description
`list[SessionState]`	List of session states matching the agent name, excluding closing sessions.

get_children ¶

get_children(session_id: str) -> list[str]

Get child session IDs for a session.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The parent session ID.	required

Returns:

Type	Description
`list[str]`	List of child session IDs.

get_or_create_session `async` ¶

get_or_create_session(
    session_id: str,
    agent_name: str | None = None,
    parent_session_id: str | None = None,
    lifecycle_policy: str | None = None,
    **metadata: Any
) -> tuple[SessionState, bool]

Get or create a session.

Uses single global lock for simplicity and safety. Session creation is infrequent - no need for DCL optimization.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Unique identifier for the session.	required
`agent_name`	`str \| None`	Name of the agent to associate with the session.	`None`
`parent_session_id`	`str \| None`	Optional parent session ID for hierarchical sessions.	`None`
`lifecycle_policy`	`str \| None`	Optional lifecycle policy override.	`None`
`**metadata`	`Any`	Arbitrary metadata to attach to the session.	`{}`

Returns:

Type	Description
`SessionState`	A tuple of (session_state, was_created) where was_created is True
`bool`	if the session was newly created, False if it already existed.

get_or_create_session_agent `async` ¶

get_or_create_session_agent(
    session_id: str, agent_name: str | None = None, input_provider: Any | None = None
) -> BaseAgent[Any, Any]

Get or create a dedicated agent for a session.

Delegates agent creation to AgentFactory.create_session_agent(), handling caching, session lookup, and config resolution locally.

NOTE: Always acquires self._lock to prevent races with close_session().

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Unique identifier for the session.	required
`agent_name`	`str \| None`	Name of the agent to use.	`None`
`input_provider`	`Any \| None`	Optional input provider for the agent.	`None`

Returns:

Type	Description
`BaseAgent[Any, Any]`	The agent instance (per-session or shared).

get_parent ¶

get_parent(session_id: str) -> SessionState | None

Get the parent session state for a session.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The child session ID.	required

Returns:

Type	Description
`SessionState \| None`	The parent session state, or None if not found.

get_session ¶

get_session(session_id: str) -> SessionState | None

Get a session by ID.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session ID to look up.	required

Returns:

Type	Description
`SessionState \| None`	The session state, or None if not found.

get_session_agent ¶

get_session_agent(session_id: str) -> BaseAgent[Any, Any] | None

Get the agent for a session.

Returns the per-session agent if one exists, otherwise the shared agent that was assigned to the session. If the session has no agent assigned yet, a warning is logged and None is returned.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session ID to look up.	required

Returns:

Type	Description
`BaseAgent[Any, Any] \| None`	The agent instance, or None if the session is unknown.

list_pending_permissions ¶

list_pending_permissions() -> list[PendingPermission]

List all pending permissions across sessions.

Returns:

Type	Description
`list[PendingPermission]`	A list of pending permissions. Currently returns an empty list.

list_pending_questions ¶

list_pending_questions() -> list[Any]

List all pending questions across sessions.

Aggregates pending questions from each session's SessionState.

Returns:

Type	Description
`list[Any]`	A list of pending question objects.

list_sessions ¶

list_sessions() -> list[SessionInfo]

List all active sessions.

Returns:

Type	Description
`list[SessionInfo]`	A list of SessionInfo DTOs for all active sessions.

revoke_inject ¶

revoke_inject(session_id: str, message_id: str) -> bool

Revoke a pending steer or followup message by ID.

In the per-prompt model, revocation is handled by SessionState.revoke() which cancels queued steer messages in feedback_queue.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session containing the message.	required
`message_id`	`str`	The ID of the message to revoke.	required

Returns:

Type	Description
`bool`	`True` if revoked or already gone (idempotent), `False`
`bool`	if the session is not found.

start_cleanup_task `async` ¶

start_cleanup_task() -> None

Start background tasks for session cleanup.

Launches two background tasks: - _cleanup_loop: periodically closes expired sessions (TTL-based). - _start_cleanup_loop: periodically expires stale deferred calls.

Both tasks are stored in _background_tasks to prevent garbage collection mid-execution (per asyncio.create_task best practice).

stop_cleanup_task `async` ¶

stop_cleanup_task() -> None

Stop both background cleanup tasks.

wait_for_completion `async` ¶

wait_for_completion(session_id: str, timeout: float | None = 300) -> str

Wait for the active run on a session to complete.

In the per-prompt model, complete_event fires when the RunHandle's single-turn generator terminates. If _consume_run() chains to a new RunHandle, the caller must re-check session.current_run_id to detect the new turn.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session to wait for.	required
`timeout`	`float \| None`	Maximum seconds to wait. Defaults to 300 seconds.	`300`

Returns:

Type	Description
`str`	The `session_id` on completion.

Raises:

Type	Description
`SessionNotFoundError`	If the session does not exist.
`TimeoutError`	If the run does not complete within `timeout` seconds.

EventBus and SessionController API Reference¶

EventBus¶

__init__ ¶

clear_replay_buffer ¶

close_session async ¶

get_subscriber_counts async ¶

publish async ¶

subscribe async ¶

unsubscribe async ¶

SessionController¶

runtime_registry property ¶

__init__ ¶

cancel_all_pending_questions ¶

cancel_run_for_session ¶

cancel_session_pending_questions ¶

close_session async ¶

find_sessions_by_agent_name ¶

get_children ¶

get_or_create_session async ¶

get_or_create_session_agent async ¶

get_parent ¶

get_session ¶

get_session_agent ¶

list_pending_permissions ¶

list_pending_questions ¶

list_sessions ¶

revoke_inject ¶

start_cleanup_task async ¶

stop_cleanup_task async ¶

wait_for_completion async ¶

init ¶

close_session `async` ¶

get_subscriber_counts `async` ¶

publish `async` ¶

subscribe `async` ¶

unsubscribe `async` ¶

runtime_registry `property` ¶

init ¶

close_session `async` ¶

get_or_create_session `async` ¶

get_or_create_session_agent `async` ¶

start_cleanup_task `async` ¶

stop_cleanup_task `async` ¶

wait_for_completion `async` ¶