diff --git a/src/tui/agent_bridge.py b/src/tui/agent_bridge.py
index 697775a6c..c46353353 100644
--- a/src/tui/agent_bridge.py
+++ b/src/tui/agent_bridge.py
@@ -51,6 +51,7 @@
     AssistantChunk,
     AssistantMessage,
     PermissionRequested,
+    QueuedPromptReady,
     ToolEventMessage,
 )
 from .state import AppState
@@ -273,10 +274,21 @@ def resume_session(self, session_id: str) -> list[Any] | None:
             return list(messages)
 
     def submit(self, prompt: str) -> bool:
-        """Queue ``prompt`` for the agent. Returns False if busy."""
+        """Start an agent run for ``prompt``. Returns False if busy.
+
+        When a run is already in flight the prompt is **enqueued** onto
+        ``app_state.queued_prompts`` *under* ``_busy_lock``, so the
+        enqueue is atomic with the busy check. This pairs with
+        :meth:`_finish`, which clears ``busy`` and reads the queue under
+        the same lock: a run can therefore never finish in the gap
+        between "is it busy?" and "enqueue", which would otherwise leave
+        the prompt stranded with no drain ever posted. The REPL drains
+        the queue one-per-turn (see ``QueuedPromptReady``).
+        """
 
         with self._busy_lock:
             if self._busy:
+                self._state.queued_prompts.append(prompt)
                 return False
             self._busy = True
             self._abort_controller = AbortController()
@@ -560,6 +572,18 @@ def _finish(self) -> None:
             # reading the context field, so replacing here doesn't
             # orphan them either.
             self._tool_context.abort_controller = AbortController()
+            # Drain decision, made under the same lock as the enqueue in
+            # ``submit`` so the two can't interleave. ``busy`` is already
+            # False above, so a ``QueuedPromptReady`` posted now is
+            # guaranteed to find the bridge idle on the UI thread.
+            has_queued = bool(self._state.queued_prompts)
+        # Post OUTSIDE the lock (``_post`` marshals to the UI thread). The
+        # worker-side non-empty check is only a filter — the REPL handler
+        # re-checks idle + non-empty before popping, so a queue cleared by
+        # ESC in the meantime is a harmless no-op. Fires on every exit
+        # path (completion, abort, error) because all route through here.
+        if has_queued:
+            self._post(QueuedPromptReady())
 
     # ---- advisor rendering ----
     def _emit_advisor_events(self) -> None:
diff --git a/src/tui/app.py b/src/tui/app.py
index 12c31fec9..4abdf552e 100644
--- a/src/tui/app.py
+++ b/src/tui/app.py
@@ -42,7 +42,7 @@
     dispatch_registry_command,
 )
 from .history_store import HistoryStore  # noqa: F401 (re-exported for tests)
-from .messages import CancelRequested
+from .messages import CancelRequested, QueuedPromptsChanged
 from .screens.cost_threshold import CostThresholdScreen
 from .screens.diff_dialog import DiffDialogScreen, FileDiff
 from .screens.effort_picker import EffortPickerScreen
@@ -1620,23 +1620,47 @@ def _ensure_command_context(self) -> Any:
         return self._command_context
 
     # ---- agent loop plumbing ----
-    def submit_to_agent(self, prompt: str) -> None:
-        try:
-            self.history_store.append(prompt)
-        except Exception:
-            pass
-        submitted = self._agent_bridge.submit(prompt)
-        if not submitted:
-            # If the bridge is busy we queue the prompt for the next
-            # turn so the user can keep typing. Phase 2 adds a visible
-            # queued-prompts pill in the status line.
-            self.app_state.queued_prompts.append(prompt)
+    def submit_to_agent(self, prompt: str, *, record_history: bool = True) -> bool:
+        """Submit ``prompt`` to the agent.
+
+        Returns ``True`` if a run started, or ``False`` if the bridge was
+        busy and the prompt was **queued** for the next turn (the bridge
+        enqueues it under its busy lock; the REPL drains the queue
+        one-per-turn). Pass ``record_history=False`` to skip the prompt-
+        history append — used by the queue drain, since a queued prompt
+        was already recorded to history when it was first typed.
+        """
+
+        if record_history:
+            try:
+                self.history_store.append(prompt)
+            except Exception:
+                pass
+        started = self._agent_bridge.submit(prompt)
+        if not started:
+            # The bridge already appended to ``queued_prompts`` under its
+            # busy lock; surface the change so the REPL refreshes the dim
+            # queued-prompts preview (and the status-line count pill).
+            self._post_to_screen(QueuedPromptsChanged())
+        return started
 
     def on_cancel_requested(self, _: CancelRequested) -> None:
