Merge pull request #100 from vstorm-co/feat/message-delivery

feat: add mid-run message delivery queues (steering & follow-up)

Author: DEENUU1

apps/cli/agent.py

@@ -12,6 +12,7 @@
 from apps.cli.reminder import _build_reminder_config
 from pydantic_deep.agent import DEFAULT_INSTRUCTIONS, create_deep_agent
 from pydantic_deep.capabilities.hooks import Hook, HookEvent, HookInput, HookResult
+from pydantic_deep.capabilities.message_queue import MessageQueue
 from pydantic_deep.deps import DeepAgentDeps
 
 
@@ -356,6 +357,8 @@ def create_cli_agent(  # noqa: C901
                 stacklevel=2,
             )
 
+    queue = MessageQueue()
+
     agent = create_deep_agent(
         model=effective_model,
         instructions=instructions,
@@ -422,6 +425,8 @@ def create_cli_agent(  # noqa: C901
         middleware=middleware or None,
         toolsets=[local_context] if local_context else None,
         capabilities=extra_capabilities or None,
+        # Message queue for mid-run steering and follow-up delivery
+        message_queue=queue,
         # Periodic reminder
         periodic_reminder=_build_reminder_config(
             periodic_reminder, reminder_mode, config, on_reminder, reminder_model
@@ -435,6 +440,7 @@ def create_cli_agent(  # noqa: C901
     deps = DeepAgentDeps(
         backend=effective_backend,
         context_middleware=context_mw,
+        message_queue=queue,
     )
     deps._task_manager = task_mgr  # type: ignore[attr-defined]
     return agent, deps

apps/cli/app.py

@@ -86,6 +86,7 @@ def __init__(
         self.message_history: list[ModelMessage] = message_history or []
         self.last_response: str = ""
         self._startup_error = startup_error
+        self.queue = getattr(deps, "message_queue", None)
 
         # Register custom themes
         from apps.cli.styles.themes import register_themes
@@ -229,6 +230,7 @@ def reconfigure_agent(self, model: str | None = None) -> None:
             )
             self.agent = agent
             self.deps = deps
+            self.queue = getattr(deps, "message_queue", None)
             self._startup_error = None
             self.model_name = effective
 

apps/cli/screens/chat.py

@@ -35,6 +35,7 @@
 from apps.cli.widgets.input_area import InputArea
 from apps.cli.widgets.message_list import MessageList
 from apps.cli.widgets.notification import notify_success, notify_warning
+from apps.cli.widgets.queued_panel import QueuedWidget
 from apps.cli.widgets.side_panel import SidePanel
 from apps.cli.widgets.status_bar import StatusBar
 from pydantic_deep.deps import DEFAULT_USAGE_LIMITS
@@ -383,20 +384,45 @@ def _show_welcome(self) -> None:
 
     # ── User input handling ───────────────────────────────────────
 
-    def on_user_submitted(self, event: UserSubmitted) -> None:
+    async def on_user_submitted(self, event: UserSubmitted) -> None:
         """Handle user submitting a prompt."""
         text = event.text
 
+        app = self.app
+        queue = app.queue
+        task = app._agent_task
+        is_running = task is not None and not task.done()
+
+        # Mid-run: route to queue. `>>` prefix = steering, plain text = follow-up.
+        # `!` keeps meaning "shell command" regardless of agent state.
+        if is_running and queue is not None and not text.startswith("!"):
+            if text.startswith(">>"):
+                steer_text = text[2:].strip()
+                if steer_text:
+                    await queue.steer(steer_text)
+                    preview = steer_text[:40] + ("…" if len(steer_text) > 40 else "")
+                    app.notify(f"steering queued: {preview}")
+                    self._increment_queue_badge(steering=True)
+            else:
+                await queue.follow_up(text)
+                app.notify("follow-up queued")
+                self._increment_queue_badge(steering=False)
+            return
+
         # Shell command
         if text.startswith("!"):
-            self.app.run_shell_command(text[1:])  # type: ignore[attr-defined]
+            app.run_shell_command(text[1:])  # type: ignore[attr-defined]
             return
 
         # Slash command (but not things like "I used /path/to/file")
         if text.startswith("/") and not text.startswith("//"):
-            self.app.handle_command(text)  # type: ignore[attr-defined]
+            app.handle_command(text)  # type: ignore[attr-defined]
             return
 
+        # User typed `>>foo` while idle — strip the steering prefix and run as a normal prompt
+        if text.startswith(">>"):
+            text = text[2:].lstrip()
+
         # Expand @file references — read files and append content to prompt
         text = self._expand_file_refs(text)
 
@@ -445,6 +471,7 @@ def _run_agent(self, text: str) -> None:
         msg_list = self.query_one(MessageList)
 
         header.is_streaming = True
+        self.query_one(InputArea).is_agent_running = True
         app.last_response = ""  # type: ignore
         assistant = msg_list.begin_assistant_message()
 
@@ -491,6 +518,7 @@ async def _agent_stream_worker(  # noqa: C901
 
         log.info("Agent run started", prompt_length=len(text), history_messages=len(history))
 
+        _follow_up_scheduled = False
         pending: dict[str, tuple[dict[str, Any], float]] = {}
         _run_cancelled = False
         _TODO_TOOLS: frozenset[str] = frozenset()  # Show all tool calls in UI
@@ -787,6 +815,21 @@ def _parse_args(raw: Any) -> dict[str, Any]:
                 # Auto-save session
                 self._save_session()
 
+                # Drain follow-up queue and schedule next run if pending.
+                _queue = app.queue
+                if _queue is not None:
+                    _follow_up_msgs = await _queue.drain_follow_up()
+                    if _follow_up_msgs:
+                        from pydantic_deep.capabilities.message_queue import (
+                            format_follow_up as _fmt_fu,
+                        )
+
+                        _follow_up_text = _fmt_fu(_follow_up_msgs)
+                        msg_list.append_user_message(_follow_up_text)
+                        self._decrement_queue_badge(len(_follow_up_msgs))
+                        _follow_up_scheduled = True
+                        self.call_later(self._run_agent, _follow_up_text)
+
         except asyncio.CancelledError:
             _run_cancelled = True
             log.info("Agent run cancelled")
@@ -814,13 +857,58 @@ def _parse_args(raw: Any) -> dict[str, Any]:
             app.is_streaming = False
             header.is_streaming = False
             header.is_thinking = False
+            with contextlib.suppress(Exception):
+                self.query_one(InputArea).is_agent_running = False
             msg_list.end_assistant_message()
             with contextlib.suppress(Exception):
                 self.query_one(InputArea).focus_input()
             with contextlib.suppress(Exception):
                 self.query_one(HintsBar).reset()
             with contextlib.suppress(Exception):
                 msg_list.scroll_end(animate=False)
+            _stale_queue = app.queue
+            if _stale_queue is not None:
+                stale = await _stale_queue.drain_steering()
+                if stale:
+                    n = len(stale)
+                    label = "steering message" if n == 1 else "steering messages"
+                    with contextlib.suppress(Exception):
+                        app.notify(
+                            f"{n} {label} not delivered — agent finished before next LLM call",
+                            severity="warning",
+                            timeout=6,
+                        )
+                # When the run was cancelled, follow-ups referring to the cancelled
+                # task are likely stale too. Discard with a count-only notification.
+                if _run_cancelled:
+                    stale_fu = await _stale_queue.drain_follow_up()
+                    if stale_fu:
+                        n = len(stale_fu)
+                        label = "follow-up" if n == 1 else "follow-ups"
+                        with contextlib.suppress(Exception):
+                            app.notify(
+                                f"{n} {label} discarded — run cancelled",
+                                severity="warning",
+                                timeout=6,
+                            )
+            if not _follow_up_scheduled:
+                self._reset_queue_badge()
+            else:
+                with contextlib.suppress(Exception):
+                    self.query_one(QueuedWidget).clear_steering()
+
+    def _increment_queue_badge(self, *, steering: bool) -> None:
+        with contextlib.suppress(Exception):
+            w = self.query_one(QueuedWidget)
+            w.increment_steering() if steering else w.increment_follow_up()
+
+    def _decrement_queue_badge(self, follow_up_count: int = 1) -> None:
+        with contextlib.suppress(Exception):
+            self.query_one(QueuedWidget).decrement_follow_up(follow_up_count)
+
+    def _reset_queue_badge(self) -> None:
+        with contextlib.suppress(Exception):
+            self.query_one(QueuedWidget).reset()
 
     def _expand_file_refs(self, text: str) -> str:
         """Expand @file references in the prompt with file contents."""

apps/cli/widgets/input_area.py

@@ -203,6 +203,7 @@ class InputArea(Vertical):
     """
 
     is_multiline: reactive[bool] = reactive(False)
+    is_agent_running: reactive[bool] = reactive(False)
 
     class ExitMultiline(Message):
         """Request to exit multiline mode."""
@@ -213,6 +214,19 @@ def compose(self) -> ComposeResult:
             yield PromptInput()
         yield HintsBar()
 
+    @staticmethod
+    def _running_hints() -> str:
+        return "[dim]>>[/dim] steer   write to queue   [dim]Esc[/dim] interrupt"
+
+    def watch_is_agent_running(self, running: bool) -> None:
+        if self.is_multiline:
+            return
+        hints = self.query_one(HintsBar)
+        if running:
+            hints.update(self._running_hints())
+        else:
+            hints.reset()
+
     def watch_is_multiline(self, multiline: bool) -> None:
         prompt_rows = self.query("PromptRow")
         multi = self.query("MultilineInput")

apps/cli/widgets/queued_panel.py

@@ -0,0 +1,68 @@
+"""Queued messages display widget for the side panel."""
+
+from __future__ import annotations
+
+from textual.app import ComposeResult
+from textual.css.query import NoMatches
+from textual.reactive import reactive
+from textual.widget import Widget
+from textual.widgets import Static
+
+
+class QueuedWidget(Widget):
+    """Pending steering / follow-up message counts."""
+
+    DEFAULT_CSS = """
+    QueuedWidget {
+        height: auto;
+        padding: 1;
+        border: tall $surface-lighten-2;
+        margin: 1 0 0 0;
+    }
+    """
+
+    steering_count: reactive[int] = reactive(0)
+    follow_up_count: reactive[int] = reactive(0)
+
+    def compose(self) -> ComposeResult:
+        yield Static("[bold]Queued[/bold]", id="queued-title")
+        yield Static("", id="queued-list")
+
+    def watch_steering_count(self) -> None:
+        self._refresh()
+
+    def watch_follow_up_count(self) -> None:
+        self._refresh()
+
+    def increment_steering(self) -> None:
+        self.steering_count += 1
+
+    def increment_follow_up(self) -> None:
+        self.follow_up_count += 1
+
+    def decrement_follow_up(self, n: int = 1) -> None:
+        self.follow_up_count = max(0, self.follow_up_count - n)
+
+    def reset(self) -> None:
+        self.steering_count = 0
+        self.follow_up_count = 0
+
+    def clear_steering(self) -> None:
+        self.steering_count = 0
+
+    def _refresh(self) -> None:
+        try:
+            content = self.query_one("#queued-list", Static)
+        except NoMatches:
+            return
+
+        lines: list[str] = []
+        if self.steering_count > 0:
+            lines.append(f"  {self.steering_count} steering")
+        if self.follow_up_count > 0:
+            lines.append(f"  {self.follow_up_count} follow-up")
+
+        if lines:
+            content.update("\n".join(lines))
+        else:
+            content.update("[dim]No queued messages[/dim]")

apps/cli/widgets/side_panel.py

@@ -5,6 +5,7 @@
 from textual.app import ComposeResult
 from textual.containers import Vertical
 
+from apps.cli.widgets.queued_panel import QueuedWidget
 from apps.cli.widgets.subagents_panel import SubagentsWidget
 from apps.cli.widgets.todos_panel import TodosWidget
 
@@ -29,6 +30,7 @@ class SidePanel(Vertical):
     def compose(self) -> ComposeResult:
         yield TodosWidget()
         yield SubagentsWidget()
+        yield QueuedWidget()
 
     def update_for_width(self, width: int) -> None:
         """Show or hide based on terminal width."""

docs/capabilities/message-queue.md

@@ -0,0 +1,80 @@
+# Message Queue
+
+The **Message Queue** lets external code push messages into a running agent loop without cancelling and restarting it — preserving in-flight tool results and the prompt cache.
+
+Two delivery semantics are supported:
+
+| Semantic | When delivered | Use case |
+|---|---|---|
+| **Steering** | Before the next LLM request, after the current tool batch | "Stop that approach, try X instead" |
+| **Follow-up** | When the agent would otherwise stop | "When you're done, also do Y" |
+
+## Quick start
+
+```python
+from pydantic_deep import create_deep_agent
+from pydantic_deep.capabilities.message_queue import MessageQueue, run_with_queue
+from pydantic_deep.deps import DeepAgentDeps
+
+queue = MessageQueue()
+agent = create_deep_agent(model="anthropic:claude-sonnet-4-6", message_queue=queue)
+deps = DeepAgentDeps(message_queue=queue)
+
+# In another coroutine / task while the agent is running:
+await queue.steer("stop digging deeper, summarise what you have")
+await queue.follow_up("when done, write a test for the result")
+
+# Run with follow-up support
+result = await run_with_queue(agent, "investigate the bug", deps=deps, queue=queue)
+```
+
+## Delivery modes
+
+Each message can be queued with one of two `delivery_mode` values:
+
+- **`"one_at_a_time"`** (default) — each `drain_*` call pops exactly one message
+- **`"all"`** — the first `drain_*` call empties the entire queue at once (mode is read from the head message)
+
+```python
+await queue.steer("first hint")
+await queue.steer("second hint")
+# drain_steering() returns only "first hint" (one_at_a_time)
+
+await queue.steer("batch A", delivery_mode="all")
+await queue.steer("batch B")
+# drain_steering() returns both (mode from head message)
+```
+
+## Subagent sharing
+
+By default, `DeepAgentDeps.clone_for_subagent()` passes the same `MessageQueue` reference to subagents. A subagent can therefore steer the parent:
+
+```python
+# Inside a tool or subagent:
+await ctx.deps.message_queue.steer("parent, change your approach")
+```
+
+Pass a fresh `MessageQueue()` to `clone_for_subagent()` override on the cloned deps if isolation is needed.
+
+## Delivery sequence
+
+```
+External caller          MessageQueue          Agent loop (pydantic-ai)
+     |                       |                         |
+     |-- await steer("X") -->|                         |
+     |                       |  [steering deque: X]    |
+     |                       |                         |
+     |                       |<-- before_model_request-|
+     |                       |  drain_steering()       |
+     |                       |-- inject UserPromptPart->|
+     |                       |                         |-- LLM sees "[steering] X"
+     |                       |                         |
+     |-- await follow_up("Y")|                         |
+     |                       |  [follow_up deque: Y]   |
+     |                       |                         |
+     |                       |                    agent stops
+     |                       |                         |
+     |             run_with_queue() drain_follow_up()  |
+     |                       |-- "Y" as next prompt -->|
+     |                       |                         |-- new run with Y
+```