feat(runtime): add stale-task recovery and manual stop actions

modules/background_command_manager.py

@@ -25,6 +25,7 @@ class BackgroundCommandManager:
     def __init__(self, project_path: str):
         self.project_path = Path(project_path).resolve()
         self._records: Dict[str, Dict[str, Any]] = {}
+        self._processes: Dict[str, subprocess.Popen] = {}
         self._lock = threading.RLock()
         self._cv = threading.Condition(self._lock)
 
@@ -257,6 +258,7 @@ class BackgroundCommandManager:
                 if rec is not None:
                     rec["pid"] = process.pid
                     rec["updated_at"] = time.time()
+                    self._processes[command_id] = process
 
             def _reader(stream, collector, rec_key: str):
                 try:
@@ -338,13 +340,193 @@ class BackgroundCommandManager:
             with self._cv:
                 rec = self._records.get(command_id)
                 if rec is not None:
-                    rec["status"] = status
+                    existing_status = rec.get("status")
-                    rec["result"] = result
+                    existing_result = rec.get("result")
-                    rec["truncated"] = truncated
+                    if existing_status == "cancelled" and isinstance(existing_result, dict):
-                    rec["updated_at"] = time.time()
+                        existing_output = str(existing_result.get("output") or "")
-                    rec["finished_at"] = time.time()
+                        if not existing_output and combined_output:
+                            existing_result["output"] = combined_output
+                        rec["result"] = existing_result
+                        rec["updated_at"] = time.time()
+                        rec["finished_at"] = rec.get("finished_at") or time.time()
+                    else:
+                        rec["status"] = status
+                        rec["result"] = result
+                        rec["truncated"] = truncated
+                        rec["updated_at"] = time.time()
+                        rec["finished_at"] = time.time()
+                self._processes.pop(command_id, None)
                 self._cv.notify_all()
 
+    @staticmethod
+    def _coerce_pid(value: Any) -> Optional[int]:
+        try:
+            pid = int(value)
+            return pid if pid > 0 else None
+        except (TypeError, ValueError):
+            return None
+
+    def _is_pid_alive(self, pid: Any) -> bool:
+        normalized = self._coerce_pid(pid)
+        if not normalized:
+            return False
+        try:
+            os.kill(normalized, 0)
+        except ProcessLookupError:
+            return False
+        except PermissionError:
+            return True
+        except Exception:
+            return False
+        return True
+
+    def _terminate_pid(self, pid: Any) -> bool:
+        normalized = self._coerce_pid(pid)
+        if not normalized:
+            return False
+        try:
+            os.killpg(normalized, signal.SIGINT)
+        except Exception:
+            try:
+                os.kill(normalized, signal.SIGINT)
+            except Exception:
+                return False
+        deadline = time.time() + 2.0
+        while time.time() < deadline:
+            if not self._is_pid_alive(normalized):
+                return True
+            time.sleep(0.05)
+        try:
+            os.killpg(normalized, signal.SIGKILL)
+        except Exception:
+            try:
+                os.kill(normalized, signal.SIGKILL)
+            except Exception:
+                return not self._is_pid_alive(normalized)
+        return not self._is_pid_alive(normalized)
+
+    @staticmethod
+    def _is_record_timeout_stale(rec: Dict[str, Any]) -> bool:
+        try:
+            created_at = float(rec.get("created_at") or 0)
+            timeout = float(rec.get("timeout") or 0)
+        except (TypeError, ValueError):
+            return False
+        if created_at <= 0 or timeout <= 0:
+            return False
+        return (time.time() - created_at) > (timeout + 120)
+
+    def reconcile_stale_records(self, conversation_id: Optional[str] = None) -> int:
+        """兜底修正后台命令卡死的 running 状态。"""
+        changed = 0
+        with self._lock:
+            for rec in self._records.values():
+                if not isinstance(rec, dict):
+                    continue
+                if conversation_id and rec.get("conversation_id") != conversation_id:
+                    continue
+                if rec.get("status") != "running":
+                    continue
+                command_id = rec.get("command_id")
+                process = self._processes.get(command_id) if command_id else None
+                pid = rec.get("pid")
+                stale_timeout = self._is_record_timeout_stale(rec)
+                if process and process.poll() is None and not stale_timeout:
+                    continue
+                if (not process) and self._is_pid_alive(pid) and not stale_timeout:
+                    continue
+                if stale_timeout and self._is_pid_alive(pid):
+                    self._terminate_pid(pid)
+                output = self._build_current_output(rec)
+                message = (
+                    "后台指令运行超时，已自动清理运行状态。"
+                    if stale_timeout
+                    else "检测到后台指令进程已退出，已自动清理运行状态。"
+                )
+                rec["status"] = "failed"
+                rec["result"] = {
+                    "success": False,
+                    "status": "failed",
+                    "command": rec.get("command"),
+                    "output": output,
+                    "return_code": rec.get("result", {}).get("return_code")
+                    if isinstance(rec.get("result"), dict)
+                    else -1,
+                    "truncated": bool(rec.get("truncated")),
+                    "timeout": rec.get("timeout"),
+                    "elapsed_ms": int(max(0.0, (time.time() - float(rec.get("created_at") or time.time())) * 1000)),
+                    "command_id": command_id,
+                    "run_in_background": True,
+                    "message": message,
+                }
+                rec["updated_at"] = time.time()
+                rec["finished_at"] = rec.get("finished_at") or time.time()
+                changed += 1
+                if command_id:
+                    self._processes.pop(command_id, None)
+            if changed:
+                self._cv.notify_all()
+        return changed
+
+    def cancel_command(self, command_id: str) -> Dict[str, Any]:
+        if not command_id:
+            return {"success": False, "status": "error", "error": "command_id 不能为空"}
+
+        with self._lock:
+            rec = self._records.get(command_id)
+            if not rec:
+                return {"success": False, "status": "error", "error": f"未找到后台命令: {command_id}"}
+            status = str(rec.get("status") or "")
+            if status in TERMINAL_STATUSES:
+                payload = dict(rec.get("result") or {})
+                if payload:
+                    return payload
+                return {
+                    "success": status == "completed",
+                    "status": status,
+                    "command_id": command_id,
+                    "message": "后台命令已结束",
+                }
+            process = self._processes.get(command_id)
+            pid = rec.get("pid")
+
+        stopped = False
+        if process and process.poll() is None:
+            stopped = self._terminate_pid(process.pid)
+        elif self._is_pid_alive(pid):
+            stopped = self._terminate_pid(pid)
+        else:
+            stopped = True
+
+        with self._cv:
+            rec = self._records.get(command_id)
+            if not rec:
+                return {"success": False, "status": "error", "error": f"未找到后台命令: {command_id}"}
+
+            output = self._build_current_output(rec)
+            now = time.time()
+            rec["status"] = "cancelled"
+            rec["updated_at"] = now
+            rec["finished_at"] = now
+            rec["notified"] = True
+            result = {
+                "success": False,
+                "status": "cancelled",
+                "command": rec.get("command"),
+                "output": output,
+                "return_code": None,
+                "truncated": bool(rec.get("truncated")),
+                "timeout": rec.get("timeout"),
+                "elapsed_ms": int(max(0.0, (now - float(rec.get("created_at") or now)) * 1000)),
+                "command_id": command_id,
+                "run_in_background": True,
+                "message": "后台命令已手动停止" if stopped else "后台命令停止请求已发送",
+            }
+            rec["result"] = result
+            self._processes.pop(command_id, None)
+            self._cv.notify_all()
+            return dict(result)
+
     def _build_current_output(self, rec: Dict[str, Any]) -> str:
         output = "".join((rec.get("stdout_chunks") or []) + (rec.get("stderr_chunks") or []))
         if MAX_RUN_COMMAND_CHARS and len(output) > MAX_RUN_COMMAND_CHARS:
