"""Interactive thread selector screen for `/threads` command.""" from __future__ import annotations import asyncio import contextlib import logging import sqlite3 from pathlib import Path from typing import TYPE_CHECKING, ClassVar, cast from rich.cells import cell_len from textual.binding import Binding, BindingType from textual.color import Color as TColor from textual.containers import Horizontal, Vertical, VerticalScroll from textual.content import Content from textual.css.query import NoMatches from textual.fuzzy import Matcher from textual.message import Message from textual.screen import ModalScreen from textual.style import Style as TStyle from textual.widgets import Checkbox, Input, Select, Static # Specialize focused Select overlay key handling; no public re-export available. from textual.widgets._select import SelectCurrent, SelectOverlay # noqa: PLC2701 if TYPE_CHECKING: from collections.abc import Callable, Mapping from textual.app import ComposeResult from textual.events import Click, Key, MouseMove from deepagents_code.sessions import ThreadInfo from deepagents_code import theme from deepagents_code.config import ( build_langsmith_thread_url, get_glyphs, is_ascii_mode, ) from deepagents_code.widgets._links import open_style_link logger = logging.getLogger(__name__) _URL_FETCH_TIMEOUT = 2.0 """Seconds to wait for LangSmith thread-URL resolution before giving up.""" _column_widths_cache: ( tuple[ tuple[tuple[str, str | None], ...], # (thread_id, checkpoint_id) fingerprint frozenset[str], # visible column keys bool, # relative_time dict[str, int | None], # computed widths ] | None ) = None """Module-level cache so repeated `/threads` opens skip column-width computation when the inputs (thread data + config) haven't changed.""" _COL_TID = 10 _COL_AGENT = 12 _COL_MSGS = 4 _COL_BRANCH = 16 _COL_TIMESTAMP = None _MAX_SEARCH_TEXT_LEN = 200 _COL_PROMPT = None _AUTO_WIDTH_COLUMNS = {"agent_name", "created_at", "updated_at", "cwd"} _COLUMN_ORDER = ( "thread_id", "agent_name", "messages", "created_at", "updated_at", "git_branch", "cwd", "initial_prompt", ) _COLUMN_WIDTHS: dict[str, int | None] = { "thread_id": _COL_TID, "agent_name": _COL_AGENT, "messages": _COL_MSGS, "created_at": _COL_TIMESTAMP, "updated_at": _COL_TIMESTAMP, "git_branch": _COL_BRANCH, "cwd": None, "initial_prompt": _COL_PROMPT, } _COLUMN_LABELS = { "thread_id": "Thread ID", "agent_name": "Agent", "messages": "Msgs", "created_at": "Created", "updated_at": "Updated", "git_branch": "Branch", "cwd": "Location", "initial_prompt": "Prompt", } _COLUMN_TOGGLE_LABELS = { "thread_id": "Thread ID", "agent_name": "Agent Name", "messages": "# Messages", "created_at": "Created At", "updated_at": "Updated At", "git_branch": "Git Branch", "cwd": "Working Directory", "initial_prompt": "Initial Prompt", } # Reserved for future right-aligned columns (e.g., message counts). _RIGHT_ALIGNED_COLUMNS: set[str] = set() _SWITCH_ID_PREFIX = "thread-column-" _SORT_SWITCH_ID = "thread-sort-toggle" _RELATIVE_TIME_SWITCH_ID = "thread-relative-time" _SCOPE_SELECT_ID = "thread-scope-select" _SCOPE_VALUE_CWD = "cwd" _SCOPE_VALUE_ALL = "all" _AGENT_SELECT_ID = "thread-agent-select" _AGENT_VALUE_ALL = "__all__" _AGENT_VALUE_LOADING = "__loading__" # Label shown in the agent dropdown while the disk load is still pending and no # agent names are known yet. The options list is derived from visible threads # (plus the configured `/agents` list), so before the load completes it would # otherwise show only the "All agents" sentinel; "Loading..." signals that more # options may appear. Uses a distinct transient value so the final "All agents" # label refreshes when the completed load swaps in the real sentinel. Matches # the "Loading threads..." empty-state copy. _AGENT_LABEL_LOADING = "Loading..." # Display label and filter key for threads with no stored `agent_name`. Shared # between the Agent column renderer, the agent dropdown options, and the filter # predicate so all three read identically. The parentheses distinguish the # synthetic "missing agent" bucket from an agent literally named "unknown". _UNKNOWN_AGENT_LABEL = "(unknown)" _CONTROLS_SCROLL_ID = "thread-controls-scroll" _CONTROLS_OVERFLOW_ID = "thread-controls-overflow" _CELL_PADDING_RIGHT = 1 class _Sentinel: """Type for module-level singleton sentinels.""" _CWD_DEFAULT = _Sentinel() """Sentinel for the default `filter_cwd` constructor arg. Distinguishes "caller did not specify" (use the current working directory) from an explicit `None` (start with no cwd filter).""" def _safe_cwd_string() -> str | None: """Return `str(Path.cwd())` or `None` if the working directory is unreadable. `Path.cwd()` raises `OSError` (typically `FileNotFoundError`) when the process's working directory has been deleted or is otherwise inaccessible. Treating that as "no cwd filter" matches the storage convention used by `build_run_metadata` and lets the picker degrade to "All directories" instead of crashing the Textual event loop. """ try: return str(Path.cwd()) except OSError: logger.warning( "Could not determine working directory for thread picker; " "falling back to All directories", exc_info=True, ) return None _FormatFns = tuple[ "Callable[[str | None], str]", # format_path "Callable[[str | None], str]", # format_relative_timestamp "Callable[[str | None], str]", # format_timestamp ] """Cached `(format_path, format_relative_timestamp, format_timestamp)`. Resolved once on first use via `_get_format_fns()` to avoid the overhead of a per-call deferred import inside the hot `_format_column_value` loop. """ _format_fns_cache: _FormatFns | None = None """Cached format functions, populated on first call to `_get_format_fns()`.""" def _get_format_fns() -> _FormatFns: """Return cached `(format_path, format_relative_timestamp, format_timestamp)`.""" global _format_fns_cache # noqa: PLW0603 if _format_fns_cache is not None: return _format_fns_cache from deepagents_code.sessions import ( format_path, format_relative_timestamp, format_timestamp, ) _format_fns_cache = (format_path, format_relative_timestamp, format_timestamp) return _format_fns_cache def _apply_column_width( cell: Static, key: str, column_widths: Mapping[str, int | None] ) -> None: """Apply an explicit width to a table cell when one is configured. Args: cell: The cell widget to size. key: Column key for the cell. column_widths: Effective column widths for the current table state. """ width = column_widths.get(key) if width is not None: cell.styles.width = width if key in _AUTO_WIDTH_COLUMNS: cell.styles.min_width = width def _active_sort_key(sort_by_updated: bool) -> str: """Return the active timestamp field used for sorting.""" return "updated_at" if sort_by_updated else "created_at" def _visible_column_keys(columns: dict[str, bool]) -> list[str]: """Return visible columns in the on-screen order. Args: columns: Column visibility settings keyed by column name. Returns: Visible column keys in display order. """ return [key for key in _COLUMN_ORDER if columns.get(key)] def _collapse_whitespace(value: str) -> str: """Normalize a text value onto a single display line. Args: value: Raw text to display in a single cell. Returns: The input text collapsed to a single line. """ return " ".join(value.split()) def _truncate_value(value: str, width: int | None) -> str: """Trim text to fit a fixed-width column. Args: value: Raw cell text. width: Maximum column width, or `None` for no truncation. Returns: The possibly truncated display string. """ if width is None: return value display = _collapse_whitespace(value) if len(display) <= width: return display glyphs = get_glyphs() ellipsis = glyphs.ellipsis if width <= len(ellipsis): return display[:width] return display[: width - len(ellipsis)] + ellipsis def _format_column_value( thread: ThreadInfo, key: str, *, relative_time: bool = False ) -> str: """Return the display text for one thread column. Args: thread: Thread metadata for the row. key: Column key to format. relative_time: Use relative timestamps instead of absolute. Returns: Formatted display text for the column cell. """ format_path, format_relative_ts, format_ts = _get_format_fns() fmt = format_relative_ts if relative_time else format_ts value: str if key == "thread_id": # Strip UUID separators in the compact table preview so truncation # never leaves a dangling trailing hyphen in the thread ID column. value = thread["thread_id"].replace("-", "") elif key == "agent_name": value = thread.get("agent_name") or _UNKNOWN_AGENT_LABEL elif key == "messages": raw_count = thread.get("message_count") value = str(raw_count) if raw_count is not None else "..." elif key == "created_at": value = fmt(thread.get("created_at")) elif key == "updated_at": value = fmt(thread.get("updated_at")) elif key == "git_branch": value = thread.get("git_branch") or "" elif key == "cwd": value = format_path(thread.get("cwd")) elif key == "initial_prompt": value = _collapse_whitespace(thread.get("initial_prompt") or "") else: value = "" return _truncate_value(value, _COLUMN_WIDTHS[key]) def _format_header_label(key: str) -> str: """Return the rendered header label for a column.""" return _truncate_value(_COLUMN_LABELS[key], _COLUMN_WIDTHS[key]) def _header_cell_classes(key: str, *, sort_key: str) -> str: """Return CSS classes for a header cell. Args: key: Column key for the header cell. sort_key: Currently active sort column. Returns: Space-delimited classes for the header cell widget. """ classes = f"thread-cell thread-cell-{key}" if key == sort_key: classes += " thread-cell-sorted" return classes class ThreadOption(Horizontal): """A clickable thread option in the selector.""" def __init__( self, thread: ThreadInfo, index: int, *, columns: dict[str, bool], column_widths: Mapping[str, int | None], selected: bool, current: bool, relative_time: bool = False, cell_text: dict[tuple[str, str], str] | None = None, classes: str = "", ) -> None: """Initialize a thread option row. Args: thread: Thread metadata for the row. index: The index of this option in the filtered list. columns: Column visibility settings. column_widths: Effective widths for the visible columns. selected: Whether the row is highlighted. current: Whether the row is the active thread. relative_time: Use relative timestamps. cell_text: Pre-formatted cell values keyed by `(thread_id, key)`. classes: CSS classes for styling. """ super().__init__(classes=classes) self.thread = thread self.thread_id = thread["thread_id"] self.index = index self._columns = dict(columns) self._column_widths = dict(column_widths) self._selected = selected self._current = current self._relative_time = relative_time self._cell_text = cell_text class Clicked(Message): """Message sent when a thread option is clicked.""" def __init__(self, thread_id: str, index: int) -> None: """Initialize the Clicked message. Args: thread_id: The thread identifier. index: The index of the clicked option. """ super().__init__() self.thread_id = thread_id self.index = index def compose(self) -> ComposeResult: """Compose the row cells. Yields: Static cells for each visible column. """ yield Static( self._cursor_text(), classes="thread-cell thread-cell-cursor", markup=False, ) tid = self.thread_id for key in _visible_column_keys(self._columns): if self._cell_text is not None and (tid, key) in self._cell_text: text = self._cell_text[tid, key] else: text = _format_column_value( self.thread, key, relative_time=self._relative_time ) cell = Static( text, classes=f"thread-cell thread-cell-{key}", expand=key == "initial_prompt", markup=False, ) _apply_column_width(cell, key, self._column_widths) yield cell def _cursor_text(self) -> str: """Return the cursor indicator for the row.""" return get_glyphs().cursor if self._selected else "" def set_selected(self, selected: bool) -> None: """Update row selection styling without rebuilding the row. Args: selected: Whether the row should be highlighted. """ self._selected = selected if selected: self.add_class("thread-option-selected") else: self.remove_class("thread-option-selected") try: cursor = self.query_one(".thread-cell-cursor", Static) except NoMatches: return cursor.update(self._cursor_text()) def on_click(self, event: Click) -> None: """Handle click on this option. Args: event: The click event. """ event.stop() self.post_message(self.Clicked(self.thread_id, self.index)) class ThreadControlsScroll(VerticalScroll): """Emit scroll changes so the parent can refresh the overflow indicator.""" class Scrolled(Message): """Message sent when the controls pane scroll position changes.""" def watch_scroll_y(self, old_value: float, new_value: float) -> None: """Notify the selector so it can update the overflow indicator.""" super().watch_scroll_y(old_value, new_value) if self.is_attached: self.post_message(self.Scrolled()) class DeleteThreadConfirmScreen(ModalScreen[bool]): """Confirmation modal shown before deleting a thread.""" BINDINGS: ClassVar[list[BindingType]] = [ Binding("enter", "confirm", "Confirm", show=False, priority=True), Binding("escape", "cancel", "Cancel", show=False, priority=True), ] """Key bindings for the delete-confirmation overlay. Enter confirms deletion, Esc cancels. Both use `priority=True` so they take precedence over bindings on the underlying thread selector. """ CSS = """ DeleteThreadConfirmScreen { align: center middle; } DeleteThreadConfirmScreen > Vertical { width: 50; height: auto; background: $surface; border: solid red; padding: 1 2; } DeleteThreadConfirmScreen .thread-confirm-text { text-align: center; margin-bottom: 1; } DeleteThreadConfirmScreen .thread-confirm-help { text-align: center; color: $text-muted; text-style: italic; } """ """Styling for the centered confirmation dialog, red border, and help text.""" def __init__(self, thread_id: str) -> None: """Initialize the confirmation modal. Args: thread_id: Thread ID the user is being asked to delete. """ super().__init__() self._delete_thread_id = thread_id def compose(self) -> ComposeResult: """Compose the confirmation dialog. Yields: Widgets for the delete confirmation prompt. """ with Vertical(id="delete-confirm"): yield Static( Content.from_markup( "Delete thread [bold]$tid[/bold]?", tid=self._delete_thread_id, ), classes="thread-confirm-text", ) yield Static( "Enter to confirm, Esc to cancel", classes="thread-confirm-help", ) def action_confirm(self) -> None: """Confirm deletion.""" self.dismiss(True) def action_cancel(self) -> None: """Cancel deletion.""" self.dismiss(False) _SCOPE_OVERLAY_CONSUMED_KEYS = frozenset( {"tab", "shift+tab", "up", "down", "pageup", "pagedown", "home", "end"} ) class ThreadScopeSelectOverlay(SelectOverlay): """Scope dropdown overlay that consumes option navigation while focused.""" def key_tab(self, event: Key) -> None: """Move to the next option without leaving the dropdown.""" event.prevent_default() event.stop() self.action_cursor_down() def key_shift_tab(self, event: Key) -> None: """Move to the previous option without leaving the dropdown.""" event.prevent_default() event.stop() self.action_cursor_up() def check_consume_key(self, key: str, character: str | None = None) -> bool: """Report option navigation consumption so ancestor bindings stay hidden. Args: key: Pressed key identifier. character: Printable character for the key, when present. Returns: `True` when the overlay consumes the key. """ if key in _SCOPE_OVERLAY_CONSUMED_KEYS: return True return super().check_consume_key(key, character) class ThreadScopeSelect(Select[str]): """Scope dropdown that keeps focus contained while its menu is open.""" def action_show_overlay(self) -> None: """Show the overlay when it has finished mounting.""" try: select_current = self.query_one(SelectCurrent) select_overlay = self.query_one(ThreadScopeSelectOverlay) except NoMatches: return select_current.has_value = True self.expanded = True if select_overlay.highlighted is None: select_overlay.action_first() def compose(self) -> ComposeResult: """Compose the select with a scope-specific overlay. Yields: Current value display and dropdown overlay widgets. """ yield SelectCurrent(self.prompt) yield ThreadScopeSelectOverlay(type_to_search=self._type_to_search).data_bind( compact=Select.compact ) def key_tab(self, event: Key) -> None: """Prevent focus traversal while the dropdown menu is open.""" if self.expanded: event.prevent_default() event.stop() def key_shift_tab(self, event: Key) -> None: """Prevent reverse focus traversal while the dropdown menu is open.""" if self.expanded: event.prevent_default() event.stop() def check_consume_key(self, key: str, character: str | None) -> bool: """Report Tab consumption while expanded so global bindings stay hidden. Args: key: Pressed key identifier. character: Printable character for the key, when present. Returns: `True` when the open dropdown consumes the key. """ if self.expanded and key in {"tab", "shift+tab"}: return True return super().check_consume_key(key, character) class ThreadSelectorScreen(ModalScreen[str | None]): """Modal dialog for browsing and resuming threads. Displays recent threads with keyboard navigation, fuzzy search, configurable columns, and delete support. Returns a `thread_id` string on selection, or `None` on cancel. """ BINDINGS: ClassVar[list[BindingType]] = [ Binding("up", "move_up", "Up", show=False, priority=True), Binding("down", "move_down", "Down", show=False, priority=True), Binding("pageup", "page_up", "Page up", show=False, priority=True), Binding("pagedown", "page_down", "Page down", show=False, priority=True), Binding("enter", "select", "Select", show=False, priority=True), Binding("escape", "cancel", "Cancel", show=False, priority=True), Binding("ctrl+d", "delete_thread", "Delete", show=False, priority=True), Binding("tab", "focus_next_filter", "Next filter", show=False, priority=True), Binding( "shift+tab", "focus_previous_filter", "Previous filter", show=False, priority=True, ), ] """Key bindings for thread navigation, selection, deletion, and filter focus. Arrows move the cursor, Page Up/Down jump by a visual page, Enter selects the highlighted thread, Ctrl+D opens the delete-confirmation overlay, Tab/Shift+Tab rotate focus through the filter input and column toggle checkboxes, and Esc dismisses. All bindings use `priority=True` so they take precedence over the embedded filter `Input` and checkbox widgets; vim-style `j`/`k` bindings are deliberately omitted because they would prevent typing those letters into the always-focused filter input. """ CSS = """ ThreadSelectorScreen { align: center middle; } ThreadSelectorScreen #thread-selector-shell { width: 100%; max-width: 98%; height: 90%; background: $surface; border: solid $primary; padding: 1 2; } ThreadSelectorScreen .thread-selector-title { text-style: bold; color: $primary; text-align: center; margin-bottom: 1; } ThreadSelectorScreen #thread-filter { margin-bottom: 1; border: solid $primary-lighten-2; } ThreadSelectorScreen #thread-filter:focus { border: solid $primary; } ThreadSelectorScreen .thread-selector-body { height: 1fr; } ThreadSelectorScreen .thread-table-pane { width: 1fr; min-width: 40; height: 1fr; } ThreadSelectorScreen .thread-controls { width: 28; min-width: 24; height: 1fr; margin-left: 1; padding-left: 1; border-left: solid $primary-lighten-2; } ThreadSelectorScreen .thread-controls-scroll { height: 1fr; min-height: 1; scrollbar-gutter: stable; } ThreadSelectorScreen .thread-controls-title { text-style: bold; color: $primary; margin-bottom: 1; } ThreadSelectorScreen .thread-controls-help { color: $text-muted; margin-bottom: 1; } ThreadSelectorScreen .thread-controls-label { color: $text-muted; margin-top: 0; } ThreadSelectorScreen .thread-scope-select, ThreadSelectorScreen .thread-agent-select { width: 1fr; height: auto; margin-bottom: 1; } ThreadSelectorScreen .thread-column-toggle { width: 1fr; height: auto; } ThreadSelectorScreen .thread-controls-overflow { height: 1; color: $text-muted; text-align: center; } ThreadSelectorScreen .thread-list-header { height: 1; padding: 0 1; color: $text-muted; text-style: bold; width: 100%; overflow-x: hidden; } ThreadSelectorScreen .thread-list-header .thread-cell-sorted { color: $primary; } ThreadSelectorScreen .thread-list { height: 1fr; min-height: 5; scrollbar-gutter: stable; background: $background; } ThreadSelectorScreen .thread-option { height: 1; width: 100%; padding: 0 1; overflow-x: hidden; } ThreadSelectorScreen .thread-option:hover { background: $surface-lighten-1; } ThreadSelectorScreen .thread-option-selected { background: $primary; color: $background; text-style: bold; } ThreadSelectorScreen .thread-option-selected:hover { background: $primary-lighten-1; } ThreadSelectorScreen .thread-option-current { text-style: italic; } ThreadSelectorScreen .thread-cell { height: 1; padding-right: 1; } ThreadSelectorScreen .thread-cell-cursor { width: 2; color: $primary; } ThreadSelectorScreen .thread-cell-thread_id { width: 10; } ThreadSelectorScreen .thread-cell-agent_name { width: auto; overflow-x: hidden; text-wrap: nowrap; text-overflow: ellipsis; } ThreadSelectorScreen .thread-cell-messages { width: 4; } ThreadSelectorScreen .thread-cell-created_at, ThreadSelectorScreen .thread-cell-updated_at { width: auto; } ThreadSelectorScreen .thread-cell-git_branch { width: 17; overflow-x: hidden; text-wrap: nowrap; text-overflow: ellipsis; } ThreadSelectorScreen .thread-cell-initial_prompt { width: 1fr; min-width: 1; overflow-x: hidden; text-wrap: nowrap; text-overflow: ellipsis; } ThreadSelectorScreen .thread-selector-help { height: auto; color: $text-muted; text-style: italic; margin-top: 1; text-align: center; } ThreadSelectorScreen .thread-empty { color: $text-muted; text-align: center; margin-top: 2; } """ """Styling for the modal shell, filter input, two-pane body (thread table and column controls), per-column widths, and help footer.""" def __init__( self, current_thread: str | None = None, *, thread_limit: int | None = None, initial_threads: list[ThreadInfo] | None = None, filter_cwd: str | _Sentinel | None = _CWD_DEFAULT, ) -> None: """Initialize the `ThreadSelectorScreen`. Args: current_thread: The currently active thread ID (to highlight). thread_limit: Maximum number of rows to fetch when querying DB. initial_threads: Optional preloaded rows to render immediately. filter_cwd: Working-directory filter for the picker. When the default sentinel, the picker mirrors Claude Code's `/resume` and scopes to the current working directory; the "Show threads from" select in the Options panel widens the view to "all directories". Pass an explicit path to scope to that directory, or `None` to start the picker with no cwd filter. """ super().__init__() self._current_thread = current_thread self._thread_limit = thread_limit from deepagents_code.model_config import load_thread_config cfg = load_thread_config() if isinstance(filter_cwd, _Sentinel): # Honor the persisted scope: "all" starts with no cwd filter, # "cwd" scopes to the current working directory. self._filter_cwd: str | None = ( None if cfg.scope == "all" else _safe_cwd_string() ) else: self._filter_cwd = filter_cwd initial = list(initial_threads) if initial_threads is not None else [] # The cached `initial_threads` are unfiltered, so apply the active cwd # filter here to avoid showing all threads on first paint when we are # about to re-query with the filter active. if self._filter_cwd is not None: initial = [t for t in initial if t.get("cwd") == self._filter_cwd] self._threads: list[ThreadInfo] = initial self._filtered_threads: list[ThreadInfo] = list(self._threads) self._has_initial_threads = initial_threads is not None self._disk_load_complete = False self._selected_index = 0 self._option_widgets: list[ThreadOption] = [] self._filter_text = "" self._filter_agent: str | None = None # Configured agent names from `~/.deepagents/` (the `/agents` list), # loaded off the event loop in `_load_threads`. Unioned with # thread-derived names so the agent filter mirrors `/agents` even for # agents that have not produced any threads yet. self._available_agent_names: list[str] = [] self._confirming_delete = False self._render_lock = asyncio.Lock() self._filter_input: Input | None = None self._filter_controls: list[Input | Checkbox | Select[str]] | None = None self._cell_text: dict[tuple[str, str], str] = {} self._columns = dict(cfg.columns) self._relative_time = cfg.relative_time self._sort_by_updated = cfg.sort_order == "updated_at" # Cached threads are pre-sorted by updated_at DESC (the only sort # order the cache stores). Skip the O(n log n) re-sort when that # matches the user's preference. if not (self._has_initial_threads and self._sort_by_updated): self._apply_sort() self._sync_selected_index() self._column_widths = self._compute_column_widths() @staticmethod def _switch_id(column_key: str) -> str: """Return the DOM id for a column toggle switch.""" return f"{_SWITCH_ID_PREFIX}{column_key}" @staticmethod def _switch_column_key(switch_id: str | None) -> str | None: """Extract the column key from a switch id. Args: switch_id: Widget id for a switch in the control panel. Returns: The corresponding column key, or `None` for unrelated ids. """ if not switch_id or not switch_id.startswith(_SWITCH_ID_PREFIX): return None return switch_id.removeprefix(_SWITCH_ID_PREFIX) def _sync_selected_index(self) -> None: """Select the current thread when it exists in the loaded rows.""" self._selected_index = 0 for i, thread in enumerate(self._filtered_threads): if thread["thread_id"] == self._current_thread: self._selected_index = i break def _build_title(self, thread_url: str | None = None) -> str | Content: """Build the title, optionally with a clickable thread ID link. Args: thread_url: LangSmith thread URL. When provided, the thread ID is rendered as a clickable hyperlink. Returns: Plain string or `Content` with an embedded hyperlink. """ if not self._current_thread: return "Select Thread" if thread_url: return Content.assemble( "Select Thread (current: ", ( self._current_thread, TStyle( foreground=TColor.parse(theme.get_theme_colors(self).primary), link=thread_url, ), ), ")", ) return f"Select Thread (current: {self._current_thread})" def _build_help_text(self) -> str: """Build the footer help text for the selector. Returns: Footer guidance for the active selector bindings. """ glyphs = get_glyphs() lines = ( f"{glyphs.arrow_up}/{glyphs.arrow_down} navigate" f" {glyphs.bullet} Enter select" f" {glyphs.bullet} Tab/Shift+Tab focus options" f" {glyphs.bullet} Space toggle option" f" {glyphs.bullet} Ctrl+D delete" f" {glyphs.bullet} Esc cancel" ) limit = self._effective_thread_limit() if len(self._threads) >= limit: lines += ( f"\nShowing last {limit} threads. " "Set DA_CLI_RECENT_THREADS to override." ) return lines def _effective_thread_limit(self) -> int: """Return the resolved thread limit for display purposes.""" if self._thread_limit is not None: return self._thread_limit from deepagents_code.sessions import get_thread_limit return get_thread_limit() def _format_sort_toggle_label(self) -> str: """Return the control-panel sort label for the toggle switch.""" label = "Updated At" if self._sort_by_updated else "Created At" return f"Sort by {label}" def _get_filter_input(self) -> Input: """Return the cached search input widget.""" if self._filter_input is None: self._filter_input = self.query_one("#thread-filter", Input) return self._filter_input def _filter_focus_order(self) -> list[Input | Checkbox | Select[str]]: """Return the cached tab order for filter controls in the side panel.""" if self._filter_controls is None: filter_input = self._get_filter_input() scope_select = self.query_one(f"#{_SCOPE_SELECT_ID}", Select) agent_select = self.query_one(f"#{_AGENT_SELECT_ID}", Select) sort_switch = self.query_one(f"#{_SORT_SWITCH_ID}", Checkbox) relative_switch = self.query_one(f"#{_RELATIVE_TIME_SWITCH_ID}", Checkbox) column_switches = [ self.query_one(f"#{self._switch_id(key)}", Checkbox) for key in _COLUMN_ORDER ] self._filter_controls = [ filter_input, scope_select, agent_select, sort_switch, relative_switch, *column_switches, ] return self._filter_controls def _collect_agent_options(self) -> list[tuple[str, str]]: """Return Select option tuples for the agent dropdown. Unions the configured agents (the `/agents` list, loaded into `self._available_agent_names`) with the agent names found in the currently loaded thread list, then prepends an "All agents" sentinel so the caller can always reset to an unfiltered view. Including the configured agents keeps the filter in sync with `/agents` so an agent that has not produced any threads yet is still selectable; including thread-derived names keeps threads from since-deleted agents (and the `_UNKNOWN_AGENT_LABEL` sentinel for threads with no agent name) filterable. While the disk load is still pending and no agent names are known yet, returns a single `("Loading...", _AGENT_VALUE_LOADING)` option so the dropdown signals that more options are on the way rather than implying "All agents" is the only choice. Returns: List of `(label, value)` pairs; the first entry is always `("All agents", _AGENT_VALUE_ALL)` once the load has completed (or `("Loading...", _AGENT_VALUE_LOADING)` while it is still pending with no known agents). """ names = {t.get("agent_name") or _UNKNOWN_AGENT_LABEL for t in self._threads} names.update(self._available_agent_names) if not names and not self._disk_load_complete: return [(_AGENT_LABEL_LOADING, _AGENT_VALUE_LOADING)] ordered = sorted(names, key=str.casefold) return [("All agents", _AGENT_VALUE_ALL), *((n, n) for n in ordered)] def _refresh_agent_select_options(self) -> bool: """Repopulate the agent dropdown after threads are reloaded. Preserves the current selection when the chosen agent is still a valid option — present in the reloaded threads or the configured `/agents` list (see `_collect_agent_options`); falls back to "All agents" otherwise. Returns: `True` when the active agent filter was cleared (selection fell back to "All agents"); `False` when the selection was preserved or the dropdown is not mounted. """ try: agent_select = self.query_one(f"#{_AGENT_SELECT_ID}", Select) except NoMatches: return False options = self._collect_agent_options() valid_values = {value for _, value in options} current = agent_select.value # `set_options` resets the value to the first option, posting a # transient `Select.Changed(_AGENT_VALUE_ALL)`. When we then restore # `current` below, a second `Changed(current)` is posted. Both are # queued, so `on_select_changed` runs them after this method returns; # the spurious intermediate rebuild they would trigger is harmless # because `_schedule_filter_and_rebuild` runs exclusively and the later # event's worker supersedes the earlier one before it paints. agent_select.set_options(options) if current in valid_values: agent_select.value = current return False agent_select.value = _AGENT_VALUE_ALL self._filter_agent = None return True async def _load_available_agent_names(self) -> None: """Cache the configured agent names (the `/agents` list) off-thread. Runs the filesystem scan in a worker thread so the Textual event loop is never blocked on directory I/O. Expected filesystem errors are swallowed by `get_available_agent_names`, which returns an empty list (logging the cause for unexpected `OSError`s, but staying silent when the agents directory simply does not exist yet), so the agent filter degrades to thread-derived names only. The caller in `_load_threads` additionally guards against any unexpected error so the picker is never stranded if the scan fails. """ from deepagents_code.agent import get_available_agent_names self._available_agent_names = await asyncio.to_thread(get_available_agent_names) def _get_select(self, select_id: str) -> Select[str]: """Return an Options panel dropdown widget by ID.""" return self.query_one(f"#{select_id}", Select) def _expanded_select_id(self) -> str | None: """Return the ID for the currently open Options panel dropdown.""" for select_id in (_SCOPE_SELECT_ID, _AGENT_SELECT_ID): try: if self._get_select(select_id).expanded: return select_id except NoMatches: continue return None def _is_select_expanded(self) -> bool: """Return whether an Options panel dropdown menu is currently open.""" return self._expanded_select_id() is not None def _restore_expanded_select_focus(self, select_id: str) -> None: """Return focus to an Options panel dropdown after background rendering.""" try: select = self._get_select(select_id) except NoMatches: return select.expanded = True with contextlib.suppress(NoMatches): select.query_one(ThreadScopeSelectOverlay).focus() return select.focus() def _close_expanded_select(self) -> None: """Close the open Options panel dropdown and restore focus.""" select_id = self._expanded_select_id() if select_id is None: return select = self._get_select(select_id) select.expanded = False select.focus() def _select_expanded_highlight(self) -> None: """Select the currently highlighted option in the open dropdown.""" action_select = getattr(self.focused, "action_select", None) if callable(action_select): action_select() return self._close_expanded_select() def _open_focused_select(self) -> bool: """Open a focused Options panel dropdown. Returns: `True` when a dropdown was focused and opened. """ try: selects = ( self._get_select(_SCOPE_SELECT_ID), self._get_select(_AGENT_SELECT_ID), ) except NoMatches: return False for select in selects: if self.focused is select: select.action_show_overlay() return True return False def compose(self) -> ComposeResult: """Compose the screen layout. Yields: Widgets for the thread selector UI. """ with Vertical(id="thread-selector-shell"): yield Static( self._build_title(), classes="thread-selector-title", id="thread-title" ) yield Input( placeholder="Type to search threads...", select_on_focus=False, id="thread-filter", ) with Horizontal(classes="thread-selector-body"): with Vertical(classes="thread-table-pane"): with Horizontal( classes="thread-list-header", id="thread-header", ): yield Static("", classes="thread-cell thread-cell-cursor") sort_key = _active_sort_key(self._sort_by_updated) for key in _visible_column_keys(self._columns): cell = Static( _format_header_label(key), classes=_header_cell_classes(key, sort_key=sort_key), expand=key == "initial_prompt", markup=False, ) _apply_column_width(cell, key, self._column_widths) yield cell with VerticalScroll(classes="thread-list"): if self._has_initial_threads and self._filtered_threads: self._option_widgets, _ = self._create_option_widgets() yield from self._option_widgets else: yield self._build_empty_state() with Vertical(classes="thread-controls"): yield Static("Options", classes="thread-controls-title") yield Static( ( "Tab through sort and column toggles. " "Column visibility persists between sessions." ), classes="thread-controls-help", markup=False, ) yield Static( "Show threads from:", classes="thread-controls-label", markup=False, ) yield ThreadScopeSelect( [ ("Current directory", _SCOPE_VALUE_CWD), ("All directories", _SCOPE_VALUE_ALL), ], value=( _SCOPE_VALUE_CWD if self._filter_cwd is not None else _SCOPE_VALUE_ALL ), allow_blank=False, compact=True, id=_SCOPE_SELECT_ID, classes="thread-scope-select", ) yield Static( "Show agent:", classes="thread-controls-label", markup=False, ) agent_options = self._collect_agent_options() agent_value = ( _AGENT_VALUE_LOADING if agent_options[0][1] == _AGENT_VALUE_LOADING else _AGENT_VALUE_ALL ) yield ThreadScopeSelect( agent_options, value=agent_value, allow_blank=False, compact=True, id=_AGENT_SELECT_ID, classes="thread-agent-select", ) with ThreadControlsScroll( id=_CONTROLS_SCROLL_ID, classes="thread-controls-scroll" ): yield Checkbox( self._format_sort_toggle_label(), self._sort_by_updated, id=_SORT_SWITCH_ID, classes="thread-column-toggle", compact=True, ) yield Checkbox( "Relative Timestamps", self._relative_time, id=_RELATIVE_TIME_SWITCH_ID, classes="thread-column-toggle", compact=True, ) for key in _COLUMN_ORDER: yield Checkbox( _COLUMN_TOGGLE_LABELS[key], self._columns.get(key, False), id=self._switch_id(key), classes="thread-column-toggle", compact=True, ) yield Static( get_glyphs().ellipsis, id=_CONTROLS_OVERFLOW_ID, classes="thread-controls-overflow", ) yield Static( self._build_help_text(), classes="thread-selector-help", id="thread-help", ) async def on_mount(self) -> None: """Fetch threads, configure border for ASCII terminals, and build the list.""" if is_ascii_mode(): container = self.query_one("#thread-selector-shell", Vertical) colors = theme.get_theme_colors(self) container.styles.border = ("ascii", colors.success) filter_input = self._get_filter_input() self._filter_focus_order() filter_input.focus() self.call_after_refresh(self._update_controls_overflow_hint) if self._has_initial_threads: self.call_after_refresh(self._scroll_selected_into_view) if self._current_thread: self._resolve_thread_url() if self._has_initial_threads: # Defer by one message cycle so Textual finishes processing # mount messages before we start the DB refresh. self.call_after_refresh(self._start_thread_load) else: # _load_threads replaces self._threads and schedules background # enrichment (message counts, initial prompts) after load # completes. Launch immediately when there are no cached rows # to render. self.run_worker( self._load_threads, exclusive=True, group="thread-selector-load" ) def on_resize(self) -> None: """Refresh the overflow indicator when the modal is resized.""" self.call_after_refresh(self._update_controls_overflow_hint) def on_thread_controls_scroll_scrolled( self, event: ThreadControlsScroll.Scrolled ) -> None: """Refresh the overflow indicator after the options pane scrolls.""" event.stop() self.call_after_refresh(self._update_controls_overflow_hint) def _start_thread_load(self) -> None: """Launch the thread-load worker after the initial layout pass.""" if not self.is_attached: return self.run_worker( self._load_threads, exclusive=True, group="thread-selector-load" ) def on_input_changed(self, event: Input.Changed) -> None: """Filter threads as user types. Args: event: The input changed event. """ self._filter_text = event.value self._schedule_filter_and_rebuild() def on_input_submitted(self, event: Input.Submitted) -> None: """Handle Enter key when filter input is focused. Args: event: The input submitted event. """ event.stop() self.action_select() def on_key(self, event: Key) -> None: """Return focus to search when letters are typed from other controls. Args: event: The key event. """ if self._confirming_delete: return if event.key in {"tab", "shift+tab"} and self._is_select_expanded(): event.prevent_default() event.stop() return filter_input = self._get_filter_input() if filter_input.has_focus: return character = event.character if not character or not character.isalpha(): return filter_input.focus() filter_input.insert_text_at_cursor(character) self.set_timer(0.01, self._collapse_search_selection) event.stop() def _collapse_search_selection(self) -> None: """Place the search cursor at the end without an active selection.""" filter_input = self._get_filter_input() filter_input.selection = type(filter_input.selection).cursor( len(filter_input.value) ) def on_checkbox_changed(self, event: Checkbox.Changed) -> None: """Route sort, relative-time, and column-visibility checkbox changes. Args: event: The checkbox change event. """ if event.checkbox.id == _SORT_SWITCH_ID: if self._sort_by_updated == event.value: return self._sort_by_updated = event.value self._apply_sort() self._sync_selected_index() self._update_help_widgets() self._schedule_list_rebuild() self._persist_sort_order("updated_at" if event.value else "created_at") return if event.checkbox.id == _RELATIVE_TIME_SWITCH_ID: if self._relative_time == event.value: return self._relative_time = event.value from deepagents_code.model_config import save_thread_relative_time self.run_worker( asyncio.to_thread(save_thread_relative_time, event.value), group="thread-selector-save", ) self._schedule_list_rebuild() return column_key = self._switch_column_key(event.checkbox.id) if column_key is None or column_key not in self._columns: return if self._columns[column_key] == event.value: return self._columns[column_key] = event.value self._apply_sort() self._sync_selected_index() self._update_help_widgets() if event.value and column_key in {"messages", "initial_prompt"}: self._schedule_checkpoint_enrichment() from deepagents_code.model_config import save_thread_columns snapshot = dict(self._columns) self.run_worker( asyncio.to_thread(save_thread_columns, snapshot), group="thread-selector-save", ) self._schedule_list_rebuild() def _agent_filtered_threads(self) -> list[ThreadInfo]: """Return `self._threads` narrowed to the active agent filter. Returns: All threads when no agent filter is active; otherwise only threads whose `agent_name` (or `_UNKNOWN_AGENT_LABEL` for missing values) matches `self._filter_agent`. """ if self._filter_agent is None: return list(self._threads) return [ t for t in self._threads if (t.get("agent_name") or _UNKNOWN_AGENT_LABEL) == self._filter_agent ] def _update_filtered_list(self) -> None: """Update filtered threads based on search text using fuzzy matching.""" base = self._agent_filtered_threads() query = self._filter_text.strip() if not query: self._filtered_threads = base self._apply_sort() self._sync_selected_index() self._column_widths = self._compute_column_widths() return tokens = query.split() try: matchers = [Matcher(token, case_sensitive=False) for token in tokens] scored: list[tuple[float, ThreadInfo]] = [] for thread in base: search_text = self._get_search_text(thread) scores = [matcher.match(search_text) for matcher in matchers] if all(score > 0 for score in scores): scored.append((min(scores), thread)) except Exception: logger.warning( "Fuzzy matcher failed for query %r, falling back to full list", query, exc_info=True, ) self._filtered_threads = base self._apply_sort() self._sync_selected_index() self._column_widths = self._compute_column_widths() return sort_key = _active_sort_key(self._sort_by_updated) self._filtered_threads = [ thread for _, thread in sorted( scored, key=lambda item: ( item[0], item[1].get(sort_key) or "", item[1].get("updated_at") or "", item[1]["thread_id"], ), reverse=True, ) ] self._selected_index = 0 self._column_widths = self._compute_column_widths() def _compute_column_widths(self) -> dict[str, int | None]: """Return effective widths for the current table state. Textual's `width: auto` computes per-widget widths, so this method derives shared widths from the visible data instead. Also populates `self._cell_text` as a side effect so that `ThreadOption.compose()` can reuse the formatted strings. Returns: Dict mapping column keys to their effective cell widths, with `None` for flex columns. """ global _column_widths_cache # noqa: PLW0603 # Module-level cache requires global statement visible_keys = _visible_column_keys(self._columns) visible = frozenset(visible_keys) fingerprint = tuple( (t["thread_id"], t.get("latest_checkpoint_id")) for t in self._filtered_threads ) if _column_widths_cache is not None: fp, vis, rel, cached_widths = _column_widths_cache if ( fp == fingerprint and vis == visible and rel == self._relative_time and self._cell_text ): return dict(cached_widths) # Pre-format every visible cell in one pass. cell_text: dict[tuple[str, str], str] = {} for thread in self._filtered_threads: tid = thread["thread_id"] for key in visible_keys: cell_text[tid, key] = _format_column_value( thread, key, relative_time=self._relative_time ) self._cell_text = cell_text # Derive auto-widths from the pre-formatted values. widths = dict(_COLUMN_WIDTHS) for key in _AUTO_WIDTH_COLUMNS: if key not in visible: continue header_len = cell_len(_format_header_label(key)) max_cell = max( ( cell_len(cell_text[t["thread_id"], key]) for t in self._filtered_threads ), default=0, ) widths[key] = max(header_len, max_cell) + _CELL_PADDING_RIGHT _column_widths_cache = (fingerprint, visible, self._relative_time, widths) return widths @staticmethod def _get_search_text(thread: ThreadInfo) -> str: """Build searchable text from thread fields. The result is capped at `_MAX_SEARCH_TEXT_LEN` characters so that Textual's fuzzy `Matcher` (which uses recursive backtracking) does not hit exponential performance on long initial prompts with repeated characters. Args: thread: Thread metadata. Returns: Concatenated searchable string, truncated to a safe length. """ # Include both the raw cwd and the home-collapsed display form so a # user typing either "/Users/chesar/foo" or "~/foo" matches the row. format_path, _, _ = _get_format_fns() cwd = thread.get("cwd") or "" parts = [ thread["thread_id"], thread.get("agent_name") or "", thread.get("git_branch") or "", cwd, format_path(cwd) if cwd else "", thread.get("initial_prompt") or "", ] text = " ".join(parts) return text[:_MAX_SEARCH_TEXT_LEN] def _schedule_filter_and_rebuild(self) -> None: """Queue a filter + rebuild, coalescing rapid keystrokes.""" self.run_worker( self._filter_and_build, exclusive=True, group="thread-selector-render", ) async def _filter_and_build(self) -> None: """Run fuzzy filtering in a thread then rebuild the list.""" query = self._filter_text.strip() threads = self._agent_filtered_threads() sort_by_updated = self._sort_by_updated filtered = await asyncio.to_thread( self._compute_filtered, query, threads, sort_by_updated ) self._filtered_threads = filtered if query: self._selected_index = 0 else: self._sync_selected_index() self._column_widths = self._compute_column_widths() await self._build_list(recompute_widths=False) @staticmethod def _compute_filtered( query: str, threads: list[ThreadInfo], sort_by_updated: bool, ) -> list[ThreadInfo]: """Compute filtered thread list off the main thread. Args: query: Current search query text. threads: Agent-pre-filtered thread list snapshot. sort_by_updated: Whether to sort by `updated_at`. Returns: Filtered and sorted thread list. """ sort_key = _active_sort_key(sort_by_updated) if not query: result = list(threads) result.sort(key=lambda t: t.get(sort_key) or "", reverse=True) return result tokens = query.split() try: matchers = [Matcher(token, case_sensitive=False) for token in tokens] scored: list[tuple[float, ThreadInfo]] = [] for thread in threads: search_text = ThreadSelectorScreen._get_search_text(thread) scores = [matcher.match(search_text) for matcher in matchers] if all(score > 0 for score in scores): scored.append((min(scores), thread)) except Exception: logger.warning( "Fuzzy matcher failed for query %r, falling back to full list", query, exc_info=True, ) result = list(threads) result.sort(key=lambda t: t.get(sort_key) or "", reverse=True) return result return [ thread for _, thread in sorted( scored, key=lambda item: ( item[0], item[1].get(sort_key) or "", item[1].get("updated_at") or "", item[1]["thread_id"], ), reverse=True, ) ] def _schedule_list_rebuild(self) -> None: """Queue a list rebuild, coalescing rapid updates.""" self.run_worker( self._build_list, exclusive=True, group="thread-selector-render", ) def _pending_checkpoint_fields(self) -> tuple[bool, bool]: """Return which visible checkpoint-derived fields still need loading.""" load_counts = self._columns.get("messages", False) and any( "message_count" not in thread for thread in self._threads ) load_prompts = self._columns.get("initial_prompt", False) and any( "initial_prompt" not in thread for thread in self._threads ) return load_counts, load_prompts async def _populate_visible_checkpoint_details(self) -> tuple[bool, bool]: """Load any still-missing checkpoint-derived fields for visible columns. Returns: Tuple indicating whether message counts and prompts were requested. """ from deepagents_code.sessions import populate_thread_checkpoint_details load_counts, load_prompts = self._pending_checkpoint_fields() if not load_counts and not load_prompts: return False, False await populate_thread_checkpoint_details( self._threads, include_message_count=load_counts, include_initial_prompt=load_prompts, ) return load_counts, load_prompts def _schedule_checkpoint_enrichment(self) -> None: """Schedule one checkpoint-enrichment pass for missing row fields.""" has_missing_counts, has_missing_prompts = self._pending_checkpoint_fields() if not has_missing_counts and not has_missing_prompts: return self.run_worker( self._load_checkpoint_details, exclusive=True, group="thread-selector-checkpoints", ) @staticmethod def _threads_match(old: list[ThreadInfo], new: list[ThreadInfo]) -> bool: """Check whether two thread lists have the same IDs and checkpoints in order. Args: old: Previous thread list. new: Fresh thread list. Returns: True if both lists have identical thread/checkpoint ID pairs. """ if len(old) != len(new): return False for a, b in zip(old, new, strict=True): if a["thread_id"] != b["thread_id"]: return False if a.get("latest_checkpoint_id") != b.get("latest_checkpoint_id"): return False return True async def _load_threads(self) -> None: """Load thread rows first, then kick off background enrichment.""" from deepagents_code.sessions import ( apply_cached_thread_initial_prompts, apply_cached_thread_message_counts, list_threads, ) old_threads = list(self._threads) try: limit = self._thread_limit if limit is None: from deepagents_code.sessions import get_thread_limit limit = get_thread_limit() sort_by = "updated" if self._sort_by_updated else "created" self._threads = await list_threads( limit=limit, include_message_count=False, sort_by=sort_by, cwd=self._filter_cwd, ) except (OSError, sqlite3.Error) as exc: logger.exception("Failed to load threads for thread selector") self._disk_load_complete = True await self._show_mount_error(str(exc)) return except Exception as exc: # An unexpected error (e.g. a malformed row raising ValueError) must # still mark the load complete and surface the failure; otherwise the # picker stays stuck on the "Loading threads..." placeholder forever. # CancelledError is a BaseException and is intentionally not caught # here, so a superseding exclusive worker keeps ownership of the flag. logger.exception("Unexpected error loading threads for thread selector") self._disk_load_complete = True await self._show_mount_error(str(exc)) return self._disk_load_complete = True apply_cached_thread_message_counts(self._threads) apply_cached_thread_initial_prompts(self._threads) if not self._has_initial_threads: try: await self._populate_visible_checkpoint_details() except (OSError, sqlite3.Error): logger.debug( "Could not preload checkpoint details for thread selector", exc_info=True, ) except Exception: logger.warning( "Unexpected error preloading checkpoint details " "for thread selector", exc_info=True, ) try: await self._load_available_agent_names() except Exception: # The configured-agent scan is non-essential: a failure here must # not strand the picker by skipping the filter refresh, DOM build, # and checkpoint enrichment below. `get_available_agent_names` # already absorbs expected filesystem errors; this guards anything # unexpected that escapes it. logger.warning( "Could not load configured agent names for thread selector", exc_info=True, ) self._update_filtered_list() self._sync_selected_index() if self._refresh_agent_select_options(): self._update_filtered_list() self._sync_selected_index() # Short-circuit: when the fresh data matches what is already rendered, # update widget references and cell labels without tearing down the DOM. if ( self._has_initial_threads and self._option_widgets and self._threads_match(old_threads, self._filtered_threads) ): for widget, thread in zip( self._option_widgets, self._filtered_threads, strict=True, ): widget.thread = thread self._refresh_cell_labels() else: await self._build_list() self._schedule_checkpoint_enrichment() if self._current_thread: self._resolve_thread_url() async def _load_checkpoint_details(self) -> None: """Populate checkpoint-derived thread fields in one background pass.""" if not self._threads: return try: _, load_prompts = await self._populate_visible_checkpoint_details() except (OSError, sqlite3.Error): logger.debug( "Could not load checkpoint details for thread selector", exc_info=True, ) return except Exception: logger.warning( "Unexpected error loading checkpoint details for thread selector", exc_info=True, ) return if load_prompts and self._filter_text.strip(): # Prompts may affect fuzzy match results; rebuild the filtered # list but preserve the user's cursor position. saved_tid = ( self._filtered_threads[self._selected_index]["thread_id"] if self._selected_index < len(self._filtered_threads) else None ) self._update_filtered_list() if saved_tid is not None: for i, thread in enumerate(self._filtered_threads): if thread["thread_id"] == saved_tid: self._selected_index = i break self._schedule_list_rebuild() else: self._refresh_cell_labels() def _refresh_cell_labels(self) -> None: """Update visible cell text in-place without rebuilding the DOM.""" visible_keys = _visible_column_keys(self._columns) # Recompute because thread data may have changed since # _compute_column_widths populated the cache. cell_text: dict[tuple[str, str], str] = {} for thread in self._filtered_threads: tid = thread["thread_id"] for key in visible_keys: cell_text[tid, key] = _format_column_value( thread, key, relative_time=self._relative_time ) self._cell_text = cell_text for widget in self._option_widgets: tid = widget.thread_id for key in visible_keys: try: cell = widget.query_one(f".thread-cell-{key}", Static) except NoMatches: continue cell.update(cell_text[tid, key]) def _resolve_thread_url(self) -> None: """Start exclusive background worker to resolve LangSmith thread URL.""" self.run_worker( self._fetch_thread_url, exclusive=True, group="thread-selector-url" ) async def _fetch_thread_url(self) -> None: """Resolve the LangSmith URL and update the title with a clickable link.""" if not self._current_thread: return loop = asyncio.get_running_loop() started = loop.time() try: thread_url = await asyncio.wait_for( asyncio.to_thread(build_langsmith_thread_url, self._current_thread), timeout=_URL_FETCH_TIMEOUT, ) except (TimeoutError, OSError): logger.debug( "Could not resolve LangSmith thread URL for '%s'", self._current_thread, exc_info=True, ) return except Exception: logger.debug( "Unexpected error resolving LangSmith thread URL for '%s'", self._current_thread, exc_info=True, ) return if loop.time() - started >= _URL_FETCH_TIMEOUT: logger.debug( "LangSmith thread URL for '%s' resolved after the timeout deadline", self._current_thread, ) return if thread_url: try: title_widget = self.query_one("#thread-title", Static) title_widget.update(self._build_title(thread_url)) except NoMatches: logger.debug( "Title widget #thread-title not found; " "thread selector may have been dismissed during URL resolution" ) async def _show_mount_error(self, detail: str) -> None: """Display an error message inside the thread list and refocus. Args: detail: Human-readable error detail to show. """ try: async with self._render_lock: scroll = self.query_one(".thread-list", VerticalScroll) await scroll.remove_children() await scroll.mount( Static( Content.from_markup( "[red]Failed to load threads: $detail. " "Press Esc to close.[/red]", detail=detail, ), classes="thread-empty", ) ) except Exception: logger.warning( "Could not display error message in thread selector UI", exc_info=True, ) self.focus() def _build_empty_state(self) -> Static: """Build the empty-list placeholder. Shows "Loading threads..." while the disk load is still pending so the picker never claims "No threads found" before the query completes. Returns: The placeholder `Static` widget to mount in the thread list. """ if not self._disk_load_complete: return Static( Content.styled("Loading threads...", "dim"), classes="thread-empty", id="thread-loading", ) return Static( Content.styled("No threads found", "dim"), classes="thread-empty", ) async def _build_list(self, *, recompute_widths: bool = True) -> None: """Build the thread option widgets. Args: recompute_widths: Whether to recalculate shared column widths first. """ expanded_select_id = self._expanded_select_id() async with self._render_lock: try: scroll = self.query_one(".thread-list", VerticalScroll) except NoMatches: return if recompute_widths: self._column_widths = self._compute_column_widths() with self.app.batch_update(): await scroll.remove_children() self._update_help_widgets() if not self._filtered_threads: self._option_widgets = [] await scroll.mount(self._build_empty_state()) return self._option_widgets, selected_widget = self._create_option_widgets() await scroll.mount(*self._option_widgets) if expanded_select_id is not None: self.call_after_refresh( lambda: self._restore_expanded_select_focus(expanded_select_id) ) elif selected_widget: self.call_after_refresh(self._scroll_selected_into_view) def _create_option_widgets(self) -> tuple[list[ThreadOption], ThreadOption | None]: """Build option widgets from filtered threads without mounting. Returns: Tuple of all option widgets and the currently selected widget. """ widgets: list[ThreadOption] = [] selected_widget: ThreadOption | None = None for i, thread in enumerate(self._filtered_threads): is_current = thread["thread_id"] == self._current_thread is_selected = i == self._selected_index classes = "thread-option" if is_selected: classes += " thread-option-selected" if is_current: classes += " thread-option-current" widget = ThreadOption( thread=thread, index=i, columns=self._columns, column_widths=self._column_widths, selected=is_selected, current=is_current, relative_time=self._relative_time, cell_text=self._cell_text or None, classes=classes, ) widgets.append(widget) if is_selected: selected_widget = widget return widgets, selected_widget def _scroll_selected_into_view(self) -> None: """Scroll selected option into view without animation.""" if not self._option_widgets: return if self._selected_index >= len(self._option_widgets): return try: scroll = self.query_one(".thread-list", VerticalScroll) except NoMatches: return if self._selected_index == 0: scroll.scroll_home(animate=False) else: self._option_widgets[self._selected_index].scroll_visible(animate=False) def _update_help_widgets(self) -> None: """Update visible header and help text after state changes.""" self._schedule_header_rebuild() self.call_after_refresh(self._update_controls_overflow_hint) try: help_widget = self.query_one("#thread-help", Static) help_widget.update(self._build_help_text()) except NoMatches: logger.debug("Help widget #thread-help not found during update") with contextlib.suppress(NoMatches): sort_checkbox = self.query_one(f"#{_SORT_SWITCH_ID}", Checkbox) sort_checkbox.label = self._format_sort_toggle_label() if sort_checkbox.value != self._sort_by_updated: sort_checkbox.value = self._sort_by_updated def _update_controls_overflow_hint(self) -> None: """Show an ellipsis when the options pane has hidden rows below.""" try: scroll = self.query_one(f"#{_CONTROLS_SCROLL_ID}", VerticalScroll) hint = self.query_one(f"#{_CONTROLS_OVERFLOW_ID}", Static) except NoMatches: return hint.display = scroll.scroll_y < scroll.max_scroll_y def _schedule_header_rebuild(self) -> None: """Queue a header rebuild to reflect column/sort changes.""" self.run_worker( self._rebuild_header, exclusive=True, group="thread-selector-header", ) async def _rebuild_header(self) -> None: """Replace header cells to match current visible columns.""" try: header = self.query_one("#thread-header", Horizontal) except NoMatches: return sort_key = _active_sort_key(self._sort_by_updated) self._column_widths = self._compute_column_widths() with self.app.batch_update(): await header.remove_children() cells: list[Static] = [Static("", classes="thread-cell thread-cell-cursor")] for key in _visible_column_keys(self._columns): cell = Static( _format_header_label(key), classes=_header_cell_classes(key, sort_key=sort_key), expand=key == "initial_prompt", markup=False, ) _apply_column_width(cell, key, self._column_widths) cells.append(cell) await header.mount(*cells) def _apply_sort(self) -> None: """Sort filtered threads by the active sort key.""" key = _active_sort_key(self._sort_by_updated) self._filtered_threads.sort( key=lambda thread: thread.get(key) or "", reverse=True ) def _move_selection(self, delta: int) -> None: """Move selection by delta, updating only the affected rows. Args: delta: Positions to move (negative for up, positive for down). """ if not self._filtered_threads or not self._option_widgets: return count = len(self._filtered_threads) old_index = self._selected_index new_index = (old_index + delta) % count self._selected_index = new_index self._option_widgets[old_index].set_selected(False) self._option_widgets[new_index].set_selected(True) if new_index == 0: scroll = self.query_one(".thread-list", VerticalScroll) scroll.scroll_home(animate=False) else: self._option_widgets[new_index].scroll_visible() def action_move_up(self) -> None: """Move selection up.""" if self._confirming_delete: return self._move_selection(-1) def action_move_down(self) -> None: """Move selection down.""" if self._confirming_delete: return self._move_selection(1) def _visible_page_size(self) -> int: """Return the number of thread options that fit in one visual page. Returns: Number of thread options per page, at least 1. """ default_page_size = 10 try: scroll = self.query_one(".thread-list", VerticalScroll) height = scroll.size.height except NoMatches: logger.debug( "Thread list widget not found in _visible_page_size; " "using default page size %d", default_page_size, ) return default_page_size if height <= 0: return default_page_size return max(1, height) def action_page_up(self) -> None: """Move selection up by one visible page.""" if self._confirming_delete or not self._filtered_threads: return page = self._visible_page_size() target = max(0, self._selected_index - page) delta = target - self._selected_index if delta != 0: self._move_selection(delta) def action_page_down(self) -> None: """Move selection down by one visible page.""" if self._confirming_delete or not self._filtered_threads: return count = len(self._filtered_threads) page = self._visible_page_size() target = min(count - 1, self._selected_index + page) delta = target - self._selected_index if delta != 0: self._move_selection(delta) def action_select(self) -> None: """Confirm the highlighted thread and dismiss the selector.""" if self._confirming_delete: return if self._is_select_expanded(): self._select_expanded_highlight() return if self._open_focused_select(): return if self._filtered_threads: thread_id = self._filtered_threads[self._selected_index]["thread_id"] self.dismiss(thread_id) def action_focus_next_filter(self) -> None: """Move focus through the filter and column-toggle controls.""" if self._confirming_delete: return if self._is_select_expanded(): return controls = self._filter_focus_order() focused = self.focused if focused not in controls: controls[0].focus() return index = controls.index(cast("Input | Checkbox | Select[str]", focused)) controls[(index + 1) % len(controls)].focus() def action_focus_previous_filter(self) -> None: """Move focus backward through the filter and column-toggle controls.""" if self._confirming_delete: return if self._is_select_expanded(): return controls = self._filter_focus_order() focused = self.focused if focused not in controls: controls[-1].focus() return index = controls.index(cast("Input | Checkbox | Select[str]", focused)) controls[(index - 1) % len(controls)].focus() def action_toggle_sort(self) -> None: """Toggle sort between updated_at and created_at.""" if self._confirming_delete: return self._sort_by_updated = not self._sort_by_updated self._apply_sort() self._sync_selected_index() self._update_help_widgets() self._schedule_list_rebuild() self._persist_sort_order( "updated_at" if self._sort_by_updated else "created_at" ) def on_select_changed(self, event: Select.Changed) -> None: """Handle the scope and agent `Select` dropdowns in the Options panel. The agent branch updates the in-memory `_filter_agent` and re-filters the loaded threads without re-querying the database; selecting the "All agents" sentinel (`_AGENT_VALUE_ALL`) clears the filter. The scope branch mirrors Claude Code's `/resume`: the picker is scoped to the current working directory by default, and choosing "All directories" widens the view to threads from every cwd (which does re-query). Args: event: The `Select.Changed` message; `event.select.id` selects the branch and `event.value` carries the new selection. """ if event.select.id == _AGENT_SELECT_ID: if self._confirming_delete: return if event.value == _AGENT_VALUE_LOADING: return new_agent = None if event.value == _AGENT_VALUE_ALL else str(event.value) if new_agent == self._filter_agent: return self._filter_agent = new_agent self._schedule_filter_and_rebuild() return if event.select.id != _SCOPE_SELECT_ID: return if self._confirming_delete: return if event.value == _SCOPE_VALUE_CWD: new_cwd = _safe_cwd_string() if new_cwd is None: # Working directory is gone or unreadable. Tell the user # rather than silently re-querying with no filter. self.app.notify( "Could not determine the current working directory; " "showing all directories instead.", severity="warning", markup=False, ) else: new_cwd = None self._persist_scope( _SCOPE_VALUE_CWD if event.value == _SCOPE_VALUE_CWD else _SCOPE_VALUE_ALL ) if new_cwd == self._filter_cwd: return self._filter_cwd = new_cwd self._update_help_widgets() self.run_worker( self._load_threads, exclusive=True, group="thread-selector-load" ) def _persist_scope(self, scope: str) -> None: """Save directory-scope preference to config, notifying on failure.""" async def _save() -> None: from deepagents_code.model_config import save_thread_scope ok = await asyncio.to_thread(save_thread_scope, scope) if not ok: self.app.notify("Could not save scope preference", severity="warning") self.run_worker(_save(), group="thread-selector-save") def _persist_sort_order(self, order: str) -> None: """Save sort-order preference to config, notifying on failure.""" async def _save() -> None: from deepagents_code.model_config import save_thread_sort_order ok = await asyncio.to_thread(save_thread_sort_order, order) if not ok: self.app.notify("Could not save sort preference", severity="warning") self.run_worker(_save(), group="thread-selector-save") def action_delete_thread(self) -> None: """Show delete confirmation for the highlighted thread.""" if self._confirming_delete: return if not self._filtered_threads: # Nothing to delete — fall through to quit. Using exit() instead of # dismiss() is intentional: dismiss() would just close the modal # silently, re-swallowing ctrl+d. self.app.exit() return self._confirming_delete = True thread = self._filtered_threads[self._selected_index] tid = thread["thread_id"] self.app.push_screen( DeleteThreadConfirmScreen(tid), lambda confirmed: self._on_delete_confirmed(tid, confirmed), ) @property def is_delete_confirmation_open(self) -> bool: """Return whether the delete confirmation overlay is visible.""" return self._confirming_delete def _on_delete_confirmed(self, thread_id: str, confirmed: bool | None) -> None: """Handle the result from the delete confirmation modal. Args: thread_id: Thread ID that was targeted. confirmed: Whether deletion was confirmed. """ self._confirming_delete = False if confirmed: self.run_worker( self._handle_delete_confirm(thread_id), group="thread-delete-execute", ) return with contextlib.suppress(NoMatches): self._get_filter_input().focus() async def _handle_delete_confirm(self, thread_id: str) -> None: """Execute thread deletion after confirmation. Args: thread_id: Thread ID to delete. """ from deepagents_code.sessions import delete_thread preferred_thread_id: str | None = None if self._selected_index + 1 < len(self._filtered_threads): preferred_thread_id = self._filtered_threads[self._selected_index + 1][ "thread_id" ] elif self._selected_index > 0: preferred_thread_id = self._filtered_threads[self._selected_index - 1][ "thread_id" ] try: await delete_thread(thread_id) except (OSError, sqlite3.Error): logger.warning("Failed to delete thread %s", thread_id, exc_info=True) self.app.notify( f"Failed to delete thread {thread_id[:8]}", severity="error", timeout=3, markup=False, ) with contextlib.suppress(NoMatches): self.query_one("#thread-filter", Input).focus() return self._threads = [ thread for thread in self._threads if thread["thread_id"] != thread_id ] self._update_filtered_list() if preferred_thread_id is not None: for index, thread in enumerate(self._filtered_threads): if thread["thread_id"] == preferred_thread_id: self._selected_index = index break if self._selected_index >= len(self._filtered_threads): self._selected_index = max(0, len(self._filtered_threads) - 1) await self._build_list() with contextlib.suppress(NoMatches): self.query_one("#thread-filter", Input).focus() def on_click(self, event: Click) -> None: # noqa: PLR6301 # Textual event handler """Open Rich-style hyperlinks on single click.""" open_style_link(event) def on_mouse_move(self, event: MouseMove) -> None: """Show a pointer over Rich-style hyperlinks.""" self.styles.pointer = "pointer" if event.style.link else "default" def on_leave(self) -> None: """Reset the pointer shape when the mouse leaves the selector.""" self.styles.pointer = "default" def on_thread_option_clicked(self, event: ThreadOption.Clicked) -> None: """Handle click on a thread option. Args: event: The clicked message with thread ID and index. """ if self._confirming_delete: return if 0 <= event.index < len(self._filtered_threads): self._selected_index = event.index self.dismiss(event.thread_id) def action_cancel(self) -> None: """Cancel the selection.""" if self._is_select_expanded(): self._close_expanded_select() return self.dismiss(None)