feat(code): offer abort in /threads cwd-switch prompt (#4583)

The `/threads` directory prompt now lets you abort a thread switch and remain on the current thread. --- When `/threads` selects a thread that was last used in another directory, the directory prompt now offers a third choice: press `a` to cancel the thread switch and keep the current thread and directory. Previously, the prompt only allowed switching to the thread’s original directory or continuing in the current directory. This matches the launch-time resume prompt while using wording specific to an in-session thread switch. Made by [Open SWE](https://openswe.vercel.app/agents/50cf0c96-66d9-b2ed-0ff1-5c67fa94300f) --------- Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
2026-07-22 01:35:28 -04:00 · 2026-07-13 02:01:58 -04:00
parent 4d2aa8a968
commit aaeac99d9f
6 changed files with 357 additions and 56 deletions
@@ -365,6 +365,7 @@ if TYPE_CHECKING:
    from deepagents_code.tui.widgets.approval import ApprovalMenu
    from deepagents_code.tui.widgets.ask_user import AskUserMenu
    from deepagents_code.tui.widgets.auth import AuthManagerScreen
+    from deepagents_code.tui.widgets.cwd_switch import CwdSwitchAbortMode
    from deepagents_code.tui.widgets.goal_review import GoalReviewMenu, GoalReviewResult
    from deepagents_code.tui.widgets.model_selector import ModelSelectorScreen
    from deepagents_code.tui.widgets.notification_center import (
@@ -3795,7 +3796,7 @@ class DeepAgentsApp(App):
                cwd_choice = await self._offer_thread_cwd_switch(
                    candidate,
                    restart_server=False,
-                    allow_abort=True,
+                    abort="resume",
                )
            except Exception:
                logger.exception(
@@ -16900,7 +16901,7 @@ class DeepAgentsApp(App):
        thread_id: str,
        *,
        restart_server: bool,
-        allow_abort: bool = False,
+        abort: CwdSwitchAbortMode | None = None,
    ) -> Literal["continue", "abort"]:
        """Offer to switch to a resumed thread's cwd when it differs.

@@ -16910,16 +16911,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.
+            abort: When set, the prompt offers a third "abort" option that
+                declines the resume/switch entirely; the mode selects its
+                wording (see `CwdSwitchAbortMode`). `None` hides the option.

        Returns:
            `"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.
+                declined the resume/switch or a requested switch was accepted but
+                failed (the caller should stop the resume). The user-declined
+                abort fires only when `abort` is set, and the switch-failed
+                abort only when `restart_server` is True.
        """
        target = await self._thread_cwd_mismatch(thread_id)
        if target is None:
@@ -16935,14 +16936,28 @@ class DeepAgentsApp(App):
                current_cwd=self._cwd,
                thread_cwd=str(target),
                project_settings_change_detected=project_settings_change_detected,
-                allow_abort=allow_abort,
+                abort=abort,
            )
        )
        if choice == "abort":
            return "abort"
        if choice == "switch":
            if restart_server:
-                return await self._replace_server_after_cwd_switch(target)
+                outcome = await self._replace_server_after_cwd_switch(target)
+                if outcome == "abort":
+                    # A failed restart returns "abort" just like a user-declined
+                    # abort, so the caller cannot tell them apart.
+                    # `_replace_server_after_cwd_switch` already rolled back and
+                    # notified, but that toast is transient -- leave a persistent
+                    # in-chat record so a failed switch is not mistaken for a
+                    # deliberate cancel.
+                    await self._mount_message(
+                        AppMessage(
+                            "Could not switch to the thread's directory; staying "
+                            "on the current thread.",
+                        )
+                    )
+                return outcome
            self._preserve_launch_relative_server_paths(Path(self._cwd))
            self._switch_process_cwd(target)
            return "continue"
@@ -17039,6 +17054,7 @@ class DeepAgentsApp(App):
            cwd_choice = await self._offer_thread_cwd_switch(
                thread_id,
                restart_server=True,
+                abort="thread_switch",
            )
            if cwd_choice == "abort":
                return
@@ -17059,7 +17075,11 @@ class DeepAgentsApp(App):
        prev_session_thread = self._session_state.thread_id
        prev_cwd = Path(self._cwd)

-        cwd_choice = await self._offer_thread_cwd_switch(thread_id, restart_server=True)
+        cwd_choice = await self._offer_thread_cwd_switch(
+            thread_id,
+            restart_server=True,
+            abort="thread_switch",
+        )
        if cwd_choice == "abort":
            return

@@ -1,8 +1,8 @@
-"""Prompt for switching cwd when resuming threads."""
+"""Prompt for switching cwd when resuming or switching threads."""

 from __future__ import annotations

-from typing import TYPE_CHECKING, ClassVar, Literal, cast
+from typing import TYPE_CHECKING, ClassVar, Literal, assert_never, cast

 from textual.binding import Binding, BindingType
 from textual.containers import Vertical
@@ -20,13 +20,26 @@ if TYPE_CHECKING:
 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".
+`"abort"` is only offered when the prompt is opened with an `abort` mode set;
+its meaning depends on that mode (see `CwdSwitchAbortMode`).
+"""
+
+CwdSwitchAbortMode = Literal["resume", "thread_switch"]
+"""Which flow opened an abort-capable prompt, selecting the abort wording.
+
+Passed as the prompt's `abort` argument; `None` there means abort is not
+offered. `"resume"` is the launch-time `-r` resume (abort starts a new
+session); `"thread_switch"` is the in-session `/threads` switcher (abort keeps
+the current thread). Its members are kept disjoint from `CwdSwitchChoice`'s as a
+naming convention -- not a type guarantee (these are distinct `Literal` types
+used at distinct sites, so a checker already keeps them apart) -- so a mode token
+is never mistaken for an outcome token in a log, test, or debugger.
+`test_abort_mode_tokens_disjoint_from_choice` enforces it.
 """


 class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
-    """Modal asking whether to switch cwd before resuming a thread."""
+    """Modal asking whether to switch cwd when resuming or switching to a thread."""

    can_focus = True
    can_focus_children = False
@@ -86,14 +99,29 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
        current_cwd: str,
        thread_cwd: str,
        project_settings_change_detected: bool = False,
-        allow_abort: bool = False,
+        abort: CwdSwitchAbortMode | None = None,
    ) -> 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
+        self._abort: CwdSwitchAbortMode | None = abort
+
+    def _title_text(self) -> str:
+        """Return the title, phrased for the flow that opened the prompt.
+
+        The in-session `/threads` switcher (`"thread_switch"`) asks about
+        switching; every other flow (launch-time resume, or no abort mode) asks
+        about resuming. Structured for `assert_never` exhaustiveness so a new
+        mode fails statically here rather than silently inheriting the resume
+        wording.
+        """
+        if self._abort is None or self._abort == "resume":
+            return "Resume from the thread's original directory?"
+        if self._abort == "thread_switch":
+            return "Switch to the thread's original directory?"
+        assert_never(self._abort)

    def _body_text(self) -> str:
        """Return the prompt body text."""
@@ -105,11 +133,12 @@ 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 ""
-        )
+        if self._abort is None or self._abort == "thread_switch":
+            abort_note = ""
+        elif self._abort == "resume":
+            abort_note = "\n\nOr abort to start a new session instead of resuming."
+        else:
+            assert_never(self._abort)
        return (
            "This thread was last used from:\n"
            f"  {target}\n\n"
@@ -121,6 +150,19 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
            f"the current directory.{settings_note}{abort_note}"
        )

+    def _help_text(self) -> str:
+        """Return the help line text, naming the mode's abort action if offered."""
+        help_text = "Enter: switch · Esc: stay in cwd"
+        if self._abort is None:
+            return help_text
+        if self._abort == "resume":
+            abort_help = "A: don't resume"
+        elif self._abort == "thread_switch":
+            abort_help = "A: don't switch"
+        else:
+            assert_never(self._abort)
+        return f"{help_text} · {abort_help}"
+
    def compose(self) -> ComposeResult:
        """Compose the confirmation dialog.

@@ -129,7 +171,7 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
        """
        with Vertical():
            yield Static(
-                "Resume from the thread's original directory?",
+                self._title_text(),
                classes="cwd-switch-title",
                markup=False,
            )
@@ -138,13 +180,8 @@ 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(
-                help_text,
+                self._help_text(),
                classes="cwd-switch-help",
                markup=False,
            )
@@ -160,16 +197,22 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
    ) -> 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.
