feat(web): send follow-ups into the active turn

src/query/agent_loop_compat.py

@@ -42,6 +42,7 @@
 
 from .query import QueryParams, StreamEvent, query
 from .transitions import Terminal, TerminalHolder
+from .turn_input import TurnInput
 
 
 # Renderer types are now canonically in ``src.tool_system.renderers``
@@ -517,6 +518,7 @@ async def run_query_as_agent_loop(
     on_text_chunk: TextChunkHandler | None = None,
     on_thinking_chunk: TextChunkHandler | None = None,
     on_message: Callable[[Message], None] | None = None,
+    turn_input: TurnInput | None = None,
     cancel_signal: AbortSignal | None = None,
     abort_controller: AbortController | None = None,
     extended_thinking: bool | None = None,
@@ -771,6 +773,7 @@ async def run_query_as_agent_loop(
         # callback raising AbortError from inside the SDK stream.
         on_text_chunk=on_text_chunk,
         on_thinking_chunk=on_thinking_chunk,
+        turn_input=turn_input,
     )
 
     holder = TerminalHolder()
@@ -898,6 +901,10 @@ async def run_query_as_agent_loop(
             continue
 
         if isinstance(msg, UserMessage):
+            if hasattr(msg, "_steering_text"):
+                if on_message is not None:
+                    on_message(msg)
+                continue
             # Tool result(s) arrive as UserMessages with ToolResultBlock
             # content. Persist the full message (so the next turn's
             # API call can pair tool_use IDs to their results) AND

src/query/query.py

@@ -9,7 +9,7 @@
 import re
 import sys
 import time
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Any, AsyncGenerator, Callable
 from uuid import uuid4
 
@@ -43,6 +43,7 @@
     detect_continuation_signal,
 )
 from .stop_hooks import StopHookResult, handle_stop_hooks_streaming
+from .turn_input import TurnInput
 from .token_budget import (
     ContinueDecision,
     check_token_budget,
@@ -138,6 +139,7 @@ class QueryParams:
     # Live thinking deltas (separate channel from on_text_chunk) for the TUI's
     # streaming thinking view. None → thinking isn't surfaced live.
     on_thinking_chunk: Callable[[str], None] | None = None
+    turn_input: TurnInput | None = None
 
     # Extended thinking ("adaptive" mode) — opt the model into a private
     # reasoning scratchpad before producing its visible answer. Defaults
@@ -1758,6 +1760,12 @@ async def query(
     _outbox_watermark = len(getattr(params.tool_use_context, "outbox", None) or [])
 
     while True:
+        if params.turn_input is not None:
+            incoming = params.turn_input.drain()
+            for message in incoming:
+                yield message
+            if incoming:
+                state = replace(state, messages=[*state.messages, *incoming])
         messages = state.messages
         if _diag:
             logger.warning(
@@ -2654,6 +2662,21 @@ def _n_images(ms: list[Message]) -> int:
                 )
                 return
 
+            if params.turn_input is not None and (
+                not params.max_turns or turn_count < params.max_turns
+            ):
+                incoming = params.turn_input.drain(finishing=True)
+                if incoming:
+                    for message in incoming:
+                        yield message
+                    state = replace(
+                        state,
+                        messages=[*messages, *assistant_messages, *incoming],
+                        turn_count=turn_count + 1,
+                        continuation_nudge_count=0,
+                    )
+                    continue
+
             set_terminal(holder, natural_termination, Terminal(reason="completed"))
             return
 

src/query/turn_input.py

@@ -0,0 +1,37 @@
+"""A thread-safe inbox for input delivered to one active query, not a new turn."""
+
+from __future__ import annotations
+
+import threading
+
+from src.types.messages import UserMessage
+
+
+class TurnInput:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._messages: list[UserMessage] = []
+        self._closed = False
+
+    def put(self, message: UserMessage) -> bool:
+        with self._lock:
+            if self._closed:
+                return False
+            self._messages.append(message)
+            return True
+
+    def drain(self, *, finishing: bool = False) -> list[UserMessage]:
+        with self._lock:
+            messages, self._messages = self._messages, []
+            # Checking for final input and closing admission are atomic. An
+            # acknowledged message can never miss the loop's last check.
+            if finishing and not messages:
+                self._closed = True
+            return messages
+
+    def close(self) -> list[UserMessage]:
+        """Return undelivered input on an error, interrupt, or limit exit."""
+        with self._lock:
+            self._closed = True
+            messages, self._messages = self._messages, []
+            return messages

src/server/agent_server.py

@@ -297,6 +297,7 @@ class _AgentSession:
     _lock: threading.Lock = field(default_factory=threading.Lock)
     _pending: dict[str, _Pending] = field(default_factory=dict)
     _current_abort: AbortController | None = None
+    _turn_input: Any = None
 
     # ─── outbound helpers (worker thread → main loop) ──────────────────────
 
@@ -423,6 +424,39 @@ async def _handle_control_request(self, msg: dict) -> None:
         if self.init_error is not None and subtype not in ("interrupt", "get_activity"):
             self._reply(request_id, {"ok": False, "error": self.init_error})
             return
+        if subtype == "steer":
+            text = inner.get("text")
+            if not isinstance(text, str) or not text.strip():
+                self._reply(request_id, {"status": "rejected", "error": "Enter a message to send now."})
+                return
+            text = text.strip()
+            # Uploads and slash commands need the normal prompt preparation
+            # path. Never deliver their literal markers to the model.
+            if text.startswith("/") or re.search(r"\[(?:Image|File) #\d+\]", text):
+                self._reply(request_id, {"status": "rejected", "error": "Queue attachments and commands for the next turn."})
+                return
+            with self._lock:
+                turn_input = self._turn_input
+                abort = self._current_abort
+            if turn_input is None or abort is None or abort.signal.aborted:
+                self._reply(request_id, {"status": "rejected", "reason": "not_running"})
+                return
+            # Steering is user input too: run the same validation hooks, off
+            # the socket loop, before admitting it into the active query.
+            ups = await asyncio.to_thread(self._run_user_prompt_submit_hooks, text)
+            if ups is not None and (ups.blocked or ups.prevented):
+                self._reply(request_id, {"status": "rejected", "error": ups.block_message if ups.blocked else ups.prevent_reason})
+                return
+            from src.types.messages import UserMessage
+
+            content = text
+            for context in (ups.additional_contexts if ups is not None else []):
+                content += f"\n<system-reminder>\n{context}\n</system-reminder>"
+            message = UserMessage(content=content)
+            message._steering_text = text
+            accepted = not abort.signal.aborted and turn_input.put(message)
+            self._reply(request_id, {"status": "queued"} if accepted else {"status": "rejected", "reason": "not_running"})
+            return
         if subtype == "interrupt":
             with self._lock:
                 abort = self._current_abort
@@ -5632,8 +5666,12 @@ def _run_turn(self, prompt, btw: bool = False, internal: bool = False) -> dict |
             prompt = _with_ultracode_reminder(prompt)
 
         abort = AbortController()
+        from src.query.turn_input import TurnInput
+
+        turn_input = TurnInput()
         with self._lock:
             self._current_abort = abort
+            self._turn_input = None if btw else turn_input
         # Wire the per-turn controller into the tool context so an interrupt
         # tears down an in-flight tool (Bash supervisor, etc.), not just the
         # model stream. A fresh controller per turn avoids a prior turn's
@@ -5690,6 +5728,11 @@ def on_message(message: Any) -> None:
                 )
             except Exception:  # noqa: BLE001
                 logger.exception("[agent-server] persist failed")
+            if hasattr(message, "_steering_text"):
+                self._emit({
+                    "type": "system", "subtype": "steer",
+                    "session_id": self.session_id, "text": message._steering_text,
+                })
             env = _sdk_envelope(message, self.session_id)
             if env is not None:
                 self._emit(env)
@@ -5729,6 +5772,7 @@ def on_message(message: Any) -> None:
                 on_text_chunk=on_text_chunk,
                 on_thinking_chunk=on_thinking_chunk,
                 on_message=on_message,
+                turn_input=turn_input,
                 abort_controller=abort,
                 extended_thinking=self._thinking,  # None = model default; True/False = ThinkingToggle
                 # /effort for BOTH provider families: resolved at the wire
@@ -5779,6 +5823,14 @@ def on_message(message: Any) -> None:
         finally:
             with self._lock:
                 self._current_abort = None
+                self._turn_input = None
+            # Return undelivered input to the client. In particular, Stop
+            # must not silently launch another turn from acknowledged input.
+            for pending in turn_input.close():
+                self._emit({
+                    "type": "system", "subtype": "steer_failed",
+                    "session_id": self.session_id, "text": pending._steering_text,
+                })
             if btw and _btw_snapshot is not None:
                 msgs = self.session.conversation.messages
                 msgs.clear()

src/server/desktop_gateway_methods.py

@@ -380,6 +380,10 @@ async def _route(self, frame: dict[str, Any]) -> None:
                 self.init_info = dict(frame)
                 self.init_seen.set()
                 await self._broadcast("session.info", _init_session_info(frame))
+            elif subtype == "steer":
+                await self._broadcast("message.steered", {"text": frame.get("text", "")})
+            elif subtype == "steer_failed":
+                await self._broadcast("message.steer_failed", {"text": frame.get("text", "")})
             return
 
         # A turn the agent started for itself (a /goal continuation, a loop
@@ -1127,6 +1131,7 @@ def __init__(self, websocket: WebSocket, state: DesktopServeState) -> None:
             "session.active_list": self.session_active_list,
             "session.list": self.session_active_list,
             "session.interrupt": self.session_interrupt,
+            "session.steer": self.session_steer,
             "session.rewind": self.session_rewind,
             "session.clear": self.session_clear,
             "session.title": self.session_title,
@@ -1676,6 +1681,12 @@ async def session_interrupt(self, params: dict[str, Any]) -> dict[str, Any]:
         await self._session(params).interrupt()
         return {"ok": True}
 
+    async def session_steer(self, params: dict[str, Any]) -> dict[str, Any]:
+        result = await self._session(params).control_query("steer", {"text": params.get("text")})
+        if not isinstance(result, dict):
+            raise ValueError("No response while sending to the active turn. Your message has been kept.")
+        return result
+
     async def session_rewind(self, params: dict[str, Any]) -> dict[str, Any]:
         """Drop the last N prompt-turns from the conversation (the CLI's /rewind).
 

tests/server/test_agent_server_steering.py

@@ -0,0 +1,120 @@
+"""Steering crosses the real worker/query boundary without ending its turn."""
+
+import asyncio
+import threading
+from types import SimpleNamespace
+
+import pytest
+
+from src.server.agent_server import AgentServerConfig
+from tests.server.test_agent_server_e2e import _TextProvider, _wait_for
+from tests.server.test_agent_server_workflows import _control, _session_of, _spawned
+
+pytestmark = pytest.mark.integration
+
+
+def held_provider():
+    started, release = threading.Event(), threading.Event()
+    calls = []
+
+    class Provider(_TextProvider):
+        def chat(self, messages, **kwargs):
+            calls.append(str(messages))
+            if len(calls) == 1:
+                started.set()
+                assert release.wait(15), "test did not release the model request"
+            return super().chat(messages, **kwargs)
+
+    return Provider, started, release, calls
+
+
+async def result_frames(gen):
+    frames = []
+    while True:
+        frame = await asyncio.wait_for(gen.__anext__(), timeout=10)
+        frames.append(frame)
+        if frame.get("type") == "result":
+            return frames
+
+
+async def test_steer_during_model_request_continues_same_turn_and_persists(tmp_path):
+    provider, started, release, calls = held_provider()
+    async with _spawned(tmp_path, provider) as (handle, gen):
+        await handle.send_to_agent({"type": "user", "message": {"content": "original task"}})
+        try:
+            assert await _wait_for(started.is_set)
+            reply = await _control(handle, gen, "steer-1", {"subtype": "steer", "text": "  also check complex functions  "})
+            assert reply == {"status": "queued"}
+            assert len(calls) == 1  # no second query while the current request is held
+        finally:
+            release.set()
+        frames = await result_frames(gen)
+        assert frames[-1]["subtype"] == "success"
+        assert frames[-1]["num_turns"] == 2
+        assert len(calls) == 2
+        assert "original task" in calls[1]
+        assert "also check complex functions" in calls[1]
+        assert "also check complex functions" not in calls[0]
+        assert sum(frame.get("subtype") == "steer" for frame in frames) == 1
+        session = _session_of(handle)
+        assert session._stats_turns == 1
+        assert any(message.role == "user" and message.content == "also check complex functions"
+                   for message in session.session.conversation.messages)
+        assert await _control(handle, gen, "late", {"subtype": "steer", "text": "too late"}) == {
+            "status": "rejected", "reason": "not_running",
+        }
+
+
+async def test_steer_honors_user_prompt_hooks_and_rejects_marker_payloads(tmp_path, monkeypatch):
+    provider, started, release, calls = held_provider()
+    async with _spawned(tmp_path, provider) as (handle, gen):
+        await handle.send_to_agent({"type": "user", "message": {"content": "original"}})
+        try:
+            assert await _wait_for(started.is_set)
+            session = _session_of(handle)
+            monkeypatch.setattr(session, "_run_user_prompt_submit_hooks", lambda text: SimpleNamespace(
+                blocked=True, prevented=False, block_message="blocked by validation",
+            ))
+            reply = await _control(handle, gen, "hook", {"subtype": "steer", "text": "blocked input"})
+            assert reply == {"status": "rejected", "error": "blocked by validation"}
+            for text in (" ", "/plan do it", "[Image #1] inspect", "[File #2] read"):
+                reply = await _control(handle, gen, "invalid", {"subtype": "steer", "text": text})
+                assert reply["status"] == "rejected"
+        finally:
+            release.set()
+        await result_frames(gen)
+        assert len(calls) == 1
+
+
+async def test_stop_returns_unread_input_without_starting_another_turn(tmp_path):
+    provider, started, release, calls = held_provider()
+    async with _spawned(tmp_path, provider) as (handle, gen):
+        await handle.send_to_agent({"type": "user", "message": {"content": "original"}})
+        try:
+            assert await _wait_for(started.is_set)
+            assert (await _control(handle, gen, "steer", {"subtype": "steer", "text": "keep this"}))["status"] == "queued"
+            await handle.send_to_agent({"type": "control_request", "request_id": "stop", "request": {"subtype": "interrupt"}})
+        finally:
+            release.set()
+        frames = []
+        while not (any(frame.get("type") == "result" for frame in frames)
+                   and any(frame.get("subtype") == "steer_failed" for frame in frames)):
+            frames.append(await asyncio.wait_for(gen.__anext__(), timeout=10))
+        returned = next(frame for frame in frames if frame.get("subtype") == "steer_failed")
+        assert returned["text"] == "keep this"
+        assert len(calls) == 1
+        assert _session_of(handle)._inbox.empty()
+
+
+async def test_step_limit_returns_unread_input_instead_of_exceeding_the_limit(tmp_path):
+    provider, started, release, calls = held_provider()
+    async with _spawned(tmp_path, provider, AgentServerConfig(max_turns=1)) as (handle, gen):
+        await handle.send_to_agent({"type": "user", "message": {"content": "original"}})
+        try:
+            assert await _wait_for(started.is_set)
+            assert (await _control(handle, gen, "steer", {"subtype": "steer", "text": "keep this"}))["status"] == "queued"
+        finally:
+            release.set()
+        frames = await result_frames(gen)
+        assert len(calls) == 1
+        assert any(frame.get("subtype") == "steer_failed" and frame.get("text") == "keep this" for frame in frames)

ui-web/README.md

@@ -79,6 +79,15 @@ command that takes an argument claims the draft as `/name ` — or as
 as it would on Enter. The rows and their arrangement are a pure function
 (`src/conversation/command-menu.ts`); the composer decides what a pick does.
 
+### Follow-ups during a turn
+
+Enter or **Queue** holds a follow-up for the next turn. **Send now**, available
+in the composer and beside each queued prompt, adds text to the active turn.
+The agent reads it at the next model/tool boundary with the current context;
+running tools continue. A final model response also checks for new input before
+the turn ends. Messages are kept if delivery fails, and input left unread by a
+stopped turn returns to the queue. Attachments and slash commands use Queue.
+
 ### Attachments
 
 An image and a file attach the same way: the bytes go over the socket