orxtra v0.13.0 /scheduler.src.orxtra.scheduler._executor
Updated
On this page

Scheduler, the task executor: runs workflows via execute_workflow, drives topological task groups, handles crash recovery, PG LISTEN/NOTIFY, pause/resume/abort, and Overseer/transport event sinks.

#scheduler.src.orxtra.scheduler._executor

#scheduler.src.orxtra.scheduler._executor

#classify_error

python
def classify_error(error: Exception) -> ErrorCategory

Classify an exception into an error category.

#Scheduler

#run_consult

python
async def run_consult(self, agent: str, question: str, variable_values: dict[str, str] | None=None) -> str

#run_workflow_check

python
async def run_workflow_check(self, execution: Execution) -> CheckResult

#execute_workflow

python
async def execute_workflow(self, config: WorkflowConfig) -> None

#_crash_recovery

python
async def _crash_recovery(self) -> None

Three-pass idempotent crash recovery startup.

Acquires the run lock FIRST so that clean_orphaned does not mark our own run as orphaned.

#_setup_pg_listener

python
async def _setup_pg_listener(self) -> tuple[asyncio.Task[None] | None, asyncpg.pool.PoolConnectionProxy[Any] | None, object]

Set up cross-process signal delivery via PG LISTEN.

Returns (listener_task, listener_conn, notification_callback).

#_cleanup_pg_listener

python
async def _cleanup_pg_listener(self, pg_listener_task: asyncio.Task[None] | None, pg_listener_conn: asyncpg.pool.PoolConnectionProxy[Any] | None, on_notification: Any) -> None

Clean up PG listener resources.

#_register_workflow_tasks

python
async def _register_workflow_tasks(self, config: WorkflowConfig) -> dict[str, UUID]

Validate config and register all workflow tasks.

#_execute_task_groups

python
async def _execute_task_groups(self, config: WorkflowConfig, task_id_map: dict[str, UUID]) -> None

Execute task groups in topological order.

#execute_task

python
async def execute_task(self, task: TaskSpec, parent_task_id: UUID | None, *, task_id: UUID | None=None, variables: dict[str, Any] | None=None) -> TaskResult

#_stop_services

python
async def _stop_services(self) -> None

Stop all running service instances.

#abort

python
async def abort(self) -> None

#pause

python
async def pause(self) -> None

#resume

python
async def resume(self) -> None

#is_paused

python
def is_paused(self) -> bool

#add_overseer_sink

python
def add_overseer_sink(self, sink: EventSink[OverseerEvent]) -> None

Add an overseer event sink.

#remove_overseer_sink

python
def remove_overseer_sink(self, sink: EventSink[OverseerEvent]) -> None

Remove an overseer event sink. No-op if not present.

#add_transport_sink

python
def add_transport_sink(self, sink: EventSink[TransportEvent]) -> None

Add a transport event sink.

Also registers the sink on all currently active sessions so that events from in-progress tasks are delivered immediately.

#remove_transport_sink

python
def remove_transport_sink(self, sink: EventSink[TransportEvent]) -> None

Remove a transport event sink. No-op if not present.

Also removes the sink from all currently active sessions.

#_handle_control_signal

python
async def _handle_control_signal(self, run_id: UUID, new_status: str) -> None

Handle a control signal from the trace layer.

#_handle_headless_event

python
async def _handle_headless_event(self, event: OverseerEvent) -> None

Dispatch a headless fallback for an event when no Overseer is configured.

Uses the same FALLBACK_BEHAVIORS/FALLBACK_HANDLERS tables as degraded mode.

#_send_overseer_event

python
async def _send_overseer_event(self, event: OverseerEvent) -> None

Send an event to the Overseer with a verify-then-accept retry loop.

#_dispatch_to_overseer_sinks

python
async def _dispatch_to_overseer_sinks(self, event: OverseerEvent) -> None

Dispatch an OverseerEvent to all registered sinks.

Errors in individual sinks are logged but do not crash the scheduler. Uses a snapshot copy to allow safe concurrent mutation of the sinks list.

#_write_coherence_summary

python
async def _write_coherence_summary(self) -> None

Ask the Overseer for a coherence summary and persist it.

Requires an Overseer session -- skipped in headless mode (no meaningful fallback).

#_check_session_handoff

python
async def _check_session_handoff(self) -> None

Check if the Overseer session needs handoff due to token usage.

Requires an Overseer session -- no-op in headless mode.

#_send_pending_advisories

python
async def _send_pending_advisories(self) -> None

Send stored structural advisories to the Overseer (or headless fallback handler).

Search