trading_agent.runtime.desk_runtime

One desk: its queue of triggers and the loop that works through them.

A desk handles one trigger at a time, so it never runs two turns at once. Different desks each have their own runtime and run side by side.

A message from the user that arrives while a model turn is running is steering: it goes to the desk's steering inbox and the running turn picks it up at its next tool round, instead of waiting for the turn to end. /stop skips the queue altogether and cancels the running turn. Commands that only read the desk, such as /status, also skip the queue and are answered beside the running turn, and so is /btw, which asks a model a side question.

A message that starts with a slash is a command and is answered by code. Anything else goes to the desk's turn runner, prefixed with a line stating the India time and what woke the desk; those facts change every turn, so they belong in the message rather than in the system prompt. A desk with a session keeper lets it resume the saved session when the desk starts, and roll over to a new session before the first turn of a new day.

Typical usage example:

desk = DeskRuntime(desk_configuration, event_bus, turn_runner, 'dry_run') task = asyncio.create_task(desk.run()) desk.submit(trigger)

DeskState

Bases: StrEnum

Whether a desk is waiting for a trigger or running a turn.

DeskRuntime

DeskRuntime(desk_configuration, event_bus, turn_runner, mode='dry_run', tool_registry=None, session_keeper=None, skill_library=None, usage_tracker=None, market_status=None, steering_inbox=None, command_parts=None)

The running state of one desk.

Attributes:

Name	Type	Description
name		The desk's name.
state		Whether the desk is stopped, idle or running a turn.
paused		Whether the desk answers without asking any model.
turn_runner		What runs each turn.
usage_tracker		Today's model usage, shown by /usage, or None.

Creates the desk without starting its loop.

Parameters:

Name	Type	Description	Default
desk_configuration	DeskConfiguration	The desk's settings.	required
event_bus	EventBus	The bus to publish the desk's events on.	required
turn_runner	TurnRunner	What runs each turn.	required
mode	str	"dry_run" or "live", reported by /status.	'dry_run'
tool_registry	ToolRegistry | None	The desk's tools, closed when the desk stops, or None.	None
session_keeper	SessionKeeper | None	Keeps the desk's sessions across restarts and days, or None to keep nothing.	None
skill_library	SkillLibrary | None	The desk's skills, invocable as /, or None.	None
usage_tracker	UsageTracker | None	Today's model usage, shown by /usage, or None.	None
market_status	MarketStatus | None	Describes the market for the context line, or None to leave it out.	None
steering_inbox	SteeringInbox | None	Where mid-turn messages wait for the running turn, shared with the turn runner; None creates one that only this desk reads.	None
command_parts	CommandParts | None	The subagents and journal some commands read, or None; the desk's own tool registry is added to it.	None

Source code in src/trading_agent/runtime/desk_runtime.py

def __init__(
    self,
    desk_configuration: agent_configuration.DeskConfiguration,
    event_bus: event_bus_module.EventBus,
    turn_runner: turn_runner_module.TurnRunner,
    mode: str = 'dry_run',
    tool_registry: tool_registry_module.ToolRegistry | None = None,
    session_keeper: session_keeper_module.SessionKeeper | None = None,
    skill_library: skill_library_module.SkillLibrary | None = None,
    usage_tracker: usage_tracker_module.UsageTracker | None = None,
    market_status: market_status_module.MarketStatus | None = None,
    steering_inbox: steering_inbox_module.SteeringInbox | None = None,
    command_parts: slash_commands.CommandParts | None = None,
):
    """Creates the desk without starting its loop.

    Args:
        desk_configuration (agent_configuration.DeskConfiguration): The desk's settings.
        event_bus (event_bus_module.EventBus): The bus to publish the desk's events on.
        turn_runner (turn_runner_module.TurnRunner): What runs each turn.
        mode (str): "dry_run" or "live", reported by /status.
        tool_registry (tool_registry_module.ToolRegistry | None): The desk's tools, closed when the desk stops, or None.
        session_keeper (session_keeper_module.SessionKeeper | None): Keeps the desk's sessions across restarts and days, or None to keep nothing.
        skill_library (skill_library_module.SkillLibrary | None): The desk's skills, invocable as /<skill-name>, or None.
        usage_tracker (usage_tracker_module.UsageTracker | None): Today's model usage, shown by /usage, or None.
        market_status (market_status_module.MarketStatus | None): Describes the market for the context line, or None to leave it out.
        steering_inbox (steering_inbox_module.SteeringInbox | None): Where mid-turn messages wait for the running turn, shared with the turn runner; None creates one that only this desk reads.
        command_parts (slash_commands.CommandParts | None): The subagents and journal some commands read, or None; the desk's own tool registry is added to it.
    """
    self.name = desk_configuration.name
    self.state = DeskState.STOPPED
    self.paused = False
    self.turn_runner = turn_runner
    self._event_bus = event_bus
    self._trigger_queue = asyncio.Queue()
    self._commands = slash_commands.SlashCommands(
        self,
        mode,
        skill_library,
        dataclasses.replace(
            command_parts or slash_commands.CommandParts(),
            tool_registry=tool_registry,
        ),
    )
    self._tool_registry = tool_registry
    self._session_keeper = session_keeper
    self.usage_tracker = usage_tracker
    self._market_status = market_status
    self.steering_inbox = (
        steering_inbox
        or steering_inbox_module.SteeringInbox(event_bus, self.name)
    )
    self._running_trigger_id: str | None = None
    self._turn_task: asyncio.Task | None = None
    self._stop_requested = False
    self._beside_tasks: set[asyncio.Task] = set()
    self._loop: asyncio.AbstractEventLoop | None = None

