Stream bash tool output incrementally via SSE

Replace subprocess.run() with Popen for bash tool execution, streaming
stdout line-by-line through a new on_tool_output_chunk callback. Web UI
renders chunks incrementally with a pulsing amber border indicator.

Core:
- Add on_tool_output_chunk(call_id, chunk) to SessionUI protocol
- Rewrite _exec_bash() with Popen, process-group kill via
  start_new_session + os.killpg, background stderr drain thread,
  threading.Event-based timeout detection
- Guard UI callback with contextlib.suppress so errors don't
  interrupt output collection

Server/CLI/eval:
- Add tool_output_chunk SSE event type in WebUI
- No-op implementations in TerminalUI, BackgroundTerminalUI, SilentUI

MQ:
- Add ToolOutputChunkEvent to mq/protocol.py and _OUTBOUND_REGISTRY
- Handle tool_output_chunk in bridge._handle_ws_event

Web UI:
- Add appendToolOutputChunk() with call_id-keyed DOM elements,
  inner auto-scroll, ARIA attributes, and empty chunk guards
- Fix appendToolOutput() streaming cleanup using adjacency matching
- Make collapsed output keyboard-accessible (tabindex, role, keydown)
- Improve stripAnsi() to handle CSI, OSC, and two-byte escapes;
  use it consistently in replayHistory, addInfoMessage, addErrorMessage
- Add .tool-output-stream CSS with soft pulse animation, mobile
  max-height cap, and consolidated prefers-reduced-motion support

Docs & diagrams:
- Document tool_output_chunk SSE event in api-reference.md
- Update SessionUI protocol (14 methods) in architecture.md
- Update Phase 3 execution flow in tools.md
- Add on_tool_output_chunk to 03-core-engine-classes.puml
- Update 04-conversation-turn.puml, 05-tool-pipeline.puml
- Add ToolOutputChunkEvent to 06-mq-protocol.puml
- Add to event list in 07-message-routing.puml
- Regenerate all 5 affected PNG diagrams

docs/api-reference.md

@@ -173,7 +173,13 @@ Each item in `items` (shared by `tool_info` and `approve_request`):
 | `needs_approval` | bool        | Whether this call requires explicit approval     |
 | `error`          | string/null | Error description if the call was malformed      |
 
-**`tool_result`** -- output from a completed tool execution.
+**`tool_output_chunk`** -- incremental streaming output from a bash tool execution. Sent line-by-line as stdout is produced. The `call_id` identifies the specific tool invocation (multiple bash tools may run in parallel).
+
+```json
+{"type": "tool_output_chunk", "call_id": "call_abc123", "chunk": "Building project...\n"}
+```
+
+**`tool_result`** -- final output from a completed tool execution. For bash tools, this arrives after all `tool_output_chunk` events and includes both stdout and stderr.
 
 ```json
 {"type": "tool_result", "name": "bash", "output": "file1.py\nfile2.py\n"}

docs/architecture.md

@@ -138,6 +138,8 @@ Phase 3: EXECUTE (parallel)
     run_one(items[0])
   else:
     ThreadPoolExecutor(max_workers=4).map(run_one, items)
+  Bash tool streams stdout line-by-line via ui.on_tool_output_chunk()
+  Final output (stdout + stderr) delivered via ui.on_tool_result()
   For plan tool: post-execution gate via ui.on_plan_review()
 ```
 
@@ -174,7 +176,7 @@ The engine emits state changes via `_emit_state()` which calls
 
 > See also: [Core Engine Classes diagram](diagrams/png/03-core-engine-classes.png)
 
