trading_agent.backends.claude_code_turn_runner

Runs a desk's turns on Claude through Claude Code and the Max subscription.

Claude Code runs the agent loop itself, so this is a turn runner rather than a ModelBackend. It keeps one Claude Code session per desk, so Claude remembers earlier turns. It is set up so that it never stops for a human:

• permission_mode is bypassPermissions and setting_sources is empty, so no "ask" rule from any settings file applies;
• the subscription token from claude setup-token is passed as CLAUDE_CODE_OAUTH_TOKEN;
• every built-in Claude Code tool is switched off, and a PreToolUse hook allows only trading-agent's tools.

Before each turn the runner asks Claude Code how full the session's context is. At 70% of the window it gives Claude a memory-flush turn and then sends /compact, so Claude saves what matters before the older turns are summarised. Claude Code's own automatic compaction stays on as a safety net for one very long turn, set to fire at about 82% by limiting the window it measures against.

Typical usage example:

runner = ClaudeCodeTurnRunner('main', 'claude', backend_configuration, bridge, event_bus, settings) result = await runner.run_turn(trigger)

ClaudeCodeSettings dataclass

ClaudeCodeSettings(prompt_builder, oauth_token, working_directory, maximum_rounds, sessions_directory, compression_threshold, emergency_compression_threshold)

The settings of one Claude Code session that do not come from the backend's table.

Attributes:

Name	Type	Description
prompt_builder	PromptBuilder	Builds the desk's instructions before every turn.
oauth_token	str	The subscription token from claude setup-token.
working_directory	Path	The folder Claude Code runs in.
maximum_rounds	int	The most model calls one turn may make.
sessions_directory	Path	Where Claude Code keeps its sessions, normally ~/.claude/projects, checked before a saved session is resumed.
compression_threshold	float	The share of the context window at which the session is flushed and compacted before a turn.
emergency_compression_threshold	float	The share of the window given to Claude Code as its own compaction window, the safety net for one long turn.

ClaudeCodeTurnRunner

ClaudeCodeTurnRunner(desk_name, backend_name, backend_configuration, bridge, event_bus, settings, steering_inbox=None)

Bases: TurnRunner

One desk's Claude Code session.

Attributes:

Name	Type	Description
session_id	str | None	Claude Code's session identifier, known after the first turn.

Creates the runner without starting Claude Code.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk the session belongs to.	required
backend_name	str	The backend's name in trading_agent.toml.	required
backend_configuration	BackendConfiguration	Its settings, of kind "claude_code".	required
bridge	ClaudeToolBridge	The desk's tools, as Claude Code sees them.	required
event_bus	EventBus	Where every step is published.	required
settings	ClaudeCodeSettings	The prompt, token, folder and round limit.	required
steering_inbox	SteeringInbox | None	The desk's mid-turn messages, handed to Claude after each tool call, or None.	None

Source code in src/trading_agent/backends/claude_code_turn_runner.py

def __init__(
    self,
    desk_name: str,
    backend_name: str,
    backend_configuration: agent_configuration.BackendConfiguration,
    bridge: claude_tool_bridge.ClaudeToolBridge,
    event_bus: event_bus_module.EventBus,
    settings: ClaudeCodeSettings,
    steering_inbox: steering_inbox_module.SteeringInbox | None = None,
):
    """Creates the runner without starting Claude Code.

    Args:
        desk_name (str): The desk the session belongs to.
        backend_name (str): The backend's name in trading_agent.toml.
        backend_configuration (agent_configuration.BackendConfiguration): Its settings, of kind "claude_code".
        bridge (claude_tool_bridge.ClaudeToolBridge): The desk's tools, as Claude Code sees them.
        event_bus (event_bus_module.EventBus): Where every step is published.
        settings (ClaudeCodeSettings): The prompt, token, folder and round limit.
        steering_inbox (steering_inbox_module.SteeringInbox | None): The desk's mid-turn messages, handed to Claude after each tool call, or None.
    """
    self._desk_name = desk_name
    self._backend_name = backend_name
    self._model = backend_configuration.model
    self._model_override: str | None = None
    self._applied_model: str | None = None
    self._effort = backend_configuration.effort
    self._configured_effort = backend_configuration.effort
    self._thinking_off = False
    self._applied_thinking: tuple[str | None, bool] = (
        self._effort,
        self._thinking_off,
    )
    self._context_window = backend_configuration.context_window
    self._bridge = bridge
    self._event_bus = event_bus
    self._settings = settings
    self._steering_inbox = steering_inbox
    self._client = None
    self._session_prompt_fingerprint = None
    self._image_preparer = image_preparer_module.ImagePreparer()
    self._context_usage = None
    self.session_id: str | None = None

run_turn async

run_turn(trigger)

Sends the trigger to Claude and waits for its final answer.

Parameters:

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

Returns:

Type	Description
TurnResult	Claude's final text.

Raises:

Type	Description
BackendUnavailableError	Claude cannot be used now: no token, an API key is set, the subscription limit is reached, or Claude Code failed.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

async def run_turn(
    self,
    trigger: trigger_module.Trigger,
) -> turn_runner.TurnResult:
    """Sends the trigger to Claude and waits for its final answer.

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

    Returns:
        turn_runner.TurnResult: Claude's final text.

    Raises:
        backend_module.BackendUnavailableError: Claude cannot be used now: no token, an API key is set, the subscription limit is reached, or Claude Code failed.
    """
    self._check_preconditions()
    self._bridge.executor.start_turn()
    self._bridge.current_trigger = trigger
    prompt = self._settings.prompt_builder.build()
    try:
        client = await self._connected_client(prompt)
        await self._compact_if_full(client, trigger)
        await client.query(self._query(prompt, trigger))
        self._session_prompt_fingerprint = prompt.fingerprint
        result = await self._read_response(client, trigger)
        await self._record_context_usage(client)
        return result
    except asyncio.CancelledError:
        await self._disconnect()
        raise
    except backend_module.BackendUnavailableError:
        await self._disconnect()
        raise
    except claude_agent_sdk.ClaudeSDKError as error:
        await self._disconnect()
        raise backend_module.BackendUnavailableError(
            f'Claude Code failed: {error!r}'
        ) from error

side_question async

side_question(question)

Answers a question in a throwaway fork of the session, so the desk's own session is not changed.

The fork runs in its own short-lived Claude Code process, beside any turn the desk is running. Its session file is deleted afterwards.

Parameters:

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

Returns:

Type	Description
str	Claude's answer.

Raises:

Type	Description
BackendUnavailableError	Claude cannot be used now, or answered with an error.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

async def side_question(self, question: str) -> str:
    """Answers a question in a throwaway fork of the session, so the desk's own session is not changed.

    The fork runs in its own short-lived Claude Code process, beside any turn the desk is running. Its session file is deleted afterwards.

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

    Returns:
        str: Claude's answer.

    Raises:
        backend_module.BackendUnavailableError: Claude cannot be used now, or answered with an error.
    """
    self._check_preconditions()
    prompt = self._settings.prompt_builder.build()
    answer = ''
    fork_id = None
    try:
        async for message in claude_agent_sdk.query(
            prompt=f'{turn_runner.SIDE_QUESTION_NOTE}\n{question}',
            options=self._side_question_options(prompt.text),
        ):
            if isinstance(message, claude_agent_sdk.AssistantMessage):
                if message.error is not None:
                    raise backend_module.BackendUnavailableError(
                        f'Claude answered with an error: {message.error}'
                    )
                text = self._text_of(message)
                if text:
                    answer = text
            elif isinstance(message, claude_agent_sdk.ResultMessage):
                fork_id = message.session_id
                if message.is_error:
                    raise backend_module.BackendUnavailableError(
                        f'Claude could not answer: {message.result or message.subtype}'
                    )
    except claude_agent_sdk.ClaudeSDKError as error:
        raise backend_module.BackendUnavailableError(
            f'Claude Code failed: {error!r}'
        ) from error
    finally:
        self._delete_fork(fork_id)
    return answer

compact async

compact(trigger)

Flushes and compacts the session 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	Claude cannot be used now.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

async def compact(self, trigger: trigger_module.Trigger) -> str:
    """Flushes and compacts the session 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: Claude cannot be used now.
    """
    if self.session_id is None:
        return 'Claude has no session yet, so there is nothing to compact.'
    self._check_preconditions()
    self._bridge.executor.start_turn()
    prompt = self._settings.prompt_builder.build()
    try:
        client = await self._connected_client(prompt)
        usage = await client.get_context_usage()
        after = await self._flush_and_compact(
            client,
            trigger,
            'command',
            usage['totalTokens'],
        )
        await self._record_context_usage(client)
    except backend_module.BackendUnavailableError:
        await self._disconnect()
        raise
    except claude_agent_sdk.ClaudeSDKError as error:
        await self._disconnect()
        raise backend_module.BackendUnavailableError(
            f'Claude Code failed: {error!r}'
        ) from error
    return f"Compacted Claude's session from {usage['totalTokens']:,} to {after:,} tokens, after a memory-flush turn."

active_model

active_model()

The Claude model the session uses now: the user's choice from the model menu, or the configured one.

Returns:

Type	Description
str | None	The model's name or alias.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

def active_model(self) -> str | None:
    """The Claude model the session uses now: the user's choice from the model menu, or the configured one.

    Returns:
        str | None: The model's name or alias.
    """
    return self._model_override or self._model

use_model

use_model(model)

Switches the session to another Claude model at its next turn, keeping the conversation, as /model does in Claude Code.

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/backends/claude_code_turn_runner.py

def use_model(self, model: str | None) -> None:
    """Switches the session to another Claude model at its next turn, keeping the conversation, as /model does in Claude Code.

    Args:
        model (str | None): The model's name, or None for the configured model.
    """
    if model == self._model:
        model = None
    self._model_override = model