submit

submit(trigger)

Adds a trigger to the end of the desk's queue.

Parameters:

Name	Type	Description	Default
trigger	Trigger	The trigger to handle.	required

A /stop is answered at once instead of being queued, a read-only command such as /status is answered beside the running turn, and a user message that arrives during a model turn goes to the steering inbox.

Raises:

Type	Description
ValueError	The trigger is addressed to a different desk.

Source code in src/trading_agent/runtime/desk_runtime.py

def submit(self, trigger: trigger_module.Trigger) -> None:
    """Adds a trigger to the end of the desk's queue.

    Args:
        trigger (trigger_module.Trigger): The trigger to handle.

    A /stop is answered at once instead of being queued, a read-only command such as /status is answered beside the running turn, and a user message that arrives during a model turn goes to the steering inbox.

    Raises:
        ValueError: The trigger is addressed to a different desk.
    """
    if trigger.desk != self.name:
        raise ValueError(
            f'Trigger for desk {trigger.desk!r} was sent to desk {self.name!r}'
        )
    if self._called_from_another_thread():
        self._loop.call_soon_threadsafe(self._accept, trigger)
        return
    self._accept(trigger)

command_catalogue

command_catalogue()

The desk's commands and usable skills, for a client's command menu.

Returns:

Type	Description
list[dict[str, str]]	The entries, as SlashCommands.catalogue describes them.

Source code in src/trading_agent/runtime/desk_runtime.py

def command_catalogue(self) -> list[dict[str, str]]:
    """The desk's commands and usable skills, for a client's command menu.

    Returns:
        list[dict[str, str]]: The entries, as SlashCommands.catalogue describes them.
    """
    return self._commands.catalogue()

queued_count property

queued_count

How many triggers are waiting to be handled.

run async

run()

Handles triggers one at a time until the task is cancelled.

Source code in src/trading_agent/runtime/desk_runtime.py

async def run(self) -> None:
    """Handles triggers one at a time until the task is cancelled."""
    self.state = DeskState.IDLE
    self._loop = asyncio.get_running_loop()
    self._publish(event_module.EventKind.DESK_STARTED, {})
    try:
        if self._session_keeper is not None:
            await self._session_keeper.start()
        while True:
            trigger = await self._trigger_queue.get()
            await self._handle(trigger)
    finally:
        for task in list(self._beside_tasks):
            task.cancel()
        await self.turn_runner.close()
        if self._tool_registry is not None:
            await self._tool_registry.close()
        self.state = DeskState.STOPPED
        self._publish(event_module.EventKind.DESK_STOPPED, {})

roll_over async

roll_over()

Ends the desk's session with a summary and starts a fresh one, for /rollover.

Returns:

Type	Description
str	What happened, for the reply.

Source code in src/trading_agent/runtime/desk_runtime.py

async def roll_over(self) -> str:
    """Ends the desk's session with a summary and starts a fresh one, for /rollover.

    Returns:
        str: What happened, for the reply.
    """
    if self._session_keeper is None:
        return (
            'This desk keeps no sessions, so there is nothing to roll over.'
        )
    session_id = await self._session_keeper.roll_over()
    return f'Rolled over: the day is summarised and session {session_id} has started.'