Streaming TUI, streaming CLI output with line-buffered rendering

cli.py

@@ -1187,6 +1187,7 @@ def __init__(
         # History file for persistent input recall across sessions
         self._history_file = Path.home() / ".hermes_history"
         self._last_invalidate: float = 0.0  # throttle UI repaints
+        self._stream_buf = ""
 
     def _invalidate(self, min_interval: float = 0.25) -> None:
         """Throttled UI repaint — prevents terminal blinking on slow/SSH connections."""
@@ -1386,6 +1387,7 @@ def _init_agent(self) -> bool:
                 platform="cli",
                 session_db=self._session_db,
                 clarify_callback=self._clarify_callback,
+                stream_delta_callback=self._stream_delta,
                 honcho_session_key=self.session_id,
                 fallback_model=self._fallback_model,
             )
@@ -2905,6 +2907,28 @@ def _reload_mcp(self):
         except Exception as e:
             print(f"  ❌ MCP reload failed: {e}")
 
+    _stream_started = False
+
+    def _stream_delta(self, text: str):
+        """Buffer streaming tokens; emit complete lines via _cprint."""
+        if not text:
+            return
+        if not self._stream_started:
+            text = text.lstrip("\n")
+            if not text:
+                return
+            self._stream_started = True
+        self._stream_buf += text
+        while "\n" in self._stream_buf:
+            line, self._stream_buf = self._stream_buf.split("\n", 1)
+            _cprint(line)
+
+    def _flush_stream(self):
+        """Emit any remaining partial line from the stream buffer."""
+        if self._stream_buf:
+            _cprint(self._stream_buf)
+        self._stream_buf = ""
+
     def _clarify_callback(self, question, choices):
         """
         Platform callback for the clarify tool. Called from the agent thread.
@@ -3076,12 +3100,12 @@ def chat(self, message, images: list = None) -> Optional[str]:
                 message if isinstance(message, str) else "", images
             )
 
-        # Add user message to history
         self.conversation_history.append({"role": "user", "content": message})
-
+        self._stream_buf = ""
+        self._stream_started = False
+
         w = shutil.get_terminal_size().columns
-        _cprint(f"{_GOLD}{'─' * w}{_RST}")
-        print(flush=True)
+        _cprint(f"\n{_GOLD}╭─ ⚕ Hermes {'─' * max(w - 15, 0)}╮{_RST}")
 
         try:
             # Run the conversation with interrupt monitoring
@@ -3127,43 +3151,28 @@ def run_agent():
 
             agent_thread.join()  # Ensure agent thread completes
 
-            # Drain any remaining agent output still in the StdoutProxy
-            # buffer so tool/status lines render ABOVE our response box.
-            # The flush pushes data into the renderer queue; the short
-            # sleep lets the renderer actually paint it before we draw.
-            import time as _time
+            self._flush_stream()
             sys.stdout.flush()
+            import time as _time
             _time.sleep(0.15)
 
-            # Update history with full conversation
             self.conversation_history = result.get("messages", self.conversation_history) if result else self.conversation_history
-
-            # Get the final response
             response = result.get("final_response", "") if result else ""
-
-            # Handle failed results (e.g., non-retryable errors like invalid model)
+
             if result and result.get("failed") and not response:
-                error_detail = result.get("error", "Unknown error")
-                response = f"Error: {error_detail}"
-
-            # Handle interrupt - check if we were interrupted
+                response = f"Error: {result.get('error', 'Unknown error')}"
+
             pending_message = None
             if result and result.get("interrupted"):
                 pending_message = result.get("interrupt_message") or interrupt_msg
-                # Add indicator that we were interrupted
                 if response and pending_message:
-                    response = response + "\n\n---\n_[Interrupted - processing new message]_"
-
-            if response:
-                w = shutil.get_terminal_size().columns
-                label = " ⚕ Hermes "
-                fill = w - 2 - len(label)  # 2 for ╭ and ╮
-                top = f"{_GOLD}╭─{label}{'─' * max(fill - 1, 0)}╮{_RST}"
-                bot = f"{_GOLD}╰{'─' * (w - 2)}╯{_RST}"
-
-                # Render box + response as a single _cprint call so
-                # nothing can interleave between the box borders.
-                _cprint(f"\n{top}\n{response}\n\n{bot}")
+                    response += "\n\n---\n_[Interrupted - processing new message]_"
+
+            if response and not (self.agent and self.agent.stream_delta_callback):
+                _cprint(f"\n{response}")
+
+            w = shutil.get_terminal_size().columns
+            _cprint(f"{_GOLD}╰{'─' * (w - 2)}╯{_RST}")
 
             # Play terminal bell when agent finishes (if enabled).
             # Works over SSH — the bell propagates to the user's terminal.
@@ -3620,7 +3629,7 @@ def _get_placeholder():
                 return ""
             if cli_ref._agent_running:
                 return "type a message + Enter to interrupt, Ctrl+C to cancel"
-            return ""
+            return "Ask Hermes anything... (Alt+Enter for newline)"
 
         input_area.control.input_processors.append(_PlaceholderProcessor(_get_placeholder))
 

gateway/platforms/base.py

@@ -319,7 +319,7 @@ class SendResult:
     raw_response: Any = None
 
 
-# Type for message handlers
+# Handler may return str (sent by base) or dict(content=..., already_sent=True).
 MessageHandler = Callable[[MessageEvent], Awaitable[Optional[str]]]
 
 
@@ -691,11 +691,20 @@ async def _process_message_background(self, event: MessageEvent, session_key: st
 
         try:
             # Call the handler (this can take a while with tool calls)
-            response = await self._message_handler(event)
+            handler_result = await self._message_handler(event)
+
+            # Normalise: handler may return str or dict(content, already_sent)
+            already_sent = False
+            if isinstance(handler_result, dict):
+                response = handler_result.get("content") or ""
+                already_sent = handler_result.get("already_sent", False)
+            else:
+                response = handler_result
 
             # Send response if any
             if not response:
-                logger.warning("[%s] Handler returned empty/None response for %s", self.name, event.source.chat_id)
+                if not already_sent:
+                    logger.warning("[%s] Handler returned empty/None response for %s", self.name, event.source.chat_id)
             if response:
                 # Extract MEDIA:<path> tags (from TTS tool) before other processing
                 media_files, response = self.extract_media(response)
@@ -706,7 +715,7 @@ async def _process_message_background(self, event: MessageEvent, session_key: st
                     logger.info("[%s] extract_images found %d image(s) in response (%d chars)", self.name, len(images), len(response))
 
                 # Send the text portion first (if any remains after extractions)
-                if text_content:
+                if text_content and not already_sent:
                     logger.info("[%s] Sending response (%d chars) to %s", self.name, len(text_content), event.source.chat_id)
                     result = await self.send(
                         chat_id=event.source.chat_id,

gateway/run.py

@@ -1296,7 +1296,9 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
 
             # Update session
             self.session_store.update_session(session_entry.session_key)
-
+
+            if agent_result.get("already_sent"):
+                return {"content": response, "already_sent": True}
             return response
 
         except Exception as e:
@@ -2455,6 +2457,83 @@ async def _run_agent(
         # Queue for progress messages (thread-safe)
         progress_queue = queue.Queue() if tool_progress_enabled else None
         last_tool = [None]  # Mutable container for tracking in closure
+
+        # Streaming token queue — same pattern as progress_queue but for
+        # assistant text deltas.  An async drain task sends/edits a single
+        # platform message with the accumulated text.
+        stream_queue = queue.Queue()
+        stream_sent = [False]  # set True once any delta was delivered
+
+        def _stream_delta(text: str):
+            stream_queue.put(text)
+
+        async def send_stream_messages():
+            """Drain stream_queue, deliver via send/edit_message."""
+            _adapter = self.adapters.get(source.platform)
+            if not _adapter:
+                return
+
+            accumulated = []
+            msg_id = None
+            can_edit = True
+            last_edit_ts = 0.0
+            EDIT_INTERVAL = 0.6  # seconds between edits (rate-limit safe)
+
+            while True:
+                try:
+                    delta = stream_queue.get_nowait()
+                    accumulated.append(delta)
+                    stream_sent[0] = True
+
+                    now = asyncio.get_event_loop().time()
+                    if now - last_edit_ts < EDIT_INTERVAL:
+                        # Coalesce — will flush on next poll cycle
+                        await asyncio.sleep(0.05)
+                        continue
+
+                    full_text = "".join(accumulated)
+                    if msg_id is None:
+                        res = await _adapter.send(
+                            chat_id=source.chat_id, content=full_text)
+                        if res.success and res.message_id:
+                            msg_id = res.message_id
+                    elif can_edit:
+                        res = await _adapter.edit_message(
+                            chat_id=source.chat_id,
+                            message_id=msg_id,
+                            content=full_text,
+                        )
+                        if not res.success:
+                            can_edit = False
+                    last_edit_ts = now
+
+                except queue.Empty:
+                    await asyncio.sleep(0.15)
+                except asyncio.CancelledError:
+                    # Final flush
+                    while not stream_queue.empty():
+                        try:
+                            accumulated.append(stream_queue.get_nowait())
+                        except Exception:
+                            break
+                    if accumulated:
+                        full_text = "".join(accumulated)
+                        if msg_id is None:
+                            await _adapter.send(
+                                chat_id=source.chat_id, content=full_text)
+                        elif can_edit:
+                            try:
+                                await _adapter.edit_message(
+                                    chat_id=source.chat_id,
+                                    message_id=msg_id,
+                                    content=full_text,
+                                )
+                            except Exception:
+                                pass
+                    return
+                except Exception as e:
+                    logger.error("Stream message error: %s", e)
+                    await asyncio.sleep(0.5)
 
         def progress_callback(tool_name: str, preview: str = None, args: dict = None):
             """Callback invoked by agent when a tool is called."""
@@ -2698,6 +2777,7 @@ def run_sync():
                 session_id=session_id,
                 tool_progress_callback=progress_callback if tool_progress_enabled else None,
                 step_callback=_step_callback_sync if _hooks_ref.loaded_hooks else None,
+                stream_delta_callback=_stream_delta,
                 platform=platform_key,
                 honcho_session_key=session_key,
                 session_db=self._session_db,
@@ -2820,12 +2900,16 @@ def run_sync():
                 "api_calls": result_holder[0].get("api_calls", 0) if result_holder[0] else 0,
                 "tools": tools_holder[0] or [],
                 "history_offset": len(agent_history),
+                "already_sent": stream_sent[0],
             }
 
         # Start progress message sender if enabled
         progress_task = None
         if tool_progress_enabled:
             progress_task = asyncio.create_task(send_progress_messages())
+
+        # Start stream message sender
+        stream_task = asyncio.create_task(send_stream_messages())
 
         # Track this agent as running for this session (for interrupt support)
         # We do this in a callback after the agent is created
@@ -2901,9 +2985,10 @@ async def monitor_for_interrupt():
                     session_key=session_key
                 )
         finally:
-            # Stop progress sender and interrupt monitor
+            # Stop progress sender, stream sender, and interrupt monitor
             if progress_task:
                 progress_task.cancel()
+            stream_task.cancel()
             interrupt_monitor.cancel()
 
             # Clean up tracking
@@ -2912,7 +2997,7 @@ async def monitor_for_interrupt():
                 del self._running_agents[session_key]
 
             # Wait for cancelled tasks
-            for task in [progress_task, interrupt_monitor, tracking_task]:
+            for task in [progress_task, stream_task, interrupt_monitor, tracking_task]:
                 if task:
                     try:
                         await task