feat(code): offer abort during resume (#4188)

`dcode -r` can now be aborted from the directory-switch prompt to start
a new session instead of resuming.

---

`dcode -r` resolves the most-recent (or `-r <id>`) thread and, when its
working directory differs from the current one, already shows the
cwd-switch modal (switch dir / stay here). This adds a third **abort**
option to that existing modal so the user can decline the resume and
start a fresh session instead. The abort option is gated behind
`allow_abort`, passed only by the launch-time `-r` resume path
(in-session thread switching is unchanged).

Like the cwd-switch prompt itself, abort is only offered when the
resumed thread's cwd differs from the current directory (the prompt
doesn't appear otherwise).

Made by [Open
SWE](https://openswe.vercel.app/agents/03bf37e9-2e0e-bb6e-12ad-528fc5fb3ad1)

---------

Co-authored-by: Mason Daugherty <61371264+mdrxy@users.noreply.github.com>
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: Mason Daugherty <mason@langchain.dev>
Co-authored-by: Mason Daugherty <github@mdrxy.com>

libs/code/deepagents_code/app.py

@@ -3525,7 +3525,10 @@ class DeepAgentsApp(App):
         """Resolve a `-r` resume intent into a concrete thread ID.
 
         Consumes `self._resume_thread_intent` and resolves it into a concrete
-        thread ID. Mutates `self._lc_thread_id` and optionally
+        thread ID. When the intent resolves to an existing thread whose cwd
+        differs from the current one, the cwd-switch prompt is shown with an
+        extra "abort" option; choosing it starts a fresh thread instead of
+        resuming. Mutates `self._lc_thread_id` and optionally
         `self._assistant_id` / `self._server_kwargs`. Does NOT touch
         `self._default_assistant_id` — a one-off resume should not redefine
         the user's persisted default agent. Falls back to a fresh thread on
@@ -3539,7 +3542,6 @@ class DeepAgentsApp(App):
             thread_exists,
         )
 
-        resumed_thread_id: str | None = None
         try:
             resume = self._resume_thread_intent
             self._resume_thread_intent = None  # consumed
@@ -3549,81 +3551,118 @@ class DeepAgentsApp(App):
 
             default_agent = DEFAULT_ASSISTANT_ID
 
-            if resume == "__MOST_RECENT__":
+            # Resolve the candidate thread id before any agent/model mutation,
+            # so an abort only needs to reset the thread id (no rollback).
+            via_most_recent = resume == "__MOST_RECENT__"
+            if via_most_recent:
                 agent_filter = (
                     self._assistant_id if self._assistant_id != default_agent else None
                 )
-                thread_id = await get_most_recent(agent_filter)
-                if thread_id:
-                    agent_name = await get_thread_agent(thread_id)
-                    if agent_name:
-                        self._assistant_id = agent_name
-                        if self._server_kwargs:
-                            self._server_kwargs["assistant_id"] = agent_name
-                    self._lc_thread_id = thread_id
-                    resumed_thread_id = thread_id
-                    self._should_adopt_resumed_model = not self._model_explicitly_set
-                else:
+                candidate = await get_most_recent(agent_filter)
+                if not candidate:
                     self._lc_thread_id = generate_thread_id()
+                    self._resuming = False
+                    self._sync_status_connection()
                     if agent_filter:
                         msg = f"No previous threads for '{agent_filter}', starting new."
                     else:
                         msg = "No previous threads, starting new."
                     self.notify(msg, severity="warning", markup=False)
+                    return
             elif await thread_exists(resume):
-                self._lc_thread_id = resume
-                resumed_thread_id = resume
-                self._should_adopt_resumed_model = not self._model_explicitly_set
-                if self._assistant_id == default_agent:
-                    agent_name = await get_thread_agent(resume)
-                    if agent_name:
-                        self._assistant_id = agent_name
-                        if self._server_kwargs:
-                            self._server_kwargs["assistant_id"] = agent_name
+                candidate = resume
             else:
                 # Thread not found — notify + fall back to new thread
                 self._lc_thread_id = generate_thread_id()
+                self._resuming = False
+                self._sync_status_connection()
                 similar = await find_similar_threads(resume)
                 hint = f"Thread '{resume}' not found."
                 if similar:
                     hint += f" Did you mean: {', '.join(str(t) for t in similar)}?"
                 self.notify(hint, severity="warning", timeout=6, markup=False)
-            if resumed_thread_id is not None:
-                # The cwd-switch offer is a post-resolution convenience. Isolate
-                # its failures so they can't fall into the resume-resolution
-                # handler below, which would discard the already-resolved thread
-                # and misleadingly report "Could not look up thread history."
-                try:
-                    await self._offer_thread_cwd_switch(
-                        resumed_thread_id,
-                        restart_server=False,
-                    )
-                except Exception:
-                    logger.exception(
-                        "cwd switch offer failed for resumed thread %s",
-                        resumed_thread_id,
-                    )
-                    self.notify(
-                        "Resumed the thread, but could not check its working "
-                        "directory. Local context may be stale.",
-                        severity="warning",
-                        markup=False,
-                    )
+                return
+
+            # Commit the resolved thread before the cwd-switch offer so a
+            # failure in that offer leaves the thread resolved (see the
+            # isolation guard below) rather than falling through to the
+            # resume-resolution handler.
+            self._lc_thread_id = candidate
+
+            # The cwd-switch prompt doubles as the resume confirmation: at
+            # launch it carries an extra "abort" option that starts a fresh
+            # session instead of resuming. Isolate its failures so they can't
+            # fall into the resume-resolution handler below, which would
+            # discard the already-resolved thread and misleadingly report
+            # "Could not look up thread history."
+            try:
+                cwd_choice = await self._offer_thread_cwd_switch(
+                    candidate,
+                    restart_server=False,
+                    allow_abort=True,
+                )
+            except Exception:
+                logger.exception(
+                    "cwd switch offer failed for resumed thread %s",
+                    candidate,
+                )
+                self.notify(
+                    "Resumed the thread, but could not check its working "
+                    "directory. Local context may be stale.",
+                    severity="warning",
+                    markup=False,
+                )
+                cwd_choice = "continue"
+
+            if cwd_choice == "abort":
+                # User declined the resume: start a fresh session and skip the
+                # agent/model adoption below so it inherits the launch default.
+                self._lc_thread_id = generate_thread_id()
+                self._resuming = False
+                self._sync_status_connection()
+                self.notify(
+                    "Starting a new session.",
+                    severity="information",
+                    markup=False,
+                )
+                return
+
+            # Confirmed: adopt the thread's agent for this session — always when
+            # resuming the most recent thread (bare `-r`, even over an
+            # explicitly pinned `-a`), and for an explicit `-r <id>` only when
+            # the user hasn't pinned a non-default agent — plus its persisted
+            # model.
+            self._should_adopt_resumed_model = not self._model_explicitly_set
+            if via_most_recent or self._assistant_id == default_agent:
+                agent_name = await get_thread_agent(candidate)
+                if agent_name:
+                    self._assistant_id = agent_name
+                    if self._server_kwargs:
+                        self._server_kwargs["assistant_id"] = agent_name
         except Exception:
             logger.exception("Failed to resolve resume thread %r", resume)
             self._lc_thread_id = generate_thread_id()
+            self._resuming = False
+            self._sync_status_connection()
             self.notify(
                 "Could not look up thread history. Starting new session.",
                 severity="warning",
             )
         finally:
+            # Sync the resolved (or fresh) thread id into session state before
+            # signaling completion. This must run in `finally` so an early
+            # return — a fallback to a new thread, or the user aborting the
+            # resume — can't leave `session_state.thread_id` pointing at a
+            # thread that was never adopted. `_init_session_state` may run
+            # concurrently and capture a not-yet-final id mid-prompt; this
+            # reconciles it whenever session state already exists. If session
+            # state hasn't been assigned yet, correctness instead relies on
+            # `_init_session_state` reading the now-final `_lc_thread_id` when
+            # it constructs the state.
+            if self._session_state:
+                self._session_state.thread_id = self._lc_thread_id
             self._resume_thread_resolved_event.set()
 
-        # Update session state if ready (may still be initializing in a
-        # concurrent worker)
-        if self._session_state:
-            self._session_state.thread_id = self._lc_thread_id
-
     async def _start_server_background(self) -> None:
         """Background worker: resolve resume-thread intent, start server + MCP preload.
 
@@ -15458,6 +15497,7 @@ class DeepAgentsApp(App):
         thread_id: str,
         *,
         restart_server: bool,
+        allow_abort: bool = False,
     ) -> Literal["continue", "abort"]:
         """Offer to switch to a resumed thread's cwd when it differs.
 
@@ -15467,11 +15507,16 @@ class DeepAgentsApp(App):
                 switch replaces the app-owned server so the backend runs in the
                 new cwd. When False (launch-time resume), the server has not
                 started yet, so only the process cwd is changed.
+            allow_abort: When True (launch-time `-r` resume), the prompt offers a
+                third "abort" option that declines the resume entirely.
 
         Returns:
-            `"continue"` when resume may proceed, or `"abort"` when a requested
-                switch was accepted but failed (the caller should stop
-                the resume).
+            `"continue"` when resume may proceed, or `"abort"` when the user
+                declined the resume or a requested switch was accepted but
+                failed (the caller should stop the resume). The two abort
+                sources are mode-exclusive: the user-declined abort fires only
+                when `allow_abort` is True, and the switch-failed abort only
+                when `restart_server` is True.
         """
         target = await self._thread_cwd_mismatch(thread_id)
         if target is None:
@@ -15487,8 +15532,11 @@ class DeepAgentsApp(App):
                 current_cwd=self._cwd,
                 thread_cwd=str(target),
                 project_settings_change_detected=project_settings_change_detected,
+                allow_abort=allow_abort,
             )
         )