-Defined in `turnstone.core.session.SessionUI` as a `typing.Protocol` with 13
+Defined in `turnstone.core.session.SessionUI` as a `typing.Protocol` with 14
 methods. Every frontend must implement all of them.
 
 ```python
@@ -186,6 +188,7 @@ class SessionUI(Protocol):
     def on_stream_end(self) -> None: ...
     def approve_tools(self, items: list[dict]) -> tuple[bool, str | None]: ...
     def on_tool_result(self, name: str, output: str) -> None: ...
+    def on_tool_output_chunk(self, call_id: str, chunk: str) -> None: ...
     def on_status(self, usage: dict, context_window: int, effort: str) -> None: ...
     def on_plan_review(self, content: str) -> str: ...
     def on_info(self, message: str) -> None: ...

docs/diagrams/03-core-engine-classes.puml

@@ -13,6 +13,7 @@ interface "SessionUI" as SessionUI <<Protocol>> {
     + on_stream_end()
     + approve_tools(items: list) → (bool, str|None)
     + on_tool_result(name: str, output: str)
+    + on_tool_output_chunk(call_id: str, chunk: str)
     + on_status(usage: dict, ctx_window: int, effort: str)
     + on_plan_review(content: str) → str
     + on_info(message: str)

docs/diagrams/04-conversation-turn.puml

@@ -109,7 +109,7 @@ group loop [while tool_calls present]
 
         note right of TP
           Parallel execution:
-          bash → subprocess.run()
+          bash → Popen + line-by-line streaming
           read_file → open().read()
           search → grep subprocess
           edit_file → string replace
@@ -121,8 +121,10 @@ group loop [while tool_calls present]
         end note
 
         note right of TP
-          on_tool_result() called
-          inside each _exec_* handler
+          bash: on_tool_output_chunk()
+          called per stdout line,
+          then on_tool_result() at end.
+          Other tools: on_tool_result() only.
         end note
 
         TP --> CS : [(call_id, output), ...]

docs/diagrams/05-tool-pipeline.puml

@@ -115,6 +115,7 @@ partition "Phase 3: Execute" #E3F2FD {
 
     :_truncate_output() on each result\n(max context_window × chars_per_token × 0.5 chars\ndefault: ~context_window × 2 chars);
 
+    :bash: ui.on_tool_output_chunk(call_id, line) per stdout line;
     :ui.on_tool_result(name, output) for each;
 
     if (plan tool was executed?) then (yes)

docs/diagrams/06-mq-protocol.puml

@@ -126,6 +126,11 @@ package "Outbound Events (Bridge → Client)" #E3F2FD {
             ..
             correlation_id = request_id
         }
+        class ToolOutputChunkEvent {
+            type = "tool_output_chunk"
+            + call_id: str
+            + chunk: str
+        }
         class ToolResultEvent {
             type = "tool_result"
             + name: str

docs/diagrams/07-message-routing.puml

@@ -36,7 +36,7 @@ ServerA --> BridgeA : {status:"ok"}
 
 BridgeA -> Redis : PUBLISH turnstone:events:abc12345\nAckEvent(status:"ok")
 
-... SSE events flow: content, tool_result, status, state_change ...
+... SSE events flow: content, tool_output_chunk, tool_result, status, state_change ...
 
 BridgeA -> Redis : PUBLISH turnstone:events:abc12345\nContentEvent, ToolResultEvent, ...
 BridgeA -> Redis : PUBLISH turnstone:events:global\nStateChangeEvent(state:"idle")

docs/diagrams/png/03-core-engine-classes.png

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:30deb9eec4cb61d9865611f3a6b10c696638f540f93683654ea8d903b2e2ac0b
-size 227161
+oid sha256:6644411ca4ddb9300842602492816e3d804ee916d6013915365ec6af6fad210d
+size 229645

docs/diagrams/png/04-conversation-turn.png

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:3bd265f9e3ecb55e93b88039f363cfd7953fd29b4a68e8830d924a34b431fa29
-size 264506
+oid sha256:a90dec1546dd0f8343e3e27cf6f235d95f3ffc375928d34d0df5d8f25d63e5ad
+size 269901

docs/diagrams/png/05-tool-pipeline.png

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:32bb7e8aa409d872a3e519a878601649b569f4af8e1ee77da940b785e834effe
-size 232985
+oid sha256:335315486c253966d9004933d8c887016a4a9e7c1b168a82c6ff339a5172b458
+size 237025

docs/diagrams/png/06-mq-protocol.png

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:3846bb799587c6b1ff8335f10eeeef874248500b36c6be0a54ee532c4772c459
-size 190947
+oid sha256:e660f453a3708d1a7d827f07cd967f122500eb1c4845f5a75cc1309091bce6af
+size 185796

docs/diagrams/png/07-message-routing.png

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:9d77472902935937f04420b35375b3a869a4cb51f8ac08dab3c1d097d549de2d
-size 221103
+oid sha256:8cc7c94d5ac4862c3c09450346f0af923818e02ea0550cc8023f039fdc701179
+size 221528

docs/tools.md

@@ -91,6 +91,10 @@ Each item's `execute` callable is invoked:
 - Single tool calls run directly on the current thread.
 - Multiple tool calls run in parallel via `ThreadPoolExecutor(max_workers=4)`.
 - Errored or denied items return their error/denial message without executing.
+- The `bash` tool streams stdout incrementally: each line calls
+  `ui.on_tool_output_chunk(call_id, line)` as it is produced, then the final
+  combined output (stdout + stderr) is delivered via `ui.on_tool_result()`.
+  Other tools deliver results atomically via `on_tool_result()` only.
 - Special post-execution gate for `plan`: the plan output is shown to the user
   for review, and the user can reject or annotate it.
 

tests/test_server_live.py

@@ -90,6 +90,9 @@ class RecordingUI:
     def on_tool_result(self, name, output):
         self.tool_results.append((name, output))
 
+    def on_tool_output_chunk(self, call_id, chunk):
+        pass
+
     def on_status(self, usage, context_window, effort):
         self.events.append(("status",))
 

tests/test_session.py

@@ -30,6 +30,9 @@ class NullUI:
     def on_tool_result(self, name, output):
         pass
 
+    def on_tool_output_chunk(self, call_id, chunk):
+        pass
+
     def on_status(self, usage, context_window, effort):
         pass
 

tests/test_workstream.py

@@ -57,6 +57,9 @@ class FakeUI:
     def on_tool_result(self, name, output):
         pass
 
+    def on_tool_output_chunk(self, call_id, chunk):
+        pass
+
     def on_status(self, usage, context_window, effort):
         pass
 

turnstone/cli.py

@@ -184,6 +184,9 @@ class TerminalUI(SessionUI):
     def on_tool_result(self, name: str, output: str) -> None:
         pass  # Optional: display summary
 
+    def on_tool_output_chunk(self, call_id: str, chunk: str) -> None:
+        pass  # Terminal shows spinner during tool execution
+
     def on_status(self, usage: dict[str, Any], context_window: int, effort: str) -> None:
         total_tok = usage["prompt_tokens"] + usage["completion_tokens"]
         pct = total_tok / context_window * 100 if context_window > 0 else 0
@@ -323,6 +326,10 @@ class WorkstreamTerminalUI(TerminalUI):
         if self.is_foreground:
             super().on_tool_result(name, output)
 
+    def on_tool_output_chunk(self, call_id: str, chunk: str) -> None:
+        if self.is_foreground:
+            super().on_tool_output_chunk(call_id, chunk)
+
     def on_plan_review(self, content: str) -> str:
         # Must wait until foregrounded to show plan review
         if not self.is_foreground:

turnstone/core/session.py

@@ -9,9 +9,11 @@ to receive events and handle approval prompts.
 from __future__ import annotations
 
 import concurrent.futures
+import contextlib
 import json
 import os
 import re
+import signal
 import subprocess
 import tempfile
 import textwrap
@@ -70,6 +72,7 @@ class SessionUI(Protocol):
     def on_stream_end(self) -> None: ...
     def approve_tools(self, items: list[dict[str, Any]]) -> tuple[bool, str | None]: ...
     def on_tool_result(self, name: str, output: str) -> None: ...
+    def on_tool_output_chunk(self, call_id: str, chunk: str) -> None: ...
     def on_status(self, usage: dict[str, Any], context_window: int, effort: str) -> None: ...
     def on_plan_review(self, content: str) -> str: ...
     def on_info(self, message: str) -> None: ...
@@ -1701,31 +1704,69 @@ class ChatSession:
     # -- Execute methods (do the work, report output via UI) -------------------
 
     def _exec_bash(self, item: dict[str, Any]) -> tuple[str, str]:
-        """Execute a bash command via temp script."""
+        """Execute a bash command via temp script, streaming stdout."""
         call_id, command = item["call_id"], item["command"]
         try:
             with tempfile.NamedTemporaryFile(mode="w", suffix=".sh", delete=False) as f:
                 f.write(command)
                 script_path = f.name
             try:
-                result = subprocess.run(
+                proc = subprocess.Popen(
                     ["bash", script_path],
-                    capture_output=True,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
                     text=True,
-                    timeout=self.tool_timeout,
+                    start_new_session=True,
                 )
+                # Drain stderr in background thread to avoid pipe deadlock
+                stderr_lines: list[str] = []
+
+                def drain_stderr() -> None:
+                    assert proc.stderr is not None
+                    for line in proc.stderr:
+                        stderr_lines.append(line)
+
+                stderr_thread = threading.Thread(target=drain_stderr, daemon=True)
+                stderr_thread.start()
+
+                # Stream stdout line-by-line with process-group timeout
+                stdout_parts: list[str] = []
+                timed_out = threading.Event()
+
+                def _on_timeout() -> None:
+                    timed_out.set()
+                    with contextlib.suppress(OSError, ProcessLookupError):
+                        os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+
+                timer = threading.Timer(self.tool_timeout, _on_timeout)
+                timer.start()
+                try:
+                    assert proc.stdout is not None
+                    for line in proc.stdout:
+                        stdout_parts.append(line)
+                        with contextlib.suppress(Exception):
+                            self.ui.on_tool_output_chunk(call_id, line)
+                finally:
+                    timer.cancel()
+
+                proc.wait()
+                stderr_thread.join(timeout=5)
             finally:
                 os.unlink(script_path)
-            output = result.stdout
-            if result.stderr:
-                output += ("\n" if output else "") + result.stderr
+
+            if timed_out.is_set():
+                raise subprocess.TimeoutExpired(cmd="bash", timeout=self.tool_timeout)
+
+            output = "".join(stdout_parts)
+            if stderr_lines:
+                output += ("\n" if output else "") + "".join(stderr_lines)
             output = output.strip()
             output = self._truncate_output(output)
 
             self.ui.on_tool_result("bash", output)
 
-            if result.returncode != 0:
-                output += f"\n[exit code: {result.returncode}]"
+            if proc.returncode != 0:
+                output += f"\n[exit code: {proc.returncode}]"
 
             return call_id, output if output else "(no output)"
 

turnstone/eval.py

@@ -68,6 +68,9 @@ class NullUI:
     def on_tool_result(self, name: str, output: str) -> None:
         pass
 
+    def on_tool_output_chunk(self, call_id: str, chunk: str) -> None:
+        pass
+
     def on_status(self, usage: dict[str, Any], context_window: int, effort: str) -> None:
         pass
 

turnstone/mq/bridge.py

@@ -39,6 +39,7 @@ from turnstone.mq.protocol import (
     StatusEvent,
     StreamEndEvent,
     ToolInfoEvent,
+    ToolOutputChunkEvent,
     ToolResultEvent,
     TurnCompleteEvent,
     WorkstreamClosedEvent,
@@ -434,6 +435,15 @@ class Bridge:
             self._handle_approval(ws_id, data)
         elif etype == "plan_review":
             self._handle_plan_review(ws_id, data)
+        elif etype == "tool_output_chunk":
+            self._publish_ws(
+                ws_id,
+                ToolOutputChunkEvent(
+                    ws_id=ws_id,
+                    call_id=data.get("call_id", ""),
+                    chunk=data.get("chunk", ""),
+                ),
+            )
         elif etype == "tool_result":
             self._publish_ws(
                 ws_id,

turnstone/mq/protocol.py

@@ -195,6 +195,15 @@ class ApprovalRequestEvent(OutboundEvent):
     items: list[dict[str, Any]] = field(default_factory=list)
 
 
+@dataclass
+class ToolOutputChunkEvent(OutboundEvent):
+    """Incremental streaming output from a bash tool."""
+
+    type: str = "tool_output_chunk"
+    call_id: str = ""
+    chunk: str = ""
+
+
 @dataclass
 class ToolResultEvent(OutboundEvent):
     """Tool execution result."""
@@ -368,6 +377,7 @@ _OUTBOUND_REGISTRY: dict[str, type[OutboundEvent]] = {
         ReasoningEvent,
         ToolInfoEvent,
         ApprovalRequestEvent,
+        ToolOutputChunkEvent,
         ToolResultEvent,
         PlanReviewEvent,
         StatusEvent,

turnstone/server.py

@@ -208,6 +208,9 @@ class WebUI:
         self._broadcast_activity()
         self._enqueue({"type": "tool_result", "name": name, "output": output})
 
+    def on_tool_output_chunk(self, call_id: str, chunk: str) -> None:
+        self._enqueue({"type": "tool_output_chunk", "call_id": call_id, "chunk": chunk})
+
     def on_status(self, usage: dict[str, Any], context_window: int, effort: str) -> None:
         total_tok = usage["prompt_tokens"] + usage["completion_tokens"]
         pct = total_tok / context_window * 100 if context_window > 0 else 0

turnstone/ui/static/app.js

@@ -1170,6 +1170,12 @@ function handleEvent(evt) {
       showInlineToolBlock(evt.items, false);
       break;
 
+    case "tool_output_chunk":
+      if (evt.call_id && evt.chunk) {
+        appendToolOutputChunk(evt.call_id, evt.chunk);
+      }
+      break;
+
     case "tool_result":
       appendToolOutput(evt.name, evt.output);
       break;
@@ -1313,9 +1319,7 @@ function replayHistory(messages) {
       }
     } else if (msg.role === "tool") {
       if (lastToolBlock) {
-        var stripped = (msg.content || "")
-          .replace(/\x1b\[[0-9;]*m/g, "")
-          .trim();
+        var stripped = stripAnsi(msg.content || "").trim();
         if (stripped) {
           var out = document.createElement("div");
           out.className = "tool-output";
@@ -1339,7 +1343,11 @@ function replayHistory(messages) {
 // --- Inline tool/approval blocks ---
 
 function stripAnsi(s) {
-  return s.replace(/\x1b\[[0-9;]*m/g, "");
+  // Strip CSI sequences, OSC sequences, and two-byte escapes
+  return s.replace(
+    /\x1b(?:\[[0-9;?]*[A-Za-z]|\][^\x07\x1b]*(?:\x07|\x1b\\)?|[()#][A-Za-z0-9]|.)/g,
+    "",
+  );
 }
 
 function buildToolDiv(item) {
@@ -1489,6 +1497,43 @@ function resolveInlineApproval(approved, always, feedback) {
   scrollToBottom();
 }
 
+function appendToolOutputChunk(callId, chunk) {
+  if (!chunk) return;
+  var stripped = stripAnsi(chunk);
+  if (!stripped) return;
+
+  // Find or create a streaming output element keyed by call_id
+  var el = messagesEl.querySelector(
+    '.tool-output-stream[data-call-id="' + callId + '"]',
+  );
+  if (!el) {
+    // Find the last approval-block and last bash tool div inside it
+    var blocks = messagesEl.querySelectorAll(".approval-block");
+    if (!blocks.length) return;
+    var block = blocks[blocks.length - 1];
+    var tools = block.querySelectorAll('.approval-tool[data-func-name="bash"]');
+    var target = tools.length ? tools[tools.length - 1] : null;
+    if (!target) {
+      // Fallback to last tool div
+      var allTools = block.querySelectorAll(".approval-tool");
+      target = allTools.length ? allTools[allTools.length - 1] : null;
+    }
+    if (!target) return;
+
+    el = document.createElement("pre");
+    el.className = "tool-output tool-output-stream";
+    el.dataset.callId = callId;
+    el.setAttribute("aria-label", "Streaming command output");
+    el.setAttribute("aria-live", "off");
+    el.textContent = "";
+    target.after(el);
+  }
+
+  el.textContent += stripped;
+  el.scrollTop = el.scrollHeight;
+  scrollToBottom();
+}
+
 function appendToolOutput(name, output) {
   // Find the last approval-block in messages
   const blocks = messagesEl.querySelectorAll(".approval-block");
@@ -1507,6 +1552,12 @@ function appendToolOutput(name, output) {
   if (!target && tools.length) target = tools[tools.length - 1];
   if (!target) return;
 
+  // Remove the streaming output element adjacent to this tool
+  var streamEl = target.nextElementSibling;
+  if (streamEl && streamEl.classList.contains("tool-output-stream")) {
+    streamEl.remove();
+  }
+
   const stripped = stripAnsi(output || "").trim();
   if (!stripped) return;
 
@@ -1514,12 +1565,28 @@ function appendToolOutput(name, output) {
   out.className = "tool-output";
   out.textContent = stripped;
 
-  // Auto-collapse long output
+  // Auto-collapse long output (keyboard-accessible)
   const lineCount = stripped.split("\n").length;
   if (lineCount > 10) {
     out.classList.add("collapsed");
-    out.addEventListener("click", function () {
+    out.setAttribute("tabindex", "0");
+    out.setAttribute("role", "button");
+    out.setAttribute(
+      "aria-label",
+      "Tool output (collapsed). Activate to expand.",
+    );
+    var expandHandler = function () {
       this.classList.remove("collapsed");
+      this.removeAttribute("tabindex");
+      this.removeAttribute("role");
+      this.removeAttribute("aria-label");
+    };
+    out.addEventListener("click", expandHandler);
+    out.addEventListener("keydown", function (e) {
+      if (e.key === "Enter" || e.key === " ") {
+        e.preventDefault();
+        expandHandler.call(this);
+      }
     });
   }
 
@@ -1531,8 +1598,7 @@ function appendToolOutput(name, output) {
 function addInfoMessage(text) {
   const el = document.createElement("div");
   el.className = "msg msg-info";
-  // Strip ANSI codes
-  el.textContent = text.replace(/\x1b\[[0-9;]*m/g, "");
+  el.textContent = stripAnsi(text);
   messagesEl.appendChild(el);
   scrollToBottom();
 }
@@ -1541,7 +1607,7 @@ function addErrorMessage(text) {
   const el = document.createElement("div");
   el.className = "msg msg-error";
   el.setAttribute("role", "alert");
-  el.textContent = text.replace(/\x1b\[[0-9;]*m/g, "");
+  el.textContent = stripAnsi(text);
   messagesEl.appendChild(el);
   scrollToBottom();
 }

turnstone/ui/static/style.css

@@ -510,6 +510,15 @@ body {
   max-height: 300px;
   overflow-y: auto;
 }
+.tool-output-stream {
+  border-left: 2px solid var(--accent);
+  max-height: 400px;
+  animation: stream-pulse 2s ease-in-out infinite;
+}
+@keyframes stream-pulse {
+  0%, 100% { border-left-color: var(--accent); }
+  50% { border-left-color: rgba(229, 160, 66, 0.15); }
+}
 .tool-output.collapsed { max-height: 150px; position: relative; }
 .tool-output.collapsed::after {
   content: 'click to expand';
@@ -826,6 +835,7 @@ body {
   :root { --dash-grid: 50px 1fr 50px; }
   .dash-col-node, .dash-cell-node, .dash-col-task, .dash-cell-task, .dash-col-ctx, .dash-cell-ctx { display: none; }
   .dash-row-sub { padding-left: 66px; }
+  .tool-output, .tool-output-stream { max-height: 200px; }
 }
 
 /* ==========================================================================
@@ -838,6 +848,7 @@ body {
   .dash-state-dot[data-state="running"],
   .dash-state-dot[data-state="thinking"],
   .dash-state-dot[data-state="attention"] { animation: none; opacity: 1; }
+  .tool-output-stream { animation: none; border-left-color: var(--accent); }
   .thinking-indicator::after { animation: none; content: '...'; }
   .ws-tab, .ws-tab .tab-close, #new-tab-btn,
   .header-btn, .hmenu-item, .dashboard-card,