@@ -353,6 +535,7 @@ class BackgroundCommandManager:
 
     def wait_for_completion(self, command_id: str, timeout_seconds: Optional[float] = None, claim: bool = False) -> Dict[str, Any]:
         """阻塞等待后台命令完成。"""
+        self.reconcile_stale_records()
         with self._cv:
             rec = self._records.get(command_id)
             if not rec:
@@ -400,6 +583,7 @@ class BackgroundCommandManager:
     def poll_updates(self, conversation_id: Optional[str] = None) -> List[Dict[str, Any]]:
         """获取未通知且未被 sleep 领取的已完成任务。"""
         updates: List[Dict[str, Any]] = []
+        self.reconcile_stale_records(conversation_id=conversation_id)
         with self._lock:
             for rec in self._records.values():
                 if conversation_id and rec.get("conversation_id") != conversation_id:
@@ -430,6 +614,7 @@ class BackgroundCommandManager:
                 rec["updated_at"] = time.time()
 
     def get_record(self, command_id: str) -> Optional[Dict[str, Any]]:
+        self.reconcile_stale_records()
         with self._lock:
             rec = self._records.get(command_id)
             if not rec:
@@ -438,6 +623,7 @@ class BackgroundCommandManager:
 
     def get_record_with_output(self, command_id: str) -> Optional[Dict[str, Any]]:
         """获取单条后台命令记录，并附带当前可读输出。"""
+        self.reconcile_stale_records()
         with self._lock:
             rec = self._records.get(command_id)
             if not rec:
@@ -453,6 +639,7 @@ class BackgroundCommandManager:
         limit: int = 200,
     ) -> List[Dict[str, Any]]:
         """列出后台命令记录（按创建时间倒序）。"""
+        self.reconcile_stale_records(conversation_id=conversation_id)
         with self._lock:
             items: List[Dict[str, Any]] = []
             for rec in self._records.values():
@@ -468,6 +655,7 @@ class BackgroundCommandManager:
     def has_pending_for_conversation(self, conversation_id: Optional[str]) -> bool:
         if not conversation_id:
             return False
+        self.reconcile_stale_records(conversation_id=conversation_id)
         with self._lock:
             for rec in self._records.values():
                 if rec.get("conversation_id") != conversation_id:
@@ -482,6 +670,7 @@ class BackgroundCommandManager:
         items: List[Dict[str, Any]] = []
         if not conversation_id:
             return items
+        self.reconcile_stale_records(conversation_id=conversation_id)
         with self._lock:
             for rec in self._records.values():
                 if rec.get("conversation_id") != conversation_id:

modules/sub_agent_manager.py

@@ -6,6 +6,7 @@ import time
 import uuid
 import os
 import shutil
+import signal
 from pathlib import Path, PurePosixPath
 from typing import Dict, List, Optional, Any, Tuple, TYPE_CHECKING
 
@@ -62,6 +63,10 @@ class SubAgentManager:
         self.conversation_agents: Dict[str, List[int]] = {}
         self.processes: Dict[str, subprocess.Popen] = {}  # task_id -> Popen对象
         self._load_state()
+        try:
+            self.reconcile_task_states()
+        except Exception:
+            pass
 
     # ------------------------------------------------------------------
     # 公共方法
@@ -267,6 +272,7 @@ class SubAgentManager:
 
         task_id = task["task_id"]
         process = self.processes.get(task_id)
+        pid = task.get("pid")
 
         if process and process.poll() is None:
             # 进程还在运行，终止它
@@ -279,17 +285,17 @@ class SubAgentManager:
                     process.wait()
             except Exception as exc:
                 return {"success": False, "error": f"终止进程失败: {exc}"}
+        elif self._is_pid_alive(pid):
+            if not self._terminate_pid(pid):
+                return {"success": False, "error": f"终止进程失败: PID {pid} 无法停止"}
 
