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

AgentExecutionMixin runs orchestrator and agent tasks end-to-end: session creation, prompt assembly, context-window checks, agent handoff, tool building, and attempt completion/timeout handling.

#scheduler.src.orxtra.scheduler._agent_execution

#scheduler.src.orxtra.scheduler._agent_execution

#_resolve_task_variables

python
def _resolve_task_variables(template: str, variables: dict[str, Any] | None) -> str

Resolve variables in a task prompt template.

Wraps compose's strict resolve_variables with two adaptations for the scheduler's workflow context:

  1. Values are coerced to str (callers pass dict[str, Any]).
  2. Unused variables are filtered out rather than rejected,

because the workflow executor accumulates all dependency outputs as variables and tasks use only a subset.

Unresolved placeholders still raise ValueError (catches typos in task prompts).

#_deferred_stub_execute

python
async def _deferred_stub_execute(args: dict[str, Any]) -> ToolOutput[Any]

Placeholder execute for deferred tool stubs.

If called, the LLM tried to use the tool before loading it.

#AgentExecutionMixin

Mixin for agent and orchestrator task execution.

#_execute_orchestrator_task

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

Execute an orchestrator task with multi-turn suspension support.

#_execute_orchestrator_or_agent_task

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

Dispatch to orchestrator or agent task execution.

#_compute_context_usage

python
def _compute_context_usage(self, session: Session) -> tuple[int, float]

Compute total tokens used and usage percentage of context limit.

Returns (tokens_used, usage_percent).

#_check_agent_context

python
async def _check_agent_context(self, session: Session, session_id: str, task_id: UUID) -> None

Check agent session context usage and emit warnings or trigger handoff.

Emits ContextWarning at 80% usage. Triggers handoff at 90%.

#_agent_handoff

python
async def _agent_handoff(self, session: Session, task_id: UUID) -> None

Perform agent context handoff: summarize conversation, create new session.

Asks the current session to summarize, then creates a fresh session with the summary as initial context. Replaces the session in _task_sessions.

#_build_agent_tools

python
def _build_agent_tools(self, agent_def: Agent, task_id: UUID, session_id: str, task_name: str, task_agent: str) -> list[Tool]

Build raw tools based on agent's allow list.

Uses the ToolRegistry for data-driven construction of standard tools, with special handling for git (subcommand resolution), consult (needs already-built tools), inline tool definitions (per-agent [[tools.define]]), deferred tools (stubs + load_tools auto-grant), and lifecycle tools (always present).

#_assemble_agent_prompt

python
async def _assemble_agent_prompt(self, task: TaskSpec, task_id: UUID, variables: dict[str, Any] | None, attempt: int, attempt_id: UUID, prior_attempts: list[dict[str, Any]]) -> str

Assemble full prompt with runtime context layers.

Uses the compose engine with fragment providers for each layer. Variable substitution in the task prompt uses strict resolution: unresolved placeholders raise ValueError. Unused variables are filtered (workflow executor accumulates dependency outputs).

#_refresh_injection_data

python
async def _refresh_injection_data(self) -> None

Refresh constraints, lessons, and notepad from trace.

Called at the start of each task attempt, before prompt assembly. Each callback replaces the scheduler's in-memory list with fresh data from the storage backend.

When no callbacks are registered (default), the lists remain as initialized (empty or manually populated).

Search