thinking

thinking()

Claude's effort level now, or "off" when thinking is switched off.

Returns:

Type	Description
str | None	The level, or "default" when trading_agent.toml sets no effort.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

def thinking(self) -> str | None:
    """Claude's effort level now, or "off" when thinking is switched off.

    Returns:
        str | None: The level, or "default" when trading_agent.toml sets no effort.
    """
    if self._thinking_off:
        return 'off'
    return self._effort or 'default'

set_thinking

set_thinking(level)

Sets Claude's effort, or switches thinking off, from the next turn on.

Claude Code takes these settings only when its process starts, so the next turn restarts the process and resumes the same session, keeping the conversation.

Parameters:

Name	Type	Description	Default
level	str | None	One of turn_runner.THINKING_LEVELS, or None for the effort in trading_agent.toml with thinking on.	required

Returns:

Type	Description
bool	Always True.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

def set_thinking(self, level: str | None) -> bool:
    """Sets Claude's effort, or switches thinking off, from the next turn on.

    Claude Code takes these settings only when its process starts, so the next turn restarts the process and resumes the same session, keeping the conversation.

    Args:
        level (str | None): One of turn_runner.THINKING_LEVELS, or None for the effort in trading_agent.toml with thinking on.

    Returns:
        bool: Always True.
    """
    if level is None:
        self._effort = self._configured_effort
        self._thinking_off = False
    elif level == 'off':
        self._thinking_off = True
    else:
        self._effort = level
        self._thinking_off = False
    return True

context_usage

context_usage()

How full Claude's context window was at the end of the last turn, as Claude Code measured it.

Returns:

Type	Description
ContextUsage | None	The snapshot, or None before the first turn of the session.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

def context_usage(self) -> context_usage_module.ContextUsage | None:
    """How full Claude's context window was at the end of the last turn, as Claude Code measured it.

    Returns:
        context_usage_module.ContextUsage | None: The snapshot, or None before the first turn of the session.
    """
    return self._context_usage

note_compaction

note_compaction()

Remembers that the session was compacted, so the next turn sends the instruction files again.

Updated instructions delivered inside earlier messages may not survive a compaction, so the next turn's message carries the complete current version.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

def note_compaction(self) -> None:
    """Remembers that the session was compacted, so the next turn sends the instruction files again.

    Updated instructions delivered inside earlier messages may not survive a compaction, so the next turn's message carries the complete current version.
    """
    self._session_prompt_fingerprint = None

saved_state

saved_state()

Claude Code's session id and the prompt it last saw, so a restart can resume the session.

Returns:

Type	Description
dict[str, Any] | None	The session id and prompt fingerprint, or None before the first turn.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

def saved_state(self) -> dict[str, Any] | None:
    """Claude Code's session id and the prompt it last saw, so a restart can resume the session.

    Returns:
        dict[str, Any] | None: The session id and prompt fingerprint, or None before the first turn.
    """
    if self.session_id is None:
        return None
    return {
        'session_id': self.session_id,
        'prompt_fingerprint': self._session_prompt_fingerprint,
    }

restore_state async

restore_state(state)

Resumes a session saved before a restart, if Claude Code still has it.

Claude Code keeps each session as a file named after its id. When that file is gone, resuming would fail, so the runner starts a new session instead and the router tells it what it missed.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	What saved_state returned before the restart.	required

Returns:

Type	Description
bool	True when the next turn will resume the saved session.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

async def restore_state(self, state: dict[str, Any]) -> bool:
    """Resumes a session saved before a restart, if Claude Code still has it.

    Claude Code keeps each session as a file named after its id. When that file is gone, resuming would fail, so the runner starts a new session instead and the router tells it what it missed.

    Args:
        state (dict[str, Any]): What saved_state returned before the restart.

    Returns:
        bool: True when the next turn will resume the saved session.
    """
    session_id = state.get('session_id')
    if not isinstance(session_id, str) or not session_id:
        return False
    if not self._session_file_exists(session_id):
        _LOGGER.warning(
            'Claude Code no longer has session %s; desk %s starts a new one',
            session_id,
            self._desk_name,
        )
        return False
    await self._disconnect()
    self.session_id = session_id
    self._session_prompt_fingerprint = state.get('prompt_fingerprint')
    return True

reset async

reset()

Stops Claude Code and forgets the session, so the next turn starts afresh.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

async def reset(self) -> None:
    """Stops Claude Code and forgets the session, so the next turn starts afresh."""
    await self._disconnect()
    self.session_id = None
    self._session_prompt_fingerprint = None
    self._context_usage = None

close async

close()

Stops the desk's Claude Code process when the desk stops.

Source code in src/trading_agent/backends/claude_code_turn_runner.py

async def close(self) -> None:
    """Stops the desk's Claude Code process when the desk stops."""
    await self._disconnect()