trading_agent.terminal.screens.chat_screen

The chat with one desk: today's turns, the live progress line and the message box.

The screen reads today's journal of the desk when it opens, then takes live events from the application. Events that arrive while the journal is still loading are held and added afterwards; ChatTranscript drops the ones the journal already had.

The terminal program's own commands (/attach, /desk, /model with no name, /theme, /keys, /exit, and one per page such as /trading) are handled here. Files are attached with @path in a message, /attach, by dragging them onto the terminal, or with Ctrl+V for a picture on the clipboard; they upload at once and go with the next message. Every other message, including every other /command, goes to the desk, which answers commands such as /status itself.

Typical usage example:

app.push_screen(ChatScreen(gateway, history, 'general'))

ChatScreen

ChatScreen(gateway, history, desk_name, settings, opener)

Bases: Screen[None]

The chat screen.

Creates the screen.

Parameters:

Name	Type	Description	Default
gateway	GatewayClient	The gateway's routes.	required
history	PromptHistory	The earlier prompts.	required
desk_name	str	The desk to open on.	required
settings	TerminalSettings	The remembered choices, such as whether the side panel is shown.	required
opener	AttachmentOpener	Downloads and opens the desk's files.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def __init__(
    self,
    gateway: gateway_client.GatewayClient,
    history: prompt_history_module.PromptHistory,
    desk_name: str,
    settings: terminal_settings.TerminalSettings,
    opener: attachment_opener.AttachmentOpener,
):
    """Creates the screen.

    Args:
        gateway (gateway_client.GatewayClient): The gateway's routes.
        history (prompt_history_module.PromptHistory): The earlier prompts.
        desk_name (str): The desk to open on.
        settings (terminal_settings.TerminalSettings): The remembered choices, such as whether the side panel is shown.
        opener (attachment_opener.AttachmentOpener): Downloads and opens the desk's files.
    """
    super().__init__()
    self.desk_name = desk_name
    self._gateway = gateway
    self._history = history
    self._settings = settings
    self._opener = opener
    self._transcript = chat_transcript.ChatTranscript()
    self._views: dict[str, Any] = {}
    self._loading = True
    self._held_events: list[dict[str, Any]] = []
    self._desks_document: dict[str, Any] = {}
    self._connection = event_stream.CONNECTING
    self._tools_expanded = False
    self._last_escape = 0.0
    self._last_interrupt = 0.0
    self._last_exit_press = 0.0
    self._commands: list[dict[str, Any]] = []
    self._names = model_names.ModelNames()

compose

compose()

Builds the screen.

Yields:

Type	Description
Widget	The status line, the transcript, the progress line, the command menu, the message box and the key hints.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def compose(self):
    """Builds the screen.

    Yields:
        Widget: The status line, the transcript, the progress line, the command menu, the message box and the key hints.
    """
    yield widgets.Static('', id='status-line')
    with containers.Horizontal(id='chat-body'):
        yield containers.VerticalScroll(id='transcript')
        yield side_panel.SidePanel(
            self._gateway, self._settings, self._opener
        )
    yield progress_line.ProgressLine()
    yield command_menu.CommandMenu()
    yield request_dock.RequestDock(self._gateway)
    yield pending_files.PendingFiles(self._gateway)
    with containers.Vertical(id='composer-box'):
        yield composer_module.Composer(self._history)
    yield widgets.Static(
        '[dim]enter send · shift+enter new line · / commands · esc stop · alt+d desk · alt+p model · ctrl+o tools · ctrl+b panel · f2 pages · ? keys[/dim]',
        id='key-hints',
    )

on_mount

on_mount()

Loads the desk's chat and focuses the message box.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_mount(self) -> None:
    """Loads the desk's chat and focuses the message box."""
    self.query_one(composer_module.Composer).focus()
    self.query_one(
        side_panel.SidePanel
    ).display = self._settings.side_panel_open
    self._show_status()
    self.load_desk(self.desk_name)
    self.set_interval(_REQUESTS_REFRESH_SECONDS, self.refresh_requests)

