trading_agent.subagents.subagent_runner

Runs the tasks a desk delegates to its subagents.

Each task gets a fresh model router over the subagent's backends (or the desk's, for "inherit"), its own tool registry from the toolbox, and a system prompt built from its definition. Tasks run side by side; each has its own time limit, and one that fails, times out or crashes never stops the others. Afterwards the task's bash session and any Claude Code process are closed.

Every task publishes subagent_started and subagent_finished, and its tool calls carry the task's trigger id, so the journal shows what each subagent did.

Typical usage example:

runner = SubagentRunner(desk_name, mode, library, toolbox, runner_factory, ['claude', 'qwen'], backends, event_bus, 3) reports = await runner.run_all([SubagentTask('news_reader', 'What moved INFY today?')])

CLAUDE_ALIASES module-attribute

CLAUDE_ALIASES = ('opus', 'sonnet', 'haiku', 'fable')

SubagentTask dataclass

SubagentTask(subagent, task)

One task for one subagent.

Attributes:

Name	Type	Description
subagent	str	The subagent's name.
task	str	What it should do, with everything it needs to know.

SubagentReport dataclass

SubagentReport(subagent, outcome, text, backend_name, seconds, notes)

How one task ended.

Attributes:

Name	Type	Description
subagent	str	The subagent's name.
outcome	str	"answered", "timed out", "failed" or "unknown subagent".
text	str	The subagent's report, or why there is none.
backend_name	str | None	The backend that answered, or None.
seconds	float	How long the task took.
notes	tuple[str, ...]	Remarks about the definition, such as tools it may not have.

SubagentRunner

SubagentRunner(desk_name, mode, library, toolbox, runner_factory, desk_backend_order, backends, event_bus, maximum_identical_tool_calls, permission_gate=None)

Runs one desk's subagent tasks.

Creates the runner.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk the subagents work for.	required
mode	str	"dry_run" or "live".	required
library	SubagentLibrary	The definitions.	required
toolbox	SubagentToolbox	Builds each task's tools.	required
runner_factory	RunnerFactory	Builds a turn runner from a backend name, its settings, an executor, a prompt builder and a round limit.	required
desk_backend_order	Sequence[str]	The desk's backends, for subagents that inherit them.	required
backends	Mapping[str, BackendConfiguration]	Every configured backend.	required
event_bus	EventBus	Where each task's events are published.	required
maximum_identical_tool_calls	int	The loop guard's limit for each task.	required
permission_gate	PermissionGate | None	Holds the tasks' calls that need the user's permission, or None when none do.	None

Source code in src/trading_agent/subagents/subagent_runner.py

def __init__(
    self,
    desk_name: str,
    mode: str,
    library: subagent_library_module.SubagentLibrary,
    toolbox: subagent_toolbox_module.SubagentToolbox,
    runner_factory: RunnerFactory,
    desk_backend_order: Sequence[str],
    backends: Mapping[str, agent_configuration.BackendConfiguration],
    event_bus: event_bus_module.EventBus,
    maximum_identical_tool_calls: int,
    permission_gate: permission_gate_module.PermissionGate | None = None,
):
    """Creates the runner.

    Args:
        desk_name (str): The desk the subagents work for.
        mode (str): "dry_run" or "live".
        library (subagent_library_module.SubagentLibrary): The definitions.
        toolbox (subagent_toolbox_module.SubagentToolbox): Builds each task's tools.
        runner_factory (RunnerFactory): Builds a turn runner from a backend name, its settings, an executor, a prompt builder and a round limit.
        desk_backend_order (Sequence[str]): The desk's backends, for subagents that inherit them.
        backends (Mapping[str, agent_configuration.BackendConfiguration]): Every configured backend.
        event_bus (event_bus_module.EventBus): Where each task's events are published.
        maximum_identical_tool_calls (int): The loop guard's limit for each task.
        permission_gate (permission_gate_module.PermissionGate | None): Holds the tasks' calls that need the user's permission, or None when none do.
    """
    self._desk_name = desk_name
    self._mode = mode
    self._library = library
    self._toolbox = toolbox
    self._runner_factory = runner_factory
    self._desk_backend_order = list(desk_backend_order)
    self._backends = backends
    self._event_bus = event_bus
    self._maximum_identical_tool_calls = maximum_identical_tool_calls
    self._permission_gate = permission_gate

