From 075d9fc608b26638c6670e9c7a124ce9ce602923 Mon Sep 17 00:00:00 2001 From: yangziyu-hhh Date: Fri, 12 Jun 2026 13:45:39 +0800 Subject: [PATCH] fix: address Bash streaming review feedback --- agent/tools/bash/bash.py | 22 +++- channel/web/static/js/console.js | 28 ++++- channel/web/web_channel.py | 2 +- tests/test_bash_streaming.py | 185 +++++++++++++++++++++++++++++++ 4 files changed, 227 insertions(+), 10 deletions(-) create mode 100644 tests/test_bash_streaming.py diff --git a/agent/tools/bash/bash.py b/agent/tools/bash/bash.py index 4fcc724c..dad6fb95 100644 --- a/agent/tools/bash/bash.py +++ b/agent/tools/bash/bash.py @@ -22,7 +22,7 @@ class Bash(BaseTool): """Tool for executing bash commands""" _IS_WIN = sys.platform == "win32" - _PROGRESS_MAX_BYTES = 32 * 1024 + _PROGRESS_MAX_BYTES = 4 * 1024 _PROGRESS_INTERVAL = 0.5 name: str = "bash" @@ -290,8 +290,9 @@ SAFETY: if process.poll() is None: self._kill_process(process) process.wait() + join_deadline = time.monotonic() + 5 for reader in readers: - reader.join() + reader.join(timeout=max(0, join_deadline - time.monotonic())) from types import SimpleNamespace return SimpleNamespace( @@ -302,12 +303,23 @@ SAFETY: def _kill_process(self, process): if self._IS_WIN: - process.kill() + try: + result = subprocess.run( + ["taskkill", "/F", "/T", "/PID", str(process.pid)], + capture_output=True, + timeout=5, + ) + if result.returncode != 0 and process.poll() is None: + process.kill() + except (OSError, subprocess.SubprocessError): + if process.poll() is None: + process.kill() else: try: os.killpg(process.pid, signal.SIGKILL) - except PermissionError: - process.kill() + except (PermissionError, ProcessLookupError): + if process.poll() is None: + process.kill() @staticmethod def _redact_progress(text: str, dotenv_vars: dict) -> str: diff --git a/channel/web/static/js/console.js b/channel/web/static/js/console.js index 31727cbf..825871a9 100644 --- a/channel/web/static/js/console.js +++ b/channel/web/static/js/console.js @@ -188,6 +188,8 @@ const I18N = { delete_session_title: '删除会话', delete_message_confirm: '确认删除这条消息？', delete_message_title: '删除消息', + edit_disabled_reply_active: '正在生成回复，暂时无法编辑。', + delete_disabled_reply_active: '正在生成回复，暂时无法删除。', untitled_session: '新对话', context_cleared: '— 以上内容已从上下文中移除 —', tip_new_chat: '新建对话', @@ -392,6 +394,8 @@ const I18N = { delete_session_title: 'Delete Session', delete_message_confirm: 'Delete this message?', delete_message_title: 'Delete Message', + edit_disabled_reply_active: 'Reply is being generated; editing is temporarily unavailable.', + delete_disabled_reply_active: 'Reply is being generated; deletion is temporarily unavailable.', untitled_session: 'New Chat', context_cleared: '— Context above has been cleared —', tip_new_chat: 'New Chat', @@ -900,11 +904,11 @@ function updateEditButtonsState() { 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'); + ? t('edit_disabled_reply_active') + : t('edit_message'); } else { btn.title = active - ? 'Reply is being generated; deletion is temporarily unavailable.' + ? t('delete_disabled_reply_active') : t('delete_message_title'); } }); @@ -2446,6 +2450,7 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) { // Add tool execution indicator (collapsible) const toolEl = document.createElement('div'); toolEl.className = 'agent-step agent-tool-step tool-streaming'; + toolEl.dataset.progressReceived = 'false'; const argsStr = formatToolArgs(item.arguments || {}); toolEl.innerHTML = `
@@ -2471,7 +2476,10 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) { } else if (item.type === 'tool_progress') { const toolEl = toolElements.get(item.tool_call_id); if (toolEl) { - toolEl.classList.add('expanded'); + if (toolEl.dataset.progressReceived !== 'true') { + toolEl.classList.add('expanded'); + toolEl.dataset.progressReceived = 'true'; + } toolEl.querySelector('.tool-live-output').textContent = String(item.content || ''); scrollChatToBottom(); } @@ -2501,6 +2509,11 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) { } toolEl.classList.remove('tool-streaming'); + toolEl.classList.remove('expanded'); + if (!item.result) { + const outputSection = toolEl.querySelector('.tool-output-section'); + if (outputSection) outputSection.remove(); + } if (isError) toolEl.classList.add('tool-failed'); toolElements.delete(item.tool_call_id); } @@ -2661,6 +2674,13 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) { // Record every event for re-attach replay (capped to avoid // unbounded growth on very long streams). + if (item.type === 'tool_progress' && item.tool_call_id) { + const previousIndex = buffer.items.findIndex( + buffered => buffered.type === 'tool_progress' + && buffered.tool_call_id === item.tool_call_id + ); + if (previousIndex >= 0) buffer.items.splice(previousIndex, 1); + } if (buffer.items.length < 5000) buffer.items.push(item); // Background session: keep the stream alive so the reply finishes diff --git a/channel/web/web_channel.py b/channel/web/web_channel.py index 3b374c5f..341c13bc 100644 --- a/channel/web/web_channel.py +++ b/channel/web/web_channel.py @@ -439,7 +439,7 @@ class WebChannel(ChatChannel): "type": "tool_progress", "tool_call_id": data.get("tool_call_id"), "tool": data.get("tool_name", "tool"), - "content": str(data.get("message", ""))[-32 * 1024:], + "content": str(data.get("message", ""))[-4 * 1024:], }) elif event_type == "tool_execution_end": diff --git a/tests/test_bash_streaming.py b/tests/test_bash_streaming.py new file mode 100644 index 00000000..ff3d3591 --- /dev/null +++ b/tests/test_bash_streaming.py @@ -0,0 +1,185 @@ +import subprocess +import sys +import time +from unittest.mock import patch + +import pytest + +from agent.tools.bash.bash import Bash + + +posix_only = pytest.mark.skipif(Bash._IS_WIN, reason="POSIX shell command") +windows_only = pytest.mark.skipif(not Bash._IS_WIN, reason="Windows integration test") + + +def _windows_pid_command(pid_file): + path = str(pid_file).replace("\\", "\\\\") + return ( + f'"{sys.executable}" -c ' + f'"import os,time;open(r\'{path}\',\'w\').write(str(os.getpid()));time.sleep(30)"' + ) + + +def _windows_pid_is_running(pid): + result = subprocess.run( + ["tasklist", "/FI", f"PID eq {pid}", "/FO", "CSV", "/NH"], + capture_output=True, + text=True, + errors="replace", + ) + return f'"{pid}"' in result.stdout + + +def _windows_kill_pid_file(pid_file): + if pid_file.exists(): + subprocess.run( + ["taskkill", "/F", "/PID", pid_file.read_text()], + capture_output=True, + ) + + +@posix_only +def test_fast_command_returns_output_without_progress(tmp_path): + tool = Bash({"cwd": str(tmp_path)}) + progress = [] + tool.progress_callback = progress.append + + result = tool.execute({"command": "printf fast"}) + + assert result.status == "success" + assert result.result["output"] == "fast" + assert progress == [] + + +@posix_only +def test_timeout_returns_promptly(tmp_path): + tool = Bash({"cwd": str(tmp_path)}) + started = time.monotonic() + + result = tool.execute({"command": "sleep 10", "timeout": 1}) + + assert result.status == "error" + assert "timed out after 1 seconds" in result.result + assert time.monotonic() - started < 3 + + +@posix_only +def test_background_process_holding_pipe_does_not_hang(tmp_path): + tool = Bash({"cwd": str(tmp_path)}) + started = time.monotonic() + + result = tool.execute({"command": "sleep 10 & printf done", "timeout": 3}) + + assert result.status == "success" + assert result.result["output"] == "done" + assert time.monotonic() - started < 7 + + +@posix_only +def test_output_without_trailing_newline_streams_progress(tmp_path): + tool = Bash({"cwd": str(tmp_path)}) + progress = [] + tool.progress_callback = progress.append + + result = tool.execute({ + "command": "for i in 1 2 3 4 5; do printf .; sleep 0.3; done", + "timeout": 5, + }) + + assert result.status == "success" + assert result.result["output"] == "....." + assert progress + assert progress[-1].endswith(".") + + +def test_windows_kill_uses_taskkill_for_process_tree(): + tool = Bash() + process = type("Process", (), {"pid": 1234, "poll": lambda self: None, "kill": lambda self: None})() + + with patch.object(tool, "_IS_WIN", True), patch.object(subprocess, "run") as run: + run.return_value.returncode = 0 + tool._kill_process(process) + + run.assert_called_once_with( + ["taskkill", "/F", "/T", "/PID", "1234"], + capture_output=True, + timeout=5, + ) + + +def test_windows_kill_falls_back_when_taskkill_fails(): + tool = Bash() + process = type("Process", (), {"pid": 1234, "poll": lambda self: None, "kill": lambda self: None})() + + with patch.object(tool, "_IS_WIN", True), \ + patch.object(subprocess, "run", side_effect=OSError), \ + patch.object(process, "kill") as kill: + tool._kill_process(process) + + kill.assert_called_once() + + +@windows_only +def test_windows_timeout_kills_process_tree(tmp_path): + pid_file = tmp_path / "timeout-child.pid" + child_command = _windows_pid_command(pid_file) + tool = Bash({"cwd": str(tmp_path)}) + started = time.monotonic() + + try: + result = tool.execute({ + "command": f'start "" /b {child_command} & ping -n 30 127.0.0.1 >nul', + "timeout": 1, + }) + + assert result.status == "error" + assert "timed out after 1 seconds" in result.result + assert time.monotonic() - started < 8 + assert pid_file.exists() + time.sleep(0.5) + assert not _windows_pid_is_running(pid_file.read_text()) + finally: + _windows_kill_pid_file(pid_file) + + +@windows_only +def test_windows_long_running_command_streams_progress(tmp_path): + tool = Bash({"cwd": str(tmp_path)}) + progress = [] + tool.progress_callback = progress.append + + result = tool.execute({ + "command": ( + f'"{sys.executable}" -u -c ' + '"import sys,time; ' + "[(sys.stdout.write('.'),sys.stdout.flush(),time.sleep(0.3)) " + 'for _ in range(5)]"' + ), + "timeout": 5, + }) + + assert result.status == "success" + assert result.result["output"] == "....." + assert progress + assert progress[-1].endswith(".") + + +@windows_only +def test_windows_background_process_holding_pipe_does_not_hang(tmp_path): + pid_file = tmp_path / "background-child.pid" + child_command = _windows_pid_command(pid_file) + tool = Bash({"cwd": str(tmp_path)}) + started = time.monotonic() + + try: + result = tool.execute({ + "command": f'start "" /b {child_command} & echo done', + "timeout": 3, + }) + + assert result.status == "success" + assert result.result["output"].strip() == "done" + assert time.monotonic() - started < 8 + assert pid_file.exists() + finally: + _windows_kill_pid_file(pid_file)