feat(tui): complete the command queue — drain queued prompts + dim preview

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:

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:

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.

src/tui/screens/repl.py

@@ -34,13 +34,16 @@
     AssistantMessage,
     PermissionRequested,
     PermissionResolved,
+    QueuedPromptReady,
+    QueuedPromptsChanged,
     StateChanged,
     ToolEventMessage,
 )
 from ..a11y import LiveRegion, aria_label
 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:

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.