From 66931c5caa5e3987f5797bf9524742669bcde10b Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Wed, 6 May 2026 16:54:04 +0800 Subject: [PATCH] feat(runtime): add stale-task recovery and manual stop actions --- modules/background_command_manager.py | 199 ++++++++++++++++- modules/sub_agent_manager.py | 206 ++++++++++++++---- server/chat_flow_task_main.py | 8 + server/conversation.py | 70 +++++- static/src/app/methods/taskPolling.ts | 112 +++++++++- .../overlay/BackgroundCommandDialog.vue | 44 +++- .../overlay/SubAgentActivityDialog.vue | 43 +++- static/src/stores/backgroundCommand.ts | 48 ++++ static/src/stores/subAgent.ts | 57 +++++ .../styles/components/overlays/_overlays.scss | 32 ++- .../styles/components/panels/_left-panel.scss | 10 +- 11 files changed, 778 insertions(+), 51 deletions(-) diff --git a/modules/background_command_manager.py b/modules/background_command_manager.py index 66c2f29..b12548c 100644 --- a/modules/background_command_manager.py +++ b/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 - rec["result"] = result - rec["truncated"] = truncated - rec["updated_at"] = time.time() - rec["finished_at"] = time.time() + existing_status = rec.get("status") + existing_result = rec.get("result") + if existing_status == "cancelled" and isinstance(existing_result, dict): + existing_output = str(existing_result.get("output") or "") + 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: diff --git a/modules/sub_agent_manager.py b/modules/sub_agent_manager.py index 977e248..488d7bb 100644 --- a/modules/sub_agent_manager.py +++ b/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" - task["updated_at"] = time.time() - task["notified"] = True - task["final_result"] = { - "success": False, - "status": "terminated", - "task_id": task_id, - "agent_id": task.get("agent_id"), - "message": "子智能体已被强制关闭。", - } + self.processes.pop(task_id, None) + self._mark_task_terminated( + task, + message="子智能体已被强制关闭。", + system_message=f"🛑 子智能体{task.get('agent_id')} 已被手动关闭。", + notified=True, + ) 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 diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index 5ffeff1..6a743ff 100644 --- a/server/chat_flow_task_main.py +++ b/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 diff --git a/server/conversation.py b/server/conversation.py index ea174f9..9f5b135 100644 --- a/server/conversation.py +++ b/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//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//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//duplicate', methods=['POST']) @api_login_required @with_terminal diff --git a/static/src/app/methods/taskPolling.ts b/static/src/app/methods/taskPolling.ts index 3fa3f45..2856b58 100644 --- a/static/src/app/methods/taskPolling.ts +++ b/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); diff --git a/static/src/components/overlay/BackgroundCommandDialog.vue b/static/src/components/overlay/BackgroundCommandDialog.vue index e7ad277..d6d1605 100644 --- a/static/src/components/overlay/BackgroundCommandDialog.vue +++ b/static/src/components/overlay/BackgroundCommandDialog.vue @@ -20,6 +20,17 @@ {{ activeDetail?.command || activeCommand.command }} +
+ + {{ stopError }} +
{{ detailError }}
@@ -33,17 +44,46 @@