run_all async

run_all(tasks)

Runs the tasks side by side.

Parameters:

Name	Type	Description	Default
tasks	Sequence[SubagentTask]	The tasks.	required

Returns:

Type	Description
list[SubagentReport]	One report per task, in the order given.

Source code in src/trading_agent/subagents/subagent_runner.py

async def run_all(
    self,
    tasks: Sequence[SubagentTask],
) -> list[SubagentReport]:
    """Runs the tasks side by side.

    Args:
        tasks (Sequence[SubagentTask]): The tasks.

    Returns:
        list[SubagentReport]: One report per task, in the order given.
    """
    return list(
        await asyncio.gather(*[self.run_one(task) for task in tasks])
    )

run_one async

run_one(task)

Runs one task from start to finish.

Parameters:

Name	Type	Description	Default
task	SubagentTask	The task.	required

Returns:

Type	Description
SubagentReport	How it ended.

Source code in src/trading_agent/subagents/subagent_runner.py

async def run_one(self, task: SubagentTask) -> SubagentReport:
    """Runs one task from start to finish.

    Args:
        task (SubagentTask): The task.

    Returns:
        SubagentReport: How it ended.
    """
    started = time.monotonic()
    definition = self._library.find(task.subagent)
    if definition is None:
        names = []
        for known in self._library.definitions():
            names.append(known.name)
        return SubagentReport(
            subagent=task.subagent,
            outcome='unknown subagent',
            text=f'There is no subagent named {task.subagent}. Subagents: {", ".join(names) or "none"}.',
            backend_name=None,
            seconds=0.0,
            notes=(),
        )
    notes = []
    routes_wanted = self._backend_choices(definition, notes)
    prepared = self._toolbox.prepare(definition)
    for name in prepared.refused:
        notes.append(f'{name} is never given to subagents')
    for name in prepared.unknown:
        notes.append(f"{name} is not one of the desk's tools")
    executor = tool_call_executor.ToolCallExecutor(
        self._desk_name,
        prepared.registry,
        self._event_bus,
        loop_guard.LoopGuard(self._maximum_identical_tool_calls),
        self._permission_gate,
    )
    prompt = subagent_prompt_builder.SubagentPromptBuilder(
        definition,
        self._desk_name,
        self._mode,
    )
    routes = []
    for backend_name, backend_configuration in routes_wanted:
        runner = self._runner_factory(
            backend_name,
            backend_configuration,
            executor,
            prompt,
            definition.maximum_turns,
        )
        routes.append(model_router.BackendRoute(backend_name, runner))
    trigger = trigger_module.Trigger.now(
        trigger_module.TriggerSource.SUBAGENT,
        self._desk_name,
        f'[Task from the {self._desk_name} desk]\n{task.task}',
    )
    self._publish(
        event_module.EventKind.SUBAGENT_STARTED,
        {
            'trigger_id': trigger.trigger_id,
            'subagent': definition.name,
            'task': task.task,
            'backends': [name for name, _ in routes_wanted],
            'tools': prepared.registry.tool_names,
        },
    )
    outcome, text, backend_name = await self._run(
        routes, trigger, definition, executor
    )
    if prepared.bash_session is not None:
        await prepared.bash_session.close()
    report = SubagentReport(
        subagent=definition.name,
        outcome=outcome,
        text=text,
        backend_name=backend_name,
        seconds=time.monotonic() - started,
        notes=tuple(notes),
    )
    self._publish(
        event_module.EventKind.SUBAGENT_FINISHED,
        {
            'trigger_id': trigger.trigger_id,
            'subagent': definition.name,
            'outcome': outcome,
            'backend': backend_name,
            'seconds': round(report.seconds, 1),
            'report_characters': len(text),
        },
    )
    return report