-        """ESC from the prompt — cancel the in-flight agent run, if any."""
+        """ESC from the prompt — TS ``useCancelRequest.handleCancel`` parity.
+
+        Priority 1: if a run is in flight, cancel it and **return** — the
+        queue is left intact (``useCancelRequest.ts:97-102``). Priority 2:
+        if idle with a non-empty queue, pop the queued prompts **back into
+        the input** for editing rather than discarding them
+        (``:104-109`` → ``popAllEditable``). ESC never destroys the queue;
+        ``clearCommandQueue`` is the separate kill-agents gesture
+        (``handleKillAgents``), not this handler.
+        """
 
         if self._agent_bridge.cancel():
             self.announcer.announce("Cancelling…", level="assertive", notify=False)
+            return
+        if self.app_state.queued_prompts and self._repl_screen is not None:
+            self._repl_screen.pop_queue_into_input()
 
     # ---- helpers ----
     def _build_default_tool_context(self) -> ToolContext:
diff --git a/src/tui/messages.py b/src/tui/messages.py
index 1d327d7e0..5931383bb 100644
--- a/src/tui/messages.py
+++ b/src/tui/messages.py
@@ -179,6 +179,37 @@ class CancelRequested(Message):
     pass
 
 
+@dataclass
+class QueuedPromptReady(Message):
+    """A queued prompt may be drained now that the bridge is idle.
+
+    Posted from :meth:`src.tui.agent_bridge.AgentBridge._finish` *after*
+    ``busy`` clears, when ``AppState.queued_prompts`` is non-empty. The
+    REPL screen re-checks on the UI thread and, if still idle + the queue
+    is non-empty, pops the oldest prompt and submits it — FIFO, one per
+    turn (the Python parity of TS ``useCommandQueue`` auto-processing).
+    The worker-side check is only a cheap filter; the UI handler is
+    authoritative, so a spurious post (e.g. a queue cleared by ESC) is a
+    safe no-op.
+    """
+
+    pass
+
+
+@dataclass
+class QueuedPromptsChanged(Message):
+    """The queued-prompts list changed; rebuild the dim preview widget.
+
+    Posted whenever ``AppState.queued_prompts`` is mutated: appended
+    (a prompt submitted while busy), popped (drain), or cleared (ESC).
+    The screen reads the live list from
+    ``src.tui.app.ClawCodexTUI.app_state.queued_prompts`` and refreshes
+    :class:`src.tui.widgets.queued_commands.QueuedCommands`.
+    """
+
+    pass
+
+
 @dataclass
 class PromptPasted(Message):
     """Bracketed-paste landed in the :class:`PromptInput` widget.
diff --git a/src/tui/screens/repl.py b/src/tui/screens/repl.py
index 4ac130775..cbf284ec6 100644
--- a/src/tui/screens/repl.py
+++ b/src/tui/screens/repl.py
@@ -34,6 +34,8 @@
     AssistantMessage,
     PermissionRequested,
     PermissionResolved,
+    QueuedPromptReady,
+    QueuedPromptsChanged,
     StateChanged,
     ToolEventMessage,
 )
@@ -41,6 +43,7 @@
 from ..widgets.fullscreen_layout import FullscreenLayout
 from ..widgets.header import StartupHeader
 from ..widgets.prompt_input import PromptInput, PromptSubmitted
+from ..widgets.queued_commands import QueuedCommands
 from ..widgets.status_line import StatusLine
 from ..widgets.transcript_view import Transcript
 
@@ -108,6 +111,10 @@ def __init__(
             suggestions_provider=suggestions_provider,
             vim_mode=initial_vim_mode(),
         )
+        # Dim preview of prompts queued while a run is in flight (parity
+        # with PromptInputQueuedCommands). Mounted directly above the
+        # prompt input; hidden while the queue is empty.
+        self.queued_commands = QueuedCommands()
         # ARIA live region — stays height: 1 and only announces the
         # most recent status change. Mounted just above the status
         # bar so it's adjacent to the prompt for single-sweep reads.
@@ -126,6 +133,10 @@ def on_mount(self) -> None:
         self._fullscreen.scroll_region().mount(self.transcript)
         self._fullscreen.bottom_region().mount(self.live_region)
         self._fullscreen.bottom_region().mount(self.status_bar)
+        # Order: live region, status line, queued preview, prompt — so
+        # the queued text sits right above the input (TS marginTop), with
+        # the at-a-glance count staying on the status line.
+        self._fullscreen.bottom_region().mount(self.queued_commands)
         self._fullscreen.bottom_region().mount(self.prompt_input)
         # Bind the status line to app state so the spinner / queue count
         # reflect the authoritative agent state.
@@ -167,10 +178,91 @@ def on_prompt_submitted(self, message: PromptSubmitted) -> None:
             # no agent turn. A bare "#" falls through to the agent.
             app.run_memory_shortcut(text[1:], self.transcript)
             return
-        self.transcript.append_user(text)
-        self.status_bar.set_busy()
-        self.status_bar.bump_turn()
-        app.submit_to_agent(text)
+        # Plain agent prompt. ``submit_to_agent`` returns False when the
+        # bridge is busy and the prompt was QUEUED — in that case it
+        # shows only in the dim queued-prompts preview (refreshed via
+        # QueuedPromptsChanged), NOT the transcript, until the drain runs
+        # it (TS parity: queued commands live above the input, not in the
+        # conversation). Only render the user row + bump the turn when a
+        # run actually started.
+        if app.submit_to_agent(text):
+            self.transcript.append_user(text)
+            self.status_bar.set_busy()
+            self.status_bar.bump_turn()
+
+    # ---- queued-prompt drain (TS useCommandQueue auto-processing) ----
+    def on_queued_prompt_ready(self, _: QueuedPromptReady) -> None:
+        """Drain the oldest queued prompt now that the bridge is idle.
+
+        Posted by ``AgentBridge._finish`` after a run ends. We re-check
+        idle + non-empty on the UI thread (authoritative — the worker
+        side is only a filter), pop the oldest prompt, and replay it like
+        a plain prompt. ``record_history=False`` because it was recorded
+        when first typed. One pop per ``QueuedPromptReady`` → FIFO, one
+        prompt per turn (the next run's ``_finish`` posts again if more
+        remain).
+        """
+
+        app = self.app
+        bridge = getattr(app, "_agent_bridge", None)
+        state = getattr(app, "app_state", None)
+        if bridge is None or state is None:
+            return
+        # A spurious post (queue cleared by ESC, or a run started in the
+        # meantime) must be a safe no-op.
+        if bridge.busy or not state.queued_prompts:
+            return
+        text = state.queued_prompts.pop(0)
+        self._refresh_queued_preview()
+        # We just confirmed the bridge idle on this single UI thread, so
+        # the submit starts a run (returns True). The defensive ``else``
+        # only matters if that invariant is ever broken — then the bridge
+        # re-queued it under its lock and it drains on the next turn.
+        if app.submit_to_agent(text, record_history=False):
+            self.transcript.append_user(text)
+            self.status_bar.set_busy()
+            self.status_bar.bump_turn()
+        else:  # pragma: no cover - unreachable on the single UI thread
+            self._refresh_queued_preview()
+
+    def pop_queue_into_input(self) -> None:
+        """ESC Priority 2: drain queued prompts back into the prompt input.
+
+        TS ``popAllEditable`` (``messageQueueManager.ts:428``): the queued
+        texts (in order) are joined with the current draft by newlines and
+        loaded into the input for editing; the queue is then emptied.
+        Prompts are moved INTO the input, never discarded — so ESC can
+        never lose what the user typed. (Images / cursor-offset handling
+        in TS are N-A for the plain-``str`` Python queue.)
+
+        Caveat: the prompt input is still single-line (Textual ``Input``;
+        a ``TextArea`` swap is Phase-2 — see ``prompt_input.py``). A merged
+        value containing ``\\n`` (only when there are multiple queued
+        prompts, or a non-empty draft) is preserved verbatim in the value
+        (correct on submit) but renders flat until that swap.
+        """
+
+        app = self.app
+        state = getattr(app, "app_state", None)
+        if state is None or not state.queued_prompts:
+            return
+        queued = list(state.queued_prompts)
+        current = self.prompt_input.current_text()
+        merged = "\n".join(t for t in [*queued, current] if t)
+        state.queued_prompts.clear()
+        self.prompt_input.set_value(merged)
+        self.prompt_input.focus_input()
+        self._refresh_queued_preview()
+
+    def on_queued_prompts_changed(self, _: QueuedPromptsChanged) -> None:
+        self._refresh_queued_preview()
+
+    def _refresh_queued_preview(self) -> None:
+        """Rebuild the dim queued-prompts widget from live app state."""
+
+        state = getattr(self.app, "app_state", None)
+        prompts = list(getattr(state, "queued_prompts", []) or [])
+        self.queued_commands.set_prompts(prompts)
 
     # ---- agent message handlers ----
     def on_agent_run_started(self, _: AgentRunStarted) -> None:
diff --git a/src/tui/widgets/prompt_input.py b/src/tui/widgets/prompt_input.py
index 57a3e2116..e43eddca1 100644
--- a/src/tui/widgets/prompt_input.py
+++ b/src/tui/widgets/prompt_input.py
@@ -198,6 +198,11 @@ def set_value(self, value: str) -> None:
         self._input.value = value or ""
         self._hide_suggestions()
 
+    def current_text(self) -> str:
+        """The current draft text (used by the ESC queue-pop, /history)."""
+
+        return self._input.value or ""
+
     # ---- bracketed paste ----
     def handle_paste(self, text: str) -> PasteInfo:
         """Insert a bracketed-paste payload as a single atomic operation.
