feat(code): resume threads in-TUI with /threads -r [ID] (#4609)

The `dcode` TUI now supports `/threads -r [ID]` to resume a previous
thread in place, and `/clear` shows the thread you just left with a
one-step resume hint.

---

After `/clear` mints a new thread, there was no low-friction path back
to the thread you just left — the id was dropped and only the modal
`/threads` selector could find it again.

This adds an in-TUI equivalent of the launch-time `-r` flag rather than
a bespoke one-level `/go-back`, keeping parity with muscle memory users
already have and composing better:

- `TextualSessionState` now records the outgoing thread as
`previous_thread_id` on every `reset_thread` (`/clear`).
- `/clear` output names the previous thread (with the same clickable
LangSmith-link upgrade as the new-thread line) and prints a resume hint.
- `/threads -r [ID]` resumes without opening the modal: bare `-r`
returns to the previous thread (falling back to the most recent thread
on disk), and `-r <ID>` resumes a specific thread. It reuses the
existing `_resume_thread` machinery, so behavior matches selecting a
thread in the modal. Because `/threads` is an `IMMEDIATE_UI` command,
only the bare form bypasses the busy queue — `/threads -r …` is queued
while the agent is running, so a resume never races an active turn.

Autocomplete gains an `[-r [ID]]` argument hint and a splash tip
advertises the shortcut.

Made by [Open
SWE](https://openswe.vercel.app/agents/b717ea48-7edd-8049-95cc-64384c56e8a8)

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

libs/code/deepagents_code/app.py

@@ -1687,6 +1687,13 @@ class TextualSessionState:
         """1-based user-turn count for the thread (coding-agent-v1 turn_number)."""
         self.turn_id: str | None = None
         """Stable id for the current user turn (coding-agent-v1 turn_id)."""
+        self.previous_thread_id: str | None = None
+        """Thread id abandoned by the most recent `reset_thread`.
+
+        Set by every `reset_thread` caller — `/clear`, `/force-clear`, and the
+        agent-switch teardown — not just `/clear`. Lets the TUI point users
+        back to the thread they just left; `None` until the first reset.
+        """
         # Assign the backing field directly: the setter reads `self._thread_id`
         # to detect a thread change, and it isn't set yet.
         self._thread_id = thread_id or _new_thread_id()
@@ -1723,9 +1730,13 @@ class TextualSessionState:
     def reset_thread(self) -> str:
         """Reset to a new thread.
 
+        Records the outgoing thread as `previous_thread_id` first, so the UI
+        can offer a one-step path back to it.
+
         Returns:
             The new thread_id.
         """
+        self.previous_thread_id = self._thread_id
         self.thread_id = _new_thread_id()  # setter resets the turn markers
         self.approval_mode_key = None
         return self.thread_id
@@ -7607,8 +7618,10 @@ class DeepAgentsApp(App):
         if cmd in BYPASS_WHEN_CONNECTING:
             return self._connecting and not (self._agent_running or self._shell_running)
         if cmd in IMMEDIATE_UI:
-            # Only bare form (no args) bypasses — /model opens selector,
-            # /model <name> does a direct switch that shouldn't race with agent.
+            # Only bare form (no args) bypasses — the selector-opening form is
+            # safe, but an argument form does a direct action that shouldn't
+            # race the agent (e.g. `/model <name>` switches models, `/threads
+            # -r <id>` resumes a thread).
             return value == cmd
         return cmd in SIDE_EFFECT_FREE
 
@@ -9718,6 +9731,51 @@ class DeepAgentsApp(App):
                     prefix="Started new thread",
                     thread_id=new_thread_id,
                 )
+                previous_thread_id = self._session_state.previous_thread_id
+                if previous_thread_id:
+                    import sqlite3
+
+                    from deepagents_code.sessions import thread_exists
+
+                    # Best-effort: on any failure just suppress the hint (never
+                    # crash `/clear`), but log unexpected errors loudly so a
+                    # real bug isn't silently read as "not resumable".
+                    try:
+                        previous_thread_is_resumable = await thread_exists(
+                            previous_thread_id
+                        )
+                    except (sqlite3.Error, OSError):
+                        logger.debug(
+                            "Could not check whether previous thread %s is resumable",
+                            previous_thread_id,
+                            exc_info=True,
+                        )
+                        previous_thread_is_resumable = False
+                    except Exception:
+                        logger.warning(
+                            "Unexpected error checking previous thread %s resumability",
+                            previous_thread_id,
+                            exc_info=True,
+                        )
+                        previous_thread_is_resumable = False
+                else:
+                    previous_thread_is_resumable = False
+                if previous_thread_id and previous_thread_is_resumable:
+                    previous_msg_widget = AppMessage(
+                        f"Previous thread: {previous_thread_id}"
+                    )
+                    await self._mount_message(previous_msg_widget)
+                    self._schedule_thread_message_link(
+                        previous_msg_widget,
+                        prefix="Previous thread",
+                        thread_id=previous_thread_id,
+                    )
+                    await self._mount_message(
+                        AppMessage(
+                            "Resume it with /threads -r"
+                            f" (or /threads -r {previous_thread_id})"
+                        )
+                    )
         elif cmd == "/copy":
             await self._mount_message(UserMessage(command))
             # Reverse-scan for the newest assistant message that has finished
@@ -9766,8 +9824,8 @@ class DeepAgentsApp(App):
         elif cmd in {"/offload", "/compact"}:
             await self._mount_message(UserMessage(command))
             await self._handle_offload()
-        elif cmd == "/threads":
-            await self._show_thread_selector()
+        elif cmd == "/threads" or cmd.startswith("/threads "):
+            await self._handle_threads_command(command)
         elif cmd == "/trace":
             await self._handle_trace_command(command)
         elif cmd == "/update" or cmd.startswith("/update "):
@@ -16238,6 +16296,144 @@ class DeepAgentsApp(App):
             if self._chat_input:
                 self._chat_input.set_cursor_active(active=not self._agent_running)
 
+    async def _handle_threads_command(self, command: str) -> None:
+        """Dispatch `/threads`, optionally resuming a thread without the modal.
+
+        Bare `/threads` opens the interactive selector. `/threads -r [ID]`
+        resumes in place: `-r` alone returns to the thread left by the most
+        recent reset (e.g. `/clear`), falling back to the most recent thread
+        for the active agent; `-r <ID>` resumes a specific thread. Both forms
+        only resume threads owned by the active agent, mirroring the
+        launch-time `-r` flag.
+
+        Args:
+            command: The raw command text, e.g. `"/threads -r abc123"`.
+        """
+        args = command.split()[1:]  # drop the leading "/threads"
+
+        if not args:
+            await self._show_thread_selector()
+            return
+
+        if args[0].lower() not in {"-r", "--resume"}:
+            await self._mount_message(UserMessage(command))
+            await self._mount_message(
+                AppMessage(
+                    "Usage: /threads (open selector) or /threads -r [ID] "
+                    "(resume the previous or a specific thread)."
+                )
+            )
+            return
+
+        max_resume_args = 2  # flag plus at most one thread id
+        if len(args) > max_resume_args:
+            await self._mount_message(UserMessage(command))
+            await self._mount_message(
+                AppMessage("Usage: /threads -r [ID] accepts at most one thread ID."),
+            )
+            return
+
+        await self._mount_message(UserMessage(command))
+        requested_id = args[1] if len(args) == max_resume_args else None
+        target = await self._resolve_threads_resume_target(requested_id)
+        if target is not None:
+            await self._resume_thread(target)
+
+    async def _resolve_threads_resume_target(
+        self, requested_id: str | None
+    ) -> str | None:
+        """Resolve a `/threads -r` argument to a concrete, resumable thread id.
+
+        Only threads owned by the active agent resolve: an explicit id owned by
+        another agent is refused, and the bare-`-r` fallback is filtered to the
+        active agent's most recent thread.
+
+        Args:
+            requested_id: Explicit thread id from `-r <ID>`, or `None` for a
+                bare `-r` (resume the previous or most-recent thread).
+
+        Returns:
+            The thread id to resume, or `None` when nothing suitable exists;
+            in that case a user-facing message has already been mounted.
+        """
+        import sqlite3
+
+        from deepagents_code.sessions import (
+            find_similar_threads,
+            get_most_recent,
+            get_thread_agent,
+            thread_exists,
+        )
+
+        try:
+            active_agent = self._assistant_id or DEFAULT_ASSISTANT_ID
+            if requested_id is not None:
+                if await thread_exists(requested_id):
+                    owner = await get_thread_agent(requested_id)
+                    if owner == active_agent:
+                        return requested_id
+                    if owner:
+                        msg = (
+                            f"Thread '{requested_id}' belongs to agent '{owner}', not "
+                            f"the active agent '{active_agent}'. Switch agents first."
+                        )
+                    else:
+                        msg = (
+                            f"Could not verify which agent owns thread "
+                            f"'{requested_id}', so it was not resumed."
+                        )
+                    await self._mount_message(AppMessage(msg))
+                    return None
+                hint = f"Thread '{requested_id}' not found."
+                similar = await find_similar_threads(requested_id)
+                if similar:
+                    hint += f" Did you mean: {', '.join(str(t) for t in similar)}?"
+                await self._mount_message(AppMessage(hint))
+                return None
+
+            # Bare `-r`: prefer the thread the session just left (e.g. via
+            # `/clear`), then fall back to the most recent inactive thread on disk.
+            previous = (
+                self._session_state.previous_thread_id if self._session_state else None
+            )
+            if previous and await thread_exists(previous):
+                owner = await get_thread_agent(previous)
+                if owner == active_agent:
+                    return previous
+
+            current = self._session_state.thread_id if self._session_state else None
+            candidate = await get_most_recent(
+                active_agent,
+                exclude_thread_id=current,
+            )
+            if candidate:
+                return candidate
+
+            msg = f"No previous threads for '{active_agent}' to resume."
+            await self._mount_message(AppMessage(msg))
+        except (sqlite3.Error, OSError):
+            # Expected thread-store failures: log for a debug session and tell
+            # the user to retry. Mirrors the launch-time resolver's handling.
+            logger.warning(
+                "Thread-history lookup failed resolving resume target %r",
+                requested_id,
+                exc_info=True,
+            )
+            await self._mount_message(
+                AppMessage("Could not look up thread history. Please try again.")
+            )
+        except Exception:
+            # Anything else (e.g. a programming error, or a widget-mount fault)
+            # is a real bug, not "database unavailable" — surface it loudly
+            # instead of masking it behind the retry message above.
+            logger.exception(
+                "Unexpected error resolving resume target %r", requested_id
+            )
+            await self._mount_message(
+                AppMessage("Something went wrong resolving that thread.")
+            )
+        return None
+
     async def _show_thread_selector(self) -> None:
         """Show interactive thread selector as a modal screen."""
         from functools import partial
@@ -16909,6 +17105,13 @@ class DeepAgentsApp(App):
                 thread_id=thread_id,
                 preloaded_payload=prefetched_payload,
             )