take_event

take_event(event)

Adds one live event to the chat, when it belongs to this desk.

Parameters:

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

Source code in src/trading_agent/terminal/screens/chat_screen.py

def take_event(self, event: dict[str, Any]) -> None:
    """Adds one live event to the chat, when it belongs to this desk.

    Args:
        event (dict[str, Any]): The event.
    """
    if event.get('desk') != self.desk_name:
        return
    if event.get('kind') in _REQUEST_KINDS:
        self.refresh_requests()
    if self._loading:
        self._held_events.append(event)
        return
    self._add_events([event])

take_desks

take_desks(document)

Updates the status line and placeholder from the desks' states.

Parameters:

Name	Type	Description	Default
document	dict[str, Any]	The answer of GET /api/desks.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def take_desks(self, document: dict[str, Any]) -> None:
    """Updates the status line and placeholder from the desks' states.

    Args:
        document (dict[str, Any]): The answer of GET /api/desks.
    """
    self._desks_document = document
    self._show_status()
    for panel in self.query(side_panel.SidePanel):
        panel.take_desk(self.desk(), document.get('mode', ''))

take_connection

take_connection(state)

Shows the event stream's state, and reloads the chat after a reconnect.

Parameters:

Name	Type	Description	Default
state	str	"connecting", "live", "reconnecting" or "refused".	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def take_connection(self, state: str) -> None:
    """Shows the event stream's state, and reloads the chat after a reconnect.

    Args:
        state (str): "connecting", "live", "reconnecting" or "refused".
    """
    was_down = self._connection == event_stream.RECONNECTING
    self._connection = state
    self._show_status()
    if was_down and state == event_stream.LIVE:
        self.load_desk(self.desk_name)

desk

desk()

The summary of the desk this chat shows.

Returns:

Type	Description
dict[str, Any] | None	The summary, or None before the desks are known.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def desk(self) -> dict[str, Any] | None:
    """The summary of the desk this chat shows.

    Returns:
        dict[str, Any] | None: The summary, or None before the desks are known.
    """
    for desk in self._desks_document.get('desks', []):
        if desk.get('name') == self.desk_name:
            return desk
    return None

load_desk async

load_desk(desk_name)

Clears the chat and fills it with the desk's journal of today.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

@work(exclusive=True, group='load')
async def load_desk(self, desk_name: str) -> None:
    """Clears the chat and fills it with the desk's journal of today.

    Args:
        desk_name (str): The desk.
    """
    self.desk_name = desk_name
    self.app.remember_desk(desk_name)
    self._loading = True
    self._held_events = []
    self._transcript = chat_transcript.ChatTranscript()
    self._views = {}
    transcript = self.query_one('#transcript', containers.VerticalScroll)
    await transcript.remove_children()
    self._show_status()
    self._load_commands(desk_name)
    self.refresh_requests()
    panel = self.query_one(side_panel.SidePanel)
    panel.set_desk(desk_name)
    panel.take_desk(self.desk(), self._desks_document.get('mode', ''))
    try:
        journal = await self._gateway.desk_journal(desk_name)
    except gateway_client.GatewayError as error:
        await transcript.mount(
            widgets.Static(
                Content.from_markup(
                    "[$error]Could not read today's journal: $message[/]",
                    message=error.message,
                ),
                classes='notice',
            )
        )
        journal = {
            'events': [],
        }
    events = list(journal.get('events', []))
    held = self._held_events
    self._held_events = []
    self._loading = False
    if not events and not held:
        await transcript.mount(
            widgets.Static(
                Content.from_markup(
                    '[dim]Nothing today yet with $desk. Say hello, or try /status.[/dim]',
                    desk=desk_name,
                ),
                classes='notice',
                id='empty-notice',
            )
        )
    self._add_events(events + held)
    transcript.anchor()

refresh_requests async

refresh_requests()

Reads the pending requests and shows this desk's first one in the dock.

