trading_agent.terminal.chat_items

The pieces of a chat: turns, the steps inside them, and events that belong to no turn.

They mirror the web page's chatTurnBuilder.ts types, and ChatTranscript fills them in from gateway events.

Typical usage example:

turn = ChatTurn(trigger_id='t1', source='user_message', text='Hello') turn.status_text() # 'waiting in the queue'

ToolStep dataclass

ToolStep(call_id, name, arguments, result_text=None, is_error=False, picture_path=None, ended_early=False)

One tool call inside a turn, with its result once it arrives.

Attributes:

Name	Type	Description
call_id	str	The call's id, which its result carries too.
name	str	The tool's name.
arguments	dict[str, Any]	The call's arguments.
result_text	str | None	The tool's answer, or None while it runs.
is_error	bool	Whether the tool reported a failure.
picture_path	str | None	A picture the tool showed the model, relative to the desk folder, or None.
ended_early	bool	Whether the turn ended, for example by a stop, before the tool answered.

arguments_text

arguments_text()

The arguments as indented JSON.

Returns:

Type	Description
str	The JSON text.

Source code in src/trading_agent/terminal/chat_items.py

def arguments_text(self) -> str:
    """The arguments as indented JSON.

    Returns:
        str: The JSON text.
    """
    return json.dumps(self.arguments, indent=2, ensure_ascii=False)

EventStep dataclass

EventStep(event)

Any other event shown inside a turn, such as a risk decision or a model switch.

Attributes:

Name	Type	Description
event	dict[str, Any]	The event, with kind, desk, payload and created_at.

SteeringStep dataclass

SteeringStep(message)

A message the user sent while the turn was running, at the point it arrived.

Attributes:

Name	Type	Description
message	ChatTurn	The message's own turn, which says whether it reached the running turn.

RequestStep dataclass

RequestStep(request)

A question, plan or permission request the desk made during a turn, with its latest status.

Attributes:

Name	Type	Description
request	dict[str, Any]	The request, as GET /api/requests returns it.

ToolRun dataclass

ToolRun(steps)

Tool calls in a row, shown as one folded line.

Attributes:

Name	Type	Description
steps	list[ToolStep]	The tool calls, in order.

ChatTurn dataclass

ChatTurn(trigger_id, source, text, attachments=list(), received_at='', started_at=None, finished_at=None, started=False, finished=False, steps=list(), answer=None, silent=False, error=None, backend=None, steering_into=None, steering_outcome=None, stopped=False, request_id=None, streaming_text='')

One trigger and everything the desk did about it.

Attributes:

Name	Type	Description
trigger_id	str	The trigger's id.
source	str	What woke the desk, such as "user_message" or "cron".
text	str	The trigger's text.
attachments	list[dict[str, Any]]	The files the user sent with it.
received_at	str	When the gateway received it, as ISO text.
started_at	str | None	When the turn started, or None.
finished_at	str | None	When the turn finished, or None.
started	bool	Whether the turn has started.
finished	bool	Whether the turn has finished.
steps	list[ToolStep | EventStep | SteeringStep | RequestStep]	Tool calls, events, steering messages and requests, in order.
answer	str | None	The desk's final answer, or None until it arrives.
silent	bool	Whether the desk decided there was nothing to say.
error	str | None	Why the turn failed, or None.
backend	str | None	The backend that answered, or None.
steering_into	str | None	The running turn this message was handed to, or None.
steering_outcome	str | None	"waiting", "delivered" or "queued" for a steering message, otherwise None.
stopped	bool	Whether the user stopped the turn.
request_id	str | None	The question, plan or permission this trigger answers, or None.
streaming_text	str	The text the desk is writing now, before the final answer arrives.

status_text

status_text()

Where the turn has got to, in the web page's words.

Returns:

Type	Description
str	A short status line, such as "answered by claude" or "waiting in the queue".

Source code in src/trading_agent/terminal/chat_items.py

def status_text(self) -> str:
    """Where the turn has got to, in the web page's words.

    Returns:
        str: A short status line, such as "answered by claude" or "waiting in the queue".
    """
    if self.stopped:
        return 'stopped by you'
    if self.finished:
        if self.backend is not None:
            return f'answered by {self.backend}'
        return 'finished'
    if self.started:
        return 'working…'
    if self.steering_outcome == 'waiting':
        return 'waiting for the running turn to pick it up at its next tool call'
    if self.steering_outcome == 'delivered':
        return 'handed to the running turn'
    if self.steering_outcome == 'queued':
        return 'the running turn ended first, so this is the next turn'
    return 'waiting in the queue'

running_tool

running_tool()

The tool call still waiting for its result, if any.

Returns:

Type	Description
ToolStep | None	The newest tool call when it has no result yet, otherwise None.

Source code in src/trading_agent/terminal/chat_items.py

def running_tool(self) -> ToolStep | None:
    """The tool call still waiting for its result, if any.

    Returns:
        ToolStep | None: The newest tool call when it has no result yet, otherwise None.
    """
    for step in reversed(self.steps):
        if isinstance(step, ToolStep):
            if step.result_text is None and not step.ended_early:
                return step
            return None
    return None

segments

segments()

The steps, with tool calls in a row grouped into one run.

Returns:

Type	Description
list[ToolRun | EventStep | SteeringStep | RequestStep]	The segments, in order.

Source code in src/trading_agent/terminal/chat_items.py

def segments(
    self,
) -> list[ToolRun | EventStep | SteeringStep | RequestStep]:
    """The steps, with tool calls in a row grouped into one run.

    Returns:
        list[ToolRun | EventStep | SteeringStep | RequestStep]: The segments, in order.
    """
    segments = []
    for step in self.steps:
        if isinstance(step, ToolStep):
            if segments and isinstance(segments[-1], ToolRun):
                segments[-1].steps.append(step)
            else:
                segments.append(ToolRun(steps=[step]))
        else:
            segments.append(step)
    return segments

LooseEvent dataclass

LooseEvent(event)

An event that belongs to no turn, such as a fill reported while the desk was idle.

Attributes:

Name	Type	Description
event	dict[str, Any]	The event.