+        if choice == "abort":
+            return "abort"
         if choice == "switch":
             if restart_server:
                 return await self._replace_server_after_cwd_switch(target)

libs/code/deepagents_code/widgets/cwd_switch.py

@@ -17,8 +17,12 @@ if TYPE_CHECKING:
     from deepagents_code.app import DeepAgentsApp
 
 
-CwdSwitchChoice = Literal["switch", "stay"]
-"""Outcome of the cwd switch prompt."""
+CwdSwitchChoice = Literal["switch", "stay", "abort"]
+"""Outcome of the cwd switch prompt.
+
+`"abort"` is only offered when the prompt is opened with `allow_abort=True`
+(launch-time `-r` resume) and means "don't resume; start a new session".
+"""
 
 
 class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
@@ -30,6 +34,7 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
     BINDINGS: ClassVar[list[BindingType]] = [
         Binding("enter", "switch", "Switch", show=False, priority=True),
         Binding("escape", "stay", "Stay", show=False, priority=True),
+        Binding("a", "abort", "Abort", show=False, priority=True),
         Binding(
             "ctrl+c",
             "quit_or_interrupt",
@@ -81,12 +86,14 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
         current_cwd: str,
         thread_cwd: str,
         project_settings_change_detected: bool = False,
+        allow_abort: bool = False,
     ) -> None:
         """Initialize the prompt."""
         super().__init__()
         self._current_cwd = current_cwd
         self._thread_cwd = thread_cwd
         self._project_settings_change_detected = project_settings_change_detected
+        self._allow_abort = allow_abort
 
     def _body_text(self) -> str:
         """Return the prompt body text."""
@@ -98,6 +105,11 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
             if self._project_settings_change_detected
             else ""
         )