Source code in src/trading_agent/terminal/screens/chat_screen.py

@work(exclusive=True, group='requests')
async def refresh_requests(self) -> None:
    """Reads the pending requests and shows this desk's first one in the dock."""
    try:
        pending = await self._gateway.requests('pending')
    except gateway_client.GatewayError:
        return
    mine = []
    for request in pending:
        if request.get('desk') == self.desk_name:
            mine.append(request)
    docks = self.query(request_dock.RequestDock)
    if not docks:
        return
    dock = docks.first(request_dock.RequestDock)
    focus_was_in_dock = (
        self.focused is not None
        and dock in self.focused.ancestors_with_self
    )
    if not dock.show(mine):
        return
    if dock.shown_request_id() is not None:
        self.notify(
            'The desk is waiting for your answer above the message box. Press tab to answer.',
            timeout=6,
        )
    if focus_was_in_dock:
        self.query_one(composer_module.Composer).focus()

on_request_dock_answered

on_request_dock_answered(event)

Reads the requests again after one was answered.

Parameters:

Name	Type	Description	Default
event	Answered	The notice.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_request_dock_answered(
    self, event: request_dock.RequestDock.Answered
) -> None:
    """Reads the requests again after one was answered.

    Args:
        event (request_dock.RequestDock.Answered): The notice.
    """
    del event
    self.refresh_requests()

on_text_area_changed

on_text_area_changed(event)

Opens or narrows the command menu as the user types.

Parameters:

Name	Type	Description	Default
event	Changed	The change.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_text_area_changed(self, event: widgets.TextArea.Changed) -> None:
    """Opens or narrows the command menu as the user types.

    Args:
        event (widgets.TextArea.Changed): The change.
    """
    menu = self.query_one(command_menu.CommandMenu)
    composer = self.query_one(composer_module.Composer)
    text = event.text_area.text
    last_word = ''
    if text and not text[-1].isspace():
        last_word = text.split()[-1]
    if last_word.startswith('@'):
        suggestions = attachable_files.AttachableFiles().suggestions(
            last_word[1:]
        )
        composer.menu_open = menu.show_paths(suggestions)
        return
    composer.menu_open = menu.filter(text)

on_composer_menu_moved

on_composer_menu_moved(event)

Moves the command menu's highlight.

Parameters:

Name	Type	Description	Default
event	MenuMoved	The move.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_composer_menu_moved(
    self, event: composer_module.Composer.MenuMoved
) -> None:
    """Moves the command menu's highlight.

    Args:
        event (composer_module.Composer.MenuMoved): The move.
    """
    self.query_one(command_menu.CommandMenu).move(event.step)

on_composer_menu_picked

on_composer_menu_picked(event)

Puts the highlighted command into the message box.

Parameters:

Name	Type	Description	Default
event	MenuPicked	The pick.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_composer_menu_picked(
    self, event: composer_module.Composer.MenuPicked
) -> None:
    """Puts the highlighted command into the message box.

    Args:
        event (composer_module.Composer.MenuPicked): The pick.
    """
    del event
    self._complete_command()

on_composer_submitted

on_composer_submitted(event)

Sends the message, or completes the highlighted command first.

Parameters:

Name	Type	Description	Default
event	Submitted	The submission.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_composer_submitted(
    self, event: composer_module.Composer.Submitted
) -> None:
    """Sends the message, or completes the highlighted command first.

    Args:
        event (composer_module.Composer.Submitted): The submission.
    """
    menu = self.query_one(command_menu.CommandMenu)
    composer = self.query_one(composer_module.Composer)
    text = event.text.strip()
    if composer.menu_open:
        name = menu.highlighted_name()
        if menu.showing_paths and name is not None:
            self._complete_command()
            return
        if name is not None and name != text:
            self._complete_command()
            return
    tray = self.query_one(pending_files.PendingFiles)
    if not text and not tray.has_files():
        return
    if text:
        self._history.add(text)
    composer.set_text('')
    menu.hide()
    composer.menu_open = False
    if text and self._run_terminal_command(text):
        return
    for word, path in attachable_files.AttachableFiles().mentioned(text):
        problem = tray.add(self.desk_name, path)
        if problem is not None:
            self.notify(problem, title='Not attached', severity='error')
            composer.set_text(text)
            return
        text = text.replace(word, '', 1).strip()
    self.query_one('#transcript', containers.VerticalScroll).anchor()
    self._send(text)

