fix: address Bash streaming review feedback

This commit is contained in:
yangziyu-hhh
2026-06-12 13:45:39 +08:00
parent 402e2bfee0
commit 075d9fc608
4 changed files with 227 additions and 10 deletions

View File

@@ -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,11 +303,22 @@ SAFETY:
def _kill_process(self, process):
if self._IS_WIN:
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:
except (PermissionError, ProcessLookupError):
if process.poll() is None:
process.kill()
@staticmethod

View File

@@ -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 = `
<div class="tool-header" onclick="this.parentElement.classList.toggle('expanded')">
@@ -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) {
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

View File

@@ -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":

View File

@@ -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)