diff --git a/src/tui/widgets/queued_commands.py b/src/tui/widgets/queued_commands.py
new file mode 100644
index 000000000..2f8c0e37f
--- /dev/null
+++ b/src/tui/widgets/queued_commands.py
@@ -0,0 +1,122 @@
+"""Dim preview of prompts queued while the agent is busy.
+
+Parity with ``components/PromptInput/PromptInputQueuedCommands.tsx``:
+when the user types a prompt while a run is in flight, it is *queued*
+for the next turn (``AppState.queued_prompts``) rather than dropped.
+This widget renders, directly above the prompt input, a dim header
+(``"N message(s) queued for next turn"``) followed by one dim line per
+queued prompt so the user can see *what* is pending — not just the
+``"queued N"`` count pill on the status line.
+
+Scope note (deliberately reduced vs TS): TS ``PromptInputQueuedCommands``
+iterates typed ``QueuedCommand`` records with a ``mode``
+(prompt / bash / task-notification), caps task-notifications, folds idle
+hints, and renders each through the full ``<Message>`` component. Python's
+``queued_prompts`` is a plain ``list[str]`` of raw *prompt* text (slash /
+bash / memory inputs never reach the queue — see
+``REPLScreen.on_prompt_submitted``), so the faithful slice is the dim
+header + per-prompt text line. The capping / folding / multi-mode system
+has no Python producer and is intentionally out of scope.
+"""
+
+from __future__ import annotations
+
+from rich.text import Text
+from textual.widgets import Static
+
+# U+2026 HORIZONTAL ELLIPSIS — one cell wide, matches the truncation
+# marker used elsewhere in the prompt chrome (prompt_input.py).
+_ELLIPSIS = "…"
+
+
+def _truncate(text: str, width: int) -> str:
+    """Clip ``text`` to ``width`` columns, ending in an ellipsis.
+
+    Uses character count as a column proxy (consistent with the rest of
+    the prompt chrome). ``width <= 0`` is treated as "no bound known"
+    and returns the text unchanged — the widget falls back to a sane
+    default width, and the Rich ``overflow="ellipsis"`` safety net on
+    the renderable clips anything that still exceeds the real console.
+    """
+
+    if width <= 0 or len(text) <= width:
+        return text
+    if width == 1:
+        return _ELLIPSIS
+    return text[: width - 1] + _ELLIPSIS
+
+
+def format_queued_preview(prompts: list[str], width: int) -> Text:
+    """Build the dim renderable for the queued-prompts preview.
+
+    Pure + deterministic so it can be unit-tested without a live layout.
+    Each prompt is reduced to its **first line**, whitespace-collapsed,
+    and truncated to ``width`` with an ellipsis so a multi-line or huge
+    pasted prompt can never blow up the footer. Returns an empty
+    ``Text`` when the queue is empty (the widget hides itself in that
+    case via CSS).
+    """
+
+    count = len(prompts)
+    if count == 0:
+        return Text("")
+    header = (
+        "1 message queued for next turn"
+        if count == 1
+        else f"{count} messages queued for next turn"
+    )
+    # Whole renderable is dim; ``no_wrap`` + ``overflow`` is a safety net
+    # for any line that still exceeds the real console width at render.
+    out = Text(no_wrap=True, overflow="ellipsis", style="dim")
+    out.append(_truncate(header, width))
+    for prompt in prompts:
+        first_line = prompt.split("\n", 1)[0]
+        collapsed = " ".join(first_line.split())
+        out.append("\n")
+        out.append(_truncate(collapsed, width))
+    return out
+
+
+class QueuedCommands(Static):
+    """Footer widget showing prompts queued for the next turn.
+
+    Hidden (``display: none``) while the queue is empty; the
+    ``-has-queue`` state class flips it visible. Re-renders on resize
+    (``render`` recomputes from the live content width) and on
+    :meth:`set_prompts`.
+    """
+
+    DEFAULT_CSS = """
+    QueuedCommands {
+        height: auto;
+        padding: 0 1;
+        color: $text-muted;
+        display: none;
+    }
+    QueuedCommands.-has-queue {
+        display: block;
+    }
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._prompts: list[str] = []
+
+    def set_prompts(self, prompts: list[str]) -> None:
+        """Replace the queued-prompt list and refresh the preview."""
+
+        self._prompts = list(prompts)
+        # State class drives visibility — no manual height juggling.
+        self.set_class(bool(self._prompts), "-has-queue")
+        self.refresh(layout=True)
+
+    def render(self) -> Text:
+        if not self._prompts:
+            return Text("")
+        # content_size is (0, 0) until first layout; fall back so the
+        # very first paint still truncates to something sane.
+        width = self.content_size.width or 80
+        return format_queued_preview(self._prompts, width)
+
+
+__all__ = ["QueuedCommands", "format_queued_preview"]
diff --git a/tests/tui/test_queued_commands.py b/tests/tui/test_queued_commands.py
new file mode 100644
index 000000000..b4ba25fe4
--- /dev/null
+++ b/tests/tui/test_queued_commands.py
@@ -0,0 +1,383 @@
+"""Tests for the TUI command queue.
+
+Parity target: ``components/PromptInput/PromptInputQueuedCommands.tsx`` +
+``hooks/useCommandQueue`` — a prompt typed while a run is in flight is
+*queued* for the next turn (shown in a dim preview above the input, not
+the transcript) and drained one-per-turn (FIFO) when the run ends; ESC
+discards the queue.
+
+Three layers:
+* ``format_queued_preview`` — the pure display formatter.
+* ``AgentBridge`` — enqueue-under-lock FIFO + ``_finish`` posting
+  ``QueuedPromptReady`` only when the queue is non-empty.
+* ``REPLScreen`` via ``App.run_test`` — the end-to-end drain, the
+  queued-not-transcript invariant, ESC clearing, and that slash / bash /
+  memory inputs never enter the queue.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+
+# --------------------------------------------------------------------------
+# 1. Pure formatter
+# --------------------------------------------------------------------------
+class TestFormatQueuedPreview:
+    def _lines(self, prompts: list[str], width: int = 80) -> list[str]:
+        from src.tui.widgets.queued_commands import format_queued_preview
+
+        return format_queued_preview(prompts, width).plain.splitlines()
+
+    def test_empty_queue_renders_nothing(self) -> None:
+        from src.tui.widgets.queued_commands import format_queued_preview
+
+        assert format_queued_preview([], 80).plain == ""
+
+    def test_singular_header(self) -> None:
+        lines = self._lines(["hello"])
+        assert lines[0] == "1 message queued for next turn"
+        assert lines[1] == "hello"
+
+    def test_plural_header_preserves_order(self) -> None:
+        lines = self._lines(["a", "b", "c"])
+        assert lines[0] == "3 messages queued for next turn"
+        assert lines[1:] == ["a", "b", "c"]
+
+    def test_only_first_line_of_multiline_prompt(self) -> None:
+        # A multi-line paste must not blow the footer up to many rows.
+        lines = self._lines(["line1\nline2\nline3"])
+        assert lines[1] == "line1"
+
+    def test_internal_whitespace_collapsed(self) -> None:
+        lines = self._lines(["  too    many   spaces  "])
+        assert lines[1] == "too many spaces"
+
+    def test_truncated_to_width_with_ellipsis(self) -> None:
+        lines = self._lines(["x" * 100], width=20)
+        assert lines[1] == "x" * 19 + "…"
+        assert len(lines[1]) == 20
+
+    def test_header_also_bounded_by_width(self) -> None:
+        lines = self._lines(["hi"], width=10)
+        assert len(lines[0]) == 10
+        assert lines[0].endswith("…")
+
+    def test_renderable_is_dim(self) -> None:
+        from src.tui.widgets.queued_commands import format_queued_preview
+
+        assert str(format_queued_preview(["hi"], 80).style) == "dim"
+
+
+# --------------------------------------------------------------------------
+# 2. Bridge enqueue + drain trigger
+# --------------------------------------------------------------------------
+class TestBridgeQueue:
+    def _bridge(self, tmp_path: Path, monkeypatch):
+        import src.services.session_storage as storage_mod
+        from src.agent.session import Session
+        from src.tool_system.context import ToolContext
+        from src.tool_system.registry import ToolRegistry
+        from src.tui.agent_bridge import AgentBridge
+        from src.tui.state import AppState
+
+        # Keep the persister off the developer's ~/.clawcodex/sessions.
+        monkeypatch.setattr(storage_mod, "SESSIONS_DIR", tmp_path / "sessions")
+
+        posted: list[object] = []
+        session = Session.create("test", "test-model")
+        bridge = AgentBridge(
+            post_message=posted.append,
+            session=session,
+            provider=MagicMock(model="m"),
+            tool_registry=ToolRegistry(),
+            tool_context=ToolContext(workspace_root=tmp_path),
+            app_state=AppState(),
+            # Worker never actually runs → the run stays "in flight"
+            # until the test calls _finish(), giving deterministic
+            # control over the busy window.
+            run_worker=lambda *a, **k: None,
+        )
+        return bridge, posted
+
+    def test_idle_submit_starts_run(self, tmp_path, monkeypatch) -> None:
+        bridge, _ = self._bridge(tmp_path, monkeypatch)
+        assert bridge.submit("first") is True
+        assert bridge.busy is True
+        assert list(bridge._state.queued_prompts) == []
+
+    def test_submit_while_busy_enqueues_fifo(self, tmp_path, monkeypatch) -> None:
+        bridge, _ = self._bridge(tmp_path, monkeypatch)
+        bridge.submit("first")  # starts a run → busy
+        assert bridge.submit("second") is False
+        assert bridge.submit("third") is False
+        assert list(bridge._state.queued_prompts) == ["second", "third"]
+
+    def test_finish_posts_ready_when_queue_nonempty(self, tmp_path, monkeypatch) -> None:
+        from src.tui.messages import QueuedPromptReady
+
+        bridge, posted = self._bridge(tmp_path, monkeypatch)
+        bridge.submit("first")
+        bridge.submit("second")  # queued
+        posted.clear()
+        bridge._finish()
+        assert bridge.busy is False
+        assert any(isinstance(m, QueuedPromptReady) for m in posted)
+
+    def test_finish_silent_when_queue_empty(self, tmp_path, monkeypatch) -> None:
+        from src.tui.messages import QueuedPromptReady
+
+        bridge, posted = self._bridge(tmp_path, monkeypatch)
+        bridge.submit("first")  # busy, nothing queued
+        posted.clear()
+        bridge._finish()
+        assert not any(isinstance(m, QueuedPromptReady) for m in posted)
+
+
+# --------------------------------------------------------------------------
+# 3. End-to-end via the real REPL screen
+# --------------------------------------------------------------------------
+pytest.importorskip("textual")
+
+
+class _FakeProvider:
+    provider_name = "fake"
+    model = "fake-model"
+
+
+def _make_app(tmp_path: Path):
+    from src.tui.app import ClawCodexTUI
+    from src.tool_system.context import ToolContext
+    from src.tool_system.registry import ToolRegistry
+
+    return ClawCodexTUI(
+        provider=_FakeProvider(),
+        provider_name="fake",
+        workspace_root=tmp_path,
+        tool_registry=ToolRegistry(),
+        tool_context=ToolContext(workspace_root=tmp_path),
+        stream=False,
+    )
+
+
+def _spy_user_rows(screen, monkeypatch) -> list[str]:
+    """Record every text appended to the transcript as a user row."""
+
+    rows: list[str] = []
+    orig = screen.transcript.append_user
+
+    def _spy(text: str) -> None:
+        rows.append(text)
+        orig(text)
+
+    monkeypatch.setattr(screen.transcript, "append_user", _spy)
+    return rows
+
+
+def _stick_busy(app) -> None:
+    """Make submitted runs stay in flight until the test calls _finish()."""
+
+    app._agent_bridge._run_worker = lambda *a, **k: None
+
+
+async def _boot(tmp_path, monkeypatch):
+    import src.services.session_storage as storage_mod
+    from src.tui.history_store import HistoryStore
+
+    monkeypatch.setattr(storage_mod, "SESSIONS_DIR", tmp_path / "sessions")
+    app = _make_app(tmp_path)
+    # Isolate prompt history off the real ~/.clawcodex/history.jsonl so
+    # test prompts ("first", "second", …) never pollute the user's file.
+    app.history_store = HistoryStore(tmp_path / "history.jsonl")
+    return app
+
+
+@pytest.mark.asyncio
+async def test_prompt_while_busy_queues_without_transcript(tmp_path, monkeypatch):
+    from src.tui.widgets.prompt_input import PromptSubmitted
+    from src.tui.widgets.queued_commands import QueuedCommands
+
+    app = await _boot(tmp_path, monkeypatch)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        screen = app.screen
+        _stick_busy(app)
+        rows = _spy_user_rows(screen, monkeypatch)
+
+        # Idle prompt → run starts, shown in the transcript.
+        screen.post_message(PromptSubmitted(text="first"))
+        await pilot.pause()
+        assert app._agent_bridge.busy is True
+        assert rows == ["first"]
+        assert app.app_state.queued_prompts == []
+
+        # Busy → the next prompt is queued, NOT shown in the transcript.
+        screen.post_message(PromptSubmitted(text="second"))
+        await pilot.pause()
+        assert app.app_state.queued_prompts == ["second"]
+        assert "second" not in rows
+        assert screen.query_one(QueuedCommands).has_class("-has-queue")
+
+
+@pytest.mark.asyncio
+async def test_finish_drains_queue_fifo(tmp_path, monkeypatch):
+    from src.tui.messages import AgentRunFinished
+    from src.tui.widgets.prompt_input import PromptSubmitted
+    from src.tui.widgets.queued_commands import QueuedCommands
+
+    app = await _boot(tmp_path, monkeypatch)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        screen = app.screen
+        _stick_busy(app)
+        rows = _spy_user_rows(screen, monkeypatch)
+
+        for text in ("first", "second", "third"):
+            screen.post_message(PromptSubmitted(text=text))
+            await pilot.pause()
+        # "first" ran; "second"/"third" queued in order.
+        assert rows == ["first"]
+        assert app.app_state.queued_prompts == ["second", "third"]
+
+        async def _end_run() -> None:
+            # Mirror the real worker order: AgentRunFinished is posted
+            # before _finish() posts QueuedPromptReady — proving the
+            # two-message ordering doesn't interfere with the drain.
+            screen.post_message(
+                AgentRunFinished(response_text="", num_turns=1, usage=None)
+            )
+            app._agent_bridge._finish()
+            await pilot.pause()
+            await pilot.pause()
+
+        # End the first run → drain exactly one (the oldest).
+        await _end_run()
+        assert rows == ["first", "second"]
+        assert app.app_state.queued_prompts == ["third"]
+
+        # End the second run → drain the last; queue empties, widget hides.
+        await _end_run()
+        assert rows == ["first", "second", "third"]
+        assert app.app_state.queued_prompts == []
+        assert not screen.query_one(QueuedCommands).has_class("-has-queue")
+
+
+@pytest.mark.asyncio
+async def test_queued_prompt_recorded_to_history_once(tmp_path, monkeypatch):
+    """A queued prompt is in input history once (typed), not twice (drain)."""
+    from src.tui.widgets.prompt_input import PromptSubmitted
+
+    app = await _boot(tmp_path, monkeypatch)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        screen = app.screen
+        _stick_busy(app)
+        _spy_user_rows(screen, monkeypatch)
+
+        screen.post_message(PromptSubmitted(text="first"))
+        await pilot.pause()
+        screen.post_message(PromptSubmitted(text="queued one"))
+        await pilot.pause()
+        app._agent_bridge._finish()
+        await pilot.pause()
+        await pilot.pause()
+
+        recorded = [r.prompt for r in app.history_store.load() if r.prompt == "queued one"]
+        assert recorded == ["queued one"]
+
+
+@pytest.mark.asyncio
+async def test_escape_while_busy_preserves_queue(tmp_path, monkeypatch):
+    """TS handleCancel Priority 1: ESC cancels the run, queue is untouched."""
+    from src.tui.messages import CancelRequested
+    from src.tui.widgets.prompt_input import PromptSubmitted
+    from src.tui.widgets.queued_commands import QueuedCommands
+
+    app = await _boot(tmp_path, monkeypatch)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        screen = app.screen
+        _stick_busy(app)
+        _spy_user_rows(screen, monkeypatch)
+
+        screen.post_message(PromptSubmitted(text="first"))
+        await pilot.pause()
+        screen.post_message(PromptSubmitted(text="kept"))
+        await pilot.pause()
+        assert app.app_state.queued_prompts == ["kept"]
+        assert app._agent_bridge.busy is True
+
+        # ESC with a run in flight → cancel the run, leave the queue.
+        app.post_message(CancelRequested())
+        await pilot.pause()
+        assert app.app_state.queued_prompts == ["kept"]
+        assert screen.query_one(QueuedCommands).has_class("-has-queue")
+
+
+@pytest.mark.asyncio
+async def test_escape_while_idle_pops_queue_into_input(tmp_path, monkeypatch):
+    """TS handleCancel Priority 2: ESC idle pops queued prompts into the input."""
+    from src.tui.messages import CancelRequested
+    from src.tui.widgets.prompt_input import PromptInput, PromptSubmitted
+    from src.tui.widgets.queued_commands import QueuedCommands
+
+    app = await _boot(tmp_path, monkeypatch)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        screen = app.screen
+        _stick_busy(app)
+        _spy_user_rows(screen, monkeypatch)
+
+        screen.post_message(PromptSubmitted(text="first"))
+        await pilot.pause()
+        screen.post_message(PromptSubmitted(text="queued one"))
+        await pilot.pause()
+        assert app.app_state.queued_prompts == ["queued one"]
+
+        # Simulate the run ending without auto-draining, and a draft typed.
+        app._agent_bridge._busy = False
+        screen.query_one(PromptInput).set_value("draft")
+
+        app.post_message(CancelRequested())
+        await pilot.pause()
+        # Queue drained back into the input (queued text then the draft),
+        # not discarded; preview hidden.
+        assert app.app_state.queued_prompts == []
+        assert screen.query_one(PromptInput).current_text() == "queued one\ndraft"
+        assert not screen.query_one(QueuedCommands).has_class("-has-queue")
+
+
+@pytest.mark.asyncio
+async def test_slash_bash_memory_never_enqueue(tmp_path, monkeypatch):
+    from src.tui.widgets.prompt_input import PromptSubmitted
+
+    app = await _boot(tmp_path, monkeypatch)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        screen = app.screen
+        _stick_busy(app)
+        _spy_user_rows(screen, monkeypatch)
+
+        # Isolate the routing: stub the prefix handlers so they don't run
+        # bash / touch memory files / open dialogs.
+        monkeypatch.setattr(app, "run_bash_mode", lambda *a, **k: None)
+        monkeypatch.setattr(app, "run_memory_shortcut", lambda *a, **k: None)
+        monkeypatch.setattr(app, "handle_local_slash_command", lambda *a, **k: True)
+
+        # Start a run so the bridge is busy (a plain prompt WOULD queue).
+        screen.post_message(PromptSubmitted(text="first"))
+        await pilot.pause()
+        assert app._agent_bridge.busy is True
+
+        for prefixed in ("/help", "!ls", "#a note"):
+            screen.post_message(PromptSubmitted(text=prefixed))
+            await pilot.pause()
+        assert app.app_state.queued_prompts == []
+
+        # Control: a plain prompt still queues.
+        screen.post_message(PromptSubmitted(text="plain"))
+        await pilot.pause()
+        assert app.app_state.queued_prompts == ["plain"]