Merge pull request #2879 from yangziyu-hhh/master

feat: stream Bash progress and guard message actions during replies

.github/workflows/test-windows-bash.yml

@@ -0,0 +1,32 @@
+name: Windows Bash Streaming Tests
+
+on:
+  workflow_dispatch:
+  pull_request:
+    paths:
+      - "agent/tools/bash/bash.py"
+      - "tests/test_bash_streaming.py"
+      - ".github/workflows/test-windows-bash.yml"
+
+jobs:
+  windows-bash-tests:
+    runs-on: windows-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: pip
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install pytest
+          python -m pip install -r requirements.txt
+
+      - name: Run Windows Bash streaming tests
+        run: python -m pytest tests/test_bash_streaming.py -v

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
+        return self.messages

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:

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 = 4 * 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,
+                timeout,
-                cwd=self.cwd,
+                env,
-                stdout=subprocess.PIPE,
+                dotenv_vars,
-                stderr=subprocess.PIPE,
-                text=True,
-                encoding="utf-8",
-                errors="replace",
-                timeout=timeout,
-                env=env,
             )
             
             logger.debug(f"[Bash] Exit code: {result.returncode}")
@@ -236,6 +235,105 @@ 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()
+            join_deadline = time.monotonic() + 5
+            for reader in readers:
+                reader.join(timeout=max(0, join_deadline - time.monotonic()))
+
+        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:
+            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, ProcessLookupError):
+                if process.poll() is None:
+                    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.

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; }
@@ -1579,3 +1591,9 @@
 .delete-msg-btn:hover {
     color: #ef4444 !important;
 }
+
+.edit-msg-btn:disabled,
+.delete-msg-btn:disabled {
+    cursor: not-allowed !important;
+    opacity: 0.35 !important;
+}

channel/web/static/js/console.js

@@ -197,6 +197,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: '新建对话',
@@ -410,6 +412,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',
@@ -907,6 +911,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
+                ? t('edit_disabled_reply_active')
+                : t('edit_message');
+        } else {
+            btn.title = active
+                ? t('delete_disabled_reply_active')
+                : 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: {} };
@@ -1195,6 +1219,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;
@@ -1215,6 +1240,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;
 
@@ -2011,6 +2037,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;
 
@@ -2061,9 +2088,9 @@ async function editUserMessage(msgEl) {
 
     // Fill input with the original content
     chatInput.value = rawContent;
-    chatInput.style.height = 'auto';
+    chatInput.dispatchEvent(new Event("input", { bubbles: true }));
-    chatInput.style.height = chatInput.scrollHeight + 'px';
     chatInput.focus();
+    chatInput.selectionStart = chatInput.selectionEnd = chatInput.value.length;
     scrollChatToBottom();
 }
 
@@ -2268,7 +2295,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;
@@ -2283,12 +2310,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];
     };
@@ -2437,10 +2466,11 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) {
                 contentEl.innerHTML = '';
 
                 // Add tool execution indicator (collapsible)
-                currentToolEl = document.createElement('div');
+                const toolEl = document.createElement('div');
-                currentToolEl.className = 'agent-step agent-tool-step';
+                toolEl.className = 'agent-step agent-tool-step tool-streaming';
+                toolEl.dataset.progressReceived = 'false';
                 const argsStr = formatToolArgs(item.arguments || {});
-                currentToolEl.innerHTML = `
+                toolEl.innerHTML = `
                     <div class="tool-header" onclick="this.parentElement.classList.toggle('expanded')">
                         <i class="fas fa-cog fa-spin text-primary-400 flex-shrink-0 tool-icon"></i>
                         <span class="tool-name">${item.tool}</span>
@@ -2451,36 +2481,59 @@ function startSSE(requestId, loadingEl, timestamp, titleInfo, replayItems) {
                             <div class="tool-detail-label">Input</div>
                             <pre class="tool-detail-content">${argsStr}</pre>
                         </div>
-                        <div class="tool-detail-section tool-output-section"></div>
+                        <div class="tool-detail-section tool-output-section">
+                            <div class="tool-detail-label tool-output-label">Output</div>
+                            <pre class="tool-detail-content tool-live-output"></pre>
+                        </div>
                     </div>`;
-                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) {
+                    if (toolEl.dataset.progressReceived !== 'true') {
+                        toolEl.classList.add('expanded');
+                        toolEl.dataset.progressReceived = 'true';
+                    }
+                    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 += ` <span class="tool-time">${item.execution_time}s</span>`;
                     }
 
                     // Fill output section
-                    const outputSection = currentToolEl.querySelector('.tool-output-section');
+                    const outputLabel = toolEl.querySelector('.tool-output-label');
-                    if (outputSection && item.result) {
+                    const outputEl = toolEl.querySelector('.tool-live-output');
-                        outputSection.innerHTML = `
+                    if (outputLabel) outputLabel.textContent = isError ? 'Error' : 'Output';
-                            <div class="tool-detail-label">${isError ? 'Error' : 'Output'}</div>
+                    if (outputEl) {
-                            <pre class="tool-detail-content ${isError ? 'tool-error-text' : ''}">${escapeHtml(String(item.result))}</pre>`;
+                        outputEl.textContent = item.result ? String(item.result) : '';
+                        outputEl.classList.toggle('tool-error-text', isError);
                     }
 
-                    if (isError) currentToolEl.classList.add('tool-failed');
+                    toolEl.classList.remove('tool-streaming');
-                    currentToolEl = null;
+                    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);
                 }
 
             } else if (item.type === 'image') {
@@ -2639,6 +2692,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
@@ -3274,6 +3334,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) {
@@ -3729,6 +3790,7 @@ function switchSession(newSessionId) {
     // Switching back re-attaches and resumes live streaming.
 
     sessionId = newSessionId;
+    updateEditButtonsState();
     localStorage.setItem(SESSION_ID_KEY, sessionId);
 
     historyPage = 0;

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", ""))[-4 * 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,

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)