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

OverseerInterface is the protocol the Scheduler uses to reach the Overseer; OverseerAdapter bridges it to the real Overseer, with degraded-mode fallbacks (trace-only, log-only, escalate-to-inbox).

#scheduler.src.orxtra.scheduler._overseer

#scheduler.src.orxtra.scheduler._overseer

#_maintain_current_allocations

python
async def _maintain_current_allocations(event: OverseerEvent, logger: logging.Logger, *, trace_writer: Any=None, run_id: UUID | None=None, principal_id: UUID | None=None) -> None

For budget events: do nothing, just log.

#_escalate_to_human_inbox

python
async def _escalate_to_human_inbox(event: OverseerEvent, logger: logging.Logger, *, trace_writer: Any=None, run_id: UUID | None=None, principal_id: UUID | None=None) -> None

Default fallback: create an inbox item for human intervention.

#_write_to_trace

python
async def _write_to_trace(event: OverseerEvent, logger: logging.Logger, *, trace_writer: Any=None, run_id: UUID | None=None, principal_id: UUID | None=None) -> None

Write the event to trace as a headless fallback record. For TaskFailed/TaskEscalated, also creates an inbox item for human review.

#_log_only

python
async def _log_only(event: OverseerEvent, logger: logging.Logger, *, trace_writer: Any=None, run_id: UUID | None=None, principal_id: UUID | None=None) -> None

Log the event, no other action needed.

#OverseerInterface

Interface the scheduler uses to communicate with the Overseer.

#send_event

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

#verify_actions

python
async def verify_actions(self, event_type: str='') -> list[str]

#send_correction

python
async def send_correction(self, message: str) -> None

#is_degraded

python
def is_degraded(self, event_type: str) -> bool

#refine_context

python
async def refine_context(self, task_name: str, raw_context: str) -> str

#OverseerAdapter

Bridges the scheduler's OverseerInterface protocol to the real Overseer instance.

#_gate_session_tools

python
def _gate_session_tools(self) -> None

Replace the Overseer session's tools with autonomy-gated versions.

#session

python
def session(self) -> SessionProtocol

Expose the overseer's session for handoff.

#update_session

python
def update_session(self, new_session: SessionProtocol) -> None

Update the overseer's session after handoff.

#gate_tools

python
def gate_tools(self, tools: list[Tool]) -> list[Tool]

Wrap tools with autonomy gating.

Tools whose action type is not autonomous at the current level get a replacement execute function that returns a blocked message instead of running the original.

#_gate_tool

python
def _gate_tool(self, tool: Tool) -> Tool

Wrap a single tool with autonomy gating.

#send_event

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

Delegate to the Overseer and capture tool calls.

#send_correction

python
async def send_correction(self, message: str) -> None

Send a correction message to the Overseer session.

#is_degraded

python
def is_degraded(self, event_type: str) -> bool

Check if the Overseer is degraded for a given event type.

#verify_actions

python
async def verify_actions(self, event_type: str='') -> list[str]

Run verification checks on the Overseer's response and return a list of error messages.

#_check_schema_validation

python
def _check_schema_validation(self) -> list[str]

Validate create_task/create_workflow tool call args against TaskSpec/WorkflowConfig schema.

#_validate_create_task_args

python
def _validate_create_task_args(args: dict[str, Any]) -> list[str]

Validate create_task tool call arguments.

#_validate_create_workflow_args

python
def _validate_create_workflow_args(args: dict[str, Any]) -> list[str]

Validate create_workflow tool call arguments.

#_check_constraint_consistency

python
def _check_constraint_consistency(self) -> list[str]

Check that no two add_constraint calls in the current response create mechanical constraints of the same kind on the same glob pattern.

#_check_repetition

python
def _check_repetition(self, event_type: str) -> list[str]

Compare current tool calls to previous calls for the same event type. Flag if identical.

#_check_proportionality

python
def _check_proportionality(self) -> list[str]

Flag create_workflow budgets that exceed the configured fraction of remaining run budget.

#refine_context

python
async def refine_context(self, task_name: str, raw_context: str) -> str

Send context to Overseer for refinement.

This is a request/response, not an event. The Overseer can read files, consult, etc. Returns the refined context text.

Search