+        Textual gates a binding's action on a truthy `check_action` result, so
+        both `False` and `None` stop `a` from dispatching `action_abort` -- in
+        neither case does the key fire the action, and it falls through the same
+        way. They differ only in footer presentation (`False` hides the binding,
+        `None` shows it grayed out), which is moot here anyway: every binding is
+        declared `show=False` and the modal renders its own help line instead of
+        a `Footer`. We return `False` to mark the disabled state explicitly; the
+        actual inertness backstop is `action_abort`'s own `self._abort is None`
+        guard, should the action ever be dispatched.

        Returns:
-            `self._allow_abort` for the `abort` action so the binding is only
-                active when abort was offered; `True` for every other action.
+            `self._abort is not None` for the `abort` action, so the binding is
+                enabled only when abort was offered; `True` for every other action.
        """
        if action == "abort":
-            return self._allow_abort
+            return self._abort is not None
        return True

    def action_switch(self) -> None:
@@ -181,8 +224,8 @@ class CwdSwitchPromptScreen(ModalScreen[CwdSwitchChoice]):
        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:
+        """Dismiss with `abort` to skip the resume/switch, when the prompt allows it."""
+        if self._abort is None:
            return
        self.dismiss("abort")

@@ -15270,7 +15270,7 @@ class TestResolveResumeThread:
            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`."""