on_composer_files_dropped

on_composer_files_dropped(event)

Attaches files dragged onto the terminal.

Parameters:

Name	Type	Description	Default
event	FilesDropped	The files.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_composer_files_dropped(
    self, event: composer_module.Composer.FilesDropped
) -> None:
    """Attaches files dragged onto the terminal.

    Args:
        event (composer_module.Composer.FilesDropped): The files.
    """
    self._attach_files(event.paths)

on_composer_empty_backspace

on_composer_empty_backspace(event)

Removes the newest attached file.

Parameters:

Name	Type	Description	Default
event	EmptyBackspace	The key press.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_composer_empty_backspace(
    self, event: composer_module.Composer.EmptyBackspace
) -> None:
    """Removes the newest attached file.

    Args:
        event (composer_module.Composer.EmptyBackspace): The key press.
    """
    del event
    self.query_one(pending_files.PendingFiles).remove_newest()

on_composer_picture_wanted

on_composer_picture_wanted(event)

Attaches a picture on the clipboard, or pastes text when there is none.

Parameters:

Name	Type	Description	Default
event	PictureWanted	The key press.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

@work(group='clipboard', thread=True)
def on_composer_picture_wanted(
    self, event: composer_module.Composer.PictureWanted
) -> None:
    """Attaches a picture on the clipboard, or pastes text when there is none.

    Args:
        event (composer_module.Composer.PictureWanted): The key press.
    """
    del event
    folder = self.app.opener.folder / 'clipboard'
    path = clipboard_picture.ClipboardPicture().save(folder)
    if path is None:
        self.app.call_from_thread(self._paste_text)
        return
    self.app.call_from_thread(self._attach_files, [path])

action_escape

action_escape()

Closes the menu, stops the running turn, or clears the box on a second press.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def action_escape(self) -> None:
    """Closes the menu, stops the running turn, or clears the box on a second press."""
    menu = self.query_one(command_menu.CommandMenu)
    composer = self.query_one(composer_module.Composer)
    if composer.menu_open:
        menu.hide()
        composer.menu_open = False
        return
    if self._desk_is_busy():
        self._send('/stop')
        self.notify('Stopping the running turn.', timeout=3)
        return
    now = time.monotonic()
    if composer.text and now - self._last_escape < _DOUBLE_PRESS_SECONDS:
        self._history.add(composer.text)
        composer.set_text('')
        return
    self._last_escape = now
    if composer.text:
        self.notify('Press esc again to clear the message box.', timeout=2)

action_interrupt

action_interrupt()

Clears the box, stops the running turn, or exits on a second press.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def action_interrupt(self) -> None:
    """Clears the box, stops the running turn, or exits on a second press."""
    composer = self.query_one(composer_module.Composer)
    if composer.text:
        composer.set_text('')
        return
    if self._desk_is_busy():
        self._send('/stop')
        self.notify('Stopping the running turn.', timeout=3)
        return
    now = time.monotonic()
    if now - self._last_interrupt < _DOUBLE_PRESS_SECONDS:
        self.app.exit()
        return
    self._last_interrupt = now
    self.notify(
        'Press ctrl+c again to exit. The desks keep running.', timeout=2
    )

action_exit_when_empty

action_exit_when_empty()

Exits on a second press while the box is empty; otherwise deletes forward.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def action_exit_when_empty(self) -> None:
    """Exits on a second press while the box is empty; otherwise deletes forward."""
    composer = self.query_one(composer_module.Composer)
    if composer.text:
        composer.action_delete_right()
        return
    now = time.monotonic()
    if now - self._last_exit_press < _DOUBLE_PRESS_SECONDS:
        self.app.exit()
        return
    self._last_exit_press = now
    self.notify('Press ctrl+d again to exit.', timeout=2)