+        abort_note = (
+            "\n\nOr abort to start a new session instead of resuming."
+            if self._allow_abort
+            else ""
+        )
         return (
             "This thread was last used from:\n"
             f"  {target}\n\n"
@@ -106,7 +118,7 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
             "Switch if you want local context, project instructions, skills, "
             "MCP config, and env files to match the original directory. Stay "
             "here if you intentionally want to continue this thread against "
-            f"the current directory.{settings_note}"
+            f"the current directory.{settings_note}{abort_note}"
         )
 
     def compose(self) -> ComposeResult:
@@ -126,8 +138,13 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
                 classes="cwd-switch-body",
                 markup=False,
             )
+            help_text = (
+                "Enter: switch · Esc: stay here · A: don't resume"
+                if self._allow_abort
+                else "Enter: switch · Esc: stay here"
+            )
             yield Static(
-                "Enter: switch · Esc: stay here",
+                help_text,
                 classes="cwd-switch-help",
                 markup=False,
             )
@@ -136,6 +153,25 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
         """Focus the modal so screen bindings work after nested modal flows."""
         self.focus()
 
+    def check_action(
+        self,
+        action: str,
+        parameters: tuple[object, ...],  # noqa: ARG002  # required by Textual's DOMNode.check_action override signature
+    ) -> bool | None:
+        """Disable the `abort` binding unless the prompt was opened for it.
+
+        Makes the disabled state first-class: when `allow_abort` is False the
+        `a` key is not bound to anything (it passes through) rather than firing
+        a no-op action.
+
+        Returns:
+            `self._allow_abort` for the `abort` action so the binding is only
+                active when abort was offered; `True` for every other action.
+        """
+        if action == "abort":
+            return self._allow_abort
+        return True
+
     def action_switch(self) -> None:
         """Dismiss with `switch`."""
         self.dismiss("switch")