+        """The launch-time cwd prompt is invoked with the `resume` abort mode."""
        app = self._make_app("agent")

        async with app.run_test() as pilot:
@@ -15291,7 +15291,7 @@ class TestResolveResumeThread:
                await app._resolve_resume_thread()

            assert offer.await_args is not None
-            assert offer.await_args.kwargs["allow_abort"] is True
+            assert offer.await_args.kwargs["abort"] == "resume"
            assert app._lc_thread_id == "some-thread"

    async def test_abort_syncs_session_state_to_fresh_thread(self) -> None:
@@ -21774,6 +21774,93 @@ class TestResumeThreadCwdSwitch:
        notify.assert_called_once()
        assert "Cached local context may be stale" in notify.call_args.args[0]

+    async def test_offer_switch_failure_returns_abort(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """An accepted server-backed switch that fails to restart returns `abort`.
+
+        With `abort="thread_switch"` set, both abort sources (user-declined and
+        switch-failed) are reachable for a single call configuration -- not
+        concurrently, but the return value alone cannot say which fired. This
+        pins the switch-failed source end-to-end through the real method (the
+        other abort tests either use `restart_server=False` or mock the whole
+        method out) and checks it leaves a persistent in-chat record.
+        """
+        current = tmp_path / "current"
+        target = tmp_path / "target"
+        current.mkdir()
+        target.mkdir()
+        monkeypatch.chdir(current)
+        app = DeepAgentsApp(thread_id="thread-1", cwd=current)
+        app._push_screen_wait = AsyncMock(return_value="switch")  # ty: ignore[invalid-assignment]
+        monkeypatch.setattr(
+            app,
+            "_preview_project_settings_change",
+            AsyncMock(return_value=False),
+        )
+        replace = AsyncMock(return_value="abort")
+        app._replace_server_after_cwd_switch = replace  # ty: ignore[invalid-assignment]
+        mount = AsyncMock()
+        app._mount_message = mount  # ty: ignore[invalid-assignment]
+
+        with patch("deepagents_code.sessions.get_thread_cwd", return_value=str(target)):
+            outcome = await app._offer_thread_cwd_switch(
+                "thread-1", restart_server=True, abort="thread_switch"
+            )
+
+        assert outcome == "abort"
+        replace.assert_awaited_once()
+        # The failed switch must be surfaced as a durable message, distinct from
+        # the transient toast `_replace_server_after_cwd_switch` already raised.
+        mount.assert_awaited_once()
+        assert mount.await_args is not None
+        # `AppMessage` stores its raw text in `_content` (its serialization
+        # field); read it directly so a rename fails loudly rather than
+        # defaulting to "" and masking the mismatch.
+        mounted = mount.await_args.args[0]
+        assert isinstance(mounted, AppMessage)
+        assert mounted._content == (
+            "Could not switch to the thread's directory; staying on the current thread."
+        )
+
+    async def test_offer_user_declined_returns_abort(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """A user-declined abort returns `abort` without attempting a switch.
+
+        Complements `test_offer_switch_failure_returns_abort`: that pins the
+        switch-failed abort source through the real method; this pins the
+        user-declined source (the `choice == "abort"` early return), which every
+        other abort test reaches only by mocking `_offer_thread_cwd_switch` out.
+        """
+        current = tmp_path / "current"
+        target = tmp_path / "target"
+        current.mkdir()
+        target.mkdir()
+        monkeypatch.chdir(current)
+        app = DeepAgentsApp(thread_id="thread-1", cwd=current)
+        app._push_screen_wait = AsyncMock(return_value="abort")  # ty: ignore[invalid-assignment]
+        monkeypatch.setattr(
+            app,
+            "_preview_project_settings_change",
+            AsyncMock(return_value=False),
+        )
+        replace = AsyncMock(return_value="continue")
+        app._replace_server_after_cwd_switch = replace  # ty: ignore[invalid-assignment]
+
+        with patch("deepagents_code.sessions.get_thread_cwd", return_value=str(target)):
+            outcome = await app._offer_thread_cwd_switch(
+                "thread-1", restart_server=True, abort="thread_switch"
+            )
+
+        assert outcome == "abort"
+        # A declined abort must not touch the server.
+        replace.assert_not_awaited()
+
    async def test_no_prompt_when_thread_cwd_matches_current(
        self,
        tmp_path: Path,
@@ -21842,6 +21929,57 @@ class TestResumeThreadCwdSwitch:
        set_spinner.assert_has_awaits([call("Loading thread"), call(None)])
        load_thread_history.assert_not_awaited()

+    async def test_threads_switch_offers_abort_and_cancels(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """The `/threads` switcher offers abort; aborting keeps the current thread."""
+        monkeypatch.chdir(tmp_path)
+        app = DeepAgentsApp(thread_id="old-thread", cwd=tmp_path)
+        app._agent = MagicMock()
+        app._session_state = TextualSessionState(thread_id="old-thread")
+        app._lc_thread_id = "old-thread"
+        app._mount_message = AsyncMock()  # ty: ignore[invalid-assignment]
+        fetch = AsyncMock()
+        app._fetch_thread_history_data = fetch  # ty: ignore[invalid-assignment]
+        offer = AsyncMock(return_value="abort")
+        app._offer_thread_cwd_switch = offer  # ty: ignore[invalid-assignment]
+
+        await app._resume_thread("new-thread")
+
+        assert offer.await_args is not None
+        assert offer.await_args.kwargs["abort"] == "thread_switch"
+        # Aborting must not switch threads or load history.
+        assert app._session_state.thread_id == "old-thread"
+        assert app._lc_thread_id == "old-thread"
+        fetch.assert_not_awaited()
+        # Abort returns before the switch lock is acquired; leaving it set would
+        # permanently block `/threads` for the session.
+        assert app._thread_switching is False
+
+    async def test_threads_reselect_offers_abort(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """Reselecting the current thread also offers abort and cancels silently."""
+        monkeypatch.chdir(tmp_path)
+        app = DeepAgentsApp(thread_id="thread-1", cwd=tmp_path)
+        app._agent = MagicMock()
+        app._session_state = TextualSessionState(thread_id="thread-1")
+        app._lc_thread_id = "thread-1"
+        mount = AsyncMock()
+        app._mount_message = mount  # ty: ignore[invalid-assignment]
+        offer = AsyncMock(return_value="abort")
+        app._offer_thread_cwd_switch = offer  # ty: ignore[invalid-assignment]
+
+        await app._resume_thread("thread-1")
+
+        assert offer.await_args is not None
+        assert offer.await_args.kwargs["abort"] == "thread_switch"
+        mount.assert_not_awaited()
+
    # --- _resolve_thread_cwd_mismatch (pure staticmethod) ---

    def test_resolve_relative_path_is_unavailable(self) -> None:
@@ -2,6 +2,7 @@

 from __future__ import annotations

+from typing import get_args
 from unittest.mock import MagicMock

 from textual.app import App, ComposeResult
@@ -9,6 +10,7 @@ from textual.binding import Binding
 from textual.widgets import Static

 from deepagents_code.tui.widgets.cwd_switch import (
+    CwdSwitchAbortMode,
    CwdSwitchChoice,
    CwdSwitchPromptScreen,
 )
@@ -105,20 +107,93 @@ class TestCwdSwitchAbortOption:

        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."""
+    def test_check_action_gates_abort_binding_by_mode(self) -> None:
+        """`check_action` enables the `a` binding only when an abort mode is set.
+
+        Guards the binding-disable against regressing to always-enabled — a
+        regression the direct `action_abort` tests would miss because they call
+        the action method directly, bypassing the binding layer.
+        """
+
+        def abort_enabled(abort: CwdSwitchAbortMode | None) -> bool | None:
+            screen = CwdSwitchPromptScreen(
+                current_cwd="/a", thread_cwd="/b", abort=abort
+            )
+            return screen.check_action("abort", ())
+
+        assert abort_enabled("resume") is True
+        assert abort_enabled("thread_switch") is True
+        assert abort_enabled(None) is False
+
+        # Non-abort actions are always allowed, regardless of mode.
+        no_abort = CwdSwitchPromptScreen(current_cwd="/a", thread_cwd="/b")
+        assert no_abort.check_action("switch", ()) is True
+
+    def test_body_mentions_abort_only_when_allowed(self) -> None:
+        """The abort affordance is described only when an abort mode is set."""
        without = CwdSwitchPromptScreen(current_cwd="/a", thread_cwd="/b")
        with_abort = CwdSwitchPromptScreen(
-            current_cwd="/a", thread_cwd="/b", allow_abort=True
+            current_cwd="/a", thread_cwd="/b", abort="resume"
        )

        assert "new session" not in without._body_text()
        assert "new session" in with_abort._body_text()

+    def test_switch_mode_omits_abort_body_note(self) -> None:
+        """The in-session `/threads` abort is described only in the help line."""
+        switch = CwdSwitchPromptScreen(
+            current_cwd="/a",
+            thread_cwd="/b",
+            abort="thread_switch",
+        )
+
+        body = switch._body_text()
+        assert "new session" not in body
+        assert "instead of switching" not in body
+        assert "keep your current thread" not in body
+
+    def test_title_reflects_flow(self) -> None:
+        """The title asks about switching for `/threads`, resuming otherwise."""
+
+        def title(abort: CwdSwitchAbortMode | None) -> str:
+            return CwdSwitchPromptScreen(
+                current_cwd="/a", thread_cwd="/b", abort=abort
+            )._title_text()
+
+        assert title("thread_switch") == "Switch to the thread's original directory?"
+        assert title("resume") == "Resume from the thread's original directory?"
+        assert title(None) == "Resume from the thread's original directory?"
+
+    def test_help_text_names_mode_specific_abort_action(self) -> None:
+        """The help line shows the mode's abort wording, or omits it entirely."""
+
+        def help_line(abort: CwdSwitchAbortMode | None) -> str:
+            return CwdSwitchPromptScreen(
+                current_cwd="/a", thread_cwd="/b", abort=abort
+            )._help_text()
+
+        assert help_line("resume") == (
+            "Enter: switch · Esc: stay in cwd · A: don't resume"
+        )
+        assert help_line("thread_switch") == (
+            "Enter: switch · Esc: stay in cwd · A: don't switch"
+        )
+        assert help_line(None) == "Enter: switch · Esc: stay in cwd"
+
+    def test_abort_mode_tokens_disjoint_from_choice(self) -> None:
+        """Abort-mode tokens never collide with prompt-outcome tokens.
+
+        The disjointness is a naming convention (input mode vs. outcome), not a
+        type guarantee. This pins it so a future member like a re-added
+        `"switch"` mode -- which would make a mode token ambiguous with a
+        `CwdSwitchChoice` outcome in logs and debuggers -- fails loudly.
+        """
+        assert not (set(get_args(CwdSwitchAbortMode)) & set(get_args(CwdSwitchChoice)))
+
    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
+            current_cwd="/a", thread_cwd="/b", abort="resume"
        )
        dismiss = MagicMock()
        screen.dismiss = dismiss  # ty: ignore[invalid-assignment]
@@ -128,7 +203,7 @@ class TestCwdSwitchAbortOption:
        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`."""
+        """Abort does nothing when the prompt was not opened with an abort mode."""
        screen = CwdSwitchPromptScreen(current_cwd="/a", thread_cwd="/b")
        dismiss = MagicMock()
        screen.dismiss = dismiss  # ty: ignore[invalid-assignment]
@@ -144,7 +219,7 @@ class TestCwdSwitchAbortOption:
            outcomes: list[CwdSwitchChoice | None] = []
            app.push_screen(
                CwdSwitchPromptScreen(
-                    current_cwd="/a", thread_cwd="/b", allow_abort=True
+                    current_cwd="/a", thread_cwd="/b", abort="resume"
                ),
                outcomes.append,
            )
@@ -102,8 +102,16 @@ class TestBranchDisplay:
            assert bar.branch == ""
            assert display.render() == ""

-    async def test_branch_display_shows_branch_name(self) -> None:
-        """Setting branch reactive should update the display widget."""
+    async def test_branch_display_shows_branch_name(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Setting branch reactive should update the display widget.
+
+        `HIDE_CWD` removes the cwd from the layout so the branch region isn't
+        starved of width by a deep pytest cwd -- otherwise this assertion flakes
+        on the actual run directory's path length rather than any real behavior.
+        """
+        monkeypatch.setenv(HIDE_CWD, "1")
        async with StatusBarApp().run_test() as pilot:
            bar = pilot.app.query_one("#status-bar", StatusBar)
            bar.branch = "main"
@@ -112,9 +120,16 @@ class TestBranchDisplay:
            rendered = str(display.render())
            assert "main" in rendered

-    async def test_branch_display_with_feature_branch(self) -> None:
-        """Feature branch names with slashes should display correctly."""
-        async with StatusBarApp().run_test(size=(120, 24)) as pilot:
+    async def test_branch_display_with_feature_branch(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Feature branch names with slashes should display correctly.
+
+        `HIDE_CWD` keeps the branch region wide enough regardless of the pytest
+        cwd path length (see `test_branch_display_shows_branch_name`).
+        """
+        monkeypatch.setenv(HIDE_CWD, "1")
+        async with StatusBarApp().run_test() as pilot:
            bar = pilot.app.query_one("#status-bar", StatusBar)
            bar.branch = "feat/new-feature"
            await pilot.pause()
@@ -230,8 +245,15 @@ class TestBranchDisplay:
            monkeypatch.delenv("UI_CHARSET_MODE", raising=False)
            reset_glyphs_cache()

-    async def test_branch_display_contains_git_icon(self) -> None:
-        """Branch display should include the git branch glyph prefix."""
+    async def test_branch_display_contains_git_icon(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Branch display should include the git branch glyph prefix.
+
+        `HIDE_CWD` keeps the branch region wide enough regardless of the pytest
+        cwd path length (see `test_branch_display_shows_branch_name`).
+        """
+        monkeypatch.setenv(HIDE_CWD, "1")
        async with StatusBarApp().run_test() as pilot:
            bar = pilot.app.query_one("#status-bar", StatusBar)
            bar.branch = "develop"
@@ -19,6 +19,7 @@ from textual.widgets._select import SelectCurrent

 from deepagents_code.app import DeepAgentsApp, _ThreadHistoryPayload
 from deepagents_code.sessions import ThreadInfo
+from deepagents_code.tui.widgets.cwd_switch import CwdSwitchAbortMode
 from deepagents_code.tui.widgets.thread_selector import (
    ContainedSelect,
    ContainedSelectOverlay,
@@ -3151,6 +3152,7 @@ class TestResumeThread:
        offer_cwd_switch.assert_awaited_once_with(
            "thread-123",
            restart_server=True,
+            abort="thread_switch",
        )
        assert len(mounted) == 1
        assert "Already on thread" in _get_widget_text(mounted[0])
@@ -3174,9 +3176,11 @@ class TestResumeThread:
            thread_id: str,
            *,
            restart_server: bool,
+            abort: CwdSwitchAbortMode | None,
        ) -> str:
            assert thread_id == "thread-123"
            assert restart_server is True
+            assert abort == "thread_switch"
            app._cwd = str(target)
            return "continue"

@@ -3217,11 +3221,10 @@ class TestResumeThread:

        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,
+            abort="thread_switch",
        )
        assert app._lc_thread_id == "new-thread"
        assert app._session_state.thread_id == "new-thread"