diff --git a/docs/api-reference.md b/docs/api-reference.md index 42f7169d..cef63bb0 100644 --- a/docs/api-reference.md +++ b/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"} diff --git a/docs/architecture.md b/docs/architecture.md index f9036c7a..afdffeb0 100644 --- a/docs/architecture.md +++ b/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: ... diff --git a/docs/diagrams/03-core-engine-classes.puml b/docs/diagrams/03-core-engine-classes.puml index 88718319..c07cc605 100644 --- a/docs/diagrams/03-core-engine-classes.puml +++ b/docs/diagrams/03-core-engine-classes.puml @@ -13,6 +13,7 @@ interface "SessionUI" as SessionUI <> { + 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) diff --git a/docs/diagrams/04-conversation-turn.puml b/docs/diagrams/04-conversation-turn.puml index 375bb80c..49790884 100644 --- a/docs/diagrams/04-conversation-turn.puml +++ b/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), ...] diff --git a/docs/diagrams/05-tool-pipeline.puml b/docs/diagrams/05-tool-pipeline.puml index f4730013..6658db69 100644 --- a/docs/diagrams/05-tool-pipeline.puml +++ b/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) diff --git a/docs/diagrams/06-mq-protocol.puml b/docs/diagrams/06-mq-protocol.puml index c78267ba..018bb7ed 100644 --- a/docs/diagrams/06-mq-protocol.puml +++ b/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 diff --git a/docs/diagrams/07-message-routing.puml b/docs/diagrams/07-message-routing.puml index e52bb7bb..c5e79b9f 100644 --- a/docs/diagrams/07-message-routing.puml +++ b/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") diff --git a/docs/diagrams/png/03-core-engine-classes.png b/docs/diagrams/png/03-core-engine-classes.png index 4c0929dd..e482105c 100644 --- a/docs/diagrams/png/03-core-engine-classes.png +++ b/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 diff --git a/docs/diagrams/png/04-conversation-turn.png b/docs/diagrams/png/04-conversation-turn.png index d2e031de..caf08cc0 100644 --- a/docs/diagrams/png/04-conversation-turn.png +++ b/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 diff --git a/docs/diagrams/png/05-tool-pipeline.png b/docs/diagrams/png/05-tool-pipeline.png index c1c331a2..42edba8e 100644 --- a/docs/diagrams/png/05-tool-pipeline.png +++ b/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 diff --git a/docs/diagrams/png/06-mq-protocol.png b/docs/diagrams/png/06-mq-protocol.png index 92823911..2da536a2 100644 --- a/docs/diagrams/png/06-mq-protocol.png +++ b/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 diff --git a/docs/diagrams/png/07-message-routing.png b/docs/diagrams/png/07-message-routing.png index e251ef65..e1dcdc77 100644 --- a/docs/diagrams/png/07-message-routing.png +++ b/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 diff --git a/docs/tools.md b/docs/tools.md index e6d3082f..2688ae44 100644 --- a/docs/tools.md +++ b/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. diff --git a/tests/test_server_live.py b/tests/test_server_live.py index f0e0cddc..084af965 100644 --- a/tests/test_server_live.py +++ b/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",)) diff --git a/tests/test_session.py b/tests/test_session.py index edce20ee..696d240d 100644 --- a/tests/test_session.py +++ b/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 diff --git a/tests/test_workstream.py b/tests/test_workstream.py index e47ac9e1..669fd424 100644 --- a/tests/test_workstream.py +++ b/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 diff --git a/turnstone/cli.py b/turnstone/cli.py index de9163b7..5609dc9a 100644 --- a/turnstone/cli.py +++ b/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: diff --git a/turnstone/core/session.py b/turnstone/core/session.py index ec6bac0f..c16574a1 100644 --- a/turnstone/core/session.py +++ b/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)" diff --git a/turnstone/eval.py b/turnstone/eval.py index 8eff87c4..25208894 100644 --- a/turnstone/eval.py +++ b/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 diff --git a/turnstone/mq/bridge.py b/turnstone/mq/bridge.py index ae558d85..eefe859c 100644 --- a/turnstone/mq/bridge.py +++ b/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, diff --git a/turnstone/mq/protocol.py b/turnstone/mq/protocol.py index 99e4638c..7c75f4b1 100644 --- a/turnstone/mq/protocol.py +++ b/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, diff --git a/turnstone/server.py b/turnstone/server.py index faf6da1b..5f114664 100644 --- a/turnstone/server.py +++ b/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 diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js index b9093e39..66fce098 100644 --- a/turnstone/ui/static/app.js +++ b/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(); } diff --git a/turnstone/ui/static/style.css b/turnstone/ui/static/style.css index 4c03f64f..2ff644ac 100644 --- a/turnstone/ui/static/style.css +++ b/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,