orxtra v0.13.0 /services.src.orxtra.services._run_manager
Updated
On this page

RunManager maps active run IDs to their Scheduler instances so AG-UI's SSE clients can subscribe transport/overseer sinks to a live run; subscribing to an inactive run returns None.

#services.src.orxtra.services._run_manager

#services.src.orxtra.services._run_manager

RunManager -- maps active run IDs to their Scheduler instances.

Allows SSE clients (AG-UI) to subscribe to a running run's event streams by registering transport and overseer sinks on the scheduler. Subscribing to a run that is not active (not registered) returns None, signaling the caller to serve a completed-run snapshot instead.

#RunManager

Instance-scoped registry of active run schedulers.

Held on the DispatchContext and constructed during API lifespan.

#register_run

python
def register_run(self, run_id: UUID, scheduler: Scheduler) -> None

Register a scheduler for an active run.

#deregister_run

python
def deregister_run(self, run_id: UUID) -> None

Remove the scheduler for a completed/failed run.

#subscribe

python
def subscribe(self, run_id: UUID, transport_sink: EventSink[TransportEvent], overseer_sink: EventSink[OverseerEvent]) -> Callable[[], None] | None

Subscribe sinks to a live run.

Returns an unsubscribe closure if the run is active, or None if the run is not registered (completed/not started).

#is_active

python
def is_active(self, run_id: UUID) -> bool

Check whether a run is currently registered.

#active_run_ids

python
def active_run_ids(self) -> frozenset[UUID]

Return the set of currently active run IDs.

Search