From 402e2bfee090c62ebbef29630d95bc4133c69ba8 Mon Sep 17 00:00:00 2001 From: yangziyu-hhh Date: Wed, 10 Jun 2026 13:46:50 +0800 Subject: [PATCH] feat: stream Bash progress and guard message actions during replies --- agent/protocol/agent_stream.py | 15 +++- agent/tools/base_tool.py | 10 +++ agent/tools/bash/bash.py | 106 ++++++++++++++++++++++++++--- channel/web/static/css/console.css | 18 +++++ channel/web/static/js/console.js | 78 ++++++++++++++++----- channel/web/web_channel.py | 11 ++- 6 files changed, 207 insertions(+), 31 deletions(-) diff --git a/agent/protocol/agent_stream.py b/agent/protocol/agent_stream.py index 1b49baa7..1eb96e50 100644 --- a/agent/protocol/agent_stream.py +++ b/agent/protocol/agent_stream.py @@ -1174,10 +1174,21 @@ class AgentStreamExecutor: # Set tool context tool.model = self.model tool.context = self.agent + tool.progress_callback = lambda message: self._emit_event( + "tool_execution_progress", + { + "tool_call_id": tool_id, + "tool_name": tool_name, + "message": message, + } + ) # Execute tool start_time = time.time() - result: ToolResult = tool.execute_tool(arguments) + try: + result: ToolResult = tool.execute_tool(arguments) + finally: + tool.progress_callback = None execution_time = time.time() - start_time result_dict = { @@ -1719,4 +1730,4 @@ class AgentStreamExecutor: not as a message. The AgentLLMModel will handle this. """ # Don't add system message here - it will be handled separately by the LLM adapter - return self.messages \ No newline at end of file + return self.messages diff --git a/agent/tools/base_tool.py b/agent/tools/base_tool.py index a3ca2625..53f7cf9c 100644 --- a/agent/tools/base_tool.py +++ b/agent/tools/base_tool.py @@ -38,6 +38,16 @@ class BaseTool: description: str = "Base tool" params: dict = {} # Store JSON Schema model: Optional[Any] = None # LLM model instance, type depends on bot implementation + progress_callback = None + + def report_progress(self, message: str): + callback = getattr(self, "progress_callback", None) + if not callback: + return + try: + callback(str(message)) + except Exception as e: + logger.debug(f"[{self.name}] progress callback failed: {e}") @classmethod def get_json_schema(cls) -> dict: diff --git a/agent/tools/bash/bash.py b/agent/tools/bash/bash.py index 5e02832e..4fcc724c 100644 --- a/agent/tools/bash/bash.py +++ b/agent/tools/bash/bash.py @@ -4,9 +4,12 @@ Bash tool - Execute bash commands import os import re +import signal import sys import subprocess import tempfile +import threading +import time from typing import Dict, Any from agent.tools.base_tool import BaseTool, ToolResult @@ -19,6 +22,8 @@ class Bash(BaseTool): """Tool for executing bash commands""" _IS_WIN = sys.platform == "win32" + _PROGRESS_MAX_BYTES = 32 * 1024 + _PROGRESS_INTERVAL = 0.5 name: str = "bash" description: str = f"""Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last {DEFAULT_MAX_LINES} lines or {DEFAULT_MAX_BYTES // 1024}KB (whichever is hit first). If truncated, full output is saved to a temp file. @@ -113,17 +118,11 @@ SAFETY: if command and not command.strip().lower().startswith("chcp"): command = f"chcp 65001 >nul 2>&1 && {command}" - result = subprocess.run( + result = self._run_streaming( command, - shell=True, - cwd=self.cwd, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - encoding="utf-8", - errors="replace", - timeout=timeout, - env=env, + timeout, + env, + dotenv_vars, ) logger.debug(f"[Bash] Exit code: {result.returncode}") @@ -236,6 +235,93 @@ SAFETY: except Exception as e: return ToolResult.fail(f"Error executing command: {str(e)}") + def _run_streaming(self, command: str, timeout: int, env: dict, dotenv_vars: dict): + process = subprocess.Popen( + command, + shell=True, + cwd=self.cwd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + start_new_session=not self._IS_WIN, + ) + stdout_chunks, stderr_chunks = [], [] + recent = bytearray() + recent_lock = threading.Lock() + + def drain(stream, chunks): + while True: + chunk = os.read(stream.fileno(), 4096) + if not chunk: + break + chunks.append(chunk) + with recent_lock: + recent.extend(chunk) + if len(recent) > self._PROGRESS_MAX_BYTES: + del recent[:-self._PROGRESS_MAX_BYTES] + + readers = [ + threading.Thread(target=drain, args=(process.stdout, stdout_chunks), daemon=True), + threading.Thread(target=drain, args=(process.stderr, stderr_chunks), daemon=True), + ] + for reader in readers: + reader.start() + + started = time.monotonic() + last_reported_at = started + last_snapshot = None + try: + while process.poll() is None: + now = time.monotonic() + elapsed = now - started + if elapsed >= timeout: + self._kill_process(process) + raise subprocess.TimeoutExpired(command, timeout) + if elapsed >= self._PROGRESS_INTERVAL and now - last_reported_at >= self._PROGRESS_INTERVAL: + with recent_lock: + snapshot = bytes(recent).decode("utf-8", errors="replace") + snapshot = self._redact_progress(snapshot, dotenv_vars) + if snapshot and snapshot != last_snapshot: + self.report_progress(snapshot) + last_snapshot = snapshot + last_reported_at = now + time.sleep(0.1) + finally: + if process.poll() is None: + self._kill_process(process) + process.wait() + for reader in readers: + reader.join() + + from types import SimpleNamespace + return SimpleNamespace( + returncode=process.returncode, + stdout=b"".join(stdout_chunks).decode("utf-8", errors="replace"), + stderr=b"".join(stderr_chunks).decode("utf-8", errors="replace"), + ) + + def _kill_process(self, process): + if self._IS_WIN: + process.kill() + else: + try: + os.killpg(process.pid, signal.SIGKILL) + except PermissionError: + process.kill() + + @staticmethod + def _redact_progress(text: str, dotenv_vars: dict) -> str: + text = re.sub( + r'(?i)\b(API_KEY|TOKEN|PASSWORD|AUTHORIZATION)\s*=\s*[^\s]+', + lambda match: f"{match.group(1)}=[REDACTED]", + text, + ) + for value in dotenv_vars.values(): + value = str(value or "") + if len(value) >= 6: + text = text.replace(value, "[REDACTED]") + return text + def _get_safety_warning(self, command: str) -> str: """ Get safety warning for absolutely catastrophic commands only. diff --git a/channel/web/static/css/console.css b/channel/web/static/css/console.css index e260bbc9..0e2a09db 100644 --- a/channel/web/static/css/console.css +++ b/channel/web/static/css/console.css @@ -605,6 +605,18 @@ color: inherit; } .tool-error-text { color: #f87171; } +.tool-live-output:empty { display: none; } +.tool-streaming .tool-live-output:not(:empty)::after { + content: ' '; + display: inline-block; + width: 0.45em; + height: 1em; + margin-left: 0.15em; + vertical-align: -0.15em; + background: currentColor; + animation: tool-cursor-blink 1s steps(1) infinite; +} +@keyframes tool-cursor-blink { 50% { opacity: 0; } } /* Log level highlighting */ .log-line { display: block; } @@ -1571,3 +1583,9 @@ .delete-msg-btn:hover { color: #ef4444 !important; } + +.edit-msg-btn:disabled, +.delete-msg-btn:disabled { + cursor: not-allowed !important; + opacity: 0.35 !important; +} diff --git a/channel/web/static/js/console.js b/channel/web/static/js/console.js index 903c7df6..31727cbf 100644 --- a/channel/web/static/js/console.js +++ b/channel/web/static/js/console.js @@ -889,6 +889,26 @@ let pollGeneration = 0; // incremented on each restart to cancel stale poll lo let loadingContainers = {}; let activeStreams = {}; // request_id -> EventSource let sessionActiveRequest = {}; // session_id -> request_id (in-flight stream per session) + +function isCurrentSessionConversationActive() { + return !!sessionActiveRequest[sessionId]; +} + +function updateEditButtonsState() { + const active = isCurrentSessionConversationActive(); + document.querySelectorAll('.edit-msg-btn, .delete-msg-btn').forEach(btn => { + btn.disabled = active; + if (btn.classList.contains('edit-msg-btn')) { + btn.title = active + ? 'Reply is being generated; editing is temporarily unavailable.' + : (currentLang === 'zh' ? '编辑消息' : 'Edit Message'); + } else { + btn.title = active + ? 'Reply is being generated; deletion is temporarily unavailable.' + : t('delete_message_title'); + } + }); +} let streamBuffers = {}; // request_id -> { items: [event...], timestamp } for re-attach replay let isComposing = false; let appConfig = { use_agent: false, title: 'CowAgent', subtitle: '', providers: {}, api_bases: {} }; @@ -1177,6 +1197,7 @@ messagesDiv.addEventListener('click', (e) => { const editBtn = e.target.closest('.edit-msg-btn'); if (editBtn) { e.preventDefault(); + if (isCurrentSessionConversationActive()) return; const msgRoot = editBtn.closest('.user-message-group'); if (msgRoot) editUserMessage(msgRoot); return; @@ -1197,6 +1218,7 @@ messagesDiv.addEventListener('click', (e) => { const deleteBtn = e.target.closest('.delete-msg-btn'); if (deleteBtn) { e.preventDefault(); + if (isCurrentSessionConversationActive()) return; const userMsgEl = deleteBtn.closest('.user-message-group'); if (!userMsgEl) return; @@ -1993,6 +2015,7 @@ function copyToClipboard(text) { // Edit user message: extract content, remove this and subsequent messages, fill input async function editUserMessage(msgEl) { + if (isCurrentSessionConversationActive()) return; const rawContent = msgEl.dataset.rawContent; if (!rawContent) return; @@ -2043,9 +2066,9 @@ async function editUserMessage(msgEl) { // Fill input with the original content chatInput.value = rawContent; - chatInput.style.height = 'auto'; - chatInput.style.height = chatInput.scrollHeight + 'px'; + chatInput.dispatchEvent(new Event("input", { bubbles: true })); chatInput.focus(); + chatInput.selectionStart = chatInput.selectionEnd = chatInput.value.length; scrollChatToBottom(); } @@ -2250,7 +2273,7 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) { let contentEl = null; // .answer-content (final streaming answer) let mediaEl = null; // .media-content (images & file attachments) let accumulatedText = ''; - let currentToolEl = null; + const toolElements = new Map(); let currentReasoningEl = null; // live reasoning bubble let reasoningText = ''; let reasoningStartTime = 0; @@ -2265,12 +2288,14 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) { const ownerSession = sessionId; const isActive = () => ownerSession === sessionId; sessionActiveRequest[ownerSession] = requestId; + updateEditButtonsState(); // Per-request event buffer used to rebuild the bubble on re-attach. const buffer = streamBuffers[requestId] || { items: [], timestamp }; streamBuffers[requestId] = buffer; const clearOwnerRequest = () => { if (sessionActiveRequest[ownerSession] === requestId) { delete sessionActiveRequest[ownerSession]; + updateEditButtonsState(); } delete streamBuffers[requestId]; }; @@ -2419,10 +2444,10 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) { contentEl.innerHTML = ''; // Add tool execution indicator (collapsible) - currentToolEl = document.createElement('div'); - currentToolEl.className = 'agent-step agent-tool-step'; + const toolEl = document.createElement('div'); + toolEl.className = 'agent-step agent-tool-step tool-streaming'; const argsStr = formatToolArgs(item.arguments || {}); - currentToolEl.innerHTML = ` + toolEl.innerHTML = `
${item.tool} @@ -2433,36 +2458,51 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) {
Input
${argsStr}
-
+
+
Output
+ 

+                        

                     `;
-                stepsEl.appendChild(currentToolEl);
+                stepsEl.appendChild(toolEl);
+                toolElements.set(item.tool_call_id, toolEl);
 
                 scrollChatToBottom();
 
+            } else if (item.type === 'tool_progress') {
+                const toolEl = toolElements.get(item.tool_call_id);
+                if (toolEl) {
+                    toolEl.classList.add('expanded');
+                    toolEl.querySelector('.tool-live-output').textContent = String(item.content || '');
+                    scrollChatToBottom();
+                }
+
             } else if (item.type === 'tool_end') {
-                if (currentToolEl) {
+                const toolEl = toolElements.get(item.tool_call_id);
+                if (toolEl) {
                     const isError = item.status !== 'success';
-                    const icon = currentToolEl.querySelector('.tool-icon');
+                    const icon = toolEl.querySelector('.tool-icon');
                     icon.className = isError
                         ? 'fas fa-times text-red-400 flex-shrink-0 tool-icon'
                         : 'fas fa-check text-primary-400 flex-shrink-0 tool-icon';
 
                     // Show execution time
-                    const nameEl = currentToolEl.querySelector('.tool-name');
+                    const nameEl = toolEl.querySelector('.tool-name');
                     if (item.execution_time !== undefined) {
                         nameEl.innerHTML += ` ${item.execution_time}s`;
                     }
 
                     // Fill output section
-                    const outputSection = currentToolEl.querySelector('.tool-output-section');
-                    if (outputSection && item.result) {
-                        outputSection.innerHTML = `
-                            
${isError ? 'Error' : 'Output'}

-                            
${escapeHtml(String(item.result))}
`;
+                    const outputLabel = toolEl.querySelector('.tool-output-label');
+                    const outputEl = toolEl.querySelector('.tool-live-output');
+                    if (outputLabel) outputLabel.textContent = isError ? 'Error' : 'Output';
+                    if (outputEl) {
+                        outputEl.textContent = item.result ? String(item.result) : '';
+                        outputEl.classList.toggle('tool-error-text', isError);
                     }
 
-                    if (isError) currentToolEl.classList.add('tool-failed');
-                    currentToolEl = null;
+                    toolEl.classList.remove('tool-streaming');
+                    if (isError) toolEl.classList.add('tool-failed');
+                    toolElements.delete(item.tool_call_id);
                 }
 
             } else if (item.type === 'image') {
@@ -3256,6 +3296,7 @@ function loadHistory(page) {
             const sentinel = document.getElementById('history-load-more');
             const insertBefore = sentinel ? sentinel.nextSibling : messagesDiv.firstChild;
             messagesDiv.insertBefore(fragment, insertBefore);
+            updateEditButtonsState();
 
             // Manage the "load more" sentinel at the very top
             if (data.has_more) {
@@ -3711,6 +3752,7 @@ function switchSession(newSessionId) {
     // Switching back re-attaches and resumes live streaming.
 
     sessionId = newSessionId;
+    updateEditButtonsState();
     localStorage.setItem(SESSION_ID_KEY, sessionId);
 
     historyPage = 0;
diff --git a/channel/web/web_channel.py b/channel/web/web_channel.py
index 1be31c6a..3b374c5f 100644
--- a/channel/web/web_channel.py
+++ b/channel/web/web_channel.py
@@ -432,7 +432,15 @@ class WebChannel(ChatChannel):
             elif event_type == "tool_execution_start":
                 tool_name = data.get("tool_name", "tool")
                 arguments = data.get("arguments", {})
-                q.put({"type": "tool_start", "tool": tool_name, "arguments": arguments})
+                q.put({"type": "tool_start", "tool_call_id": data.get("tool_call_id"), "tool": tool_name, "arguments": arguments})
+
+            elif event_type == "tool_execution_progress":
+                q.put({
+                    "type": "tool_progress",
+                    "tool_call_id": data.get("tool_call_id"),
+                    "tool": data.get("tool_name", "tool"),
+                    "content": str(data.get("message", ""))[-32 * 1024:],
+                })
 
             elif event_type == "tool_execution_end":
                 tool_name = data.get("tool_name", "tool")
@@ -445,6 +453,7 @@ class WebChannel(ChatChannel):
                     result_str = result_str[:2000] + "…"
                 q.put({
                     "type": "tool_end",
+                    "tool_call_id": data.get("tool_call_id"),
                     "tool": tool_name,
                     "status": status,
                     "result": result_str,