-        task["status"] = "terminated"
+        self.processes.pop(task_id, None)
-        task["updated_at"] = time.time()
+        self._mark_task_terminated(
-        task["notified"] = True
+            task,
-        task["final_result"] = {
+            message="子智能体已被强制关闭。",
-            "success": False,
+            system_message=f"🛑 子智能体{task.get('agent_id')} 已被手动关闭。",
-            "status": "terminated",
+            notified=True,
-            "task_id": task_id,
+        )
-            "agent_id": task.get("agent_id"),
-            "message": "子智能体已被强制关闭。",
-        }
         self._save_state()
 
         return {
@@ -662,6 +668,155 @@ class SubAgentManager:
             if migrated:
                 self._save_state()
 
+    @staticmethod
+    def _coerce_pid(value: Any) -> Optional[int]:
+        try:
+            pid = int(value)
+            return pid if pid > 0 else None
+        except (TypeError, ValueError):
+            return None
+
+    def _is_pid_alive(self, pid: Any) -> bool:
+        normalized = self._coerce_pid(pid)
+        if not normalized:
+            return False
+        try:
+            os.kill(normalized, 0)
+        except ProcessLookupError:
+            return False
+        except PermissionError:
+            return True
+        except Exception:
+            return False
+        return True
+
+    def _terminate_pid(self, pid: Any) -> bool:
+        normalized = self._coerce_pid(pid)
+        if not normalized:
+            return False
+        try:
+            os.killpg(normalized, signal.SIGTERM)
+        except Exception:
+            try:
+                os.kill(normalized, signal.SIGTERM)
+            except Exception:
+                return False
+        deadline = time.time() + 5
+        while time.time() < deadline:
+            if not self._is_pid_alive(normalized):
+                return True
+            time.sleep(0.1)
+        try:
+            os.killpg(normalized, signal.SIGKILL)
+        except Exception:
+            try:
+                os.kill(normalized, signal.SIGKILL)
+            except Exception:
+                return not self._is_pid_alive(normalized)
+        return not self._is_pid_alive(normalized)
+
+    def _mark_task_terminated(
+        self,
+        task: Dict[str, Any],
+        *,
+        message: str,
+        system_message: Optional[str] = None,
+        notified: bool = False,
+    ) -> Dict[str, Any]:
+        task["status"] = "terminated"
+        task["updated_at"] = time.time()
+        if notified:
+            task["notified"] = True
+        result = {
+            "success": False,
+            "status": "terminated",
+            "task_id": task.get("task_id"),
+            "agent_id": task.get("agent_id"),
+            "message": message,
+            "system_message": system_message or message,
+        }
+        task["final_result"] = result
+        return result
+
+    def _should_force_cleanup_stale_task(self, task: Dict[str, Any]) -> bool:
+        try:
+            created_at = float(task.get("created_at") or 0)
+        except (TypeError, ValueError):
+            created_at = 0
+        if created_at <= 0:
+            return False
+        timeout_seconds = int(task.get("timeout_seconds") or SUB_AGENT_DEFAULT_TIMEOUT or 0)
+        timeout_seconds = max(timeout_seconds, 1)
+        grace_seconds = 120
+        elapsed = time.time() - created_at
+        if elapsed <= (timeout_seconds + grace_seconds):
+            return False
+        output_file = Path(task.get("output_file", ""))
+        if output_file.exists():
+            return False
+        progress_file = Path(task.get("progress_file", ""))
+        if progress_file.exists():
+            try:
+                stale_span = time.time() - progress_file.stat().st_mtime
+                if stale_span <= grace_seconds:
+                    return False
+            except Exception:
+                pass
+        return True
+
+    def _refresh_task_runtime_state(self, task: Dict[str, Any]) -> Dict[str, Any]:
+        """刷新单个任务运行态（用于进程句柄丢失/重启后的兜底修正）。"""
+        status = task.get("status")
+        if status in TERMINAL_STATUSES.union({"terminated"}):
+            return {"status": status, "task_id": task.get("task_id")}
+
+        task_id = task.get("task_id")
+        process = self.processes.get(task_id) if task_id else None
+        if process:
+            poll_result = process.poll()
+            if poll_result is None:
+                return {"status": "running", "task_id": task_id}
+            return self._check_task_status(task)
+
+        output_file = Path(task.get("output_file", ""))
+        if output_file.exists():
+            return self._check_task_status(task)
+
+        pid = task.get("pid")
+        if self._is_pid_alive(pid):
+            if self._should_force_cleanup_stale_task(task):
+                return self._mark_task_terminated(
+                    task,
+                    message="子智能体疑似僵尸任务，已超时自动清理运行状态。",
+                    system_message="⚠️ 子智能体长时间未结束，系统已自动清理运行状态。",
+                    notified=True,
+                )
+            return {"status": "running", "task_id": task_id}
+
+        return self._mark_task_terminated(
+            task,
+            message="检测到子智能体进程已退出，已自动清理运行状态。",
+            system_message="⚠️ 子智能体进程异常退出，系统已自动清理运行状态。",
+            notified=True,
+        )
+
+    def reconcile_task_states(self, conversation_id: Optional[str] = None) -> int:
+        """修正运行态任务状态，返回修正条目数。"""
+        changed = 0
+        for task in self.tasks.values():
+            if not isinstance(task, dict):
+                continue
+            if conversation_id and task.get("conversation_id") != conversation_id:
+                continue
+            before_status = task.get("status")
+            before_notified = task.get("notified")
+            self._refresh_task_runtime_state(task)
+            if task.get("status") != before_status or task.get("notified") != before_notified:
+                changed += 1
+        if changed:
+            self._save_state()
+        return changed
+
     def _save_state(self):
         payload = {
             "tasks": self.tasks,
@@ -674,6 +829,7 @@ class SubAgentManager:
         return f"sub_{agent_id}_{int(time.time())}_{suffix}"
 
     def _active_task_count(self, conversation_id: Optional[str] = None) -> int:
+        self.reconcile_task_states(conversation_id=conversation_id)
         active = [
             t for t in self.tasks.values()
             if t.get("status") in {"pending", "running"}
@@ -729,6 +885,7 @@ class SubAgentManager:
         return candidate
 
     def _select_task(self, task_id: Optional[str], agent_id: Optional[int]) -> Optional[Dict]:
+        self.reconcile_task_states()
         if task_id:
             return self.tasks.get(task_id)
 
@@ -761,6 +918,7 @@ class SubAgentManager:
     def poll_updates(self) -> List[Dict]:
         """检查运行中的子智能体任务，返回新完成的结果。"""
         updates: List[Dict] = []
+        self.reconcile_task_states()
         pending_tasks = [
             task for task in self.tasks.values()
             if task.get("status") not in TERMINAL_STATUSES.union({"terminated"})
@@ -1082,8 +1240,8 @@ class SubAgentManager:
 
     def get_overview(self, conversation_id: Optional[str] = None) -> List[Dict[str, Any]]:
         """返回子智能体任务概览，用于前端展示。"""
+        self.reconcile_task_states(conversation_id=conversation_id)
         overview: List[Dict[str, Any]] = []
-        state_changed = False
         for task_id, task in self.tasks.items():
             if conversation_id and task.get("conversation_id") != conversation_id:
                 continue
@@ -1103,30 +1261,6 @@ class SubAgentManager:
                 "sub_conversation_id": task.get("sub_conversation_id"),
             }
 
-            # 运行中的任务检查进程状态
-            if snapshot["status"] not in TERMINAL_STATUSES and snapshot["status"] != "terminated":
-                # 检查进程是否还在运行
-                process = self.processes.get(task_id)
-                if process:
-                    poll_result = process.poll()
-                    if poll_result is not None:
-                        # 进程已结束，检查输出
-                        status_result = self._check_task_status(task)
-                        snapshot["status"] = status_result.get("status", "failed")
-                        if status_result.get("status") in TERMINAL_STATUSES:
-                            task["status"] = status_result["status"]
-                            task["final_result"] = status_result
-                            state_changed = True
-                else:
-                    # 进程句柄丢失（重启后常见），尝试直接检查输出文件
-                    logger.debug("[SubAgentManager] 进程句柄缺失，尝试读取输出文件: %s", task_id)
-                    status_result = self._check_task_status(task)
-                    snapshot["status"] = status_result.get("status", snapshot["status"])
-                    if status_result.get("status") in TERMINAL_STATUSES:
-                        task["status"] = status_result["status"]
-                        task["final_result"] = status_result
-                        state_changed = True
-
             if snapshot["status"] in TERMINAL_STATUSES or snapshot["status"] == "terminated":
                 # 已结束的任务带上最终结果/系统消息，方便前端展示
                 final_result = task.get("final_result") or {}
@@ -1136,6 +1270,4 @@ class SubAgentManager:
             overview.append(snapshot)
 
         overview.sort(key=lambda item: item.get("created_at") or 0, reverse=True)
-        if state_changed:
-            self._save_state()
         return overview

server/chat_flow_task_main.py

@@ -1350,6 +1350,10 @@ async def handle_task_with_sender(
     bg_manager = getattr(web_terminal, "background_command_manager", None)
     has_running_background_commands = False
     if manager:
+        try:
+            manager.reconcile_task_states(conversation_id=conversation_id)
+        except Exception as exc:
+            debug_log(f"[SubAgent] reconcile_task_states failed: {exc}")
         if not hasattr(web_terminal, "_announced_sub_agent_tasks"):
             web_terminal._announced_sub_agent_tasks = set()
         running_tasks = [
@@ -1396,6 +1400,10 @@ async def handle_task_with_sender(
 
     # 检查是否有后台 run_command 或待通知任务
     if bg_manager and conversation_id:
+        try:
+            bg_manager.reconcile_stale_records(conversation_id=conversation_id)
+        except Exception as exc:
+            debug_log(f"[BgCmdDebug] reconcile_stale_records failed: {exc}")
         waiting_items = bg_manager.list_waiting_items(conversation_id)
         if waiting_items:
             has_running_background_commands = True

server/conversation.py

@@ -97,6 +97,10 @@ def _terminate_running_sub_agents(terminal: WebTerminal, reason: str = "") -> in
     current_conv_id = getattr(getattr(terminal, "context_manager", None), "current_conversation_id", None)
     if not current_conv_id:
         return 0
+    try:
+        manager.reconcile_task_states(conversation_id=current_conv_id)
+    except Exception:
+        pass
     running_tasks = [
         task for task in manager.tasks.values()
         if task.get("status") not in TERMINAL_STATUSES.union({"terminated"})
@@ -1166,6 +1170,37 @@ def get_sub_agent_activity(task_id: str, terminal: WebTerminal, workspace: UserW
         return jsonify({"success": False, "error": str(exc)}), 500
 
 
+@conversation_bp.route('/api/sub_agents/<task_id>/terminate', methods=['POST'])
+@api_login_required
+@with_terminal
+def terminate_sub_agent(task_id: str, terminal: WebTerminal, workspace: UserWorkspace, username: str):
+    """手动停止指定子智能体。"""
+    manager = getattr(terminal, "sub_agent_manager", None)
+    if not manager:
+        return jsonify({"success": False, "error": "子智能体管理器不可用"}), 404
+    try:
+        try:
+            manager._load_state()
+        except Exception:
+            pass
+
+        task = manager.tasks.get(task_id)
+        if not task:
+            return jsonify({"success": False, "error": "未找到对应子智能体任务"}), 404
+
+        current_conv_id = terminal.context_manager.current_conversation_id
+        task_conv_id = task.get("conversation_id")
+        if current_conv_id and task_conv_id and task_conv_id != current_conv_id:
+            return jsonify({"success": False, "error": "无权限停止该子智能体任务"}), 403
+
+        result = manager.terminate_sub_agent(task_id=task_id)
+        if not result.get("success"):
+            return jsonify({"success": False, "error": result.get("error") or "停止子智能体失败"}), 400
+        return jsonify({"success": True, "data": result})
+    except Exception as exc:
+        return jsonify({"success": False, "error": str(exc)}), 500
+
+
 @conversation_bp.route('/api/background_commands', methods=['GET'])
 @api_login_required
 @with_terminal
@@ -1184,11 +1219,18 @@ def list_background_commands(terminal: WebTerminal, workspace: UserWorkspace, us
 
         records = manager.list_records(conversation_id=conversation_id, limit=limit_num)
         data = []
+        terminal_statuses = {"completed", "failed", "timeout", "cancelled"}
         for rec in records:
             result = rec.get("result") if isinstance(rec.get("result"), dict) else {}
+            status = rec.get("status")
+            notice_pending = bool(
+                status in terminal_statuses
+                and not rec.get("notified")
+                and not rec.get("claimed_by_sleep")
+            )
             data.append({
                 "command_id": rec.get("command_id"),
-                "status": rec.get("status"),
+                "status": status,
                 "command": rec.get("command"),
                 "conversation_id": rec.get("conversation_id"),
                 "created_at": rec.get("created_at"),
@@ -1197,6 +1239,7 @@ def list_background_commands(terminal: WebTerminal, workspace: UserWorkspace, us
                 "timeout": rec.get("timeout"),
                 "return_code": result.get("return_code"),
                 "run_in_background": True,
+                "notice_pending": notice_pending,
             })
         return jsonify({"success": True, "data": data})
     except Exception as exc:
@@ -1241,6 +1284,31 @@ def get_background_command_detail(command_id: str, terminal: WebTerminal, worksp
         return jsonify({"success": False, "error": str(exc)}), 500
 
 
+@conversation_bp.route('/api/background_commands/<command_id>/cancel', methods=['POST'])
+@api_login_required
+@with_terminal
+def cancel_background_command(command_id: str, terminal: WebTerminal, workspace: UserWorkspace, username: str):
+    """手动停止指定后台 run_command。"""
+    manager = getattr(terminal, "background_command_manager", None)
+    if not manager:
+        return jsonify({"success": False, "error": "后台命令管理器不可用"}), 404
+    try:
+        rec = manager.get_record(command_id)
+        if not rec:
+            return jsonify({"success": False, "error": "未找到对应后台命令"}), 404
+        current_conv_id = terminal.context_manager.current_conversation_id
+        rec_conv_id = rec.get("conversation_id")
+        if current_conv_id and rec_conv_id and rec_conv_id != current_conv_id:
+            return jsonify({"success": False, "error": "无权限停止该后台命令"}), 403
+
+        result = manager.cancel_command(command_id)
+        if not result.get("status"):
+            return jsonify({"success": False, "error": result.get("error") or "停止后台命令失败"}), 400
+        return jsonify({"success": True, "data": result})
+    except Exception as exc:
+        return jsonify({"success": False, "error": str(exc)}), 500
+
+
 @conversation_bp.route('/api/conversations/<conversation_id>/duplicate', methods=['POST'])
 @api_login_required
 @with_terminal

static/src/app/methods/taskPolling.ts

@@ -146,6 +146,66 @@ export const taskPollingMethods = {
       clearInterval(this.waitingTaskProbeTimer);
       this.waitingTaskProbeTimer = null;
     }
+    this._waitingProbeStableEmptyCount = 0;
+  },
+
+  async inspectWaitingCompletionState() {
+    const terminalStatuses = new Set(['completed', 'failed', 'timeout', 'terminated', 'cancelled']);
+    let hasSubAgentPending = false;
+    let hasBackgroundPending = false;
+    let resolved = true;
+
+    try {
+      const [subResp, bgResp] = await Promise.all([
+        fetch('/api/sub_agents'),
+        fetch('/api/background_commands?limit=200')
+      ]);
+
+      if (subResp.ok) {
+        const subResult = await subResp.json().catch(() => ({}));
+        const tasks = Array.isArray(subResult?.data) ? subResult.data : [];
+        const relevantTasks = tasks.filter((task: any) => {
+          if (!task) return false;
+          if (task.conversation_id && task.conversation_id !== this.currentConversationId) return false;
+          return true;
+        });
+        hasSubAgentPending = relevantTasks.some((task: any) => {
+          if (!task?.run_in_background) return false;
+          const status = String(task?.status || '').toLowerCase();
+          return !terminalStatuses.has(status) || !!task?.notice_pending;
+        });
+      } else {
+        resolved = false;
+      }
+
+      if (bgResp.ok) {
+        const bgResult = await bgResp.json().catch(() => ({}));
+        const commands = Array.isArray(bgResult?.data) ? bgResult.data : [];
+        hasBackgroundPending = commands.some((command: any) => {
+          if (!command) return false;
+          const status = String(command?.status || '').toLowerCase();
+          return !terminalStatuses.has(status) || !!command?.notice_pending;
+        });
+      } else {
+        resolved = false;
+      }
+    } catch (error) {
+      debugNotifyLog('[DEBUG_NOTIFY][ui] inspectWaitingCompletionState:error', {
+        error: error?.message || String(error)
+      });
+      return {
+        resolved: false,
+        hasPending: true,
+        hasBackgroundPending: this.waitingForBackgroundCommand
+      };
+    }
+
+    return {
+      resolved,
+      hasPending:
+        !resolved || hasSubAgentPending || hasBackgroundPending,
+      hasBackgroundPending
+    };
   },
 
   async tryResumeWaitingNoticeTask() {
@@ -239,6 +299,30 @@ export const taskPollingMethods = {
         debugNotifyLog('[DEBUG_NOTIFY][ui] waitingTaskProbe:resumed-and-stop');
         keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] waitingTaskProbe:resumed-and-stop');
         this.stopWaitingTaskProbe();
+        return;
+      }
+
+      const state = await this.inspectWaitingCompletionState();
+      if (state.hasPending) {
+        this._waitingProbeStableEmptyCount = 0;
+        this.waitingForBackgroundCommand = !!state.hasBackgroundPending;
+        return;
+      }
+
+      this._waitingProbeStableEmptyCount = (this._waitingProbeStableEmptyCount || 0) + 1;
+      if (this._waitingProbeStableEmptyCount >= 2) {
+        debugLog('[TaskPolling] 等待态兜底清理：未发现运行中的后台子智能体/后台指令');
+        this.streamingMessage = false;
+        this.taskInProgress = false;
+        this.stopRequested = false;
+        this.waitingForSubAgent = false;
+        this.waitingForBackgroundCommand = false;
+        if (typeof this.clearPendingTools === 'function') {
+          this.clearPendingTools('waiting_probe_auto_clear');
+        }
+        this.clearTaskState();
+        this.stopWaitingTaskProbe();
+        this.$forceUpdate();
       }
     }, 1000);
   },
@@ -2052,11 +2136,27 @@ export const taskPollingMethods = {
         (task: any) => task?.run_in_background && !terminalStatuses.has(task?.status)
       );
       const pendingNotice = relevant.filter((task: any) => task?.notice_pending);
+      let backgroundPending = false;
+      try {
+        const bgResp = await fetch('/api/background_commands?limit=200');
+        if (bgResp.ok) {
+          const bgResult = await bgResp.json().catch(() => ({}));
+          const bgCommands = Array.isArray(bgResult?.data) ? bgResult.data : [];
+          backgroundPending = bgCommands.some((command: any) => {
+            if (!command) return false;
+            const status = String(command?.status || '').toLowerCase();
+            return !terminalStatuses.has(status) || !!command?.notice_pending;
+          });
+        }
+      } catch (bgErr) {
+        console.warn('[TaskPolling] 获取后台指令状态失败:', bgErr);
+      }
 
-      if (running.length || pendingNotice.length) {
+      if (running.length || pendingNotice.length || backgroundPending) {
         debugLog('[TaskPolling] 恢复子智能体等待状态', {
           running: running.length,
           pendingNotice: pendingNotice.length,
+          backgroundPending,
           runningTasks: running.map((task: any) => ({
             task_id: task?.task_id,
             status: task?.status
@@ -2067,12 +2167,20 @@ export const taskPollingMethods = {
           }))
         });
         this.waitingForSubAgent = true;
-        this.waitingForBackgroundCommand = false;
+        this.waitingForBackgroundCommand = backgroundPending;
         this.taskInProgress = true;
         this.streamingMessage = false;
         this.stopRequested = false;
         this.startWaitingTaskProbe();
         this.$forceUpdate();
+      } else {
+        this.waitingForSubAgent = false;
+        this.waitingForBackgroundCommand = false;
+        this.stopWaitingTaskProbe();
+        if (!this.streamingMessage) {
+          this.taskInProgress = false;
+        }
+        this.$forceUpdate();
       }
     } catch (error) {
       console.error('[TaskPolling] 恢复子智能体等待状态失败:', error);

static/src/components/overlay/BackgroundCommandDialog.vue

@@ -20,6 +20,17 @@
             {{ activeDetail?.command || activeCommand.command }}
           </span>
         </div>
+        <div class="subagent-activity-actions" v-if="canStop">
+          <button
+            type="button"
+            class="subagent-stop-btn"
+            :disabled="stopLoading"
+            @click="handleStop"
+          >
+            {{ stopLoading ? '停止中...' : '手动停止' }}
+          </button>
+          <span v-if="stopError" class="subagent-activity-error">{{ stopError }}</span>
+        </div>
         <div class="subagent-activity-body">
           <div v-if="detailError" class="subagent-activity-error">{{ detailError }}</div>
           <div v-else-if="detailLoading && !displayOutput" class="subagent-activity-empty">
@@ -33,17 +44,46 @@
 </template>
 
 <script setup lang="ts">
-import { computed } from 'vue';
+import { computed, ref } from 'vue';
 import { storeToRefs } from 'pinia';
 import { useBackgroundCommandStore } from '@/stores/backgroundCommand';
 
 const commandStore = useBackgroundCommandStore();
-const { activeCommand, activeDetail, detailLoading, detailError } = storeToRefs(commandStore);
+const { activeCommand, activeDetail, detailLoading, detailError, stoppingCommandIds } =
+  storeToRefs(commandStore);
+const stopError = ref('');
 
 const close = () => {
   commandStore.closeCommand();
 };
 
+const isTerminalStatus = (status?: string) => {
+  const normalized = (status || '').toString().toLowerCase();
+  return ['completed', 'failed', 'timeout', 'cancelled'].includes(normalized);
+};
+
+const canStop = computed(() => {
+  const status = activeDetail.value?.status || activeCommand.value?.status;
+  if (!activeCommand.value?.command_id) return false;
+  return !isTerminalStatus(status);
+});
+
+const stopLoading = computed(() => {
+  const commandId = activeCommand.value?.command_id;
+  if (!commandId) return false;
+  return !!stoppingCommandIds.value?.[commandId];
+});
+
+const handleStop = async () => {
+  const commandId = activeCommand.value?.command_id;
+  if (!commandId || stopLoading.value) return;
+  stopError.value = '';
+  const result = await commandStore.cancelCommand(commandId);
+  if (!result?.success) {
+    stopError.value = result?.error || '停止失败';
+  }
+};
+
 const displayOutput = computed(() => {
   const text = (activeDetail.value?.output || '').toString();
   return text;

static/src/components/overlay/SubAgentActivityDialog.vue

@@ -16,6 +16,17 @@
             activeAgent.summary
           }}</span>
         </div>
+        <div class="subagent-activity-actions" v-if="canStop">
+          <button
+            type="button"
+            class="subagent-stop-btn"
+            :disabled="stopLoading"
+            @click="handleStop"
+          >
+            {{ stopLoading ? '停止中...' : '手动停止' }}
+          </button>
+          <span v-if="stopError" class="subagent-activity-error">{{ stopError }}</span>
+        </div>
         <div class="subagent-activity-body">
           <div v-if="activityError" class="subagent-activity-error">{{ activityError }}</div>
           <div v-else-if="!displayItems.length" class="subagent-activity-empty">
@@ -34,7 +45,7 @@
 </template>
 
 <script setup lang="ts">
-import { computed } from 'vue';
+import { computed, ref } from 'vue';
 import { storeToRefs } from 'pinia';
 import { useSubAgentStore } from '@/stores/subAgent';
 
@@ -47,7 +58,9 @@ type ActivityEntry = {
 };
 
 const subAgentStore = useSubAgentStore();
-const { activeAgent, activityEntries, activityLoading, activityError } = storeToRefs(subAgentStore);
+const { activeAgent, activityEntries, activityLoading, activityError, stoppingTaskIds } =
+  storeToRefs(subAgentStore);
+const stopError = ref('');
 
 const close = () => {
   subAgentStore.closeSubAgent();
@@ -98,6 +111,32 @@ const buildText = (entry: ActivityEntry) => {
   return `${tool || '工具'}`;
 };
 
+const isTerminalStatus = (status?: string) => {
+  const normalized = (status || '').toString().toLowerCase();
+  return ['completed', 'failed', 'timeout', 'terminated', 'cancelled'].includes(normalized);
+};
+
+const canStop = computed(() => {
+  if (!activeAgent.value?.task_id) return false;
+  return !isTerminalStatus(activeAgent.value.status);
+});
+
+const stopLoading = computed(() => {
+  const taskId = activeAgent.value?.task_id;
+  if (!taskId) return false;
+  return !!stoppingTaskIds.value?.[taskId];
+});
+
+const handleStop = async () => {
+  const taskId = activeAgent.value?.task_id;
+  if (!taskId || stopLoading.value) return;
+  stopError.value = '';
+  const result = await subAgentStore.terminateSubAgent(taskId);
+  if (!result?.success) {
+    stopError.value = result?.error || '停止失败';
+  }
+};
+
 const displayItems = computed(() => {
   const order: string[] = [];
   const map = new Map<string, ActivityEntry>();

static/src/stores/backgroundCommand.ts

@@ -5,6 +5,7 @@ interface BackgroundCommand {
   command_id: string;
   status?: string;
   command?: string;
+  notice_pending?: boolean;
   created_at?: number;
   updated_at?: number;
   finished_at?: number | null;
@@ -22,6 +23,7 @@ interface BackgroundCommandState {
   pollTimer: ReturnType<typeof setInterval> | null;
   detailPollTimer: ReturnType<typeof setInterval> | null;
   activeCommand: BackgroundCommand | null;
+  stoppingCommandIds: Record<string, boolean>;
   activeDetail: BackgroundCommandDetail | null;
   detailLoading: boolean;
   detailError: string | null;
@@ -35,6 +37,7 @@ export const useBackgroundCommandStore = defineStore('backgroundCommand', {
     pollTimer: null,
     detailPollTimer: null,
     activeCommand: null,
+    stoppingCommandIds: {},
     activeDetail: null,
     detailLoading: false,
     detailError: null
@@ -49,6 +52,13 @@ export const useBackgroundCommandStore = defineStore('backgroundCommand', {
         const data = await resp.json();
         if (data.success) {
           this.commands = Array.isArray(data.data) ? data.data : [];
+          const activeId = this.activeCommand?.command_id;
+          if (activeId) {
+            const latest = this.commands.find((item) => item.command_id === activeId);
+            if (latest) {
+              this.activeCommand = { ...latest };
+            }
+          }
         }
       } catch (error) {
         console.error('获取后台指令列表失败:', error);
@@ -120,6 +130,12 @@ export const useBackgroundCommandStore = defineStore('backgroundCommand', {
           if (current) {
             Object.assign(current, data.data);
           }
+          if (this.activeCommand && this.activeCommand.command_id === commandId) {
+            this.activeCommand = {
+              ...this.activeCommand,
+              ...data.data
+            };
+          }
           const status = (data.data.status || '').toString();
           if (TERMINAL_STATUSES.has(status)) {
             this.stopDetailPolling();
@@ -131,6 +147,38 @@ export const useBackgroundCommandStore = defineStore('backgroundCommand', {
       } finally {
         this.detailLoading = false;
       }
+    },
+    async cancelCommand(commandId: string) {
+      const normalizedId = (commandId || '').toString().trim();
+      if (!normalizedId) {
+        return { success: false, error: 'command_id 不能为空' };
+      }
+      this.stoppingCommandIds = {
+        ...this.stoppingCommandIds,
+        [normalizedId]: true
+      };
+      try {
+        const resp = await fetch(
+          `/api/background_commands/${encodeURIComponent(normalizedId)}/cancel`,
+          { method: 'POST' }
+        );
+        const data = await resp.json().catch(() => ({}));
+        if (!resp.ok || !data?.success) {
+          throw new Error(data?.error || `HTTP ${resp.status}`);
+        }
+        await this.fetchCommands();
+        if (this.activeCommand?.command_id === normalizedId) {
+          await this.fetchCommandDetail(normalizedId);
+        }
+        return { success: true, data: data?.data || null };
+      } catch (error: any) {
+        const message = error?.message || String(error);
+        return { success: false, error: message };
+      } finally {
+        const next = { ...this.stoppingCommandIds };
+        delete next[normalizedId];
+        this.stoppingCommandIds = next;
+      }
     }
   }
 });

static/src/stores/subAgent.ts

@@ -8,6 +8,7 @@ interface SubAgent {
   summary?: string;
   last_tool?: string;
   conversation_id?: string;
+  notice_pending?: boolean;
 }
 
 interface SubAgentActivityEntry {
@@ -24,17 +25,21 @@ interface SubAgentState {
   pollTimer: ReturnType<typeof setInterval> | null;
   activityTimer: ReturnType<typeof setInterval> | null;
   activeAgent: SubAgent | null;
+  stoppingTaskIds: Record<string, boolean>;
   activityEntries: SubAgentActivityEntry[];
   activityLoading: boolean;
   activityError: string | null;
 }
 
+const TERMINAL_STATUSES = new Set(['completed', 'failed', 'timeout', 'terminated']);
+
 export const useSubAgentStore = defineStore('subAgent', {
   state: (): SubAgentState => ({
     subAgents: [],
     pollTimer: null,
     activityTimer: null,
     activeAgent: null,
+    stoppingTaskIds: {},
     activityEntries: [],
     activityLoading: false,
     activityError: null
@@ -49,6 +54,13 @@ export const useSubAgentStore = defineStore('subAgent', {
         const data = await resp.json();
         if (data.success) {
           this.subAgents = Array.isArray(data.data) ? data.data : [];
+          const activeTaskId = this.activeAgent?.task_id;
+          if (activeTaskId) {
+            const latest = this.subAgents.find((item) => item.task_id === activeTaskId);
+            if (latest) {
+              this.activeAgent = { ...latest };
+            }
+          }
         }
       } catch (error) {
         console.error('获取子智能体列表失败:', error);
@@ -117,6 +129,16 @@ export const useSubAgentStore = defineStore('subAgent', {
         if (data && data.success && data.data) {
           const entries = Array.isArray(data.data.entries) ? data.data.entries : [];
           this.activityEntries = entries;
+          if (this.activeAgent && this.activeAgent.task_id === taskId) {
+            this.activeAgent = {
+              ...this.activeAgent,
+              status: data.data.status || this.activeAgent.status
+            };
+          }
+          const status = (data.data.status || '').toString();
+          if (TERMINAL_STATUSES.has(status)) {
+            this.stopActivityPolling();
+          }
         }
       } catch (error: any) {
         this.activityError = error?.message || String(error);
@@ -125,6 +147,41 @@ export const useSubAgentStore = defineStore('subAgent', {
         this.activityLoading = false;
       }
     },
+    async terminateSubAgent(taskId: string) {
+      const normalizedId = (taskId || '').toString().trim();
+      if (!normalizedId) {
+        return { success: false, error: 'task_id 不能为空' };
+      }
+      this.stoppingTaskIds = {
+        ...this.stoppingTaskIds,
+        [normalizedId]: true
+      };
+      try {
+        const resp = await fetch(`/api/sub_agents/${encodeURIComponent(normalizedId)}/terminate`, {
+          method: 'POST'
+        });
+        const data = await resp.json().catch(() => ({}));
+        if (!resp.ok || !data?.success) {
+          throw new Error(data?.error || `HTTP ${resp.status}`);
+        }
+        await this.fetchSubAgents();
+        if (this.activeAgent?.task_id === normalizedId) {
+          await this.fetchSubAgentActivity(normalizedId);
+          this.activeAgent = {
+            ...this.activeAgent,
+            status: 'terminated'
+          };
+        }
+        return { success: true, data: data?.data || null };
+      } catch (error: any) {
+        const message = error?.message || String(error);
+        return { success: false, error: message };
+      } finally {
+        const next = { ...this.stoppingTaskIds };
+        delete next[normalizedId];
+        this.stoppingTaskIds = next;
+      }
+    },
     stripConversationPrefix(conversationId: string) {
       if (!conversationId) return '';
       return conversationId.startsWith('conv_') ? conversationId.slice(5) : conversationId;

static/src/styles/components/overlays/_overlays.scss

@@ -2005,6 +2005,34 @@
     flex-wrap: wrap;
 }
 
+.subagent-activity-actions {
+    display: flex;
+    align-items: center;
+    gap: 10px;
+    flex-wrap: wrap;
+}
+
+.subagent-stop-btn {
+    border: 1px solid rgba(239, 68, 68, 0.4);
+    background: rgba(239, 68, 68, 0.12);
+    color: #ef4444;
+    border-radius: 10px;
+    font-size: 12px;
+    padding: 4px 10px;
+    cursor: pointer;
+    transition: background 0.2s ease, border-color 0.2s ease;
+}
+
+.subagent-stop-btn:hover:not(:disabled) {
+    background: rgba(239, 68, 68, 0.18);
+    border-color: rgba(239, 68, 68, 0.55);
+}
+
+.subagent-stop-btn:disabled {
+    opacity: 0.55;
+    cursor: not-allowed;
+}
+
 .subagent-activity-status {
     padding: 2px 8px;
     border-radius: 999px;
@@ -2021,7 +2049,9 @@
 }
 
 .subagent-activity-status.failed,
-.subagent-activity-status.timeout {
+.subagent-activity-status.timeout,
+.subagent-activity-status.terminated,
+.subagent-activity-status.cancelled {
     background: rgba(239, 68, 68, 0.15);
     color: #ef4444;
 }

static/src/styles/components/panels/_left-panel.scss

@@ -404,6 +404,12 @@
 
 .sub-agent-panel {
     padding: 16px 20px 24px;
+    display: flex;
+    flex-direction: column;
+    gap: 10px;
+    flex: 1 1 auto;
+    min-height: 0;
+    overflow-y: auto;
 }
 
 .sub-agent-cards {
@@ -458,7 +464,9 @@
 }
 
 .sub-agent-status.failed,
-.sub-agent-status.timeout {
+.sub-agent-status.timeout,
+.sub-agent-status.terminated,
+.sub-agent-status.cancelled {
     color: #d63031;
 }