+
+            # The switch succeeded: record the thread we just left so a
+            # subsequent bare `/threads -r` steps back to it rather than
+            # resolving `previous == current` and reporting "Already on
+            # thread". Set only after the last statement that can raise, so a
+            # failed switch (handled below) never leaves a stale pointer.
+            self._session_state.previous_thread_id = prev_session_thread
         except Exception as exc:
             if prefetched_payload is None:
                 logger.exception("Failed to prefetch history for thread %s", thread_id)

libs/code/deepagents_code/command_registry.py

@@ -166,7 +166,8 @@ COMMANDS: tuple[SlashCommand, ...] = (
         name="/threads",
         description="Browse and resume past threads",
         bypass_tier=BypassTier.IMMEDIATE_UI,
-        hidden_keywords="continue history sessions",
+        hidden_keywords="continue history sessions resume back previous",
+        argument_hint="[-r [ID]]",
     ),
     SlashCommand(
         name="/trace",

libs/code/deepagents_code/sessions.py

@@ -1261,24 +1261,49 @@ def _coerce_prompt_text(content: object) -> str | None:
     return str(content)
 
 
-async def get_most_recent(agent_name: str | None = None) -> str | None:
-    """Get most recent thread_id, optionally filtered by agent.
+async def get_most_recent(
+    agent_name: str | None = None,
+    *,
+    exclude_thread_id: str | None = None,
+) -> str | None:
+    """Get the most recent thread, optionally agent-filtered and/or excluding a thread.
+
+    Args:
+        agent_name: Return only threads created by this agent.
+        exclude_thread_id: Ignore this thread when selecting the most recent one.
 
     Returns:
-        Most recent thread_id or None if no threads exist.
+        Most recent thread ID, or `None` if no matching threads exist.
     """
     async with _connect() as conn:
         if not await _table_exists(conn, "checkpoints"):
             return None
 
-        if agent_name:
+        if agent_name and exclude_thread_id:
+            query = """
+                SELECT thread_id FROM checkpoints
+                WHERE json_extract(metadata, '$.agent_name') = ?
+                  AND thread_id != ?
+                ORDER BY checkpoint_id DESC
+                LIMIT 1
+            """
+            params: tuple[str, ...] = (agent_name, exclude_thread_id)
+        elif agent_name:
             query = """
                 SELECT thread_id FROM checkpoints
                 WHERE json_extract(metadata, '$.agent_name') = ?
                 ORDER BY checkpoint_id DESC
                 LIMIT 1
             """
-            params: tuple = (agent_name,)
+            params = (agent_name,)
+        elif exclude_thread_id:
+            query = """
+                SELECT thread_id FROM checkpoints
+                WHERE thread_id != ?
+                ORDER BY checkpoint_id DESC
+                LIMIT 1
+            """
+            params = (exclude_thread_id,)
         else:
             query = (
                 "SELECT thread_id FROM checkpoints ORDER BY checkpoint_id DESC LIMIT 1"

libs/code/deepagents_code/tui/widgets/startup_tip.py

@@ -13,6 +13,7 @@ from deepagents_code._env_vars import HIDE_SPLASH_TIPS, is_env_truthy
 _TIPS: dict[str, int] = {
     "Use @ to reference files and / for commands": 3,
     "Try /threads to resume a previous conversation": 2,
+    "Use /threads -r to jump back to the thread you just left with /clear": 1,
     "Use /offload when your conversation gets long": 2,
     "Use /copy to copy the latest assistant message": 3,
     "Use /mcp to search your MCP servers and inspect tool parameters": 1,

libs/code/tests/unit_tests/test_app.py

@@ -4808,15 +4808,74 @@ class TestClearCommand:
                 str(widget._content) == "Started new thread: new-thread"
                 for widget in app_msgs
             )
-            schedule.assert_called_once()
-            widget = schedule.call_args.args[0]
-            assert isinstance(widget, AppMessage)
-            assert widget in app_msgs
-            assert schedule.call_args.kwargs == {
+            started_widget = schedule.call_args_list[0].args[0]
+            assert isinstance(started_widget, AppMessage)
+            assert started_widget in app_msgs
+            assert schedule.call_args_list[0].kwargs == {
                 "prefix": "Started new thread",
                 "thread_id": "new-thread",
             }
 
+    async def test_clear_points_back_to_previous_thread(self) -> None:
+        """/clear should surface the abandoned thread and a resume hint."""
+        app = DeepAgentsApp(thread_id="old-thread")
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._session_state = TextualSessionState(thread_id="old-thread")
+            app._lc_thread_id = "old-thread"
+
+            with (
+                patch("deepagents_code.app._new_thread_id", return_value="new-thread"),
+                patch(
+                    "deepagents_code.sessions.thread_exists",
+                    AsyncMock(return_value=True),
+                ),
+                patch.object(app, "_schedule_thread_message_link") as schedule,
+            ):
+                await app._handle_command("/clear")
+                await pilot.pause()
+
+            assert app._session_state.previous_thread_id == "old-thread"
+
+            app_msgs = list(app.query(AppMessage))
+            assert any(
+                str(widget._content) == "Previous thread: old-thread"
+                for widget in app_msgs
+            )
+            assert any(
+                str(widget._content)
+                == "Resume it with /threads -r (or /threads -r old-thread)"
+                for widget in app_msgs
+            )
+            assert schedule.call_args_list[1].kwargs == {
+                "prefix": "Previous thread",
+                "thread_id": "old-thread",
+            }
+
+    async def test_clear_omits_previous_thread_without_checkpoint(self) -> None:
+        """/clear should not advertise a thread that cannot be resumed."""
+        app = DeepAgentsApp(thread_id="old-thread")
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._session_state = TextualSessionState(thread_id="old-thread")
+            app._lc_thread_id = "old-thread"
+
+            with (
+                patch("deepagents_code.app._new_thread_id", return_value="new-thread"),
+                patch(
+                    "deepagents_code.sessions.thread_exists",
+                    AsyncMock(return_value=False),
+                ),
+                patch.object(app, "_schedule_thread_message_link") as schedule,
+            ):
+                await app._handle_command("/clear")
+                await pilot.pause()
+
+            contents = [str(widget._content) for widget in app.query(AppMessage)]
+            assert "Previous thread: old-thread" not in contents
+            assert not any("Resume it with /threads -r" in text for text in contents)
+            schedule.assert_called_once()
+
 
 class TestCopyCommand:
     """Tests for `/copy` command behavior."""

libs/code/tests/unit_tests/test_sessions.py

@@ -246,6 +246,23 @@ class TestThreadFunctions:
             tid = asyncio.run(sessions.get_most_recent(agent_name="agent2"))
             assert tid == "thread2"
 
+    def test_get_most_recent_excludes_thread(self, temp_db):
+        """Get most recent skips the excluded thread."""
+        with patch.object(sessions, "get_db_path", return_value=temp_db):
+            tid = asyncio.run(sessions.get_most_recent(exclude_thread_id="thread3"))
+            assert tid == "thread2"
+
+    def test_get_most_recent_excludes_thread_with_agent_filter(self, temp_db):
+        """Thread exclusion composes with the agent filter."""
+        with patch.object(sessions, "get_db_path", return_value=temp_db):
+            tid = asyncio.run(
+                sessions.get_most_recent(
+                    agent_name="agent1",
+                    exclude_thread_id="thread3",
+                )
+            )
+            assert tid == "thread1"
+
     def test_get_most_recent_empty(self, tmp_path):
         """Get most recent returns None when empty."""
         db_path = tmp_path / "empty.db"

libs/code/tests/unit_tests/test_threads_resume.py

@@ -0,0 +1,350 @@
+"""Unit tests for in-TUI `/threads -r` resume and previous-thread tracking."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from deepagents_code.app import DeepAgentsApp, TextualSessionState
+
+
+class TestSessionStatePreviousThread:
+    """`reset_thread` should record the outgoing thread as `previous_thread_id`."""
+
+    def test_previous_thread_starts_none(self) -> None:
+        state = TextualSessionState(thread_id="thread-a")
+        assert state.previous_thread_id is None
+
+    def test_reset_thread_records_previous(self) -> None:
+        state = TextualSessionState(thread_id="thread-a")
+        first = state.thread_id
+        new = state.reset_thread()
+        assert state.previous_thread_id == first
+        assert new != first
+        assert state.thread_id == new
+
+    def test_reset_thread_updates_previous_each_time(self) -> None:
+        state = TextualSessionState(thread_id="thread-a")
+        second = state.reset_thread()
+        assert state.previous_thread_id == "thread-a"
+        state.reset_thread()
+        assert state.previous_thread_id == second
+
+
+def _make_app() -> DeepAgentsApp:
+    app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-1")
+    app._mount_message = AsyncMock()  # ty: ignore
+    app._show_thread_selector = AsyncMock()  # ty: ignore
+    app._resume_thread = AsyncMock()  # ty: ignore
+    return app
+
+
+class TestHandleThreadsCommand:
+    """`/threads` dispatch: bare opens the selector, `-r` resumes in place."""
+
+    async def test_bare_opens_selector(self) -> None:
+        app = _make_app()
+        await app._handle_threads_command("/threads")
+        app._show_thread_selector.assert_awaited_once()  # ty: ignore
+        app._resume_thread.assert_not_awaited()  # ty: ignore
+
+    async def test_resume_flag_resolves_and_resumes(self) -> None:
+        app = _make_app()
+        app._resolve_threads_resume_target = AsyncMock(return_value="thread-x")  # ty: ignore
+        await app._handle_threads_command("/threads -r")
+        app._resolve_threads_resume_target.assert_awaited_once_with(None)  # ty: ignore
+        app._resume_thread.assert_awaited_once_with("thread-x")  # ty: ignore
+        app._show_thread_selector.assert_not_awaited()  # ty: ignore
+
+    async def test_resume_specific_id(self) -> None:
+        app = _make_app()
+        app._resolve_threads_resume_target = AsyncMock(return_value="abc")  # ty: ignore
+        await app._handle_threads_command("/threads -r abc")
+        app._resolve_threads_resume_target.assert_awaited_once_with("abc")  # ty: ignore
+        app._resume_thread.assert_awaited_once_with("abc")  # ty: ignore
+
+    async def test_resume_long_form_flag(self) -> None:
+        app = _make_app()
+        app._resolve_threads_resume_target = AsyncMock(return_value="abc")  # ty: ignore
+        await app._handle_threads_command("/threads --resume abc")
+        app._resolve_threads_resume_target.assert_awaited_once_with("abc")  # ty: ignore
+
+    async def test_no_resume_when_target_none(self) -> None:
+        app = _make_app()
+        app._resolve_threads_resume_target = AsyncMock(return_value=None)  # ty: ignore
+        await app._handle_threads_command("/threads -r missing")
+        app._resume_thread.assert_not_awaited()  # ty: ignore
+
+    async def test_unknown_flag_shows_usage(self) -> None:
+        app = _make_app()
+        await app._handle_threads_command("/threads --nope")
+        app._show_thread_selector.assert_not_awaited()  # ty: ignore
+        app._resume_thread.assert_not_awaited()  # ty: ignore
+        message = app._mount_message.await_args.args[0]  # ty: ignore
+        assert "Usage: /threads" in str(message._content)
+
+    async def test_too_many_args_shows_usage(self) -> None:
+        app = _make_app()
+        app._resolve_threads_resume_target = AsyncMock()  # ty: ignore
+        await app._handle_threads_command("/threads -r a b")
+        app._resolve_threads_resume_target.assert_not_awaited()  # ty: ignore
+        app._resume_thread.assert_not_awaited()  # ty: ignore
+        message = app._mount_message.await_args.args[0]  # ty: ignore
+        assert "at most one thread ID" in str(message._content)
+
+
+class TestResolveResumeTarget:
+    """`-r` argument resolution against the checkpoint store and session state."""
+
+    async def test_specific_id_exists(self) -> None:
+        app = _make_app()
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=True),
+            ),
+            patch(
+                "deepagents_code.sessions.get_thread_agent",
+                AsyncMock(return_value="agent"),
+            ),
+        ):
+            target = await app._resolve_threads_resume_target("abc")
+        assert target == "abc"
+
+    async def test_specific_id_for_another_agent_is_rejected(self) -> None:
+        app = _make_app()
+        app._assistant_id = "coder"
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=True),
+            ),
+            patch(
+                "deepagents_code.sessions.get_thread_agent",
+                AsyncMock(return_value="researcher"),
+            ),
+        ):
+            target = await app._resolve_threads_resume_target("abc")
+        assert target is None
+        message = app._mount_message.await_args.args[0]  # ty: ignore
+        assert "belongs to agent 'researcher'" in str(message._content)
+
+    async def test_specific_id_missing_notifies(self) -> None:
+        app = _make_app()
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=False),
+            ),
+            patch(
+                "deepagents_code.sessions.find_similar_threads",
+                AsyncMock(return_value=[]),
+            ),
+        ):
+            target = await app._resolve_threads_resume_target("abc")
+        assert target is None
+        message = app._mount_message.await_args.args[0]  # ty: ignore
+        assert "Thread 'abc' not found." in str(message._content)
+
+    async def test_specific_id_missing_suggests_similar(self) -> None:
+        """A near-miss id surfaces the `find_similar_threads` suggestions."""
+        app = _make_app()
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=False),
+            ),
+            patch(
+                "deepagents_code.sessions.find_similar_threads",
+                AsyncMock(return_value=["abc123", "abc456"]),
+            ),
+        ):
+            target = await app._resolve_threads_resume_target("abc")
+        assert target is None
+        message = str(app._mount_message.await_args.args[0]._content)  # ty: ignore
+        assert "Did you mean: abc123, abc456?" in message
+
+    async def test_specific_id_database_failure_notifies(self) -> None:
+        """An expected thread-store error asks the user to retry."""
+        import sqlite3
+
+        app = _make_app()
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=False),
+            ),
+            patch(
+                "deepagents_code.sessions.find_similar_threads",
+                AsyncMock(side_effect=sqlite3.OperationalError("db locked")),
+            ),
+        ):
+            target = await app._resolve_threads_resume_target("abc")
+        assert target is None
+        message = app._mount_message.await_args.args[0]  # ty: ignore
+        assert "Could not look up thread history" in str(message._content)
+
+    async def test_specific_id_unexpected_error_notifies(self) -> None:
+        """A non-DB error is surfaced distinctly from a lookup failure."""
+        app = _make_app()
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=False),
+            ),
+            patch(
+                "deepagents_code.sessions.find_similar_threads",
+                AsyncMock(side_effect=RuntimeError("boom")),
+            ),
+        ):
+            target = await app._resolve_threads_resume_target("abc")
+        assert target is None
+        message = app._mount_message.await_args.args[0]  # ty: ignore
+        assert "Something went wrong resolving that thread." in str(message._content)
+
+    async def test_bare_prefers_previous_thread(self) -> None:
+        app = _make_app()
+        state = TextualSessionState(thread_id="cur")
+        state.previous_thread_id = "prev"
+        app._session_state = state
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=True),
+            ),
+            patch(
+                "deepagents_code.sessions.get_thread_agent",
+                AsyncMock(return_value="agent"),
+            ),
+        ):
+            target = await app._resolve_threads_resume_target(None)
+        assert target == "prev"
+
+    async def test_bare_skips_previous_thread_from_another_agent(self) -> None:
+        app = _make_app()
+        app._assistant_id = "coder"
+        state = TextualSessionState(thread_id="cur")
+        state.previous_thread_id = "research-thread"
+        app._session_state = state
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=True),
+            ),
+            patch(
+                "deepagents_code.sessions.get_thread_agent",
+                AsyncMock(return_value="researcher"),
+            ),
+            patch(
+                "deepagents_code.sessions.get_most_recent",
+                AsyncMock(return_value="coder-thread"),
+            ) as most_recent,
+        ):
+            target = await app._resolve_threads_resume_target(None)
+        assert target == "coder-thread"
+        most_recent.assert_awaited_once_with(
+            "coder",
+            exclude_thread_id="cur",
+        )
+
+    async def test_bare_falls_back_to_most_recent(self) -> None:
+        app = _make_app()
+        app._session_state = TextualSessionState(thread_id="cur")
+        app._assistant_id = "coder"
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=False),
+            ),
+            patch(
+                "deepagents_code.sessions.get_most_recent",
+                AsyncMock(return_value="recent"),
+            ) as most_recent,
+        ):
+            target = await app._resolve_threads_resume_target(None)
+        assert target == "recent"
+        most_recent.assert_awaited_once_with(
+            "coder",
+            exclude_thread_id="cur",
+        )
+
+    async def test_bare_previous_deleted_falls_back(self) -> None:
+        """A `previous_thread_id` pruned since `/clear` falls through to recent."""
+        app = _make_app()
+        app._assistant_id = "coder"
+        state = TextualSessionState(thread_id="cur")
+        state.previous_thread_id = "prev"
+        app._session_state = state
+        with (
+            # previous exists no more; the fallback thread does.
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=False),
+            ),
+            patch(
+                "deepagents_code.sessions.get_thread_agent",
+                AsyncMock(return_value="coder"),
+            ) as thread_agent,
+            patch(
+                "deepagents_code.sessions.get_most_recent",
+                AsyncMock(return_value="recent"),
+            ) as most_recent,
+        ):
+            target = await app._resolve_threads_resume_target(None)
+        assert target == "recent"
+        # The deleted previous never reaches the ownership check.
+        thread_agent.assert_not_awaited()
+        most_recent.assert_awaited_once_with("coder", exclude_thread_id="cur")
+
+    async def test_bare_none_when_no_threads(self) -> None:
+        app = _make_app()
+        app._session_state = TextualSessionState(thread_id="cur")
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=False),
+            ),
+            patch(
+                "deepagents_code.sessions.get_most_recent",
+                AsyncMock(return_value=None),
+            ),
+        ):
+            target = await app._resolve_threads_resume_target(None)
+        assert target is None
+        message = app._mount_message.await_args.args[0]  # ty: ignore
+        assert "No previous threads for 'agent' to resume." in str(message._content)
+
+    async def test_bare_default_agent_fallback_is_filtered(self) -> None:
+        app = _make_app()
+        app._session_state = TextualSessionState(thread_id="cur")
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=False),
+            ),
+            patch(
+                "deepagents_code.sessions.get_most_recent",
+                AsyncMock(return_value=None),
+            ) as most_recent,
+        ):
+            await app._resolve_threads_resume_target(None)
+        most_recent.assert_awaited_once_with(
+            "agent",
+            exclude_thread_id="cur",
+        )
+
+    async def test_bare_database_failure_notifies(self) -> None:
+        app = _make_app()
+        app._session_state = TextualSessionState(thread_id="cur")
+        with (
+            patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(return_value=False),
+            ),
+            patch(
+                "deepagents_code.sessions.get_most_recent",
+                AsyncMock(side_effect=RuntimeError("db unavailable")),
+            ),
+        ):
+            target = await app._resolve_threads_resume_target(None)
+        assert target is None
+        app._mount_message.assert_awaited_once()  # ty: ignore

libs/code/tests/unit_tests/tui/widgets/test_thread_selector.py

@@ -3279,6 +3279,21 @@ class TestResumeThread:
 
         assert app._should_adopt_resumed_model is False
 
+    async def test_successful_switch_records_previous_thread(self) -> None:
+        """A successful switch records the outgoing thread as previous_thread_id.
+
+        Lets a follow-up bare `/threads -r` step back to the thread just left
+        rather than resolving `previous == current` and reporting "Already on
+        thread".
+        """
+        app = self._switch_app()
+
+        await app._resume_thread("new-thread")
+
+        session_state = app._session_state
+        assert session_state is not None
+        assert session_state.previous_thread_id == "old-thread"
+
     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