"""Tests for ThreadSelectorScreen.""" import asyncio from collections.abc import Coroutine from pathlib import Path from typing import Any, ClassVar, cast from unittest.mock import AsyncMock, MagicMock, patch import pytest from rich.cells import cell_len from rich.style import Style from textual.app import App, ComposeResult from textual.binding import Binding, BindingType from textual.containers import Container, Horizontal, Vertical, VerticalScroll from textual.css.query import NoMatches from textual.screen import ModalScreen from textual.widgets import Checkbox, Input, Select, Static from textual.widgets._select import SelectCurrent from deepagents_code.app import DeepAgentsApp, _ThreadHistoryPayload from deepagents_code.sessions import ThreadInfo from deepagents_code.widgets.thread_selector import ( DeleteThreadConfirmScreen, ThreadScopeSelect, ThreadScopeSelectOverlay, ThreadSelectorScreen, ) MOCK_THREADS: list[ThreadInfo] = [ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", "message_count": 5, "created_at": "2025-01-15T09:00:00", "git_branch": "main", "cwd": "/home/user/project-a", "initial_prompt": "Hello world", }, { "thread_id": "def67890", "agent_name": "other-agent", "updated_at": "2025-01-14T08:00:00", "message_count": 12, "created_at": "2025-01-14T07:00:00", "git_branch": "feature-x", "cwd": "/tmp/workspace", "initial_prompt": "Fix the bug", }, { "thread_id": "ghi11111", "agent_name": "my-agent", "updated_at": "2025-01-13T15:45:00", "message_count": 3, "created_at": "2025-01-13T14:00:00", "git_branch": None, "cwd": None, "initial_prompt": None, }, ] def _patch_list_threads(threads: list[ThreadInfo] | None = None) -> Any: # noqa: ANN401 """Return a patch context manager for `list_threads`. Args: threads: Thread list to return. Defaults to `MOCK_THREADS`. """ data = threads if threads is not None else MOCK_THREADS return patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, return_value=data, ) def _patch_columns(columns: dict[str, bool] | None = None) -> Any: # noqa: ANN401 """Patch thread config loaders for tests.""" import contextlib from deepagents_code.model_config import THREAD_COLUMN_DEFAULTS, ThreadConfig cols = columns if columns is not None else THREAD_COLUMN_DEFAULTS @contextlib.contextmanager def _ctx() -> Any: # noqa: ANN401 with ( patch( "deepagents_code.model_config.load_thread_columns", return_value=dict(cols), ), patch( "deepagents_code.model_config.load_thread_sort_order", return_value="updated_at", ), patch( "deepagents_code.model_config.load_thread_config", return_value=ThreadConfig( columns=dict(cols), relative_time=True, sort_order="updated_at", scope="cwd", ), ), ): yield return _ctx() def _patch_available_agents(names: list[str] | None = None) -> Any: # noqa: ANN401 """Return a patch context manager for `get_available_agent_names`. Args: names: Configured agent names to return. Defaults to an empty list so the agent filter depends only on the (deterministic) thread data. """ return patch( "deepagents_code.agent.get_available_agent_names", return_value=list(names) if names is not None else [], ) @pytest.fixture(autouse=True) def _isolate_available_agents() -> Any: # noqa: ANN401 """Keep the agent filter independent of the machine's `~/.deepagents/`. `_load_threads` scans the user's agents directory to populate the filter dropdown. Without isolation the option set would vary by developer/CI host, so default it to empty; tests that need specific agents patch it locally. """ with _patch_available_agents(): yield def _style_scalar_value(value: object) -> int: """Return the integer value from a Textual style scalar. Args: value: Style value that may be a scalar-like object. Returns: Integer scalar value. """ scalar = getattr(value, "value", None) assert isinstance(scalar, float) return int(scalar) class ThreadSelectorTestApp(App): """Test app for ThreadSelectorScreen.""" def __init__(self, current_thread: str | None = "abc12345") -> None: super().__init__() self.result: str | None = None self.dismissed = False self._current_thread = current_thread def compose(self) -> ComposeResult: yield Container(id="main") def show_selector(self) -> None: """Show the thread selector screen.""" def handle_result(result: str | None) -> None: self.result = result self.dismissed = True # Disable the default cwd filter so tests don't need to populate the # `cwd` field on every mock thread fixture. screen = ThreadSelectorScreen( current_thread=self._current_thread, filter_cwd=None ) self.push_screen(screen, handle_result) class AppWithEscapeBinding(App): """Test app with a conflicting escape binding.""" BINDINGS: ClassVar[list[BindingType]] = [ Binding("escape", "interrupt", "Interrupt", show=False, priority=True), ] def __init__(self) -> None: super().__init__() self.result: str | None = None self.dismissed = False self.interrupt_called = False def compose(self) -> ComposeResult: yield Container(id="main") def action_interrupt(self) -> None: """Handle escape.""" if isinstance(self.screen, ModalScreen): self.screen.dismiss(None) return self.interrupt_called = True def show_selector(self) -> None: """Show the thread selector screen.""" def handle_result(result: str | None) -> None: self.result = result self.dismissed = True screen = ThreadSelectorScreen(current_thread="abc12345", filter_cwd=None) self.push_screen(screen, handle_result) class TestThreadScopeSelect: """Tests for the custom thread filter select.""" def test_open_before_overlay_mount_noops(self) -> None: """Opening before composition should not leave the select expanded.""" select = ThreadScopeSelect( [("All", "all")], value="all", allow_blank=False, compact=True ) select.action_show_overlay() assert not select.expanded class TestThreadSelectorEscapeKey: """Tests for ESC key dismissing the modal.""" async def test_escape_dismisses_modal(self) -> None: """Pressing ESC should dismiss the modal with None result.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.press("escape") await pilot.pause() assert app.dismissed is True assert app.result is None async def test_escape_with_conflicting_app_binding(self) -> None: """ESC should dismiss modal even when app has its own escape binding.""" with _patch_list_threads(): app = AppWithEscapeBinding() async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.press("escape") await pilot.pause() assert app.dismissed is True assert app.result is None assert app.interrupt_called is False class TestThreadSelectorKeyboardNavigation: """Tests for keyboard navigation in the modal.""" async def test_down_arrow_moves_selection(self) -> None: """Down arrow should move selection down.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) initial_index = screen._selected_index await pilot.press("down") await pilot.pause() assert screen._selected_index == initial_index + 1 async def test_up_arrow_wraps_from_top(self) -> None: """Up arrow at index 0 should wrap to last thread.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) count = len(screen._threads) await pilot.press("up") await pilot.pause() expected = (0 - 1) % count assert screen._selected_index == expected async def test_enter_selects_thread(self) -> None: """Enter should select the current thread and dismiss.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.press("enter") await pilot.pause() assert app.dismissed is True assert app.result == "abc12345" class TestThreadSelectorCurrentThread: """Tests for current thread highlighting and preselection.""" async def test_current_thread_is_preselected(self) -> None: """Opening the selector should pre-select the current thread.""" with _patch_list_threads(): app = ThreadSelectorTestApp(current_thread="def67890") async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) # def67890 is at index 1 in MOCK_THREADS assert screen._selected_index == 1 async def test_unknown_current_thread_defaults_to_zero(self) -> None: """Unknown current thread should default to index 0.""" with _patch_list_threads(): app = ThreadSelectorTestApp(current_thread="nonexistent") async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._selected_index == 0 async def test_no_current_thread_defaults_to_zero(self) -> None: """No current thread should default to index 0.""" with _patch_list_threads(): app = ThreadSelectorTestApp(current_thread=None) async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._selected_index == 0 class TestThreadSelectorEmptyState: """Tests for empty thread list.""" async def test_no_threads_shows_empty_message(self) -> None: """Empty thread list should show a message and escape still works.""" with _patch_list_threads(threads=[]): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert len(screen._threads) == 0 # Enter with no threads should be a no-op (not crash) await pilot.press("enter") await pilot.pause() # Escape should still dismiss if not app.dismissed: await pilot.press("escape") await pilot.pause() assert app.dismissed is True assert app.result is None async def test_arrow_keys_on_empty_list_do_not_crash(self) -> None: """Arrow keys and page keys on empty list should be no-ops.""" with _patch_list_threads(threads=[]): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert len(screen._threads) == 0 for key in ("up", "down", "pageup", "pagedown"): await pilot.press(key) await pilot.pause() assert screen._selected_index == 0 await pilot.press("escape") await pilot.pause() assert app.dismissed is True class TestThreadSelectorNavigateAndSelect: """Tests for navigating then selecting a specific thread.""" async def test_navigate_down_and_select(self) -> None: """Navigate to second thread and select it.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.press("down") await pilot.pause() await pilot.press("enter") await pilot.pause() assert app.dismissed is True assert app.result == "def67890" class TestThreadSelectorTabSort: """Tests for sort toggling and focus traversal in the selector.""" async def test_sort_switch_toggles_sort(self) -> None: """The sort switch should highlight the active header column.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._sort_by_updated is True original_columns = dict(screen._columns) header = screen.query_one("#thread-header", Horizontal) updated_cell = header.query_one(".thread-cell-updated_at", Static) created_cell = header.query_one(".thread-cell-created_at", Static) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) assert str(updated_cell._Static__content) == "Updated" assert updated_cell.has_class("thread-cell-sorted") assert not created_cell.has_class("thread-cell-sorted") assert sort_switch.value is True assert "Sort by Updated" in str(sort_switch.label) sort_switch.toggle() await pilot.pause() assert screen._sort_by_updated is False assert screen._columns == original_columns created_cell = header.query_one(".thread-cell-created_at", Static) updated_cell = header.query_one(".thread-cell-updated_at", Static) assert str(created_cell._Static__content) == "Created" assert created_cell.has_class("thread-cell-sorted") assert not updated_cell.has_class("thread-cell-sorted") assert sort_switch.value is False assert "Sort by Created" in str(sort_switch.label) async def test_sorted_header_column_is_highlighted(self) -> None: """The active sort column should be highlighted without extra text.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test(size=(100, 24)) as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) header = screen.query_one("#thread-header", Horizontal) updated_cell = header.query_one(".thread-cell-updated_at", Static) created_cell = header.query_one(".thread-cell-created_at", Static) assert updated_cell.render_line(0).text.rstrip() == "Updated" assert created_cell.render_line(0).text.rstrip() == "Created" assert updated_cell.has_class("thread-cell-sorted") assert not created_cell.has_class("thread-cell-sorted") async def test_tab_moves_focus_into_column_switches(self) -> None: """Tab should move focus from the search input into the controls.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) filter_input = screen.query_one("#thread-filter", Input) scope_select = screen.query_one("#thread-scope-select", Select) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) thread_id_switch = screen.query_one( f"#{ThreadSelectorScreen._switch_id('thread_id')}", Checkbox, ) agent_name_switch = screen.query_one( f"#{ThreadSelectorScreen._switch_id('agent_name')}", Checkbox, ) messages_switch = screen.query_one( f"#{ThreadSelectorScreen._switch_id('messages')}", Checkbox, ) agent_select = screen.query_one("#thread-agent-select", Select) assert filter_input.has_focus screen.action_focus_next_filter() await pilot.pause() assert scope_select.has_focus screen.action_focus_next_filter() await pilot.pause() assert agent_select.has_focus screen.action_focus_next_filter() await pilot.pause() assert sort_switch.has_focus relative_time_switch = screen.query_one( "#thread-relative-time", Checkbox ) screen.action_focus_next_filter() await pilot.pause() assert relative_time_switch.has_focus screen.action_focus_next_filter() await pilot.pause() assert thread_id_switch.has_focus screen.action_focus_next_filter() await pilot.pause() assert agent_name_switch.has_focus screen.action_focus_next_filter() await pilot.pause() assert messages_switch.has_focus async def test_shift_tab_moves_focus_backward_through_controls(self) -> None: """Shift+Tab should move focus backward through the controls.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) filter_input = screen.query_one("#thread-filter", Input) scope_select = screen.query_one("#thread-scope-select", Select) agent_select = screen.query_one("#thread-agent-select", Select) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) assert filter_input.has_focus await pilot.press("tab") await pilot.pause() assert scope_select.has_focus await pilot.press("tab") await pilot.pause() assert agent_select.has_focus await pilot.press("tab") await pilot.pause() assert sort_switch.has_focus await pilot.press("shift+tab") await pilot.pause() assert agent_select.has_focus await pilot.press("shift+tab") await pilot.pause() assert scope_select.has_focus await pilot.press("shift+tab") await pilot.pause() assert filter_input.has_focus async def test_thread_load_preserves_open_agent_dropdown_focus(self) -> None: """Async thread loading should not move focus away from an open dropdown.""" started = asyncio.Event() release = asyncio.Event() async def delayed_list_threads( *_args: object, **_kwargs: object ) -> list[ThreadInfo]: started.set() await release.wait() return MOCK_THREADS mock_list = AsyncMock(side_effect=delayed_list_threads) with ( patch("deepagents_code.sessions.list_threads", mock_list), _patch_columns(), ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() await asyncio.wait_for(started.wait(), timeout=1) screen = app.screen assert isinstance(screen, ThreadSelectorScreen) await pilot.press("tab") await pilot.press("tab") await pilot.press("enter") await pilot.pause() agent_select = screen.query_one("#thread-agent-select", Select) assert agent_select.expanded assert isinstance(screen.focused, ThreadScopeSelectOverlay) with patch.object(screen, "_scroll_selected_into_view") as mock_scroll: release.set() for _ in range(20): await pilot.pause() if screen._disk_load_complete and screen._option_widgets: break assert agent_select.expanded assert isinstance(screen.focused, ThreadScopeSelectOverlay) mock_scroll.assert_not_called() async def test_cached_filter_controls_handle_tab_and_typing(self) -> None: """Tab traversal and type-to-search should use cached control lookups.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) filter_input = screen.query_one("#thread-filter", Input) scope_select = screen.query_one("#thread-scope-select", Select) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) controls = screen._filter_focus_order() event = MagicMock() event.character = "f" cached_input = MagicMock(spec=Input) cached_input.has_focus = False screen._filter_input = cached_input with ( patch.object( screen, "query_one", side_effect=AssertionError("unexpected DOM query"), ), patch.object(screen, "set_timer"), ): assert screen._filter_focus_order() == controls assert controls[0] is filter_input assert controls[1] is scope_select assert controls[3] is sort_switch screen.on_key(event) cached_input.focus.assert_called_once() cached_input.insert_text_at_cursor.assert_called_once_with("f") event.stop.assert_called_once() async def test_switch_toggle_keeps_focus_on_current_control(self) -> None: """Toggling a switch should not bounce focus back to the search input.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) filter_input = screen.query_one("#thread-filter", Input) await pilot.press("tab") await pilot.press("tab") await pilot.press("tab") await pilot.pause() assert sort_switch.has_focus sort_switch.toggle() await pilot.pause() assert sort_switch.has_focus assert not filter_input.has_focus async def test_typing_letter_from_controls_refocuses_search(self) -> None: """Typing a letter on a control should jump back to fuzzy search.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) filter_input = screen.query_one("#thread-filter", Input) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) await pilot.press("tab") await pilot.press("tab") await pilot.press("tab") await pilot.pause() assert sort_switch.has_focus await pilot.press("f") await pilot.pause() assert filter_input.has_focus assert filter_input.value == "f" assert screen._filter_text == "f" assert len(screen._filtered_threads) == 1 assert screen._filtered_threads[0]["thread_id"] == "def67890" async def test_typing_multiple_letters_from_controls_appends_search(self) -> None: """Typing multiple letters after refocus should append, not replace.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) filter_input = screen.query_one("#thread-filter", Input) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) await pilot.press("tab") await pilot.press("tab") await pilot.press("tab") await pilot.pause() assert sort_switch.has_focus await pilot.press("f") await pilot.pause() await pilot.press("i") await pilot.pause() assert filter_input.has_focus assert filter_input.value == "fi" assert screen._filter_text == "fi" assert len(screen._filtered_threads) == 1 assert screen._filtered_threads[0]["thread_id"] == "def67890" async def test_space_from_controls_does_not_refocus_search(self) -> None: """Space on a control should keep switch behavior instead of search focus.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) filter_input = screen.query_one("#thread-filter", Input) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) await pilot.press("tab") await pilot.press("tab") await pilot.press("tab") await pilot.pause() assert sort_switch.has_focus assert sort_switch.value is True await pilot.press("space") await pilot.pause() assert sort_switch.has_focus assert not filter_input.has_focus assert filter_input.value == "" assert sort_switch.value is False class TestThreadSelectorDownWrap: """Tests for wrapping from bottom to top.""" async def test_down_arrow_wraps_from_bottom(self) -> None: """Down arrow at last index should wrap to first thread.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) count = len(screen._threads) # Navigate to the last item for _ in range(count - 1): await pilot.press("down") await pilot.pause() assert screen._selected_index == count - 1 # One more down should wrap to 0 await pilot.press("down") await pilot.pause() assert screen._selected_index == 0 class TestThreadSelectorPageNavigation: """Tests for pageup/pagedown navigation.""" async def test_pagedown_moves_selection(self) -> None: """Pagedown should move selection forward.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) await pilot.press("pagedown") await pilot.pause() # Should move forward (clamped to last item with 3 threads) assert screen._selected_index == len(MOCK_THREADS) - 1 async def test_pageup_at_top_is_noop(self) -> None: """Pageup at index 0 should be a no-op.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._selected_index == 0 await pilot.press("pageup") await pilot.pause() assert screen._selected_index == 0 class _ThreadSelectorScopedTestApp(App): """Test app that mounts the picker with an explicit `filter_cwd`.""" def __init__(self, filter_cwd: str | None) -> None: super().__init__() self._filter_cwd = filter_cwd self.result: str | None = None self.dismissed = False def compose(self) -> ComposeResult: yield Container(id="main") def show_selector(self) -> None: """Mount the selector with a caller-supplied cwd filter.""" def handle_result(result: str | None) -> None: self.result = result self.dismissed = True screen = ThreadSelectorScreen(current_thread=None, filter_cwd=self._filter_cwd) self.push_screen(screen, handle_result) class TestThreadSelectorScopePersistedDefault: """Tests that the picker honors the persisted scope preference on open.""" def test_persisted_all_scope_starts_unfiltered(self) -> None: """A persisted scope of "all" should start the picker with no cwd filter.""" from deepagents_code.model_config import THREAD_COLUMN_DEFAULTS, ThreadConfig with patch( "deepagents_code.model_config.load_thread_config", return_value=ThreadConfig( columns=dict(THREAD_COLUMN_DEFAULTS), relative_time=True, sort_order="updated_at", scope="all", ), ): screen = ThreadSelectorScreen(current_thread=None) assert screen._filter_cwd is None def test_persisted_cwd_scope_starts_filtered(self) -> None: """A persisted scope of "cwd" should scope the picker to the cwd.""" from deepagents_code.model_config import THREAD_COLUMN_DEFAULTS, ThreadConfig with ( patch( "deepagents_code.model_config.load_thread_config", return_value=ThreadConfig( columns=dict(THREAD_COLUMN_DEFAULTS), relative_time=True, sort_order="updated_at", scope="cwd", ), ), patch( "deepagents_code.widgets.thread_selector._safe_cwd_string", return_value="/home/user/project-a", ), ): screen = ThreadSelectorScreen(current_thread=None) assert screen._filter_cwd == "/home/user/project-a" class TestThreadSelectorScopeSelect: """Tests for the cwd scope `Select` in the Options panel.""" async def test_enter_opens_scope_select_without_resuming_thread(self) -> None: """Enter on the focused scope control should open its dropdown.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) scope_select = screen.query_one("#thread-scope-select", Select) await pilot.press("tab") await pilot.pause() assert scope_select.has_focus await pilot.press("enter") await pilot.pause() assert scope_select.expanded assert not app.dismissed async def test_tab_keys_move_open_scope_select_highlight(self) -> None: """Tab and Shift+Tab should move the dropdown highlight while open.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) filter_input = screen.query_one("#thread-filter", Input) scope_select = screen.query_one("#thread-scope-select", Select) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) await pilot.press("tab") await pilot.press("enter") await pilot.pause() assert scope_select.expanded overlay = scope_select.query_one(ThreadScopeSelectOverlay) assert overlay.highlighted == 1 await pilot.press("shift+tab") await pilot.pause() assert scope_select.expanded assert overlay.highlighted == 0 assert not filter_input.has_focus assert not app.dismissed await pilot.press("tab") await pilot.pause() assert scope_select.expanded assert overlay.highlighted == 1 assert not sort_switch.has_focus assert not app.dismissed async def test_arrow_keys_move_open_scope_select_not_thread_list(self) -> None: """Normal dropdown navigation should not move the thread highlight.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) scope_select = screen.query_one("#thread-scope-select", Select) screen._selected_index = 1 await pilot.press("tab") await pilot.press("enter") await pilot.pause() assert scope_select.expanded overlay = scope_select.query_one(ThreadScopeSelectOverlay) assert overlay.highlighted == 1 await pilot.press("up") await pilot.pause() assert overlay.highlighted == 0 assert screen._selected_index == 1 await pilot.press("down") await pilot.pause() assert overlay.highlighted == 1 assert screen._selected_index == 1 await pilot.press("pageup") await pilot.pause() assert overlay.highlighted == 0 assert screen._selected_index == 1 await pilot.press("pagedown") await pilot.pause() assert overlay.highlighted == 1 assert screen._selected_index == 1 assert not app.dismissed async def test_escape_closes_open_scope_select_without_dismissing(self) -> None: """Esc should close the dropdown before it cancels the selector.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) scope_select = screen.query_one("#thread-scope-select", Select) await pilot.press("tab") await pilot.press("enter") await pilot.pause() assert scope_select.expanded await pilot.press("escape") await pilot.pause() assert not scope_select.expanded assert scope_select.has_focus assert not app.dismissed async def test_enter_selects_open_scope_select_without_resuming(self) -> None: """Enter should choose the highlighted dropdown option while open.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) scope_select = screen.query_one("#thread-scope-select", Select) await pilot.press("tab") await pilot.press("enter") await pilot.pause() assert scope_select.expanded await pilot.press("enter") await pilot.pause() assert not scope_select.expanded assert scope_select.has_focus assert not app.dismissed async def test_select_toggle_requeries_with_new_cwd(self) -> None: """Switching the scope dropdown reloads threads with the new cwd kwarg.""" starting_cwd = "/home/user/project-a" mock_list = AsyncMock(return_value=MOCK_THREADS) with ( patch("deepagents_code.sessions.list_threads", mock_list), _patch_columns(), patch( "deepagents_code.widgets.thread_selector._safe_cwd_string", return_value=starting_cwd, ), ): app = _ThreadSelectorScopedTestApp(filter_cwd=starting_cwd) async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._filter_cwd == starting_cwd scope_select = screen.query_one("#thread-scope-select", Select) # Sanity: initial query carried the cwd filter. assert mock_list.await_count >= 1 initial_kwargs = [c.kwargs for c in mock_list.await_args_list] assert any(kw.get("cwd") == starting_cwd for kw in initial_kwargs), ( f"expected starting cwd, got {initial_kwargs}" ) mock_list.reset_mock() scope_select.value = "all" await pilot.pause() await pilot.pause() assert screen._filter_cwd is None toggled_kwargs = [c.kwargs for c in mock_list.await_args_list] assert any(kw.get("cwd") is None for kw in toggled_kwargs), ( f"expected re-query with cwd=None, got {toggled_kwargs}" ) mock_list.reset_mock() scope_select.value = "cwd" await pilot.pause() await pilot.pause() assert screen._filter_cwd == starting_cwd back_kwargs = [c.kwargs for c in mock_list.await_args_list] assert any(kw.get("cwd") == starting_cwd for kw in back_kwargs), ( f"expected re-query with cwd={starting_cwd!r}, got {back_kwargs}" ) async def test_select_same_value_does_not_requery(self) -> None: """Setting the dropdown to its current value is a no-op.""" mock_list = AsyncMock(return_value=MOCK_THREADS) with ( patch("deepagents_code.sessions.list_threads", mock_list), _patch_columns(), ): app = _ThreadSelectorScopedTestApp(filter_cwd=None) async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) scope_select = screen.query_one("#thread-scope-select", Select) mock_list.reset_mock() # Re-setting to "all" (the active value) must not re-query. scope_select.value = "all" await pilot.pause() mock_list.assert_not_awaited() async def test_scope_change_persists_preference(self) -> None: """Switching the scope dropdown should persist the new preference.""" starting_cwd = "/home/user/project-a" mock_list = AsyncMock(return_value=MOCK_THREADS) mock_save = MagicMock(return_value=True) with ( patch("deepagents_code.sessions.list_threads", mock_list), _patch_columns(), patch( "deepagents_code.widgets.thread_selector._safe_cwd_string", return_value=starting_cwd, ), patch( "deepagents_code.model_config.save_thread_scope", mock_save, ), ): app = _ThreadSelectorScopedTestApp(filter_cwd=starting_cwd) async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) scope_select = screen.query_one("#thread-scope-select", Select) scope_select.value = "all" await pilot.pause() await pilot.pause() mock_save.assert_any_call("all") scope_select.value = "cwd" await pilot.pause() await pilot.pause() mock_save.assert_any_call("cwd") async def test_scope_persists_even_when_cwd_unresolvable(self) -> None: """Selecting "Current directory" persists "cwd" even if the cwd is gone. When `_safe_cwd_string()` returns `None`, the resolved filter stays `None` and the reload short-circuits, but the user's explicit "cwd" choice must still be persisted. This pins the intentional ordering of the persist call ahead of the `new_cwd == self._filter_cwd` early return. """ mock_list = AsyncMock(return_value=MOCK_THREADS) mock_save = MagicMock(return_value=True) with ( patch("deepagents_code.sessions.list_threads", mock_list), _patch_columns(), patch( "deepagents_code.widgets.thread_selector._safe_cwd_string", return_value=None, ), patch( "deepagents_code.model_config.save_thread_scope", mock_save, ), ): # Start unfiltered ("all"); the cwd is unresolvable below. app = _ThreadSelectorScopedTestApp(filter_cwd=None) async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._filter_cwd is None scope_select = screen.query_one("#thread-scope-select", Select) scope_select.value = "cwd" await pilot.pause() await pilot.pause() mock_save.assert_any_call("cwd") # Filter stays unfiltered (cwd unresolvable), yet the preference # was still persisted before the early return fired. assert screen._filter_cwd is None async def test_scope_save_failure_notifies(self) -> None: """A failed scope save should surface a warning notification.""" starting_cwd = "/home/user/project-a" mock_list = AsyncMock(return_value=MOCK_THREADS) mock_save = MagicMock(return_value=False) with ( patch("deepagents_code.sessions.list_threads", mock_list), _patch_columns(), patch( "deepagents_code.widgets.thread_selector._safe_cwd_string", return_value=starting_cwd, ), patch( "deepagents_code.model_config.save_thread_scope", mock_save, ), ): app = _ThreadSelectorScopedTestApp(filter_cwd=starting_cwd) async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) scope_select = screen.query_one("#thread-scope-select", Select) with patch.object(app, "notify") as mock_notify: scope_select.value = "all" await app.workers.wait_for_complete() await pilot.pause() mock_notify.assert_any_call( "Could not save scope preference", severity="warning" ) class TestThreadSelectorClickHandling: """Tests for mouse click handling.""" async def test_click_selects_thread(self) -> None: """Clicking a thread option should select and dismiss.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) from deepagents_code.widgets.thread_selector import ThreadOption assert len(screen._option_widgets) > 1, ( "Expected option widgets to be built" ) second = screen._option_widgets[1] second.post_message( ThreadOption.Clicked(second.thread_id, second.index) ) await pilot.pause() assert app.dismissed is True assert app.result == "def67890" _WEBBROWSER_OPEN = "deepagents_code.widgets._links.webbrowser.open" class TestThreadSelectorOnClickOpensLink: """Tests for `ThreadSelectorScreen.on_click` opening Rich-style hyperlinks.""" def test_click_on_link_opens_browser(self) -> None: """Clicking a Rich link should call `webbrowser.open`.""" screen = ThreadSelectorScreen(current_thread=None) event = MagicMock() event.style = Style(link="https://example.com") with patch(_WEBBROWSER_OPEN) as mock_open: screen.on_click(event) mock_open.assert_called_once_with("https://example.com") event.stop.assert_called_once() def test_click_without_link_is_noop(self) -> None: """Clicking on non-link text should not open the browser.""" screen = ThreadSelectorScreen(current_thread=None) event = MagicMock() event.style = Style() with patch(_WEBBROWSER_OPEN) as mock_open: screen.on_click(event) mock_open.assert_not_called() event.stop.assert_not_called() def test_click_with_browser_error_is_graceful(self) -> None: """Browser failure should not crash the widget.""" screen = ThreadSelectorScreen(current_thread=None) event = MagicMock() event.style = Style(link="https://example.com") with patch(_WEBBROWSER_OPEN, side_effect=OSError("no display")): screen.on_click(event) # should not raise event.stop.assert_not_called() class TestThreadSelectorPointer: """Tests for `ThreadSelectorScreen` link hover affordance.""" def test_hover_over_link_uses_pointer(self) -> None: """Hovering a Rich-style link sets a pointer cursor.""" screen = ThreadSelectorScreen(current_thread=None) event = MagicMock() event.style = Style(link="https://example.com") screen.on_mouse_move(event) assert screen.styles.pointer == "pointer" def test_hover_off_link_uses_default(self) -> None: """Hovering non-link text keeps the default cursor.""" screen = ThreadSelectorScreen(current_thread=None) event = MagicMock() event.style = Style() screen.on_mouse_move(event) assert screen.styles.pointer == "default" def test_leave_resets_pointer(self) -> None: """Leaving the selector resets the cursor to default.""" screen = ThreadSelectorScreen(current_thread=None) link_event = MagicMock() link_event.style = Style(link="https://example.com") screen.on_mouse_move(link_event) screen.on_leave() assert screen.styles.pointer == "default" class TestThreadSelectorBuildTitle: """Tests for _build_title with clickable thread ID.""" def test_no_current_thread(self) -> None: """Title without current thread should be plain text.""" screen = ThreadSelectorScreen(current_thread=None) assert screen._build_title() == "Select Thread" def test_current_thread_no_url(self) -> None: """Title with current thread but no URL should be a plain string.""" screen = ThreadSelectorScreen(current_thread="abc12345") title = screen._build_title() assert isinstance(title, str) assert "abc12345" in title def test_current_thread_with_url(self) -> None: """Title with a LangSmith URL should produce Content with a link.""" from textual.color import Color as TColor from textual.content import Content from textual.style import Style as TStyle screen = ThreadSelectorScreen(current_thread="abc12345") title = screen._build_title( thread_url="https://smith.langchain.com/p/t/abc12345" ) assert isinstance(title, Content) assert "abc12345" in title.plain spans = [ s for s in title._spans if isinstance(s.style, TStyle) and s.style.link ] assert len(spans) > 0 style = spans[0].style assert isinstance(style, TStyle) from deepagents_code.theme import DARK_COLORS assert style.foreground == TColor.parse(DARK_COLORS.primary) async def test_title_widget_has_id(self) -> None: """Title widget should be queryable by ID for URL updates.""" with _patch_list_threads(): app = ThreadSelectorTestApp(current_thread="abc12345") async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) title_widget = screen.query_one("#thread-title", Static) assert title_widget is not None class TestFetchThreadUrl: """Tests for _fetch_thread_url background worker.""" async def test_successful_url_updates_title(self) -> None: """Background worker should update the title with a clickable link.""" from textual.content import Content with ( _patch_list_threads(), patch( "deepagents_code.widgets.thread_selector.build_langsmith_thread_url", return_value="https://smith.langchain.com/p/t/abc12345", ), ): app = ThreadSelectorTestApp(current_thread="abc12345") async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) title_widget = screen.query_one("#thread-title", Static) content = title_widget._Static__content assert isinstance(content, Content) assert "abc12345" in content.plain async def test_timeout_leaves_title_unchanged(self) -> None: """Timeout during URL resolution should not crash or change the title.""" import time def _blocking(_tid: str) -> str: time.sleep(0.1) return "https://example.com" with ( _patch_list_threads(), patch( "deepagents_code.widgets.thread_selector._URL_FETCH_TIMEOUT", 0.01, ), patch( "deepagents_code.widgets.thread_selector.build_langsmith_thread_url", side_effect=_blocking, ), ): app = ThreadSelectorTestApp(current_thread="abc12345") async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) title_widget = screen.query_one("#thread-title", Static) assert isinstance(title_widget._Static__content, str) async def test_oserror_leaves_title_unchanged(self) -> None: """OSError during URL resolution should not crash or change the title.""" with ( _patch_list_threads(), patch( "deepagents_code.widgets.thread_selector.build_langsmith_thread_url", side_effect=OSError("network failure"), ), ): app = ThreadSelectorTestApp(current_thread="abc12345") async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) title_widget = screen.query_one("#thread-title", Static) assert isinstance(title_widget._Static__content, str) async def test_unexpected_exception_leaves_title_unchanged(self) -> None: """Unexpected exception should not crash the thread selector.""" with ( _patch_list_threads(), patch( "deepagents_code.widgets.thread_selector.build_langsmith_thread_url", side_effect=AttributeError("SDK changed"), ), ): app = ThreadSelectorTestApp(current_thread="abc12345") async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) title_widget = screen.query_one("#thread-title", Static) assert isinstance(title_widget._Static__content, str) async def test_none_url_leaves_title_unchanged(self) -> None: """When build returns None the title should remain a plain string.""" with ( _patch_list_threads(), patch( "deepagents_code.widgets.thread_selector.build_langsmith_thread_url", return_value=None, ), ): app = ThreadSelectorTestApp(current_thread="abc12345") async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) title_widget = screen.query_one("#thread-title", Static) content = title_widget._Static__content assert isinstance(content, str) assert "abc12345" in content class TestThreadSelectorColumnHeader: """Tests for the anchored column header.""" def test_header_contains_default_column_names(self) -> None: """Column header labels should contain visible column names.""" from deepagents_code.widgets.thread_selector import _format_header_label assert "Created" in _format_header_label("created_at") assert "Msgs" in _format_header_label("messages") assert "Updated" in _format_header_label("updated_at") assert "Prompt" in _format_header_label("initial_prompt") async def test_header_widget_is_mounted(self) -> None: """Column header widget should be present in the mounted screen.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) screen.query_one(".thread-list-header", Horizontal) async def test_header_stays_outside_scroll(self) -> None: """Header should be outside VerticalScroll (anchored, not scrollable).""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) header = screen.query_one(".thread-list-header", Horizontal) assert isinstance(header.parent, Vertical) async def test_timestamp_columns_share_width_with_rows(self) -> None: """Timestamp header cells should use the same width as row cells.""" from deepagents_code.widgets.thread_selector import ( _format_column_value, _format_header_label, ) with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) header = screen.query_one("#thread-header", Horizontal) row = screen._option_widgets[0] for key in ("created_at", "updated_at"): header_cell = header.query_one(f".thread-cell-{key}", Static) row_cell = row.query_one(f".thread-cell-{key}", Static) expected_width = ( max( cell_len(_format_header_label(key)), *( cell_len( _format_column_value( thread, key, relative_time=screen._relative_time, ) ) for thread in screen._filtered_threads ), ) + 1 ) assert header_cell.size.width == row_cell.size.width assert ( _style_scalar_value(header_cell.styles.width) == expected_width ) assert _style_scalar_value(row_cell.styles.width) == expected_width class TestThreadSelectorPromptOverflow: """Tests for prompt-cell overflow handling.""" async def test_prompt_cell_renders_ellipsis_when_constrained(self) -> None: """Prompt cells should use ellipsis instead of hard clipping.""" columns = { "thread_id": False, "messages": False, "created_at": False, "updated_at": True, "git_branch": False, "cwd": False, "initial_prompt": True, "agent_name": False, } thread = ThreadInfo(**MOCK_THREADS[0]) thread["initial_prompt"] = ( "This is a very long prompt that should be truncated " "with an ellipsis inside the prompt column" ) threads: list[ThreadInfo] = [thread] with _patch_list_threads(threads), _patch_columns(columns): app = ThreadSelectorTestApp(current_thread=None) async with app.run_test(size=(80, 24)) as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) row = screen._option_widgets[0] prompt_cell = row.query_one(".thread-cell-initial_prompt", Static) rendered = prompt_cell.render_line(0).text.rstrip() assert rendered.endswith("…") class TestThreadSelectorBranchOverflow: """Tests for git-branch overflow handling.""" async def test_branch_cell_renders_ellipsis_when_truncated(self) -> None: """Git branch cells should keep the ellipsis visible when clipped.""" columns = { "thread_id": False, "messages": False, "created_at": False, "updated_at": False, "git_branch": True, "cwd": False, "initial_prompt": False, "agent_name": False, } thread = ThreadInfo(**MOCK_THREADS[0]) thread["git_branch"] = "feature/very-long-branch-name" threads: list[ThreadInfo] = [thread] with _patch_list_threads(threads), _patch_columns(columns): app = ThreadSelectorTestApp(current_thread=None) async with app.run_test(size=(80, 24)) as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) row = screen._option_widgets[0] branch_cell = row.query_one(".thread-cell-git_branch", Static) rendered = branch_cell.render_line(0).text.rstrip() assert rendered.endswith("…") class TestThreadSelectorAutoWidthColumns: """Tests for shared widths on auto-sized columns.""" async def test_agent_name_column_uses_shared_width_capped_at_twelve(self) -> None: """Agent column should size to visible content up to the 12-char cap.""" from deepagents_code.widgets.thread_selector import ( _format_column_value, _format_header_label, ) columns = { "thread_id": False, "messages": False, "created_at": False, "updated_at": False, "git_branch": False, "cwd": False, "initial_prompt": False, "agent_name": True, } with _patch_list_threads(), _patch_columns(columns): app = ThreadSelectorTestApp(current_thread=None) async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) header = screen.query_one("#thread-header", Horizontal) row = screen._option_widgets[0] header_cell = header.query_one(".thread-cell-agent_name", Static) row_cell = row.query_one(".thread-cell-agent_name", Static) expected_width = ( max( cell_len(_format_header_label("agent_name")), *( cell_len( _format_column_value( thread, "agent_name", relative_time=screen._relative_time, ) ) for thread in screen._filtered_threads ), ) + 1 ) assert header_cell.size.width == row_cell.size.width assert _style_scalar_value(header_cell.styles.width) == expected_width assert _style_scalar_value(row_cell.styles.width) == expected_width assert ( _style_scalar_value(header_cell.styles.min_width) == expected_width ) assert _style_scalar_value(row_cell.styles.min_width) == expected_width assert expected_width <= 13 class TestThreadSelectorErrorHandling: """Tests for error handling when loading threads fails.""" async def test_list_threads_error_still_dismissable(self) -> None: """Database error should not crash; Escape still works.""" with patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, side_effect=OSError("database is locked"), ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert len(screen._threads) == 0 assert len(screen._option_widgets) == 0 # A failed load is still a completed load: the flag must flip so # the picker never strands on the "Loading threads..." placeholder. assert screen._disk_load_complete is True await pilot.press("escape") await pilot.pause() assert app.dismissed is True assert app.result is None async def test_unexpected_load_error_surfaces_and_completes(self) -> None: """A non-OSError/sqlite3 error must surface and not strand the UI.""" with patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, side_effect=ValueError("malformed row"), ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) # The catch-all handler marks the load complete and replaces the # loading placeholder with the error message instead of leaving a # perpetual "Loading threads..." spinner. assert screen._disk_load_complete is True with pytest.raises(NoMatches): screen.query_one("#thread-loading", Static) await pilot.press("escape") await pilot.pause() assert app.dismissed is True assert app.result is None class TestThreadSelectorLimit: """Tests for thread limit via get_thread_limit().""" async def test_custom_limit_is_forwarded(self) -> None: """get_thread_limit() return value should be forwarded to list_threads.""" with ( patch( "deepagents_code.sessions.get_thread_limit", return_value=5, ), _patch_list_threads() as mock_lt, ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() mock_lt.assert_awaited_once() call_kwargs = mock_lt.await_args.kwargs assert call_kwargs["limit"] == 5 assert call_kwargs["include_message_count"] is False assert call_kwargs["sort_by"] in {"updated", "created"} async def test_checkpoint_details_are_loaded_for_initial_render(self) -> None: """Visible checkpoint fields should be loaded before first non-cached render.""" threads_without_details: list[ThreadInfo] = [ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", } ] async def _populate( threads: list[ThreadInfo], *, include_message_count: bool, include_initial_prompt: bool, ) -> list[ThreadInfo]: await asyncio.sleep(0) assert include_message_count is True assert include_initial_prompt is True for thread in threads: thread["message_count"] = 9 thread["initial_prompt"] = "loaded prompt" return threads with ( patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, return_value=threads_without_details, ) as mock_lt, _patch_columns(), patch( "deepagents_code.sessions.populate_thread_checkpoint_details", new_callable=AsyncMock, side_effect=_populate, ) as mock_populate, ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() for _ in range(10): if mock_populate.await_count >= 1: break await pilot.pause(0.05) mock_lt.assert_awaited_once_with( limit=20, include_message_count=False, sort_by="updated", cwd=None, ) mock_populate.assert_awaited_once() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert len(screen._option_widgets) == 1 assert screen._threads[0]["message_count"] == 9 assert screen._threads[0]["initial_prompt"] == "loaded prompt" async def test_cached_counts_skip_background_population(self) -> None: """If cache fills counts before paint, background populate is skipped.""" threads_without_counts: list[ThreadInfo] = [ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", "initial_prompt": "prompt", "latest_checkpoint_id": "cp_1", } ] def _apply_cached(threads: list[ThreadInfo]) -> int: threads[0]["message_count"] = 11 return 1 with ( patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, return_value=threads_without_counts, ), patch( "deepagents_code.sessions.apply_cached_thread_message_counts", side_effect=_apply_cached, ) as mock_apply_cached, patch( "deepagents_code.sessions.populate_thread_checkpoint_details", new_callable=AsyncMock, ) as mock_populate, ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() await pilot.pause(0.1) mock_apply_cached.assert_called_once() mock_populate.assert_not_awaited() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._threads[0]["message_count"] == 11 class TestThreadSelectorCheckpointDetailErrors: """Tests for thread selector checkpoint-detail load error handling.""" async def test_unexpected_checkpoint_detail_error_logs_warning(self) -> None: """Unexpected checkpoint-load errors should be visible at warning level.""" screen = ThreadSelectorScreen( initial_threads=[ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", } ], filter_cwd=None, ) with ( patch( "deepagents_code.sessions.populate_thread_checkpoint_details", new_callable=AsyncMock, side_effect=RuntimeError("unexpected type mismatch"), ), patch( "deepagents_code.widgets.thread_selector.logger.warning" ) as mock_warning, ): await screen._load_checkpoint_details() mock_warning.assert_called_once() class TestThreadSelectorPrefetchedRows: """Tests for rendering with prefetched rows from startup cache.""" async def test_prefetched_rows_render_without_loading_state(self) -> None: """Prefetched rows should render immediately, then refresh from SQLite.""" prefetched: list[ThreadInfo] = [ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", "message_count": 5, } ] refreshed: list[ThreadInfo] = [ { "thread_id": "new12345", "agent_name": "my-agent", "updated_at": "2025-01-16T12:00:00", "message_count": 6, }, { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", "message_count": 5, }, ] app = ThreadSelectorTestApp(current_thread="abc12345") gate = asyncio.Event() async def _list_threads(*_args: object, **_kwargs: object) -> list[ThreadInfo]: await gate.wait() return refreshed with patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, side_effect=_list_threads, ) as mock_list_threads: async with app.run_test() as pilot: app.push_screen( ThreadSelectorScreen( current_thread="abc12345", thread_limit=20, initial_threads=prefetched, filter_cwd=None, ) ) await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert len(screen._option_widgets) == 1 with pytest.raises(NoMatches): screen.query_one("#thread-loading", Static) gate.set() for _ in range(10): if mock_list_threads.await_count >= 1 and len(screen._threads) == 2: break await pilot.pause(0.05) mock_list_threads.assert_awaited_once() assert mock_list_threads.await_args is not None kw = mock_list_threads.await_args.kwargs assert kw["limit"] == 20 assert kw["include_message_count"] is False assert kw["sort_by"] in {"updated", "created"} assert len(screen._threads) == 2 assert screen._threads[0]["thread_id"] == "new12345" async def test_prefetched_prompt_is_preserved_during_refresh(self) -> None: """Refreshing prefetched rows should not blank the prompt column first.""" prefetched: list[ThreadInfo] = [ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", "latest_checkpoint_id": "cp_1", "initial_prompt": "cached prompt", } ] refreshed: list[ThreadInfo] = [ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", "latest_checkpoint_id": "cp_1", } ] from deepagents_code import sessions sessions._initial_prompt_cache.clear() sessions._initial_prompt_cache["abc12345"] = ("cp_1", "cached prompt") try: with patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, return_value=refreshed, ): app = ThreadSelectorTestApp(current_thread="abc12345") async with app.run_test() as pilot: app.push_screen( ThreadSelectorScreen( current_thread="abc12345", thread_limit=20, initial_threads=prefetched, filter_cwd=None, ) ) await pilot.pause() await pilot.pause(0.1) screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._threads[0]["initial_prompt"] == "cached prompt" finally: sessions._initial_prompt_cache.clear() async def test_empty_prefetched_snapshot_still_refreshes(self) -> None: """An empty cached snapshot should still hydrate from SQLite in background.""" refreshed: list[ThreadInfo] = [ { "thread_id": "new12345", "agent_name": "my-agent", "updated_at": "2025-01-16T12:00:00", "message_count": 6, } ] app = ThreadSelectorTestApp(current_thread="abc12345") with patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, return_value=refreshed, ) as mock_list_threads: async with app.run_test() as pilot: app.push_screen( ThreadSelectorScreen( current_thread="abc12345", thread_limit=20, initial_threads=[], filter_cwd=None, ) ) await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) with pytest.raises(NoMatches): screen.query_one("#thread-loading", Static) for _ in range(10): if mock_list_threads.await_count >= 1 and len(screen._threads) == 1: break await pilot.pause(0.05) mock_list_threads.assert_awaited_once() assert mock_list_threads.await_args is not None kw = mock_list_threads.await_args.kwargs assert kw["limit"] == 20 assert kw["include_message_count"] is False assert kw["sort_by"] in {"updated", "created"} assert len(screen._threads) == 1 assert screen._threads[0]["thread_id"] == "new12345" async def test_empty_snapshot_shows_loading_until_disk_load_completes( self, ) -> None: """An empty snapshot must not claim "No threads found" while loading.""" refreshed: list[ThreadInfo] = [ { "thread_id": "new12345", "agent_name": "my-agent", "updated_at": "2025-01-16T12:00:00", "message_count": 6, } ] app = ThreadSelectorTestApp(current_thread="abc12345") gate = asyncio.Event() async def _list_threads(*_args: object, **_kwargs: object) -> list[ThreadInfo]: await gate.wait() return refreshed with patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, side_effect=_list_threads, ): async with app.run_test() as pilot: app.push_screen( ThreadSelectorScreen( current_thread="abc12345", thread_limit=20, initial_threads=[], filter_cwd=None, ) ) await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) # While the disk load is in flight, show the loading placeholder # rather than "No threads found". assert not screen._disk_load_complete screen.query_one("#thread-loading", Static) assert not screen._option_widgets gate.set() for _ in range(10): if len(screen._threads) == 1: break await pilot.pause(0.05) assert screen._disk_load_complete assert len(screen._option_widgets) == 1 async def test_empty_snapshot_resolves_to_no_threads_found(self) -> None: """An empty disk load must flip the placeholder to "No threads found".""" app = ThreadSelectorTestApp(current_thread="abc12345") gate = asyncio.Event() async def _list_threads(*_args: object, **_kwargs: object) -> list[ThreadInfo]: await gate.wait() return [] with patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, side_effect=_list_threads, ): async with app.run_test() as pilot: app.push_screen( ThreadSelectorScreen( current_thread="abc12345", thread_limit=20, initial_threads=[], filter_cwd=None, ) ) await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert not screen._disk_load_complete screen.query_one("#thread-loading", Static) gate.set() for _ in range(10): if screen._disk_load_complete: break await pilot.pause(0.05) # Once the load completes with zero rows, the loading placeholder # must resolve to the real empty state, not a perpetual spinner. assert screen._disk_load_complete assert not screen._option_widgets with pytest.raises(NoMatches): screen.query_one("#thread-loading", Static) empty = screen.query_one(".thread-empty", Static) assert "No threads found" in str(empty.content) def test_build_empty_state_reflects_disk_load_flag(self) -> None: """`_build_empty_state` chooses its message from `_disk_load_complete`.""" screen = ThreadSelectorScreen( current_thread="abc12345", thread_limit=20, initial_threads=[], filter_cwd=None, ) loading = screen._build_empty_state() assert loading.id == "thread-loading" assert "Loading threads..." in str(loading.content) screen._disk_load_complete = True resolved = screen._build_empty_state() assert resolved.id is None assert "No threads found" in str(resolved.content) class TestThreadSelectorInitialSortOrder: """Tests for initial sort order applied to prefetched rows.""" async def test_initial_threads_sorted_by_created_at_preference(self) -> None: """Prefetched rows should respect the user's sort preference on first render.""" # Threads ordered by updated_at (default cache order from list_threads) prefetched: list[ThreadInfo] = [ { "thread_id": "newer-updated", "agent_name": "agent", "updated_at": "2025-01-16T12:00:00", "created_at": "2025-01-10T08:00:00", }, { "thread_id": "older-updated", "agent_name": "agent", "updated_at": "2025-01-14T08:00:00", "created_at": "2025-01-15T10:00:00", }, ] import contextlib from deepagents_code.model_config import THREAD_COLUMN_DEFAULTS, ThreadConfig @contextlib.contextmanager def _patch_sort_created() -> Any: # noqa: ANN401 with ( patch( "deepagents_code.model_config.load_thread_columns", return_value=dict(THREAD_COLUMN_DEFAULTS), ), patch( "deepagents_code.model_config.load_thread_sort_order", return_value="created_at", ), patch( "deepagents_code.model_config.load_thread_config", return_value=ThreadConfig( columns=dict(THREAD_COLUMN_DEFAULTS), relative_time=True, sort_order="created_at", scope="cwd", ), ), ): yield gate = asyncio.Event() async def _list_threads(*_a: object, **_kw: object) -> list[ThreadInfo]: await gate.wait() return prefetched with ( patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, side_effect=_list_threads, ), _patch_sort_created(), ): app = ThreadSelectorTestApp(current_thread=None) async with app.run_test() as pilot: app.push_screen( ThreadSelectorScreen( current_thread=None, thread_limit=20, initial_threads=prefetched, filter_cwd=None, ) ) await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) # With sort by created_at, "older-updated" (created 2025-01-15) # should come before "newer-updated" (created 2025-01-10) assert len(screen._option_widgets) == 2 assert screen._option_widgets[0].thread_id == "older-updated" assert screen._option_widgets[1].thread_id == "newer-updated" gate.set() class TestThreadSelectorSearch: """Tests for fuzzy search filtering.""" async def test_search_filters_threads(self) -> None: """Typing in search should filter threads by initial prompt.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert len(screen._filtered_threads) == 3 screen._filter_text = "Hello" screen._update_filtered_list() assert len(screen._filtered_threads) == 1 assert screen._filtered_threads[0]["thread_id"] == "abc12345" def test_empty_search_returns_all(self) -> None: """Empty search text should return all threads.""" screen = ThreadSelectorScreen( current_thread=None, initial_threads=MOCK_THREADS, filter_cwd=None, ) screen._filter_text = "" screen._update_filtered_list() assert len(screen._filtered_threads) == 3 def test_search_by_thread_id(self) -> None: """Search should match thread IDs.""" screen = ThreadSelectorScreen( current_thread=None, initial_threads=MOCK_THREADS, filter_cwd=None, ) screen._filter_text = "def678" screen._update_filtered_list() assert len(screen._filtered_threads) == 1 assert screen._filtered_threads[0]["thread_id"] == "def67890" async def test_typing_in_search_filters_without_crashing(self) -> None: """Typing into the live search box should filter by initial prompt.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() for char in "fix": await pilot.press(char) await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert len(screen._filtered_threads) == 1 assert screen._filtered_threads[0]["thread_id"] == "def67890" assert app.dismissed is False def test_equal_match_scores_do_not_crash_sorting(self) -> None: """Fuzzy search should handle tied scores without comparing dict rows.""" threads: list[ThreadInfo] = [ { "thread_id": "thread-a", "agent_name": "agent", "updated_at": "2026-03-08T02:00:00+00:00", "created_at": "2026-03-08T01:00:00+00:00", "initial_prompt": "prompt one", }, { "thread_id": "thread-b", "agent_name": "agent", "updated_at": "2026-03-08T03:00:00+00:00", "created_at": "2026-03-08T01:30:00+00:00", "initial_prompt": "prompt two", }, ] screen = ThreadSelectorScreen( current_thread=None, initial_threads=threads, filter_cwd=None ) screen._filter_text = "p" screen._update_filtered_list() assert [thread["thread_id"] for thread in screen._filtered_threads] == [ "thread-b", "thread-a", ] async def test_filter_and_build_reuses_precomputed_widths(self) -> None: """Filter rebuilds should not recompute column widths twice.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) with ( patch.object( screen, "_compute_column_widths", wraps=screen._compute_column_widths, ) as mock_widths, patch.object(screen, "_update_help_widgets"), ): screen._filter_text = "fix" await screen._filter_and_build() assert mock_widths.call_count == 1 class TestThreadSelectorDelete: """Tests for ctrl+d delete functionality.""" async def test_delete_shows_confirmation(self) -> None: """Ctrl+D should show a delete confirmation overlay.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._confirming_delete is False await pilot.press("ctrl+d") await pilot.pause() await pilot.pause() assert screen._confirming_delete is True async def test_delete_confirmation_uses_screen_overlay(self) -> None: """Delete confirmation should open as a modal above the selector.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) await pilot.press("ctrl+d") await pilot.pause() await pilot.pause() assert isinstance(app.screen, DeleteThreadConfirmScreen) async def test_delete_escape_cancels(self) -> None: """Escape during delete confirmation should cancel.""" with _patch_list_threads(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) await pilot.press("ctrl+d") await pilot.pause() await pilot.pause() assert screen._confirming_delete is True assert isinstance(app.screen, DeleteThreadConfirmScreen) await pilot.press("escape") await pilot.pause() await pilot.pause() assert screen._confirming_delete is False assert app.screen is screen assert app.dismissed is False async def test_delete_keeps_selection_on_next_thread(self) -> None: """Deleting a row should move selection to the next visible thread.""" with ( _patch_list_threads(), patch( "deepagents_code.sessions.delete_thread", new_callable=AsyncMock, return_value=None, ), ): app = ThreadSelectorTestApp(current_thread=None) async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) await pilot.press("down") await pilot.pause() assert screen._selected_index == 1 assert screen._filtered_threads[1]["thread_id"] == "def67890" await pilot.press("ctrl+d") await pilot.pause() await pilot.pause() assert isinstance(app.screen, DeleteThreadConfirmScreen) await pilot.press("enter") await pilot.pause() await pilot.pause() assert app.screen is screen assert screen._selected_index == 1 selected_thread = screen._filtered_threads[screen._selected_index] assert selected_thread["thread_id"] == "ghi11111" async def test_delete_last_remaining_thread(self) -> None: """Deleting the only thread should leave an empty list without errors.""" single_thread: list[ThreadInfo] = [ { "thread_id": "only-one", "agent_name": "agent", "updated_at": "2025-01-15T10:30:00", } ] with ( _patch_list_threads(single_thread), patch( "deepagents_code.sessions.delete_thread", new_callable=AsyncMock, return_value=None, ), ): app = ThreadSelectorTestApp(current_thread=None) async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert len(screen._filtered_threads) == 1 await pilot.press("ctrl+d") await pilot.pause() await pilot.pause() assert isinstance(app.screen, DeleteThreadConfirmScreen) await pilot.press("enter") await pilot.pause() await pilot.pause() assert app.screen is screen assert screen._filtered_threads == [] assert screen._selected_index == 0 async def test_delete_last_item_in_list_moves_selection_backward(self) -> None: """Deleting the bottom thread should move selection to the previous one.""" with ( _patch_list_threads(), patch( "deepagents_code.sessions.delete_thread", new_callable=AsyncMock, return_value=None, ), ): app = ThreadSelectorTestApp(current_thread=None) async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) last_index = len(screen._filtered_threads) - 1 for _ in range(last_index): await pilot.press("down") await pilot.pause() assert screen._selected_index == last_index await pilot.press("ctrl+d") await pilot.pause() await pilot.pause() assert isinstance(app.screen, DeleteThreadConfirmScreen) await pilot.press("enter") await pilot.pause() await pilot.pause() assert app.screen is screen assert screen._selected_index < last_index assert screen._selected_index == len(screen._filtered_threads) - 1 async def test_delete_failure_keeps_thread_in_list(self) -> None: """DB failure during delete should keep the thread visible.""" with ( _patch_list_threads(), patch( "deepagents_code.sessions.delete_thread", new_callable=AsyncMock, side_effect=OSError("disk full"), ), ): app = ThreadSelectorTestApp(current_thread=None) async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) original_count = len(screen._filtered_threads) await pilot.press("ctrl+d") await pilot.pause() await pilot.pause() assert isinstance(app.screen, DeleteThreadConfirmScreen) await pilot.press("enter") await pilot.pause() await pilot.pause() assert app.screen is screen assert len(screen._filtered_threads) == original_count class TestThreadSelectorColumnConfig: """Tests for column visibility configuration.""" def test_default_columns(self) -> None: """Default column config should match THREAD_COLUMN_DEFAULTS.""" from deepagents_code.model_config import THREAD_COLUMN_DEFAULTS with _patch_columns(): screen = ThreadSelectorScreen(current_thread=None) assert screen._columns == THREAD_COLUMN_DEFAULTS async def test_switch_toggles_column_and_persists(self) -> None: """Clicking a column switch should hide the column and save the choice.""" with ( _patch_list_threads(), _patch_columns(), patch( "deepagents_code.model_config.save_thread_columns", return_value=True, ) as mock_save, ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._columns["initial_prompt"] is True prompt_switch = screen.query_one( f"#{screen._switch_id('initial_prompt')}", Checkbox, ) prompt_switch.value = False await pilot.pause() assert screen._columns["initial_prompt"] is False mock_save.assert_called() assert mock_save.call_args.args[0]["initial_prompt"] is False async def test_enabling_prompt_column_triggers_prompt_load(self) -> None: """Turning on the prompt column should fetch missing prompt data.""" threads_without_prompt: list[ThreadInfo] = [ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", "message_count": 5, } ] columns = { "thread_id": True, "messages": True, "created_at": True, "updated_at": True, "git_branch": True, "cwd": False, "initial_prompt": False, "agent_name": True, } async def _populate( rows: list[ThreadInfo], *, include_message_count: bool, include_initial_prompt: bool, ) -> list[ThreadInfo]: await asyncio.sleep(0) assert include_message_count is False assert include_initial_prompt is True rows[0]["initial_prompt"] = "loaded prompt" return rows with ( _patch_list_threads(threads_without_prompt), _patch_columns(columns), patch( "deepagents_code.sessions.populate_thread_checkpoint_details", new_callable=AsyncMock, side_effect=_populate, ) as mock_populate, ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert mock_populate.await_count == 0 prompt_switch = screen.query_one( f"#{screen._switch_id('initial_prompt')}", Checkbox, ) prompt_switch.value = True for _ in range(10): if mock_populate.await_count >= 1: break await pilot.pause(0.05) mock_populate.assert_awaited_once() assert screen._threads[0]["initial_prompt"] == "loaded prompt" class TestThreadSelectorControlsOverflow: """Tests for short-window overflow handling in the options pane.""" async def test_options_overflow_shows_ellipsis(self) -> None: """A short options pane should show an ellipsis when controls are hidden.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test(size=(100, 12)) as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) scroll = screen.query_one("#thread-controls-scroll", VerticalScroll) hint = screen.query_one("#thread-controls-overflow", Static) assert scroll.max_scroll_y > 0 assert hint.display is True async def test_options_overflow_ellipsis_hides_at_end(self) -> None: """The options ellipsis should disappear after scrolling to the end.""" with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test(size=(100, 12)) as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) scroll = screen.query_one("#thread-controls-scroll", VerticalScroll) hint = screen.query_one("#thread-controls-overflow", Static) scroll.scroll_end(animate=False) await pilot.pause() assert scroll.scroll_y == scroll.max_scroll_y assert hint.display is False def _app_test_double(app: DeepAgentsApp) -> Any: # noqa: ANN401 """Return `app` as dynamic for test-only Textual method patching. Textual apps expose real methods at type-check time, but these tests replace them with `MagicMock`/`AsyncMock` instances to isolate thread-switching logic. Keeping the dynamic escape hatch here avoids broad casts at each call site. """ return app def _get_widget_text(widget: Static) -> str: """Extract text content from a message widget. Args: widget: A message widget (e.g., `AppMessage`). Returns: The text content of the widget. """ return str(getattr(widget, "_content", "")) class TestResumeThread: """Tests for DeepAgentsApp._resume_thread.""" async def test_no_agent_shows_error(self) -> None: """_resume_thread with no agent should show an error message.""" app = DeepAgentsApp() mounted: list[Static] = [] _app_test_double(app)._mount_message = AsyncMock( side_effect=lambda w: mounted.append(w) ) app._agent = None await app._resume_thread("thread-123") assert len(mounted) == 1 assert "no active agent" in _get_widget_text(mounted[0]) async def test_no_session_state_shows_error(self) -> None: """_resume_thread with no session state should show an error message.""" app = DeepAgentsApp() mounted: list[Static] = [] _app_test_double(app)._mount_message = AsyncMock( side_effect=lambda w: mounted.append(w) ) app._agent = MagicMock() app._session_state = None await app._resume_thread("thread-123") assert len(mounted) == 1 assert "no active session" in _get_widget_text(mounted[0]) async def test_already_switching_shows_message(self) -> None: """_resume_thread should reject concurrent thread switches.""" app = DeepAgentsApp() mounted: list[Static] = [] _app_test_double(app)._mount_message = AsyncMock( side_effect=lambda w: mounted.append(w) ) app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "thread-123" app._thread_switching = True await app._resume_thread("thread-999") assert len(mounted) == 1 assert "already in progress" in _get_widget_text(mounted[0]) async def test_already_on_thread_shows_message(self) -> None: """_resume_thread when already on the thread should show info message.""" app = DeepAgentsApp() mounted: list[Static] = [] _app_test_double(app)._mount_message = AsyncMock( side_effect=lambda w: mounted.append(w) ) offer_cwd_switch = AsyncMock(return_value="continue") _app_test_double(app)._offer_thread_cwd_switch = offer_cwd_switch app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "thread-123" await app._resume_thread("thread-123") offer_cwd_switch.assert_awaited_once_with( "thread-123", restart_server=True, ) assert len(mounted) == 1 assert "Already on thread" in _get_widget_text(mounted[0]) async def test_already_on_thread_reports_cwd_switch( self, tmp_path: Path, ) -> None: """Same-thread resumes should not say unchanged after cwd switches.""" current = tmp_path / "current" target = tmp_path / "target" current.mkdir() target.mkdir() app = DeepAgentsApp(cwd=current) mounted: list[Static] = [] _app_test_double(app)._mount_message = AsyncMock( side_effect=lambda w: mounted.append(w) ) async def offer_cwd_switch( # noqa: RUF029 # must be async: awaited as _offer_thread_cwd_switch thread_id: str, *, restart_server: bool, ) -> str: assert thread_id == "thread-123" assert restart_server is True app._cwd = str(target) return "continue" _app_test_double(app)._offer_thread_cwd_switch = offer_cwd_switch app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "thread-123" await app._resume_thread("thread-123") assert len(mounted) == 1 assert "Switched to thread directory" in _get_widget_text(mounted[0]) assert "Already on thread" not in _get_widget_text(mounted[0]) async def test_successful_switch_updates_ids(self) -> None: """Successful _resume_thread should update thread IDs and load history.""" from textual.css.query import NoMatches as _NoMatches app = DeepAgentsApp(thread_id="old-thread") app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "old-thread" app._pending_messages = MagicMock() app._queued_widgets = MagicMock() _app_test_double(app)._clear_messages = AsyncMock() _app_test_double(app)._update_status = MagicMock() mock_payload = MagicMock() mock_payload.messages = [] mock_payload.context_tokens = 0 _app_test_double(app)._fetch_thread_history_data = AsyncMock( return_value=mock_payload ) _app_test_double(app)._load_thread_history = AsyncMock() _app_test_double(app)._mount_message = AsyncMock() _app_test_double(app).query_one = MagicMock(side_effect=_NoMatches()) await app._resume_thread("new-thread") assert app._lc_thread_id == "new-thread" assert app._session_state.thread_id == "new-thread" app._pending_messages.clear.assert_called_once() app._queued_widgets.clear.assert_called_once() _app_test_double(app)._clear_messages.assert_awaited_once() assert app._context_tokens == 0 app._fetch_thread_history_data.assert_awaited_once_with("new-thread") app._load_thread_history.assert_awaited_once_with( thread_id="new-thread", preloaded_payload=mock_payload, ) @staticmethod def _switch_app() -> DeepAgentsApp: from textual.css.query import NoMatches as _NoMatches app = DeepAgentsApp(thread_id="old-thread") app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "old-thread" app._pending_messages = MagicMock() app._queued_widgets = MagicMock() _app_test_double(app)._clear_messages = AsyncMock() _app_test_double(app)._update_status = MagicMock() mock_payload = MagicMock() mock_payload.messages = [] mock_payload.context_tokens = 0 _app_test_double(app)._fetch_thread_history_data = AsyncMock( return_value=mock_payload ) _app_test_double(app)._load_thread_history = AsyncMock() _app_test_double(app)._mount_message = AsyncMock() _app_test_double(app).query_one = MagicMock(side_effect=_NoMatches()) return app async def test_switch_arms_model_adoption(self) -> None: """An in-session thread switch arms session-only model adoption. Mirrors launch-time `-r`: `_load_thread_history` (real, here mocked) consumes the flag and adopts the switched-to thread's model. """ app = self._switch_app() await app._resume_thread("new-thread") assert app._should_adopt_resumed_model is True async def test_explicit_model_suppresses_switch_adoption(self) -> None: """`--model` keeps the session pinned across in-session switches.""" app = self._switch_app() app._model_explicitly_set = True await app._resume_thread("new-thread") assert app._should_adopt_resumed_model is False async def test_failure_restores_previous_thread_ids(self) -> None: """If _clear_messages raises, thread IDs should be restored.""" from textual.css.query import NoMatches as _NoMatches app = DeepAgentsApp(thread_id="old-thread") app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "old-thread" app._pending_messages = MagicMock() app._queued_widgets = MagicMock() from deepagents_code.app import _ThreadHistoryPayload mock_payload = _ThreadHistoryPayload( messages=[], context_tokens=0, model_spec="" ) _app_test_double(app)._fetch_thread_history_data = AsyncMock( return_value=mock_payload ) _app_test_double(app)._clear_messages = AsyncMock( side_effect=RuntimeError("UI gone") ) _app_test_double(app)._update_status = MagicMock() _app_test_double(app)._mount_message = AsyncMock() _app_test_double(app).query_one = MagicMock(side_effect=_NoMatches()) await app._resume_thread("new-thread") assert app._lc_thread_id == "old-thread" assert app._session_state.thread_id == "old-thread" assert any( "Failed to switch" in _get_widget_text(call.args[0]) for call in _app_test_double(app)._mount_message.call_args_list ) _app_test_double(app)._update_status.assert_any_call("") async def test_failure_during_load_history_restores_ids(self) -> None: """If _load_thread_history raises, thread IDs should be rolled back.""" from textual.css.query import NoMatches as _NoMatches app = DeepAgentsApp(thread_id="old-thread") app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "old-thread" app._pending_messages = MagicMock() app._queued_widgets = MagicMock() mock_payload = MagicMock() mock_payload.messages = [] mock_payload.context_tokens = 0 _app_test_double(app)._fetch_thread_history_data = AsyncMock( return_value=mock_payload ) _app_test_double(app)._clear_messages = AsyncMock() _app_test_double(app)._update_status = MagicMock() _app_test_double(app)._load_thread_history = AsyncMock( side_effect=[RuntimeError("checkpoint corrupt"), None] ) _app_test_double(app)._mount_message = AsyncMock() _app_test_double(app).query_one = MagicMock(side_effect=_NoMatches()) await app._resume_thread("new-thread") assert app._lc_thread_id == "old-thread" assert app._session_state.thread_id == "old-thread" assert any( "Failed to switch" in _get_widget_text(call.args[0]) for call in _app_test_double(app)._mount_message.call_args_list ) async def test_prefetch_failure_keeps_current_thread_visible(self) -> None: """Failed prefetch should not clear current conversation state.""" app = DeepAgentsApp(thread_id="old-thread") app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "old-thread" fetch_history_mock = AsyncMock( side_effect=RuntimeError("checkpoint read failed") ) clear_messages_mock = AsyncMock() mount_message_mock = AsyncMock() _app_test_double(app)._fetch_thread_history_data = fetch_history_mock _app_test_double(app)._clear_messages = clear_messages_mock _app_test_double(app)._mount_message = mount_message_mock await app._resume_thread("new-thread") assert app._session_state.thread_id == "old-thread" assert app._lc_thread_id == "old-thread" clear_messages_mock.assert_not_awaited() assert any( "Failed to switch" in _get_widget_text(call.args[0]) for call in mount_message_mock.call_args_list ) async def test_prefetch_failure_clears_switch_lock_and_restores_input(self) -> None: """Prefetch failures should release switch lock and restore input state.""" app = DeepAgentsApp(thread_id="old-thread") app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "old-thread" app._chat_input = MagicMock() _app_test_double(app)._mount_message = AsyncMock() with patch.object( app, "_fetch_thread_history_data", new_callable=AsyncMock, side_effect=RuntimeError("checkpoint read failed"), ): await app._resume_thread("new-thread") assert app._thread_switching is False app._chat_input.set_cursor_active.assert_any_call(active=False) app._chat_input.set_cursor_active.assert_any_call(active=True) async def test_double_failure_surfaces_restore_failure_hint(self) -> None: """If rollback restore fails, user-facing error should mention it.""" from textual.css.query import NoMatches as _NoMatches app = DeepAgentsApp(thread_id="old-thread") app._agent = MagicMock() app._session_state = MagicMock() app._session_state.thread_id = "old-thread" app._pending_messages = MagicMock() app._queued_widgets = MagicMock() _app_test_double(app)._fetch_thread_history_data = AsyncMock(return_value=[]) _app_test_double(app)._clear_messages = AsyncMock() _app_test_double(app)._load_thread_history = AsyncMock( side_effect=RuntimeError("checkpoint corrupt") ) mount_message_mock = AsyncMock() _app_test_double(app)._mount_message = mount_message_mock _app_test_double(app).query_one = MagicMock(side_effect=_NoMatches()) with patch.object(app, "_update_status") as update_status_mock: await app._resume_thread("new-thread") assert any( "Previous thread history could not be restored" in _get_widget_text(call.args[0]) for call in mount_message_mock.call_args_list ) update_status_mock.assert_any_call("") class TestFetchThreadHistoryData: """Tests for DeepAgentsApp._fetch_thread_history_data.""" async def test_returns_empty_when_agent_missing(self) -> None: """No active agent should return an empty history payload.""" app = DeepAgentsApp() app._agent = None payload = await app._fetch_thread_history_data("tid-1") assert payload.messages == [] assert payload.context_tokens == 0 async def test_returns_empty_when_state_missing(self) -> None: """Missing checkpoint state should return an empty history payload.""" app = DeepAgentsApp() app._agent = MagicMock() app._agent.aget_state = AsyncMock(return_value=None) payload = await app._fetch_thread_history_data("tid-1") assert payload.messages == [] assert payload.context_tokens == 0 app._agent.aget_state.assert_awaited_once_with( {"configurable": {"thread_id": "tid-1"}} ) async def test_returns_empty_when_messages_missing(self) -> None: """State with no messages should return an empty history payload.""" app = DeepAgentsApp() app._agent = MagicMock() state = MagicMock() state.values = {} app._agent.aget_state = AsyncMock(return_value=state) payload = await app._fetch_thread_history_data("tid-1") assert payload.messages == [] assert payload.context_tokens == 0 async def test_offloads_conversion_to_thread(self) -> None: """Message conversion should be offloaded via `asyncio.to_thread`.""" from deepagents_code.widgets.message_store import MessageData, MessageType app = DeepAgentsApp() app._agent = MagicMock() raw_messages = [object()] state = MagicMock() state.values = {"messages": raw_messages} app._agent.aget_state = AsyncMock(return_value=state) converted = [MessageData(type=MessageType.USER, content="hello")] with patch( "deepagents_code.app.asyncio.to_thread", new_callable=AsyncMock, return_value=converted, ) as to_thread_mock: payload = await app._fetch_thread_history_data("tid-1") assert payload.messages == converted to_thread_mock.assert_awaited_once() await_args = to_thread_mock.await_args assert await_args is not None assert await_args.args[1] == raw_messages async def test_extracts_nonzero_context_tokens(self) -> None: """Persisted _context_tokens should propagate to the payload.""" from deepagents_code.widgets.message_store import MessageData, MessageType app = DeepAgentsApp() app._agent = MagicMock() raw_messages = [object()] state = MagicMock() state.values = {"messages": raw_messages, "_context_tokens": 12000} app._agent.aget_state = AsyncMock(return_value=state) converted = [MessageData(type=MessageType.USER, content="hello")] with patch( "deepagents_code.app.asyncio.to_thread", new_callable=AsyncMock, return_value=converted, ): payload = await app._fetch_thread_history_data("tid-1") assert payload.context_tokens == 12000 async def test_extracts_model_spec(self) -> None: """Persisted `_model_spec` should propagate to the payload.""" from deepagents_code.widgets.message_store import MessageData, MessageType app = DeepAgentsApp() app._agent = MagicMock() raw_messages = [object()] state = MagicMock() state.values = { "messages": raw_messages, "_model_spec": "anthropic:claude-sonnet-4-5", } app._agent.aget_state = AsyncMock(return_value=state) converted = [MessageData(type=MessageType.USER, content="hello")] with patch( "deepagents_code.app.asyncio.to_thread", new_callable=AsyncMock, return_value=converted, ): payload = await app._fetch_thread_history_data("tid-1") assert payload.model_spec == "anthropic:claude-sonnet-4-5" async def test_missing_model_spec_is_empty(self) -> None: """A legacy thread without `_model_spec` yields `model_spec=""`.""" from deepagents_code.widgets.message_store import MessageData, MessageType app = DeepAgentsApp() app._agent = MagicMock() raw_messages = [object()] state = MagicMock() state.values = {"messages": raw_messages} app._agent.aget_state = AsyncMock(return_value=state) converted = [MessageData(type=MessageType.USER, content="hello")] with patch( "deepagents_code.app.asyncio.to_thread", new_callable=AsyncMock, return_value=converted, ): payload = await app._fetch_thread_history_data("tid-1") assert payload.model_spec == "" async def test_none_context_tokens_coerced_to_zero(self) -> None: """`_context_tokens: None` in checkpoint should coerce to 0.""" from deepagents_code.widgets.message_store import MessageData, MessageType app = DeepAgentsApp() app._agent = MagicMock() raw_messages = [object()] state = MagicMock() state.values = {"messages": raw_messages, "_context_tokens": None} app._agent.aget_state = AsyncMock(return_value=state) converted = [MessageData(type=MessageType.USER, content="hello")] with patch( "deepagents_code.app.asyncio.to_thread", new_callable=AsyncMock, return_value=converted, ): payload = await app._fetch_thread_history_data("tid-1") assert payload.context_tokens == 0 class TestLoadThreadHistory: """Tests for DeepAgentsApp._load_thread_history.""" async def test_preloaded_history_skips_fetch_and_schedules_link(self) -> None: """Preloaded history should render without state fetch round-trip.""" from deepagents_code.widgets.message_store import MessageData, MessageType app = DeepAgentsApp(thread_id="tid-1") app._agent = MagicMock() fetch_history_mock = AsyncMock() mount_message_mock = AsyncMock() schedule_link_mock = MagicMock() _app_test_double(app)._fetch_thread_history_data = fetch_history_mock _app_test_double(app)._remove_spacer = AsyncMock() _app_test_double(app)._mount_message = mount_message_mock _app_test_double(app)._schedule_thread_message_link = schedule_link_mock _app_test_double(app).set_timer = MagicMock() messages_container = MagicMock() messages_container.mount = AsyncMock() _app_test_double(app).query_one = MagicMock(return_value=messages_container) from deepagents_code.app import _ThreadHistoryPayload preloaded = _ThreadHistoryPayload( messages=[MessageData(type=MessageType.USER, content="hello")], context_tokens=0, model_spec="", ) await app._load_thread_history(thread_id="tid-1", preloaded_payload=preloaded) fetch_history_mock.assert_not_awaited() messages_container.mount.assert_awaited_once() mount_message_mock.assert_awaited_once() schedule_link_mock.assert_called_once() async def test_resume_seeds_context_tokens_from_state(self) -> None: """Resuming a thread with persisted tokens should seed the local cache.""" from deepagents_code.widgets.message_store import MessageData, MessageType app = DeepAgentsApp(thread_id="tid-1") mount_message_mock = AsyncMock() schedule_link_mock = MagicMock() _app_test_double(app)._remove_spacer = AsyncMock() _app_test_double(app)._mount_message = mount_message_mock _app_test_double(app)._schedule_thread_message_link = schedule_link_mock _app_test_double(app).set_timer = MagicMock() messages_container = MagicMock() messages_container.mount = AsyncMock() _app_test_double(app).query_one = MagicMock(return_value=messages_container) from deepagents_code.app import _ThreadHistoryPayload preloaded = _ThreadHistoryPayload( messages=[MessageData(type=MessageType.USER, content="hello")], context_tokens=8500, model_spec="", ) await app._load_thread_history(thread_id="tid-1", preloaded_payload=preloaded) assert app._context_tokens == 8500 async def test_zero_context_tokens_does_not_overwrite_cache(self) -> None: """Loading a payload with 0 tokens should not reset an existing cache.""" from deepagents_code.widgets.message_store import MessageData, MessageType app = DeepAgentsApp(thread_id="tid-1") app._context_tokens = 5000 # pre-existing cache from a previous thread mount_message_mock = AsyncMock() schedule_link_mock = MagicMock() _app_test_double(app)._remove_spacer = AsyncMock() _app_test_double(app)._mount_message = mount_message_mock _app_test_double(app)._schedule_thread_message_link = schedule_link_mock _app_test_double(app).set_timer = MagicMock() messages_container = MagicMock() messages_container.mount = AsyncMock() _app_test_double(app).query_one = MagicMock(return_value=messages_container) from deepagents_code.app import _ThreadHistoryPayload preloaded = _ThreadHistoryPayload( messages=[MessageData(type=MessageType.USER, content="hello")], context_tokens=0, model_spec="", ) await app._load_thread_history(thread_id="tid-1", preloaded_payload=preloaded) assert app._context_tokens == 5000 async def test_fallback_fetch_path_used_without_preloaded_data(self) -> None: """History should be fetched when preloaded data is not provided.""" from deepagents_code.widgets.message_store import MessageData, MessageType app = DeepAgentsApp(thread_id="tid-1") app._agent = MagicMock() from deepagents_code.app import _ThreadHistoryPayload fetched = _ThreadHistoryPayload( messages=[MessageData(type=MessageType.USER, content="hello")], context_tokens=0, model_spec="", ) fetch_history_mock = AsyncMock(return_value=fetched) mount_message_mock = AsyncMock() schedule_link_mock = MagicMock() _app_test_double(app)._fetch_thread_history_data = fetch_history_mock _app_test_double(app)._remove_spacer = AsyncMock() _app_test_double(app)._mount_message = mount_message_mock _app_test_double(app)._schedule_thread_message_link = schedule_link_mock _app_test_double(app).set_timer = MagicMock() messages_container = MagicMock() messages_container.mount = AsyncMock() _app_test_double(app).query_one = MagicMock(return_value=messages_container) await app._load_thread_history(thread_id="tid-1") fetch_history_mock.assert_awaited_once_with("tid-1") messages_container.mount.assert_awaited_once() mount_message_mock.assert_awaited_once() schedule_link_mock.assert_called_once() async def test_assistant_render_failure_does_not_abort_history_load(self) -> None: """A single assistant render failure should not abort history loading.""" from deepagents_code.widgets.message_store import MessageData, MessageType from deepagents_code.widgets.messages import AssistantMessage app = DeepAgentsApp(thread_id="tid-1") app._agent = MagicMock() mount_message_mock = AsyncMock() schedule_link_mock = MagicMock() _app_test_double(app)._remove_spacer = AsyncMock() _app_test_double(app)._mount_message = mount_message_mock _app_test_double(app)._schedule_thread_message_link = schedule_link_mock _app_test_double(app).set_timer = MagicMock() messages_container = MagicMock() messages_container.mount = AsyncMock() _app_test_double(app).query_one = MagicMock(return_value=messages_container) from deepagents_code.app import _ThreadHistoryPayload preloaded = _ThreadHistoryPayload( messages=[ MessageData(type=MessageType.ASSISTANT, content="ok"), MessageData(type=MessageType.ASSISTANT, content="fail"), ], context_tokens=0, model_spec="", ) def _set_content_side_effect(content: str) -> None: if content == "fail": msg = "markdown update failed" raise RuntimeError(msg) with patch.object( AssistantMessage, "set_content", new_callable=AsyncMock, side_effect=_set_content_side_effect, ) as set_content_mock: await app._load_thread_history( thread_id="tid-1", preloaded_payload=preloaded ) assert set_content_mock.await_count == 2 mount_message_mock.assert_awaited_once() schedule_link_mock.assert_called_once() async def test_early_return_without_thread_id_logs_debug(self) -> None: """Missing thread ID should early-return with a debug log entry.""" app = DeepAgentsApp() app._lc_thread_id = None app._agent = MagicMock() with patch("deepagents_code.app.logger.debug") as debug_mock: await app._load_thread_history() debug_mock.assert_called_once_with( "Skipping history load: no thread ID available" ) async def test_early_return_without_agent_logs_debug(self) -> None: """No agent and no preloaded payload should early-return with debug log.""" app = DeepAgentsApp(thread_id="tid-1") app._agent = None with patch("deepagents_code.app.logger.debug") as debug_mock: await app._load_thread_history(thread_id="tid-1") debug_mock.assert_called_once_with( "Skipping history load for %s: no active agent and no preloaded data", "tid-1", ) class TestResumeModelAdoption: """Tests for adopting a resumed thread's persisted model on load.""" @staticmethod def _make_app() -> DeepAgentsApp: app = DeepAgentsApp(thread_id="tid-1") _app_test_double(app)._remove_spacer = AsyncMock() _app_test_double(app)._mount_message = AsyncMock() _app_test_double(app)._schedule_thread_message_link = MagicMock() _app_test_double(app).set_timer = MagicMock() messages_container = MagicMock() messages_container.mount = AsyncMock() _app_test_double(app).query_one = MagicMock(return_value=messages_container) return app @staticmethod def _payload( model_spec: str, *, with_messages: bool = True ) -> _ThreadHistoryPayload: from deepagents_code.widgets.message_store import MessageData, MessageType messages = ( [MessageData(type=MessageType.USER, content="hello")] if with_messages else [] ) return _ThreadHistoryPayload( messages=messages, context_tokens=0, model_spec=model_spec, ) async def test_adopts_persisted_model_session_only(self) -> None: """Armed flag + persisted spec switches the model without persisting it.""" app = self._make_app() switch_mock = AsyncMock() _app_test_double(app)._switch_model = switch_mock app._should_adopt_resumed_model = True await app._load_thread_history( thread_id="tid-1", preloaded_payload=self._payload("anthropic:claude-sonnet-4-5"), ) switch_mock.assert_awaited_once() call = switch_mock.await_args assert call is not None assert call.args[0] == "anthropic:claude-sonnet-4-5" assert call.kwargs["persist"] is False assert call.kwargs["announce_unchanged"] is False assert call.kwargs["from_resume"] is True # One-shot: the flag is consumed so later loads don't re-adopt. assert app._should_adopt_resumed_model is False async def test_no_adoption_when_flag_unset(self) -> None: """Without the armed flag (e.g. in-session switch), model is untouched.""" app = self._make_app() switch_mock = AsyncMock() _app_test_double(app)._switch_model = switch_mock app._should_adopt_resumed_model = False await app._load_thread_history( thread_id="tid-1", preloaded_payload=self._payload("anthropic:claude-sonnet-4-5"), ) switch_mock.assert_not_awaited() async def test_no_adoption_for_legacy_thread_without_spec(self) -> None: """Armed flag but no persisted spec (legacy thread) leaves the model alone.""" app = self._make_app() switch_mock = AsyncMock() _app_test_double(app)._switch_model = switch_mock app._should_adopt_resumed_model = True await app._load_thread_history( thread_id="tid-1", preloaded_payload=self._payload(""), ) switch_mock.assert_not_awaited() # The flag is consumed even without a spec, so a later in-session # thread switch can't accidentally re-trigger adoption. assert app._should_adopt_resumed_model is False async def test_consumes_flag_even_when_history_empty(self) -> None: """An empty-history resume still adopts and consumes the one-shot flag. Adoption runs before the empty-`messages` early return, so the flag can't leak into a later in-session `/threads` switch. """ app = self._make_app() switch_mock = AsyncMock() _app_test_double(app)._switch_model = switch_mock app._should_adopt_resumed_model = True await app._load_thread_history( thread_id="tid-1", preloaded_payload=self._payload( "anthropic:claude-sonnet-4-5", with_messages=False ), ) switch_mock.assert_awaited_once() assert app._should_adopt_resumed_model is False class TestResumeAdoptionFailureMessage: """Tests for DeepAgentsApp._mount_resume_adoption_failure.""" async def test_names_desired_reason_and_fallback(self) -> None: """The notice states the desired model, the reason, and the fallback.""" app = DeepAgentsApp() app._model_override = "openai:gpt-5.1" # the model we fall back to mounted: list[Static] = [] _app_test_double(app)._mount_message = AsyncMock( side_effect=lambda w: mounted.append(w) ) await app._mount_resume_adoption_failure( "anthropic:claude-opus-4-8", "missing credentials for 'anthropic'", hint="Run `/auth` to use it.", ) assert len(mounted) == 1 text = _get_widget_text(mounted[0]) assert "anthropic:claude-opus-4-8" in text # desired assert "missing credentials" in text # reason assert "openai:gpt-5.1" in text # fallback assert "Run `/auth` to use it." in text # hint async def test_omits_fallback_when_no_current_model(self) -> None: """With no resolvable current model, the fallback clause is dropped.""" from deepagents_code.config import settings app = DeepAgentsApp() app._model_override = None mounted: list[Static] = [] _app_test_double(app)._mount_message = AsyncMock( side_effect=lambda w: mounted.append(w) ) with ( patch.object(settings, "model_provider", ""), patch.object(settings, "model_name", ""), ): await app._mount_resume_adoption_failure( "anthropic:claude-opus-4-8", "the model could not be initialized" ) text = _get_widget_text(mounted[0]) assert "continuing on" not in text assert "anthropic:claude-opus-4-8" in text class TestEffectiveModelSpec: """Tests for DeepAgentsApp._effective_model_spec.""" async def test_prefers_session_override(self) -> None: """A `/model` override wins over the startup default in `settings`.""" from deepagents_code.config import settings app = DeepAgentsApp() app._model_override = "openai:gpt-5.1" with ( patch.object(settings, "model_provider", "anthropic"), patch.object(settings, "model_name", "claude-sonnet-4-5"), ): assert app._effective_model_spec() == "openai:gpt-5.1" async def test_falls_back_to_settings_spec(self) -> None: """With no override, the resolved `provider:model` from settings is used.""" from deepagents_code.config import settings app = DeepAgentsApp() app._model_override = None with ( patch.object(settings, "model_provider", "anthropic"), patch.object(settings, "model_name", "claude-sonnet-4-5"), ): assert app._effective_model_spec() == "anthropic:claude-sonnet-4-5" async def test_none_when_spec_incomplete(self) -> None: """No override and a blank model yields `None` (no malformed spec).""" from deepagents_code.config import settings app = DeepAgentsApp() app._model_override = None with ( patch.object(settings, "model_provider", "anthropic"), patch.object(settings, "model_name", ""), ): assert app._effective_model_spec() is None class TestUpgradeThreadMessageLink: """Tests for DeepAgentsApp._upgrade_thread_message_link.""" async def test_noop_when_link_does_not_resolve(self) -> None: """Plain-string result should leave widget content unchanged.""" app = DeepAgentsApp() _app_test_double(app)._build_thread_message = AsyncMock( return_value="Resumed thread: tid-1" ) widget = MagicMock() widget.parent = object() widget._content = "Resumed thread: tid-1" await app._upgrade_thread_message_link( widget, prefix="Resumed thread", thread_id="tid-1", ) widget.update.assert_not_called() assert widget._content == "Resumed thread: tid-1" async def test_noop_when_widget_unmounted(self) -> None: """Unmounted widget should not be updated even when link resolves.""" from textual.content import Content app = DeepAgentsApp() _app_test_double(app)._build_thread_message = AsyncMock( return_value=Content("Resumed thread: tid-1") ) widget = MagicMock() widget.parent = None widget._content = "Resumed thread: tid-1" await app._upgrade_thread_message_link( widget, prefix="Resumed thread", thread_id="tid-1", ) widget.update.assert_not_called() async def test_updates_widget_when_link_resolves(self) -> None: """Resolved Content should replace widget content.""" from textual.content import Content app = DeepAgentsApp() linked = Content("Resumed thread: tid-1") _app_test_double(app)._build_thread_message = AsyncMock(return_value=linked) widget = MagicMock() widget.parent = object() widget._content = "Resumed thread: tid-1" await app._upgrade_thread_message_link( widget, prefix="Resumed thread", thread_id="tid-1", ) assert widget._content == linked widget.update.assert_called_once_with(linked) class TestBuildThreadMessage: """Tests for DeepAgentsApp._build_thread_message.""" async def test_plain_text_when_tracing_not_configured(self) -> None: """Returns plain string when LangSmith URL is not available.""" app = DeepAgentsApp() target = "deepagents_code.config.build_langsmith_thread_url" with patch(target, return_value=None): result = await app._build_thread_message("Resumed thread", "tid-123") assert result == "Resumed thread: tid-123" assert isinstance(result, str) async def test_hyperlinked_when_tracing_configured(self) -> None: """Returns Content with hyperlink when LangSmith URL is available.""" from textual.content import Content from textual.style import Style as TStyle app = DeepAgentsApp() url = "https://smith.langchain.com/o/org/projects/p/proj/t/tid-123" target = "deepagents_code.config.build_langsmith_thread_url" with patch(target, return_value=url): result = await app._build_thread_message("Resumed thread", "tid-123") assert isinstance(result, Content) assert "Resumed thread: " in result.plain assert "tid-123" in result.plain spans = [ s for s in result._spans if isinstance(s.style, TStyle) and s.style.link ] assert len(spans) == 1 style = spans[0].style assert isinstance(style, TStyle) assert style.link == url async def test_fallback_on_timeout(self) -> None: """Returns plain string when URL resolution times out.""" app = DeepAgentsApp() async def _raise_timeout( # noqa: RUF029 # async signature required to match asyncio.wait_for coro: Coroutine[Any, Any, Any], *_: Any, **__: Any ) -> None: coro.close() raise TimeoutError with patch("deepagents_code.app.asyncio.wait_for", new=_raise_timeout): result = await app._build_thread_message("Resumed thread", "t-1") assert isinstance(result, str) assert result == "Resumed thread: t-1" async def test_fallback_on_exception(self) -> None: """Returns plain string when URL resolution raises an exception.""" app = DeepAgentsApp() with patch( "deepagents_code.config.build_langsmith_thread_url", side_effect=OSError("network error"), ): result = await app._build_thread_message("Resumed thread", "t-1") assert isinstance(result, str) assert result == "Resumed thread: t-1" class TestConvertMessagesToData: """Tests for DeepAgentsApp._convert_messages_to_data.""" def _make_human(self, content: str) -> object: """Create a HumanMessage.""" from langchain_core.messages import HumanMessage return HumanMessage(content=content) def _make_ai( self, content: str | list[dict[str, str]] = "", tool_calls: list[dict[str, Any]] | None = None, ) -> object: """Create an AIMessage.""" from langchain_core.messages import AIMessage # LangChain accepts `tool_calls` dynamically, but its overloads don't # model this simplified test helper shape. return AIMessage( content=cast("Any", content), tool_calls=cast("Any", tool_calls or []) ) def _make_tool( self, content: str, tool_call_id: str, status: str = "success", ) -> object: """Create a ToolMessage.""" from langchain_core.messages import ToolMessage return ToolMessage(content=content, tool_call_id=tool_call_id, status=status) def test_human_message_conversion(self) -> None: """HumanMessage should become a USER MessageData.""" from deepagents_code.widgets.message_store import MessageType msgs = [self._make_human("Hello")] result = DeepAgentsApp._convert_messages_to_data(msgs) assert len(result) == 1 assert result[0].type == MessageType.USER assert result[0].content == "Hello" def test_system_prefix_skipped(self) -> None: """HumanMessages starting with [SYSTEM] should be skipped.""" msgs = [ self._make_human("[SYSTEM] Auto-injected context"), self._make_human("Real user message"), ] result = DeepAgentsApp._convert_messages_to_data(msgs) assert len(result) == 1 assert result[0].content == "Real user message" def test_ai_message_text_content(self) -> None: """AIMessage with string content should become ASSISTANT MessageData.""" from deepagents_code.widgets.message_store import MessageType msgs = [self._make_ai("Here is the answer.")] result = DeepAgentsApp._convert_messages_to_data(msgs) assert len(result) == 1 assert result[0].type == MessageType.ASSISTANT assert result[0].content == "Here is the answer." def test_ai_message_content_block_list(self) -> None: """AIMessage with list-of-blocks content should extract text.""" from deepagents_code.widgets.message_store import MessageType blocks: list[dict[str, str]] = [ {"type": "text", "text": "Part 1. "}, {"type": "text", "text": "Part 2."}, ] msgs = [self._make_ai(blocks)] result = DeepAgentsApp._convert_messages_to_data(msgs) assert len(result) == 1 assert result[0].type == MessageType.ASSISTANT assert result[0].content == "Part 1. Part 2." def test_ai_message_empty_text_skipped(self) -> None: """AIMessage with empty text should not produce an ASSISTANT entry.""" msgs = [self._make_ai(" ")] result = DeepAgentsApp._convert_messages_to_data(msgs) assert len(result) == 0 def test_tool_call_matching(self) -> None: """ToolMessage should be matched to its AIMessage tool call by ID.""" from deepagents_code.widgets.message_store import MessageType, ToolStatus msgs = [ self._make_ai( tool_calls=[ {"id": "tc-1", "name": "read_file", "args": {"path": "/a.py"}} ] ), self._make_tool("file contents", tool_call_id="tc-1"), ] result = DeepAgentsApp._convert_messages_to_data(msgs) assert len(result) == 1 assert result[0].type == MessageType.TOOL assert result[0].tool_name == "read_file" assert result[0].tool_status == ToolStatus.SUCCESS assert result[0].tool_output == "file contents" def test_tool_call_error_status(self) -> None: """ToolMessage with error status should set ERROR on the tool data.""" from deepagents_code.widgets.message_store import ToolStatus msgs = [ self._make_ai( tool_calls=[{"id": "tc-2", "name": "bash", "args": {"cmd": "fail"}}] ), self._make_tool("command failed", tool_call_id="tc-2", status="error"), ] result = DeepAgentsApp._convert_messages_to_data(msgs) assert result[0].tool_status == ToolStatus.ERROR assert result[0].tool_output == "command failed" def test_unmatched_tool_call_rejected(self) -> None: """Tool calls with no matching ToolMessage should be REJECTED.""" from deepagents_code.widgets.message_store import ToolStatus msgs = [ self._make_ai(tool_calls=[{"id": "tc-3", "name": "bash", "args": {}}]), ] result = DeepAgentsApp._convert_messages_to_data(msgs) assert len(result) == 1 assert result[0].tool_status == ToolStatus.REJECTED def test_mixed_message_sequence(self) -> None: """Full conversation with mixed message types should convert correctly.""" from deepagents_code.widgets.message_store import MessageType, ToolStatus msgs = [ self._make_human("What files are here?"), self._make_ai( "Let me check.", tool_calls=[{"id": "tc-a", "name": "list_files", "args": {"dir": "."}}], ), self._make_tool("file1.py\nfile2.py", tool_call_id="tc-a"), self._make_ai("I found 2 files."), ] result = DeepAgentsApp._convert_messages_to_data(msgs) assert len(result) == 4 assert result[0].type == MessageType.USER assert result[1].type == MessageType.ASSISTANT assert result[1].content == "Let me check." assert result[2].type == MessageType.TOOL assert result[2].tool_status == ToolStatus.SUCCESS assert result[3].type == MessageType.ASSISTANT assert result[3].content == "I found 2 files." def test_empty_messages(self) -> None: """Empty input should return empty output.""" result = DeepAgentsApp._convert_messages_to_data([]) assert result == [] def test_skill_message_from_additional_kwargs(self) -> None: """HumanMessage with __skill in additional_kwargs → SKILL MessageData.""" from langchain_core.messages import HumanMessage from deepagents_code.widgets.message_store import MessageType msg = HumanMessage( content="I'm invoking the skill `web-research`.\n---\n# Body\n---", additional_kwargs={ "__skill": { "name": "web-research", "description": "Research topics", "source": "user", "args": "find quantum", }, }, ) result = DeepAgentsApp._convert_messages_to_data([msg]) assert len(result) == 1 assert result[0].type == MessageType.SKILL assert result[0].skill_name == "web-research" assert result[0].skill_description == "Research topics" assert result[0].skill_source == "user" assert result[0].skill_args == "find quantum" # Full prompt envelope stored as body for expand view assert "web-research" in (result[0].skill_body or "") def test_skill_without_name_falls_back_to_user(self) -> None: """__skill dict missing name should fall back to USER.""" from langchain_core.messages import HumanMessage from deepagents_code.widgets.message_store import MessageType msg = HumanMessage( content="some text", additional_kwargs={"__skill": {"description": "no name"}}, ) result = DeepAgentsApp._convert_messages_to_data([msg]) assert len(result) == 1 assert result[0].type == MessageType.USER class TestColumnKeyConsistency: """Verify all column dicts stay in sync.""" def test_all_column_dicts_share_same_keys(self) -> None: """All parallel column dicts must have the same key set.""" from deepagents_code.model_config import THREAD_COLUMN_DEFAULTS from deepagents_code.widgets.thread_selector import ( _COLUMN_LABELS, _COLUMN_ORDER, _COLUMN_TOGGLE_LABELS, _COLUMN_WIDTHS, ) order_keys = set(_COLUMN_ORDER) assert order_keys == set(_COLUMN_WIDTHS), ( f"_COLUMN_WIDTHS keys differ: " f"missing={order_keys - set(_COLUMN_WIDTHS)}, " f"extra={set(_COLUMN_WIDTHS) - order_keys}" ) assert order_keys == set(_COLUMN_LABELS), ( f"_COLUMN_LABELS keys differ: " f"missing={order_keys - set(_COLUMN_LABELS)}, " f"extra={set(_COLUMN_LABELS) - order_keys}" ) assert order_keys == set(_COLUMN_TOGGLE_LABELS), ( f"_COLUMN_TOGGLE_LABELS keys differ: " f"missing={order_keys - set(_COLUMN_TOGGLE_LABELS)}, " f"extra={set(_COLUMN_TOGGLE_LABELS) - order_keys}" ) assert order_keys == set(THREAD_COLUMN_DEFAULTS), ( f"THREAD_COLUMN_DEFAULTS keys differ: " f"missing={order_keys - set(THREAD_COLUMN_DEFAULTS)}, " f"extra={set(THREAD_COLUMN_DEFAULTS) - order_keys}" ) class TestThreadsMatch: """Tests for _threads_match short-circuit comparison.""" @staticmethod def _thread(tid: str, cp: str | None = None) -> ThreadInfo: t: ThreadInfo = { "thread_id": tid, "agent_name": "a", "updated_at": "x", } if cp is not None: t["latest_checkpoint_id"] = cp return t def test_identical_lists_match(self) -> None: """Identical thread lists should match.""" a = [self._thread("t1", "cp1"), self._thread("t2", "cp2")] b = [self._thread("t1", "cp1"), self._thread("t2", "cp2")] assert ThreadSelectorScreen._threads_match(a, b) is True def test_different_lengths_do_not_match(self) -> None: """Different-length lists should not match.""" a = [self._thread("t1")] b = [self._thread("t1"), self._thread("t2")] assert ThreadSelectorScreen._threads_match(a, b) is False def test_different_thread_ids_do_not_match(self) -> None: """Different thread IDs at same position should not match.""" a = [self._thread("t1", "cp1")] b = [self._thread("t2", "cp1")] assert ThreadSelectorScreen._threads_match(a, b) is False def test_different_checkpoint_ids_do_not_match(self) -> None: """Lists with different checkpoint IDs should not match.""" a = [self._thread("t1", "cp1")] b = [self._thread("t1", "cp2")] assert ThreadSelectorScreen._threads_match(a, b) is False def test_reordered_threads_do_not_match(self) -> None: """Positional comparison means reordered lists fail.""" a = [self._thread("t1", "cp1"), self._thread("t2", "cp2")] b = [self._thread("t2", "cp2"), self._thread("t1", "cp1")] assert ThreadSelectorScreen._threads_match(a, b) is False def test_empty_lists_match(self) -> None: """Two empty lists should match.""" assert ThreadSelectorScreen._threads_match([], []) is True class TestThreadSelectorDomSkip: """Tests for skipping DOM rebuild when data matches prewarm cache.""" async def test_matching_refresh_skips_dom_rebuild(self) -> None: """When refreshed threads match prefetched, DOM should not be rebuilt.""" prefetched: list[ThreadInfo] = [ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", "latest_checkpoint_id": "cp_1", "message_count": 5, } ] # Same thread and checkpoint refreshed: list[ThreadInfo] = [ { "thread_id": "abc12345", "agent_name": "my-agent", "updated_at": "2025-01-15T10:30:00", "latest_checkpoint_id": "cp_1", } ] app = ThreadSelectorTestApp(current_thread="abc12345") with patch( "deepagents_code.sessions.list_threads", new_callable=AsyncMock, return_value=refreshed, ): async with app.run_test() as pilot: app.push_screen( ThreadSelectorScreen( current_thread="abc12345", thread_limit=20, initial_threads=prefetched, filter_cwd=None, ) ) await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) initial_widgets = list(screen._option_widgets) assert len(initial_widgets) == 1 # Wait for background refresh for _ in range(10): await pilot.pause(0.05) # Same widget objects should still be mounted (no rebuild) assert screen._option_widgets == initial_widgets class TestThreadSelectorAgentFilter: """Tests for the agent filter dropdown in the Options panel.""" def test_agent_filtered_threads_no_filter(self) -> None: """No agent filter returns all threads.""" screen = ThreadSelectorScreen( current_thread=None, initial_threads=MOCK_THREADS, filter_cwd=None, ) assert screen._filter_agent is None result = screen._agent_filtered_threads() assert len(result) == 3 def test_agent_filtered_threads_with_filter(self) -> None: """Agent filter returns only matching threads.""" screen = ThreadSelectorScreen( current_thread=None, initial_threads=MOCK_THREADS, filter_cwd=None, ) screen._filter_agent = "my-agent" result = screen._agent_filtered_threads() assert len(result) == 2 assert all(t["agent_name"] == "my-agent" for t in result) def test_agent_filtered_threads_unknown_agent(self) -> None: """Threads with no agent_name are matched by the '(unknown)' sentinel.""" threads: list[ThreadInfo] = [ { "thread_id": "t1", "agent_name": None, "updated_at": "2025-01-01T00:00:00", "created_at": "2025-01-01T00:00:00", }, { "thread_id": "t2", "agent_name": "my-agent", "updated_at": "2025-01-01T00:00:00", "created_at": "2025-01-01T00:00:00", }, ] screen = ThreadSelectorScreen( current_thread=None, initial_threads=threads, filter_cwd=None, ) screen._filter_agent = "(unknown)" result = screen._agent_filtered_threads() assert len(result) == 1 assert result[0]["thread_id"] == "t1" def test_update_filtered_list_respects_agent_filter(self) -> None: """_update_filtered_list applies agent filter before fuzzy search.""" screen = ThreadSelectorScreen( current_thread=None, initial_threads=MOCK_THREADS, filter_cwd=None, ) screen._filter_agent = "other-agent" screen._filter_text = "" screen._update_filtered_list() assert len(screen._filtered_threads) == 1 assert screen._filtered_threads[0]["thread_id"] == "def67890" def test_update_filtered_list_agent_and_text_combined(self) -> None: """Agent filter and text filter are ANDed together.""" screen = ThreadSelectorScreen( current_thread=None, initial_threads=MOCK_THREADS, filter_cwd=None, ) screen._filter_agent = "my-agent" screen._filter_text = "Hello" screen._update_filtered_list() assert len(screen._filtered_threads) == 1 assert screen._filtered_threads[0]["thread_id"] == "abc12345" def test_collect_agent_options_all_sentinel_first(self) -> None: """collect_agent_options always returns 'All agents' as the first option.""" from deepagents_code.widgets.thread_selector import _AGENT_VALUE_ALL screen = ThreadSelectorScreen( current_thread=None, initial_threads=MOCK_THREADS, filter_cwd=None, ) options = screen._collect_agent_options() assert options[0] == ("All agents", _AGENT_VALUE_ALL) def test_collect_agent_options_loading_while_pending(self) -> None: """While loading with no known agents, the dropdown shows 'Loading...'.""" from deepagents_code.widgets.thread_selector import ( _AGENT_LABEL_LOADING, _AGENT_VALUE_ALL, _AGENT_VALUE_LOADING, ) screen = ThreadSelectorScreen( current_thread=None, initial_threads=None, filter_cwd=None, ) assert screen._disk_load_complete is False assert screen._collect_agent_options() == [ (_AGENT_LABEL_LOADING, _AGENT_VALUE_LOADING) ] # Once the disk load completes with no threads, fall back to "All agents". screen._disk_load_complete = True assert screen._collect_agent_options() == [("All agents", _AGENT_VALUE_ALL)] async def test_agent_select_label_refreshes_after_empty_load(self) -> None: """The loading placeholder is replaced by the final all-agents label.""" from deepagents_code.widgets.thread_selector import ( _AGENT_SELECT_ID, _AGENT_VALUE_ALL, _AGENT_VALUE_LOADING, ) load_started = asyncio.Event() load_finished = asyncio.Event() async def list_threads_after_signal(**_: object) -> list[ThreadInfo]: load_started.set() await load_finished.wait() return [] with ( patch("deepagents_code.sessions.list_threads", list_threads_after_signal), _patch_columns(), _patch_available_agents([]), ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await asyncio.wait_for(load_started.wait(), timeout=1) await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) assert agent_select.value == _AGENT_VALUE_LOADING assert str(agent_select.query_one(SelectCurrent).label) == "Loading..." load_finished.set() await pilot.pause() await pilot.pause() assert agent_select.value == _AGENT_VALUE_ALL assert str(agent_select.query_one(SelectCurrent).label) == "All agents" def test_collect_agent_options_sorted_unique(self) -> None: """collect_agent_options returns sorted unique agent names.""" screen = ThreadSelectorScreen( current_thread=None, initial_threads=MOCK_THREADS, filter_cwd=None, ) # Inject mixed-case configured names so the order assertion distinguishes # the production case-insensitive `str.casefold` key from a plain # case-sensitive sort: the latter would place "Zebra" (uppercase) ahead # of the lowercase names rather than last. screen._available_agent_names = ["Zebra", "apple"] options = screen._collect_agent_options() # First entry is the sentinel; the rest are unique agents in # case-insensitive order with the thread/config overlap de-duplicated. labels = [label for label, _ in options[1:]] assert labels == ["apple", "my-agent", "other-agent", "Zebra"] assert len(set(labels)) == len(labels) # no duplicates def test_collect_agent_options_includes_configured_agents(self) -> None: """Configured agents (the /agents list) appear even without threads.""" screen = ThreadSelectorScreen( current_thread=None, initial_threads=MOCK_THREADS, filter_cwd=None, ) # Simulate the off-thread scan of ~/.deepagents/ surfacing an agent that # has not produced any threads yet, plus one that overlaps a thread. screen._available_agent_names = ["my-agent", "zeta-agent"] labels = [label for label, _ in screen._collect_agent_options()[1:]] # Threadless configured agent is selectable... assert "zeta-agent" in labels # ...alongside thread-derived names, with the overlap de-duplicated. assert "my-agent" in labels assert "other-agent" in labels assert labels.count("my-agent") == 1 assert labels == sorted(labels, key=str.casefold) async def test_load_threads_populates_configured_agents(self) -> None: """`/threads` agent filter mirrors `/agents`, not just agents w/ threads.""" from deepagents_code.widgets.thread_selector import _AGENT_SELECT_ID with ( _patch_list_threads(), _patch_columns(), _patch_available_agents(["brand-new-agent"]), ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._available_agent_names == ["brand-new-agent"] agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) values = set(agent_select._legal_values) # Configured-but-threadless agent is offered as a filter option. assert "brand-new-agent" in values # Thread-derived agents remain selectable too. assert "my-agent" in values assert "other-agent" in values async def test_agent_select_widget_present(self) -> None: """The agent Select widget is rendered inside the Options panel.""" from deepagents_code.widgets.thread_selector import _AGENT_SELECT_ID with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) assert agent_select is not None async def test_agent_select_filters_thread_list(self) -> None: """Changing the agent dropdown filters the visible thread list.""" from deepagents_code.widgets.thread_selector import _AGENT_SELECT_ID with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert len(screen._filtered_threads) == 3 # Set agent filter directly and re-run filtering screen._filter_agent = "my-agent" screen._update_filtered_list() assert len(screen._filtered_threads) == 2 assert all( t["agent_name"] == "my-agent" for t in screen._filtered_threads ) async def test_reload_clears_stale_agent_filter_and_refilters(self) -> None: """Reloading without the selected agent should show all remaining rows.""" from deepagents_code.widgets.thread_selector import ( _AGENT_SELECT_ID, _AGENT_VALUE_ALL, ) reloaded_threads = [MOCK_THREADS[1]] with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) agent_select.value = "my-agent" screen._filter_agent = "my-agent" screen._update_filtered_list() assert len(screen._filtered_threads) == 2 with _patch_list_threads(reloaded_threads): await screen._load_threads() await pilot.pause() assert screen._filter_agent is None assert agent_select.value == _AGENT_VALUE_ALL assert screen._filtered_threads == reloaded_threads async def test_enter_opens_agent_select_without_resuming_thread(self) -> None: """Enter on the focused agent control should open its dropdown.""" from deepagents_code.widgets.thread_selector import _AGENT_SELECT_ID with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) await pilot.press("tab") await pilot.press("tab") await pilot.pause() assert agent_select.has_focus await pilot.press("enter") await pilot.pause() assert agent_select.expanded assert not app.dismissed async def test_tab_keys_move_open_agent_select_highlight(self) -> None: """Tab and Shift+Tab should move the agent dropdown highlight while open.""" from deepagents_code.widgets.thread_selector import _AGENT_SELECT_ID with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) sort_switch = screen.query_one("#thread-sort-toggle", Checkbox) scope_select = screen.query_one("#thread-scope-select", Select) await pilot.press("tab") await pilot.press("tab") await pilot.press("enter") await pilot.pause() assert agent_select.expanded overlay = agent_select.query_one(ThreadScopeSelectOverlay) assert overlay.highlighted == 0 await pilot.press("tab") await pilot.pause() assert agent_select.expanded assert overlay.highlighted == 1 assert not sort_switch.has_focus assert not app.dismissed await pilot.press("shift+tab") await pilot.pause() assert agent_select.expanded assert overlay.highlighted == 0 assert not scope_select.has_focus assert not app.dismissed async def test_escape_closes_open_agent_select_without_dismissing(self) -> None: """Esc should close the agent dropdown before it cancels the selector.""" from deepagents_code.widgets.thread_selector import _AGENT_SELECT_ID with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) await pilot.press("tab") await pilot.press("tab") await pilot.press("enter") await pilot.pause() assert agent_select.expanded await pilot.press("escape") await pilot.pause() assert not agent_select.expanded assert agent_select.has_focus assert not app.dismissed async def test_agent_name_load_failure_does_not_strand_picker(self) -> None: """An unexpected agent-scan error must not abort the post-load build. `_load_available_agent_names` is awaited after `_load_threads`'s own try/except, so a raise there would otherwise skip the filter refresh, DOM build, and checkpoint enrichment, leaving the picker half-rendered. """ with ( _patch_list_threads(), _patch_columns(), patch( "deepagents_code.agent.get_available_agent_names", side_effect=RuntimeError("agent scan blew up"), ), ): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() screen: object = None for _ in range(20): await pilot.pause() screen = app.screen if ( isinstance(screen, ThreadSelectorScreen) and screen._disk_load_complete and screen._option_widgets ): break assert isinstance(screen, ThreadSelectorScreen) # The load completed and the list was built despite the failure. assert screen._disk_load_complete assert len(screen._filtered_threads) == 3 assert screen._option_widgets # The filter degraded to thread-derived names only (the scan # never assigned, so the cache keeps its empty default). assert screen._available_agent_names == [] async def test_reload_preserves_present_agent_filter(self) -> None: """Reloading with the selected agent still present keeps the filter.""" from deepagents_code.widgets.thread_selector import _AGENT_SELECT_ID with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) agent_select.value = "my-agent" screen._filter_agent = "my-agent" screen._update_filtered_list() assert len(screen._filtered_threads) == 2 # Reload with a thread set that still contains my-agent. with _patch_list_threads(MOCK_THREADS): await screen._load_threads() for _ in range(20): await pilot.pause() if ( screen._filter_agent == "my-agent" and len(screen._filtered_threads) == 2 ): break # Selection and filter survive the reload (no fallback to All). assert screen._filter_agent == "my-agent" assert agent_select.value == "my-agent" assert len(screen._filtered_threads) == 2 assert all( t["agent_name"] == "my-agent" for t in screen._filtered_threads ) async def test_agent_dropdown_change_filters_via_event(self) -> None: """Selecting an agent via the dropdown event filters the list. Drives the real `Select.Changed` handler rather than poking `_filter_agent`, covering the `_AGENT_VALUE_ALL` translation and the async `_filter_and_build` path the handler schedules. """ from deepagents_code.widgets.thread_selector import ( _AGENT_SELECT_ID, _AGENT_VALUE_ALL, ) with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) assert screen._filter_agent is None assert len(screen._filtered_threads) == 3 agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) agent_select.value = "my-agent" for _ in range(20): await pilot.pause() if ( screen._filter_agent == "my-agent" and len(screen._filtered_threads) == 2 ): break assert screen._filter_agent == "my-agent" assert len(screen._filtered_threads) == 2 # Returning to the sentinel clears the filter. agent_select.value = _AGENT_VALUE_ALL for _ in range(20): await pilot.pause() if ( screen._filter_agent is None and len(screen._filtered_threads) == 3 ): break assert screen._filter_agent is None assert len(screen._filtered_threads) == 3 async def test_confirming_delete_ignores_agent_select_change(self) -> None: """Agent dropdown changes are ignored while a delete is being confirmed.""" from deepagents_code.widgets.thread_selector import _AGENT_SELECT_ID with _patch_list_threads(), _patch_columns(): app = ThreadSelectorTestApp() async with app.run_test() as pilot: app.show_selector() await pilot.pause() screen = app.screen assert isinstance(screen, ThreadSelectorScreen) agent_select = screen.query_one(f"#{_AGENT_SELECT_ID}", Select) screen._confirming_delete = True agent_select.value = "my-agent" for _ in range(10): await pilot.pause() # The guard short-circuits before mutating the filter state. assert screen._filter_agent is None assert len(screen._filtered_threads) == 3