trading_agent.agent.agent_turn_runner

The turn runner that drives a model through the agent loop.

It keeps the desk's conversation between turns, so the desk remembers what was said earlier, and rebuilds the system prompt before every turn, so an edit to a workspace file takes effect on the next turn.

Typical usage example:

runner = AgentTurnRunner(agent_loop, prompt_builder) result = await runner.run_turn(trigger)

AgentTurnRunner

AgentTurnRunner(agent_loop, prompt_builder, context_window=0)

Bases: TurnRunner

Runs each trigger as a turn of the agent loop, in one ongoing conversation.

Attributes:

Name	Type	Description
conversation		The conversation so far.

Creates the runner with an empty conversation.

Parameters:

Name	Type	Description	Default
agent_loop	AgentLoop	The loop to run each turn with.	required
prompt_builder	PromptBuilder	Builds the desk's system prompt.	required
context_window	int	How many tokens the model is given, for context_usage; 0 when it is not known.	0

Source code in src/trading_agent/agent/agent_turn_runner.py

def __init__(
    self,
    agent_loop: agent_loop_module.AgentLoop,
    prompt_builder: prompt_builder_module.PromptBuilder,
    context_window: int = 0,
):
    """Creates the runner with an empty conversation.

    Args:
        agent_loop (agent_loop_module.AgentLoop): The loop to run each turn with.
        prompt_builder (prompt_builder_module.PromptBuilder): Builds the desk's system prompt.
        context_window (int): How many tokens the model is given, for context_usage; 0 when it is not known.
    """
    self._agent_loop = agent_loop
    self._prompt_builder = prompt_builder
    self._context_window = context_window
    self.conversation = conversation_module.Conversation(
        prompt_builder.build().text
    )

run_turn async

run_turn(trigger)

Refreshes the system prompt and runs one turn of the agent loop.

Parameters:

Name	Type	Description	Default
trigger	Trigger	The reason for the turn.	required

Returns:

Type	Description
TurnResult	The model's final text.

Source code in src/trading_agent/agent/agent_turn_runner.py

async def run_turn(
    self,
    trigger: trigger_module.Trigger,
) -> turn_runner.TurnResult:
    """Refreshes the system prompt and runs one turn of the agent loop.

    Args:
        trigger (trigger_module.Trigger): The reason for the turn.

    Returns:
        turn_runner.TurnResult: The model's final text.
    """
    self.conversation.system_prompt = self._prompt_builder.build().text
    return await self._agent_loop.run_turn(self.conversation, trigger)

compact async

compact(trigger)

Flushes and compresses the conversation now, for the /compact command.

Parameters:

Name	Type	Description	Default
trigger	Trigger	The command's trigger, used to tag events.	required

Returns:

Type	Description
str	What happened, for the user.

Raises:

Type	Description
BackendUnavailableError	The model could not be reached.

Source code in src/trading_agent/agent/agent_turn_runner.py

async def compact(self, trigger: trigger_module.Trigger) -> str:
    """Flushes and compresses the conversation now, for the /compact command.

    Args:
        trigger (trigger_module.Trigger): The command's trigger, used to tag events.

    Returns:
        str: What happened, for the user.

    Raises:
        backend_module.BackendUnavailableError: The model could not be reached.
    """
    before = self.conversation.estimated_tokens()
    compressed = await self._agent_loop.compact(self.conversation, trigger)
    if not compressed:
        return 'There was nothing to compact.'
    return f'Compressed the conversation from about {before:,} tokens into a summary, after a memory-flush turn.'

active_model

active_model()

The model the loop's backend uses now.

Returns:

Type	Description
str | None	The model's name.

Source code in src/trading_agent/agent/agent_turn_runner.py

def active_model(self) -> str | None:
    """The model the loop's backend uses now.

    Returns:
        str | None: The model's name.
    """
    return self._agent_loop.backend.model

use_model

use_model(model)

Switches the loop's backend to another model on the same server, keeping the conversation.

Parameters:

Name	Type	Description	Default
model	str | None	The model's name, or None for the configured model.	required

Source code in src/trading_agent/agent/agent_turn_runner.py

def use_model(self, model: str | None) -> None:
    """Switches the loop's backend to another model on the same server, keeping the conversation.

    Args:
        model (str | None): The model's name, or None for the configured model.
    """
    self._agent_loop.backend.use_model(model)

side_question async

side_question(question)

Asks the backend once, with a copy of the conversation and no tools, so nothing is kept.

Parameters:

Name	Type	Description	Default
question	str	The user's question.	required

Returns:

Type	Description
str	The model's answer.

Raises:

Type	Description
BackendUnavailableError	The backend could not answer.

Source code in src/trading_agent/agent/agent_turn_runner.py

async def side_question(self, question: str) -> str:
    """Asks the backend once, with a copy of the conversation and no tools, so nothing is kept.

    Args:
        question (str): The user's question.

    Returns:
        str: The model's answer.

    Raises:
        backend.BackendUnavailableError: The backend could not answer.
    """
    conversation = self.conversation.copy()
    conversation.system_prompt = self._prompt_builder.build().text
    conversation.append(
        messages.ConversationMessage.user(
            f'{turn_runner.SIDE_QUESTION_NOTE}\n{question}'
        )
    )
    reply = await self._agent_loop.backend.complete(
        conversation,
        [],
        self._ignore_text,
    )
    return reply.text

thinking

thinking()

The thinking level of the loop's backend.

Returns:

Type	Description
str | None	The level, or None when the backend has no thinking setting.

Source code in src/trading_agent/agent/agent_turn_runner.py

def thinking(self) -> str | None:
    """The thinking level of the loop's backend.

    Returns:
        str | None: The level, or None when the backend has no thinking setting.
    """
    return self._agent_loop.backend.thinking()

set_thinking

set_thinking(level)

Changes the thinking level of the loop's backend, keeping the conversation.

Parameters:

Name	Type	Description	Default
level	str | None	One of turn_runner.THINKING_LEVELS, or None for the configured setting.	required

Returns:

Type	Description
bool	True when the backend applied it, False when it has no thinking setting.

Source code in src/trading_agent/agent/agent_turn_runner.py

def set_thinking(self, level: str | None) -> bool:
    """Changes the thinking level of the loop's backend, keeping the conversation.

    Args:
        level (str | None): One of turn_runner.THINKING_LEVELS, or None for the configured setting.

    Returns:
        bool: True when the backend applied it, False when it has no thinking setting.
    """
    return self._agent_loop.backend.set_thinking(level)

context_usage

context_usage()

Estimates how full the model's context window is.

Returns:

Type	Description
ContextUsage | None	The estimate, or None when the context window is not known.

Source code in src/trading_agent/agent/agent_turn_runner.py

def context_usage(self) -> context_usage_module.ContextUsage | None:
    """Estimates how full the model's context window is.

    Returns:
        context_usage_module.ContextUsage | None: The estimate, or None when the context window is not known.
    """
    if self._context_window <= 0:
        return None
    backend = self._agent_loop.backend
    return context_usage_module.ContextUsage.estimate(
        backend.name,
        backend.model,
        self._context_window,
        self.conversation,
        self._agent_loop.tool_definitions(),
    )

reset async

reset()

Starts a new, empty conversation.

Source code in src/trading_agent/agent/agent_turn_runner.py

async def reset(self) -> None:
    """Starts a new, empty conversation."""
    self.conversation = conversation_module.Conversation(
        self._prompt_builder.build().text
    )