@@ -144,6 +180,12 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
         """Dismiss with `stay`."""
         self.dismiss("stay")
 
+    def action_abort(self) -> None:
+        """Dismiss with `abort` to skip the resume, when the prompt allows it."""
+        if not self._allow_abort:
+            return
+        self.dismiss("abort")
+
     def action_cancel(self) -> None:
         """Treat cancellation as staying in the current cwd."""
         self.action_stay()

libs/code/tests/unit_tests/test_app.py

@@ -14017,11 +14017,17 @@ class TestResolveResumeThread:
             assert app._should_adopt_resumed_model is False
 
     async def test_no_previous_thread_leaves_adoption_flag_unset(self) -> None:
-        """Falling back to a fresh thread must not arm model adoption."""
+        """Falling back to a fresh thread must not arm model adoption.
+
+        Also verifies the status bar stops showing "Resuming" when no
+        prior thread was found to resume.
+        """
         app = self._make_app("agent")
 
         async with app.run_test() as pilot:
             await pilot.pause()
+            app._connecting = True
+            app._resuming = True
             app._resume_thread_intent = "__MOST_RECENT__"
             with patch(
                 "deepagents_code.sessions.get_most_recent",
@@ -14030,6 +14036,9 @@ class TestResolveResumeThread:
                 await app._resolve_resume_thread()
 
             assert app._should_adopt_resumed_model is False
+            assert app._resuming is False
+            assert app._status_bar is not None
+            assert app._status_bar.connection_state == "connecting"
 
     async def test_most_recent_resume_arms_adoption_flag(self) -> None:
         """`-r` (most recent) resolving to a thread also arms model adoption.
@@ -14056,6 +14065,244 @@ class TestResolveResumeThread:
 
             assert app._should_adopt_resumed_model is True
 
+    async def test_abort_starts_new_thread(self) -> None:
+        """Choosing abort in the cwd prompt starts a fresh thread.
+
+        Also verifies the status bar drops the "Resuming" label in favor of
+        "Connecting" so the footer doesn't lie while the server boots.
+        """
+        app = self._make_app("agent")
+
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._connecting = True
+            app._resuming = True
+            app._resume_thread_intent = "some-thread"
+            app._offer_thread_cwd_switch = AsyncMock(  # ty: ignore[invalid-assignment]
+                return_value="abort"
+            )
+            with (
+                patch(
+                    "deepagents_code.sessions.thread_exists",
+                    AsyncMock(return_value=True),
+                ),
+                patch(
+                    "deepagents_code.sessions.get_thread_agent",
+                    AsyncMock(return_value="coder"),
+                ),
+            ):
+                await app._resolve_resume_thread()
+
+            assert app._lc_thread_id != "some-thread"
+            assert app._assistant_id == "agent"
+            assert app._should_adopt_resumed_model is False
+            assert app._resuming is False
+            assert app._status_bar is not None
+            assert app._status_bar.connection_state == "connecting"
+
+    async def test_abort_most_recent_starts_new_thread(self) -> None:
+        """Aborting a bare `-r` cwd prompt also starts a fresh thread.
+
+        Mirrors `test_abort_starts_new_thread` for the `__MOST_RECENT__` branch
+        and asserts the status bar flips from "Resuming" to "Connecting".
+        """
+        app = self._make_app("agent")
+
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._connecting = True
+            app._resuming = True
+            app._resume_thread_intent = "__MOST_RECENT__"
+            app._offer_thread_cwd_switch = AsyncMock(  # ty: ignore[invalid-assignment]
+                return_value="abort"
+            )
+            with (
+                patch(
+                    "deepagents_code.sessions.get_most_recent",
+                    AsyncMock(return_value="recent-thread"),
+                ),
+                patch(
+                    "deepagents_code.sessions.get_thread_agent",
+                    AsyncMock(return_value="coder"),
+                ),
+            ):
+                await app._resolve_resume_thread()
+
+            assert app._lc_thread_id != "recent-thread"
+            assert app._should_adopt_resumed_model is False
+            assert app._resuming is False
+            assert app._status_bar is not None
+            assert app._status_bar.connection_state == "connecting"
+
+    async def test_resume_offers_abort_option_at_launch(self) -> None:
+        """The launch-time cwd prompt is invoked with `allow_abort=True`."""
+        app = self._make_app("agent")
+
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._resume_thread_intent = "some-thread"
+            offer = AsyncMock(return_value="continue")
+            app._offer_thread_cwd_switch = offer  # ty: ignore[invalid-assignment]
+            with (
+                patch(
+                    "deepagents_code.sessions.thread_exists",
+                    AsyncMock(return_value=True),
+                ),
+                patch(
+                    "deepagents_code.sessions.get_thread_agent",
+                    AsyncMock(return_value=None),
+                ),
+            ):
+                await app._resolve_resume_thread()
+
+            assert offer.await_args is not None
+            assert offer.await_args.kwargs["allow_abort"] is True
+            assert app._lc_thread_id == "some-thread"
+
+    async def test_abort_syncs_session_state_to_fresh_thread(self) -> None:
+        """Aborting points session state at the fresh id, not the declined thread.
+
+        Guards the race where `_init_session_state` captures the candidate
+        thread into `session_state.thread_id` while the cwd prompt is open: the
+        finally-block sync must overwrite it with the fresh id on abort.
+        """
+        app = self._make_app("agent")
+
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            assert app._session_state is not None
+            # Simulate session state having captured the candidate mid-prompt.
+            app._session_state.thread_id = "some-thread"
+            app._resume_thread_intent = "some-thread"
+            app._offer_thread_cwd_switch = AsyncMock(  # ty: ignore[invalid-assignment]
+                return_value="abort"
+            )
+            with (
+                patch(
+                    "deepagents_code.sessions.thread_exists",
+                    AsyncMock(return_value=True),
+                ),
+                patch(
+                    "deepagents_code.sessions.get_thread_agent",
+                    AsyncMock(return_value="coder"),
+                ),
+            ):
+                await app._resolve_resume_thread()
+
+            assert app._lc_thread_id != "some-thread"
+            assert app._session_state.thread_id == app._lc_thread_id
+
+    async def test_thread_not_found_clears_resuming_status(self) -> None:
+        """A non-existent thread id falls back to fresh and drops "Resuming".
+
+        Covers the `elif await thread_exists(resume)` → `else` branch where
+        the thread id is never adopted.
+        """
+        app = self._make_app("agent")
+
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._connecting = True
+            app._resuming = True
+            app._resume_thread_intent = "ghost-thread"
+            with (
+                patch(
+                    "deepagents_code.sessions.thread_exists",
+                    AsyncMock(return_value=False),
+                ),
+                patch(
+                    "deepagents_code.sessions.find_similar_threads",
+                    AsyncMock(return_value=[]),
+                ),
+            ):
+                await app._resolve_resume_thread()
+
+            assert app._lc_thread_id != "ghost-thread"
+            assert app._should_adopt_resumed_model is False
+            assert app._resuming is False
+            assert app._status_bar is not None
+            assert app._status_bar.connection_state == "connecting"
+
+    async def test_resolve_exception_clears_resuming_status(self) -> None:
+        """An unhandled error in resume resolution drops "Resuming".
+
+        Guards the `except Exception` handler path.
+        """
+        app = self._make_app("agent")
+
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._connecting = True
+            app._resuming = True
+            app._resume_thread_intent = "some-thread"
+            with patch(
+                "deepagents_code.sessions.thread_exists",
+                AsyncMock(side_effect=RuntimeError("db offline")),
+            ):
+                await app._resolve_resume_thread()
+
+            assert app._resuming is False
+            assert app._status_bar is not None
+            assert app._status_bar.connection_state == "connecting"
+
+    async def test_offer_failure_still_resumes_and_adopts(self) -> None:
+        """A raising cwd-switch offer must not abandon the resolved thread.
+
+        Locks the central invariant of the resolve refactor: the candidate is
+        committed to `_lc_thread_id` *before* the offer, and a failure in the
+        offer is isolated to `"continue"` so agent + model adoption still run.
+        The failure must not fall through to the outer handler (which would
+        discard the thread and start a fresh session).
+        """
+        app = self._make_app("agent")
+
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._resume_thread_intent = "some-thread"
+            app._offer_thread_cwd_switch = AsyncMock(  # ty: ignore[invalid-assignment]
+                side_effect=RuntimeError("cwd probe failed")
+            )
+            with (
+                patch(
+                    "deepagents_code.sessions.thread_exists",
+                    AsyncMock(return_value=True),
+                ),
+                patch(
+                    "deepagents_code.sessions.get_thread_agent",
+                    AsyncMock(return_value="coder"),
+                ),
+            ):
+                await app._resolve_resume_thread()
+
+            assert app._lc_thread_id == "some-thread"
+            assert app._assistant_id == "coder"
+            assert app._should_adopt_resumed_model is True
+
+    async def test_fallback_syncs_session_state_to_fresh_thread(self) -> None:
+        """A non-abort fallback early return still syncs session state.
+
+        Mirrors `test_abort_syncs_session_state_to_fresh_thread` for the
+        "no previous threads" fallback: the `finally`-block sync is the only
+        thing that stops `session_state.thread_id` from being left pointing at
+        a stale id after the early return.
+        """
+        app = self._make_app("agent")
+
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            assert app._session_state is not None
+            # Simulate session state having captured a stale id before resolve.
+            app._session_state.thread_id = "stale-thread"
+            app._resume_thread_intent = "__MOST_RECENT__"
+            with patch(
+                "deepagents_code.sessions.get_most_recent",
+                AsyncMock(return_value=None),
+            ):
+                await app._resolve_resume_thread()
+
+            assert app._lc_thread_id != "stale-thread"
+            assert app._session_state.thread_id == app._lc_thread_id
+
 
 def _missing_dep_entry(
     tool: str = "ripgrep",

libs/code/tests/unit_tests/test_cwd_switch.py

@@ -4,9 +4,16 @@ from __future__ import annotations
 
 from unittest.mock import MagicMock
 
+from textual.app import App, ComposeResult
 from textual.binding import Binding
+from textual.widgets import Static
 
-from deepagents_code.widgets.cwd_switch import CwdSwitchPromptScreen
+from deepagents_code.widgets.cwd_switch import CwdSwitchChoice, CwdSwitchPromptScreen
+
+
+class _CwdSwitchTestApp(App[None]):
+    def compose(self) -> ComposeResult:
+        yield Static("base")
 
 
 class TestCwdSwitchPromptScreen:
@@ -83,3 +90,64 @@ class TestCwdSwitchPromptScreen:
         screen, dismiss = self._screen()
         screen.action_cancel()
         dismiss.assert_called_once_with("stay")
+
+
+class TestCwdSwitchAbortOption:
+    """The launch-time `-r` resume prompt adds a third `abort` option."""
+
+    def test_abort_binding_present(self) -> None:
+        """The modal binds `a` to the abort action."""
+        bindings = [b for b in CwdSwitchPromptScreen.BINDINGS if isinstance(b, Binding)]
+        bindings_by_key = {b.key: b for b in bindings}
+
+        assert bindings_by_key["a"].action == "abort"
+
+    def test_help_and_body_mention_abort_only_when_allowed(self) -> None:
+        """The abort affordance is described only when `allow_abort` is set."""
+        without = CwdSwitchPromptScreen(current_cwd="/a", thread_cwd="/b")
+        with_abort = CwdSwitchPromptScreen(
+            current_cwd="/a", thread_cwd="/b", allow_abort=True
+        )
+
+        assert "new session" not in without._body_text()
+        assert "new session" in with_abort._body_text()
+
+    def test_action_abort_dismisses_abort_when_allowed(self) -> None:
+        """Abort resolves the prompt to `abort` when offered."""
+        screen = CwdSwitchPromptScreen(
+            current_cwd="/a", thread_cwd="/b", allow_abort=True
+        )
+        dismiss = MagicMock()
+        screen.dismiss = dismiss  # ty: ignore[invalid-assignment]
+
+        screen.action_abort()
+
+        dismiss.assert_called_once_with("abort")
+
+    def test_action_abort_is_noop_when_not_allowed(self) -> None:
+        """Abort does nothing when the prompt was not opened with `allow_abort`."""
+        screen = CwdSwitchPromptScreen(current_cwd="/a", thread_cwd="/b")
+        dismiss = MagicMock()
+        screen.dismiss = dismiss  # ty: ignore[invalid-assignment]
+
+        screen.action_abort()
+
+        dismiss.assert_not_called()
+
+    async def test_pressing_a_aborts_when_allowed(self) -> None:
+        """Pressing `a` resolves the prompt to `abort` when offered."""
+        app = _CwdSwitchTestApp()
+        async with app.run_test() as pilot:
+            outcomes: list[CwdSwitchChoice | None] = []
+            app.push_screen(
+                CwdSwitchPromptScreen(
+                    current_cwd="/a", thread_cwd="/b", allow_abort=True
+                ),
+                outcomes.append,
+            )
+            await pilot.pause()
+
+            await pilot.press("a")
+            await pilot.pause()
+
+            assert outcomes == ["abort"]

libs/code/tests/unit_tests/test_thread_selector.py

@@ -3099,9 +3099,17 @@ class TestResumeThread:
         _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())
+        offer_cwd_switch = AsyncMock(return_value="continue")
+        _app_test_double(app)._offer_thread_cwd_switch = offer_cwd_switch
 
         await app._resume_thread("new-thread")
 
+        # In-session switches never offer abort — that is launch-time only.
+        # Exact-args match fails if `allow_abort=True` ever leaks in here.
+        offer_cwd_switch.assert_awaited_once_with(
+            "new-thread",
+            restart_server=True,
+        )
         assert app._lc_thread_id == "new-thread"
         assert app._session_state.thread_id == "new-thread"
         app._pending_messages.clear.assert_called_once()