action_pick_desk

action_pick_desk()

Opens the desk list.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def action_pick_desk(self) -> None:
    """Opens the desk list."""
    desks = self._desks_document.get('desks', [])
    if not desks:
        self.notify('The desks are not known yet.', severity='warning')
        return
    self.app.push_screen(
        desk_picker.DeskPicker(desks, self.desk_name), self._switch_desk
    )

action_pick_model

action_pick_model()

Opens the model list for this desk.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def action_pick_model(self) -> None:
    """Opens the model list for this desk."""
    desk = self.desk()
    if desk is None or not desk.get('backends'):
        self.notify(
            'This desk has no models to choose from.', severity='warning'
        )
        return
    self.app.push_screen(model_picker.ModelPicker(desk), self._choose_model)

action_toggle_side_panel

action_toggle_side_panel()

Shows or hides the side panel, and remembers the choice.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def action_toggle_side_panel(self) -> None:
    """Shows or hides the side panel, and remembers the choice."""
    panel = self.query_one(side_panel.SidePanel)
    panel.display = not panel.display
    self._settings.side_panel_open = bool(panel.display)
    self._settings.save()

action_focus_dock

action_focus_dock()

Moves to the waiting request's card, or to the next widget when nothing waits.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def action_focus_dock(self) -> None:
    """Moves to the waiting request's card, or to the next widget when nothing waits."""
    dock = self.query_one(request_dock.RequestDock)
    if not dock.focus_card():
        self.focus_next()

on_request_dock_left

on_request_dock_left(event)

Goes back to the message box when the user leaves a card with Escape.

Parameters:

Name	Type	Description	Default
event	Left	The notice.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_request_dock_left(
    self, event: request_dock.RequestDock.Left
) -> None:
    """Goes back to the message box when the user leaves a card with Escape.

    Args:
        event (request_dock.RequestDock.Left): The notice.
    """
    del event
    self.query_one(composer_module.Composer).focus()

action_toggle_tools

action_toggle_tools()

Opens or folds every tool call in the chat.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def action_toggle_tools(self) -> None:
    """Opens or folds every tool call in the chat."""
    transcript = self.query_one('#transcript', containers.VerticalScroll)
    at_bottom = (
        transcript.max_scroll_y - transcript.scroll_y <= _BOTTOM_SLACK_LINES
    )
    self._tools_expanded = not self._tools_expanded
    for view in self._views.values():
        if isinstance(view, turn_view.TurnView):
            view.expand_tools(self._tools_expanded)
    if at_bottom:
        transcript.anchor()
        transcript.call_after_refresh(transcript.scroll_end, animate=False)

action_search_history

action_search_history()

Opens the search of earlier prompts.

Source code in src/trading_agent/terminal/screens/chat_screen.py

def action_search_history(self) -> None:
    """Opens the search of earlier prompts."""
    self.app.push_screen(
        history_search.HistorySearch(self._history), self._use_prompt
    )

on_composer_keys_wanted

on_composer_keys_wanted(event)

Shows the keys when ? is typed into the empty message box.

Parameters:

Name	Type	Description	Default
event	KeysWanted	The request.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def on_composer_keys_wanted(
    self, event: composer_module.Composer.KeysWanted
) -> None:
    """Shows the keys when ? is typed into the empty message box.

    Args:
        event (composer_module.Composer.KeysWanted): The request.
    """
    del event
    self.app.push_screen(keys_screen.KeysScreen())

show_desk

show_desk(desk_name)

Shows another desk's chat, as the Desks page asks.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Source code in src/trading_agent/terminal/screens/chat_screen.py

def show_desk(self, desk_name: str) -> None:
    """Shows another desk's chat, as the Desks page asks.

    Args:
        desk_name (str): The desk.
    """
    self._switch_desk(desk_name)