From a04eca3aab9914ea5ae02cde014166e28e4a574d Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Wed, 8 Apr 2026 16:13:40 +0800 Subject: [PATCH] fix: unify background run_command notifications with sub-agent flow --- agentskills/run-command-guide/SKILL.md | 103 ++++ agentskills/sub-agent-guide/SKILL.md | 5 + core/main_terminal.py | 5 + core/main_terminal_parts/tools_definition.py | 59 ++- core/main_terminal_parts/tools_execution.py | 243 +++++++-- core/main_terminal_parts/tools_policy.py | 3 + modules/background_command_manager.py | 500 ++++++++++++++++++ modules/personalization_manager.py | 8 + server/chat_flow_task_main.py | 281 +++++++--- server/chat_flow_task_support.py | 134 ++++- server/chat_flow_tool_loop.py | 15 +- server/conversation.py | 75 +++ server/tasks.py | 17 + static/src/App.vue | 1 + static/src/app.ts | 6 + static/src/app/components.ts | 2 + static/src/app/lifecycle.ts | 3 + static/src/app/methods/conversation.ts | 8 +- static/src/app/methods/history.ts | 16 +- static/src/app/methods/taskPolling.ts | 158 +++++- static/src/app/methods/ui.ts | 32 +- static/src/app/state.ts | 4 + static/src/components/chat/ChatArea.vue | 3 +- .../overlay/BackgroundCommandDialog.vue | 47 ++ static/src/components/panels/LeftPanel.vue | 57 +- .../personalization/PersonalizationDrawer.vue | 17 + static/src/stores/backgroundCommand.ts | 136 +++++ static/src/stores/personalization.ts | 3 + static/src/stores/task.ts | 48 ++ static/src/stores/tutorial.ts | 4 +- static/src/stores/ui.ts | 2 +- .../styles/components/overlays/_overlays.scss | 20 + utils/tool_result_formatter.py | 45 +- 33 files changed, 1895 insertions(+), 165 deletions(-) create mode 100644 agentskills/run-command-guide/SKILL.md create mode 100644 modules/background_command_manager.py create mode 100644 static/src/components/overlay/BackgroundCommandDialog.vue create mode 100644 static/src/stores/backgroundCommand.ts diff --git a/agentskills/run-command-guide/SKILL.md b/agentskills/run-command-guide/SKILL.md new file mode 100644 index 0000000..ca6116d --- /dev/null +++ b/agentskills/run-command-guide/SKILL.md @@ -0,0 +1,103 @@ +--- +name: run-command-guide +description: run_command 工具使用指南。介绍前台与后台模式、后台模式的等待方式（command_id + sleep(wait_runcommand_id)）、互斥参数约束以及常见坑与示例。 +--- + +# run_command 使用指南 + +## 核心原则 + +- `run_command` 是 **一次性执行** 的终端命令，适合查询文件信息（`ls`、`file`、`grep -n`）、触发 CLI 工具、做简单的数据转换或运行非交互脚本。它**不能**用于启动交互式程序（如 `python` REPL、`vim`、`top`）。 +- 每次调用必须提供 `timeout`（单位秒，必须大于 0），系统会在超时后强制打断命令并返回当前输出。 +- 输出内容有字符数限制（默认 10000 字符），超过时会被截断或遭拒，因此尽量控制命令输出量，必要时拆分多个命令或保存到文件再读取。 +- 如果需要持续交互、长时间运行、反复检查进度，优先使用 `terminal_session`/`terminal_input`；`run_command` 用于“一次性”场景。 + +## 执行模式：前台 vs 后台 + +### 前台模式（默认，`run_in_background=false`） + +- timeout 上限 30 秒。命令在工具完成前会阻塞整个模型线程，返回后即可马上读取 `output`、`return_code` 等字段。 +- 适合“快照式”场景：查看目录、检查某个文件、运行一条短命令并立刻需要结果。 +- 要求命令在 30 秒内完成，否则会中断并返回 `status: "timeout"`。 + +### 后台模式（`run_in_background=true`） + +- timeout 最长 3600 秒。命令在后台继续执行，工具会等约 5 秒收集当前输出并返回，然后 **立即释放控制权**，让你继续处理其他任务。 +- 返回值里会包含 `command_id`，`status` 可能是 `running_background`。你会拿到已经产生的输出片段，但最终结果要靠系统通知或手动等待。 +- 成功或失败完成后，系统会自动发送一条 `system` 消息（内容类似 `后台命令已完成（completed）`，并附带 `command_id`、`command`、`output` 摘要）通知你。不要再依赖 `wait_sub_agent`/手动轮询。 +- 适合长时间任务（大规模日志分析、编译/安装、数据处理），或者你想发起命令后继续进行其他工作。 +- 返回的 `command_id` 需要保留，用于后续通过 `sleep(wait_runcommand_id=...)` 等待结果或根据通知定位输出。 + +## 等待后台命令完成：command_id + sleep(wait_runcommand_id) + +1. 调用后台命令后记录 `command_id`（形如 `cmd_<时间戳>_<8位hash>`）； +2. 当你确实需要知道最终输出/返回码时，调用 `sleep(wait_runcommand_id=command_id)`； +3. `sleep` 工具会阻塞直到命令完成，返回值的 `result` 字段就是后台命令的最终 payload，`success`/`return_code` 等都可以直接使用； +4. `sleep` 默认会验证该 `command_id` 是否属于当前对话；跨会话调用会报错。 + +注意：即使你不调用 `sleep`，后台命令完成后也会有系统通知，只要收到消息就可以去交叉检查输出或 `sleep` 结果。 + +## `sleep` 工具的互斥参数规则 + +- `sleep` 工具支持三种等待方式：`seconds`（纯等待）、`wait_sub_agent_ids`（等待子智能体）、`wait_runcommand_id`（等待后台 `run_command`）。 +- **只能选择其中一个等待参数**，否则会报错 “sleep 的等待参数互斥”； +- 如果你需要短暂停顿而非等待任务完成，传 `seconds` 并可选填 `reason`； +- 需要等待后台命令完成时仅填写 `wait_runcommand_id`，其余两个参数必须为 `null`/不传。 +- `wait_runcommand_id` 返回： + ``` + { + "success": true, + "mode": "wait_runcommand_id", + "command_id": "...", + "result": {...}, # 与 run_command 结果结构一致 + "message": "后台 run_command 等待完成" + } + ``` + +## 常见坑与注意事项 + +1. **忘记设置 `timeout` 或给值 ≤ 0**：前台/后台都会立即报错。后台模式还有限制（>0 且 ≤3600）。 +2. **误以为后台模式会自动输出最终结果**：后台只返回 `running_background` 的片段，必须通过系统通知或 `sleep(wait_runcommand_id=...)` 才能拿到完整状态。 +3. **没有保存 `command_id` 即失去追踪**：`command_id` 是等待、复查、排错的唯一标识符，出错后可直接用它查询后台管理器。 +4. **在 wait/run_command 之间同时传多个 `sleep` 参数**：会被客户端拒绝，报错明确指出互斥约束。 +5. **命令输出过大（超过 10000 字符）**：会被截断并提示降低输出量，可考虑写入文件再用 `read_file` 读取，或分步执行。 +6. **尝试运行交互式程序**：`run_command` 明确禁止，命令会在 `_validate_command` 阶段失败。 +7. **后台命令属于别的对话**：`sleep(wait_runcommand_id=...)` 会报 “该后台命令不属于当前对话”，只能在原 conversation 中等待。 +8. **误以为 `sleep` 会自动拼接 system message**：`sleep` 只是等待，系统通知仍旧按背景轮询插入，你收到 `system` 消息即可确认状态。 + +## 示例 + +### 前台快速查询 + +```python +run_command(command="ls -a docs | head", timeout=5) +# 立即得到输出和 return_code +``` + +### 后台安装并等待 + +```python +# 发起后台安装 +bg = run_command( + command="python -m pip install -r requirements.txt", + timeout=600, + run_in_background=True +) +command_id = bg.get("command_id") + +# 可先去做别的事情，或立即等待最终状态 +sleep(wait_runcommand_id=command_id, reason="等待 pip 安装完成") +# 返回 result["result"]["output"], return_code 等最终内容 +``` + +### 结合系统通知与 sleep + +- 系统会插入类似 `后台命令已完成（completed）` 的 `system` 消息，包含 `command_id`。 +- 如果你优先希望第一时间处理输出，可以在收到消息后再 `sleep(wait_runcommand_id=command_id)` 来确认成功状态（即使已有通知，也可再次 `sleep` 获得 `result` 结构）。 + +## 总结 + +- 前台模式适用于“立刻需要结果”的单步命令，后台模式适合“发起后继续干别的”的长任务。 +- 后台结果需要 `command_id + sleep(wait_runcommand_id=...)` 或系统通知来确认最终状态。 +- `sleep` 的等待参数互斥，务必只传一个。 +- 避免交互式命令、超出字符/超时限制，必要时将输出先写到文件再读。 diff --git a/agentskills/sub-agent-guide/SKILL.md b/agentskills/sub-agent-guide/SKILL.md index 026c6d7..acc97d5 100644 --- a/agentskills/sub-agent-guide/SKILL.md +++ b/agentskills/sub-agent-guide/SKILL.md @@ -121,6 +121,10 @@ create_sub_agent( - 子智能体在后台运行，完成后系统会**自动插入一条 user 消息**通知你 - 你会在对话中看到类似："这是一句系统自动发送的user消息，用于通知你子智能体已经运行完成..." +**完成通知机制说明：** +- 后台子智能体完成的通知是系统自动插入的 user/system 消息，不依赖手动调用 `wait_sub_agent` 或其它轮询接口。 +- 你只需在收到通知后直接查看交付目录即可，系统会把所有必需信息一并带上。 + **何时使用：** - ✅ 耗时任务（预计 > 1 分钟） - ✅ 多个并行任务 @@ -160,6 +164,7 @@ create_sub_agent(agent_id=1, task="大规模日志分析", run_in_background=tru - 系统会自动处理通知，你只需要在收到通知消息后处理结果即可 - 如果你正在执行工具调用，系统会延迟通知，避免打断你的工作流 - 通知类型取决于你是否继续工作：继续工作时收到 system 消息，停止输出时收到 user 消息 +- 后台完成由上述系统通知负责，不要再依赖 `wait_sub_agent` 或类似的手动等待流，系统会在能插入消息时自动通知你。 ### 模式选择决策树 diff --git a/core/main_terminal.py b/core/main_terminal.py index 5bd36df..bb8c89d 100644 --- a/core/main_terminal.py +++ b/core/main_terminal.py @@ -39,6 +39,7 @@ from modules.memory_manager import MemoryManager from modules.terminal_manager import TerminalManager from modules.todo_manager import TodoManager from modules.sub_agent_manager import SubAgentManager +from modules.background_command_manager import BackgroundCommandManager from modules.ocr_client import OCRClient from modules.easter_egg_manager import EasterEggManager from modules.custom_tool_registry import CustomToolRegistry, build_default_tool_category @@ -127,6 +128,9 @@ class MainTerminal(MainTerminalCommandMixin, MainTerminalContextMixin, MainTermi data_dir=str(self.data_dir), container_session=container_session, ) + self.background_command_manager = BackgroundCommandManager( + project_path=self.project_path, + ) self.easter_egg_manager = EasterEggManager() self._announced_sub_agent_tasks = set() self.silent_tool_disable = False # 是否静默工具禁用提示 @@ -168,6 +172,7 @@ class MainTerminal(MainTerminalCommandMixin, MainTerminalContextMixin, MainTermi # Skill 强约束开关（按个人空间配置） self.skill_strict_terminal_enabled: bool = False self.skill_strict_sub_agent_enabled: bool = False + self.skill_strict_run_command_background_enabled: bool = False # 当前生效的启用 skills（用于强约束判定） self.enabled_skill_ids: Set[str] = set() self.apply_personalization_preferences() diff --git a/core/main_terminal_parts/tools_definition.py b/core/main_terminal_parts/tools_definition.py index 4ff0838..e6753e1 100644 --- a/core/main_terminal_parts/tools_definition.py +++ b/core/main_terminal_parts/tools_definition.py @@ -190,22 +190,31 @@ class MainTerminalToolsDefinitionMixin: tools = [ { "type": "function", - "function": { - "name": "sleep", - "description": "等待指定的秒数，用于短暂延迟/节奏控制（例如让终端产生更多输出、或在两次快照之间留出间隔）。命令是否完成必须用 terminal_snapshot 确认；需要强制超时终止请使用 run_command。", - "parameters": { - "type": "object", - "properties": self._inject_intent({ - "seconds": { - "type": "number", - "description": "等待的秒数，可以是小数（如0.2秒）。建议范围：0.1-10秒" - }, - "reason": { - "type": "string", - "description": "等待的原因说明（可选）" - } + "function": { + "name": "sleep", + "description": "等待工具。三种模式三选一：1) seconds：短暂延迟；2) wait_sub_agent_ids：等待指定子智能体全部结束并直接返回结果；3) wait_runcommand_id：等待指定后台 run_command 结束并直接返回结果。若同时提供多个等待参数会报错。", + "parameters": { + "type": "object", + "properties": self._inject_intent({ + "seconds": { + "type": "number", + "description": "等待的秒数，可以是小数（如0.2秒）。建议范围：0.1-10秒" + }, + "wait_sub_agent_ids": { + "type": "array", + "items": {"type": "integer"}, + "description": "等待这些子智能体全部结束后返回（可提供一个或多个编号）。" + }, + "wait_runcommand_id": { + "type": "string", + "description": "等待指定后台 run_command 的 command_id 结束后返回。" + }, + "reason": { + "type": "string", + "description": "等待的原因说明（可选）" + } }), - "required": ["seconds"] + "required": [] } } }, @@ -584,16 +593,20 @@ class MainTerminalToolsDefinitionMixin: }, { "type": "function", - "function": { - "name": "run_command", - "description": "执行一次性终端命令，适合查看文件信息（file/ls/stat/iconv 等）、转换编码或调用 CLI 工具。禁止启动交互式程序；对已聚焦文件仅允许使用 grep -n 等定位命令。必须提供 timeout（最长30秒）；一旦超时，命令**一定会被打断**且无法继续执行（需要重新运行），并返回已捕获输出；输出超过10000字符将被截断或拒绝。", - "parameters": { - "type": "object", - "properties": self._inject_intent({ - "command": {"type": "string", "description": "终端命令"}, + "function": { + "name": "run_command", + "description": "执行一次性终端命令，适合查看文件信息（file/ls/stat/iconv 等）、转换编码或调用 CLI 工具。禁止启动交互式程序；对已聚焦文件仅允许使用 grep -n 等定位命令。必须提供 timeout。前台模式（run_in_background=false，默认）上限30秒，超时会打断；后台模式（run_in_background=true）上限3600秒，会先等待5秒返回已有输出并继续在后台运行，完成后由系统通知。", + "parameters": { + "type": "object", + "properties": self._inject_intent({ + "command": {"type": "string", "description": "终端命令"}, "timeout": { "type": "number", - "description": "超时时长（秒），必填，最大30" + "description": "超时时长（秒），必填。前台最大30；后台最大3600。" + }, + "run_in_background": { + "type": "boolean", + "description": "是否后台运行。true 时先等待5秒返回已有输出，并在后台持续执行直至结束。" } }), "required": ["command", "timeout"] diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index 0fc7718..d0ac5fd 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -160,6 +160,8 @@ class MainTerminalToolsExecutionMixin: self.context_manager._set_meta_flag(self._skill_meta_key("terminal-guide"), True) if normalized.endswith("skills/sub-agent-guide/skill.md"): self.context_manager._set_meta_flag(self._skill_meta_key("sub-agent-guide"), True) + if normalized.endswith("skills/run-command-guide/skill.md"): + self.context_manager._set_meta_flag(self._skill_meta_key("run-command-guide"), True) def _get_visited_read_files(self) -> Set[str]: raw = self.context_manager._get_meta_flag(self._file_read_meta_key(), []) @@ -258,15 +260,21 @@ class MainTerminalToolsExecutionMixin: except Exception: return False - def _required_skill_for_tool(self, tool_name: str) -> Optional[str]: + def _required_skill_for_tool(self, tool_name: str, arguments: Optional[Dict[str, Any]] = None) -> Optional[str]: if tool_name in self._TERMINAL_SERIES_TOOLS and bool(getattr(self, "skill_strict_terminal_enabled", False)): return "terminal-guide" if tool_name in self._SUB_AGENT_SERIES_TOOLS and bool(getattr(self, "skill_strict_sub_agent_enabled", False)): return "sub-agent-guide" + if ( + tool_name == "run_command" + and bool(getattr(self, "skill_strict_run_command_background_enabled", False)) + and bool((arguments or {}).get("run_in_background", False)) + ): + return "run-command-guide" return None - def _check_skill_strict_prerequisite(self, tool_name: str) -> Optional[Dict[str, Any]]: - skill_id = self._required_skill_for_tool(tool_name) + def _check_skill_strict_prerequisite(self, tool_name: str, arguments: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]: + skill_id = self._required_skill_for_tool(tool_name, arguments) if not skill_id: return None # 仅当该 skill 在个人空间中启用时才执行强约束 @@ -343,7 +351,7 @@ class MainTerminalToolsExecutionMixin: custom_tool = self.custom_tool_registry.get_tool(tool_name) # Skill 强约束：未阅读对应 SKILL.md 时，阻止 terminal/sub-agent 系列工具执行 - strict_error = self._check_skill_strict_prerequisite(tool_name) + strict_error = self._check_skill_strict_prerequisite(tool_name, arguments) if strict_error: return json.dumps(strict_error, ensure_ascii=False) @@ -480,39 +488,155 @@ class MainTerminalToolsExecutionMixin: # sleep工具 elif tool_name == "sleep": - seconds = arguments.get("seconds", 1) + seconds = arguments.get("seconds") + wait_sub_agent_ids = arguments.get("wait_sub_agent_ids") + wait_runcommand_id = arguments.get("wait_runcommand_id") reason = arguments.get("reason", "等待操作完成") - # 限制最大等待时间 - max_sleep = 3600 # 最多等待3600秒（1小时） - if seconds > max_sleep: + provided = 0 + if seconds is not None: + provided += 1 + if wait_sub_agent_ids: + provided += 1 + if wait_runcommand_id: + provided += 1 + if provided == 0: result = { "success": False, - "error": f"等待时间过长，最多允许 {max_sleep} 秒", - "suggestion": f"建议分多次等待或减少等待时间" + "error": "sleep 至少需要提供一个参数：seconds / wait_sub_agent_ids / wait_runcommand_id" } + elif provided > 1: + result = { + "success": False, + "error": "sleep 的等待参数互斥：seconds / wait_sub_agent_ids / wait_runcommand_id 只能提供一个" + } + elif wait_sub_agent_ids: + if not isinstance(wait_sub_agent_ids, list) or not wait_sub_agent_ids: + result = {"success": False, "error": "wait_sub_agent_ids 必须是非空数组"} + else: + normalized_ids = [] + invalid = [] + for item in wait_sub_agent_ids: + try: + agent_id = int(item) + if agent_id <= 0: + raise ValueError() + normalized_ids.append(agent_id) + except Exception: + invalid.append(item) + if invalid: + result = {"success": False, "error": f"wait_sub_agent_ids 含非法值: {invalid}"} + else: + manager = getattr(self, "sub_agent_manager", None) + if not manager: + result = {"success": False, "error": "子智能体管理器不可用"} + else: + task_ids = [] + missing = [] + for aid in normalized_ids: + task = manager.lookup_task(agent_id=aid) + if not task: + missing.append(aid) + else: + task_ids.append(task.get("task_id")) + if missing: + result = {"success": False, "error": f"未找到对应子智能体: {missing}"} + else: + wait_results = [] + waited_task_ids = [] + for tid in task_ids: + wait_result = await asyncio.to_thread( + manager.wait_for_completion, + task_id=tid, + timeout_seconds=None, + ) + wait_results.append(wait_result) + waited_task_ids.append(tid) + try: + task = manager.tasks.get(tid) + if isinstance(task, dict): + task["notified"] = True + task["updated_at"] = time.time() + except Exception: + pass + try: + manager._save_state() + except Exception: + pass + result = { + "success": True, + "mode": "wait_sub_agent_ids", + "agent_ids": normalized_ids, + "waited_task_ids": waited_task_ids, + "results": wait_results, + "message": f"已等待 {len(normalized_ids)} 个子智能体结束" + } + try: + if not hasattr(self, "_announced_sub_agent_tasks"): + self._announced_sub_agent_tasks = set() + for tid in waited_task_ids: + if tid: + self._announced_sub_agent_tasks.add(tid) + except Exception: + pass + elif wait_runcommand_id: + bg_manager = getattr(self, "background_command_manager", None) + if not bg_manager: + result = {"success": False, "error": "后台命令管理器不可用"} + else: + rec = bg_manager.get_record(str(wait_runcommand_id)) + current_conv = getattr(self.context_manager, "current_conversation_id", None) + if not rec: + result = {"success": False, "error": f"未找到后台命令: {wait_runcommand_id}"} + elif rec.get("conversation_id") and current_conv and rec.get("conversation_id") != current_conv: + result = {"success": False, "error": "该后台命令不属于当前对话"} + else: + wait_result = await asyncio.to_thread( + bg_manager.wait_for_completion, + str(wait_runcommand_id), + None, + True, + ) + bg_manager.mark_claimed(str(wait_runcommand_id)) + result = { + "success": bool(wait_result.get("success", False)), + "mode": "wait_runcommand_id", + "command_id": str(wait_runcommand_id), + "result": wait_result, + "message": "后台 run_command 等待完成" + } else: - # 确保秒数为正数 - if seconds <= 0: + max_sleep = 3600 # 最多等待3600秒（1小时） + seconds_parse_ok = True + try: + seconds = float(seconds) + except Exception: + result = {"success": False, "error": "seconds 必须是数字"} + seconds_parse_ok = False + seconds = 0.0 + if seconds_parse_ok and seconds > max_sleep: result = { "success": False, - "error": "等待时间必须大于0" + "error": f"等待时间过长，最多允许 {max_sleep} 秒", + "suggestion": f"建议分多次等待或减少等待时间" } - else: - print(f"{OUTPUT_FORMATS['info']} 等待 {seconds} 秒: {reason}") - - # 执行等待 - import asyncio - await asyncio.sleep(seconds) - - result = { - "success": True, - "message": f"已等待 {seconds} 秒", - "reason": reason, - "timestamp": datetime.now().isoformat() - } - - print(f"{OUTPUT_FORMATS['success']} 等待完成") + elif seconds_parse_ok: + # 确保秒数为正数 + if seconds <= 0: + result = { + "success": False, + "error": "等待时间必须大于0" + } + else: + print(f"{OUTPUT_FORMATS['info']} 等待 {seconds} 秒: {reason}") + await asyncio.sleep(seconds) + result = { + "success": True, + "message": f"已等待 {seconds} 秒", + "reason": reason, + "timestamp": datetime.now().isoformat() + } + print(f"{OUTPUT_FORMATS['success']} 等待完成") elif tool_name == "create_file": result = self.file_manager.create_file( @@ -768,22 +892,59 @@ class MainTerminalToolsExecutionMixin: ) elif tool_name == "run_command": - result = await self.terminal_ops.run_command( - arguments["command"], - timeout=arguments.get("timeout") - ) - - # 字符数检查 - if result.get("success") and "output" in result: - char_count = len(result["output"]) - if char_count > MAX_RUN_COMMAND_CHARS: + run_in_background = bool(arguments.get("run_in_background", False)) + timeout_value = arguments.get("timeout") + if run_in_background: + if timeout_value is None or float(timeout_value) <= 0: result = { "success": False, - "error": f"结果内容过大，有{char_count}字符，请使用限制字符数的获取内容方式，根据程度选择10k以内的数", - "char_count": char_count, - "limit": MAX_RUN_COMMAND_CHARS, - "command": arguments["command"] + "error": "后台模式下 timeout 参数必填且需大于0" } + elif float(timeout_value) > 3600: + result = { + "success": False, + "error": "后台模式下 timeout 最大为 3600 秒" + } + else: + bg_manager = getattr(self, "background_command_manager", None) + if not bg_manager: + result = {"success": False, "error": "后台命令管理器不可用"} + else: + result = bg_manager.create_background_command( + terminal_ops=self.terminal_ops, + command=arguments["command"], + timeout=timeout_value, + conversation_id=getattr(self.context_manager, "current_conversation_id", None), + wait_seconds=5.0, + ) + else: + if timeout_value is None or float(timeout_value) <= 0: + result = { + "success": False, + "error": "timeout 参数必填且需大于0" + } + elif float(timeout_value) > 30: + result = { + "success": False, + "error": "前台模式下 timeout 最大为 30 秒" + } + else: + result = await self.terminal_ops.run_command( + arguments["command"], + timeout=timeout_value + ) + + # 字符数检查 + if result.get("success") and "output" in result: + char_count = len(result["output"]) + if char_count > MAX_RUN_COMMAND_CHARS: + result = { + "success": False, + "error": f"结果内容过大，有{char_count}字符，请使用限制字符数的获取内容方式，根据程度选择10k以内的数", + "char_count": char_count, + "limit": MAX_RUN_COMMAND_CHARS, + "command": arguments["command"] + } elif tool_name == "update_memory": operation = arguments["operation"] diff --git a/core/main_terminal_parts/tools_policy.py b/core/main_terminal_parts/tools_policy.py index 0ecc7c4..5019a1d 100644 --- a/core/main_terminal_parts/tools_policy.py +++ b/core/main_terminal_parts/tools_policy.py @@ -97,6 +97,9 @@ class MainTerminalToolsPolicyMixin: # Skill 强约束开关 self.skill_strict_terminal_enabled = bool(effective_config.get("skill_strict_terminal_enabled", False)) self.skill_strict_sub_agent_enabled = bool(effective_config.get("skill_strict_sub_agent_enabled", False)) + self.skill_strict_run_command_background_enabled = bool( + effective_config.get("skill_strict_run_command_background_enabled", False) + ) # 解析当前启用的 skills（用于强约束“仅对已启用 skill 生效”） try: diff --git a/modules/background_command_manager.py b/modules/background_command_manager.py new file mode 100644 index 0000000..66c2f29 --- /dev/null +++ b/modules/background_command_manager.py @@ -0,0 +1,500 @@ +"""后台 run_command 任务管理。""" + +from __future__ import annotations + +import os +import re +import shutil +import signal +import subprocess +import threading +import time +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional + +from config import MAX_RUN_COMMAND_CHARS + + +TERMINAL_STATUSES = {"completed", "failed", "timeout", "cancelled"} + + +class BackgroundCommandManager: + """管理 run_command 的后台执行、轮询与等待。""" + + def __init__(self, project_path: str): + self.project_path = Path(project_path).resolve() + self._records: Dict[str, Dict[str, Any]] = {} + self._lock = threading.RLock() + self._cv = threading.Condition(self._lock) + + def create_background_command( + self, + *, + terminal_ops, + command: str, + timeout: Optional[int], + conversation_id: Optional[str], + wait_seconds: float = 5.0, + ) -> Dict[str, Any]: + """启动后台命令；先等待一小段时间返回已有输出。""" + if timeout is None or float(timeout) <= 0: + return { + "success": False, + "error": "timeout 参数必填且需大于0", + "status": "error", + "output": "timeout 参数缺失", + "return_code": -1, + } + + timeout_value = min(int(timeout), 3600) + if timeout_value <= 0: + timeout_value = 1 + + session_override = None + if not getattr(terminal_ops, "container_session", None): + session_override = terminal_ops._resolve_active_container_session() + + execution_in_container = terminal_ops._will_use_container(session_override) + python_rewrite = terminal_ops.container_python_cmd if execution_in_container else terminal_ops.python_cmd + pip_rewrite = terminal_ops._derive_pip_from_python(python_rewrite) + final_command = command + if re.search(r"\bpython3?\b", final_command): + final_command = re.sub(r"\bpython3?\b", python_rewrite, final_command) + if re.search(r"(? None: + start_ts = time.time() + process: Optional[subprocess.Popen] = None + stdout_buf: List[str] = [] + stderr_buf: List[str] = [] + status = "failed" + return_code = -1 + message: Optional[str] = None + + try: + exec_cmd: Optional[List[str]] = None + use_shell = True + env = os.environ.copy() + env.setdefault("PYTHONUNBUFFERED", "1") + if python_env: + env.update(python_env) + + if session and getattr(session, "mode", None) == "docker": + container_name = getattr(session, "container_name", None) + mount_path = getattr(session, "mount_path", "/workspace") or "/workspace" + docker_bin = shutil.which("docker") or "docker" + if not container_name: + raise RuntimeError("容器模式下缺少 container_name") + try: + relative = work_path.relative_to(self.project_path).as_posix() + except ValueError: + relative = "" + container_workdir = mount_path.rstrip("/") + if relative: + container_workdir = f"{container_workdir}/{relative}" + exec_cmd = [ + docker_bin, + "exec", + "-e", + "PATH=/opt/agent-venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", + "-e", + "VIRTUAL_ENV=/opt/agent-venv", + "-w", + container_workdir, + container_name, + "/bin/bash", + "-lc", + command, + ] + use_shell = False + + if use_shell: + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=str(work_path), + shell=True, + env=env, + start_new_session=True, + text=True, + errors="replace", + bufsize=1, + ) + else: + process = subprocess.Popen( + exec_cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + start_new_session=True, + text=True, + errors="replace", + bufsize=1, + ) + + with self._lock: + rec = self._records.get(command_id) + if rec is not None: + rec["pid"] = process.pid + rec["updated_at"] = time.time() + + def _reader(stream, collector, rec_key: str): + try: + while True: + line = stream.readline() + if line == "": + break + collector.append(line) + with self._lock: + rec = self._records.get(command_id) + if rec is not None: + rec[rec_key].append(line) + rec["updated_at"] = time.time() + except Exception: + return + + t_out = threading.Thread(target=_reader, args=(process.stdout, stdout_buf, "stdout_chunks"), daemon=True) + t_err = threading.Thread(target=_reader, args=(process.stderr, stderr_buf, "stderr_chunks"), daemon=True) + t_out.start() + t_err.start() + + try: + process.wait(timeout=timeout) + return_code = process.returncode if process.returncode is not None else -1 + if return_code == 0: + status = "completed" + else: + status = "failed" + message = f"命令执行失败 (返回码: {return_code})" + except subprocess.TimeoutExpired: + status = "timeout" + message = f"命令执行超时 ({timeout}秒)" + try: + os.killpg(process.pid, signal.SIGINT) + except Exception: + pass + try: + process.wait(timeout=2) + except subprocess.TimeoutExpired: + try: + os.killpg(process.pid, signal.SIGKILL) + except Exception: + try: + process.kill() + except Exception: + pass + process.wait(timeout=2) + return_code = process.returncode if process.returncode is not None else -1 + + t_out.join(timeout=1) + t_err.join(timeout=1) + + except Exception as exc: + status = "failed" + message = f"执行失败: {exc}" + finally: + combined_output = "".join(stdout_buf + stderr_buf) + truncated = False + if MAX_RUN_COMMAND_CHARS and len(combined_output) > MAX_RUN_COMMAND_CHARS: + combined_output = combined_output[-MAX_RUN_COMMAND_CHARS:] + truncated = True + + success = status == "completed" + result = { + "success": success, + "status": status, + "command": command, + "output": combined_output, + "return_code": return_code, + "truncated": truncated, + "timeout": timeout, + "elapsed_ms": int((time.time() - start_ts) * 1000), + "command_id": command_id, + "run_in_background": True, + } + if message: + result["message"] = message + + 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() + self._cv.notify_all() + + 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: + return output[-MAX_RUN_COMMAND_CHARS:] + return output + + def wait_for_completion(self, command_id: str, timeout_seconds: Optional[float] = None, claim: bool = False) -> Dict[str, Any]: + """阻塞等待后台命令完成。""" + with self._cv: + rec = self._records.get(command_id) + if not rec: + return {"success": False, "error": f"未找到后台命令: {command_id}", "status": "error"} + + status = rec.get("status") + if status in TERMINAL_STATUSES and isinstance(rec.get("result"), dict): + if claim: + rec["claimed_by_sleep"] = True + return dict(rec["result"]) + + wait_limit = timeout_seconds + if wait_limit is None: + created = float(rec.get("created_at") or time.time()) + timeout_val = float(rec.get("timeout") or 0) + if timeout_val > 0: + wait_limit = max(1.0, (created + timeout_val + 5.0) - time.time()) + else: + wait_limit = 3605.0 + + deadline = time.time() + max(0.1, float(wait_limit)) + while time.time() < deadline: + rec = self._records.get(command_id) + if not rec: + return {"success": False, "error": f"未找到后台命令: {command_id}", "status": "error"} + status = rec.get("status") + if status in TERMINAL_STATUSES and isinstance(rec.get("result"), dict): + if claim: + rec["claimed_by_sleep"] = True + return dict(rec["result"]) + self._cv.wait(timeout=min(0.5, deadline - time.time())) + + rec = self._records.get(command_id) + if not rec: + return {"success": False, "error": f"未找到后台命令: {command_id}", "status": "error"} + return { + "success": False, + "status": "timeout", + "command_id": command_id, + "message": "等待后台命令完成超时", + "output": self._build_current_output(rec), + "return_code": rec.get("result", {}).get("return_code") if isinstance(rec.get("result"), dict) else None, + } + + def poll_updates(self, conversation_id: Optional[str] = None) -> List[Dict[str, Any]]: + """获取未通知且未被 sleep 领取的已完成任务。""" + updates: List[Dict[str, Any]] = [] + with self._lock: + for rec in self._records.values(): + if conversation_id and rec.get("conversation_id") != conversation_id: + continue + if rec.get("status") not in TERMINAL_STATUSES: + continue + if rec.get("notified") or rec.get("claimed_by_sleep"): + continue + payload = rec.get("result") + if isinstance(payload, dict): + updates.append(dict(payload)) + + updates.sort(key=lambda item: self._records.get(item.get("command_id"), {}).get("updated_at", 0)) + return updates + + def mark_notified(self, command_id: str): + with self._lock: + rec = self._records.get(command_id) + if rec: + rec["notified"] = True + rec["updated_at"] = time.time() + + def mark_claimed(self, command_id: str): + with self._lock: + rec = self._records.get(command_id) + if rec: + rec["claimed_by_sleep"] = True + rec["updated_at"] = time.time() + + def get_record(self, command_id: str) -> Optional[Dict[str, Any]]: + with self._lock: + rec = self._records.get(command_id) + if not rec: + return None + return dict(rec) + + def get_record_with_output(self, command_id: str) -> Optional[Dict[str, Any]]: + """获取单条后台命令记录，并附带当前可读输出。""" + with self._lock: + rec = self._records.get(command_id) + if not rec: + return None + payload = dict(rec) + payload["output"] = self._build_current_output(rec) + return payload + + def list_records( + self, + *, + conversation_id: Optional[str] = None, + limit: int = 200, + ) -> List[Dict[str, Any]]: + """列出后台命令记录（按创建时间倒序）。""" + with self._lock: + items: List[Dict[str, Any]] = [] + for rec in self._records.values(): + if conversation_id and rec.get("conversation_id") != conversation_id: + continue + item = dict(rec) + item["output"] = self._build_current_output(rec) + items.append(item) + items.sort(key=lambda x: float(x.get("created_at") or 0), reverse=True) + max_limit = max(1, min(int(limit or 200), 1000)) + return items[:max_limit] + + def has_pending_for_conversation(self, conversation_id: Optional[str]) -> bool: + if not conversation_id: + return False + with self._lock: + for rec in self._records.values(): + if rec.get("conversation_id") != conversation_id: + continue + if rec.get("status") == "running": + return True + if rec.get("status") in TERMINAL_STATUSES and (not rec.get("notified")) and (not rec.get("claimed_by_sleep")): + return True + return False + + def list_waiting_items(self, conversation_id: Optional[str]) -> List[Dict[str, Any]]: + items: List[Dict[str, Any]] = [] + if not conversation_id: + return items + with self._lock: + for rec in self._records.values(): + if rec.get("conversation_id") != conversation_id: + continue + if rec.get("status") == "running" or ( + rec.get("status") in TERMINAL_STATUSES + and not rec.get("notified") + and not rec.get("claimed_by_sleep") + ): + items.append({ + "command_id": rec.get("command_id"), + "command": rec.get("command"), + "status": rec.get("status"), + }) + items.sort(key=lambda x: x.get("command_id") or "") + return items diff --git a/modules/personalization_manager.py b/modules/personalization_manager.py index 8b31f4f..657bf83 100644 --- a/modules/personalization_manager.py +++ b/modules/personalization_manager.py @@ -55,6 +55,7 @@ DEFAULT_PERSONALIZATION_CONFIG: Dict[str, Any] = { "skill_hints_enabled": False, # Skill 提示系统开关（默认关闭） "skill_strict_terminal_enabled": False, # 强约束：terminal 系列工具需先阅读 terminal-guide "skill_strict_sub_agent_enabled": False, # 强约束：子智能体系列工具需先阅读 sub-agent-guide + "skill_strict_run_command_background_enabled": False, # 强约束：run_command 后台模式需先阅读 run-command-guide "default_model": "kimi-k2.5", "image_compression": "original", # original / 1080p / 720p / 540p "auto_shallow_compress_enabled": False, @@ -187,6 +188,13 @@ def sanitize_personalization_payload( else: base["skill_strict_sub_agent_enabled"] = bool(base.get("skill_strict_sub_agent_enabled", False)) + if "skill_strict_run_command_background_enabled" in data: + base["skill_strict_run_command_background_enabled"] = bool(data.get("skill_strict_run_command_background_enabled")) + else: + base["skill_strict_run_command_background_enabled"] = bool( + base.get("skill_strict_run_command_background_enabled", False) + ) + if "disabled_tool_categories" in data: base["disabled_tool_categories"] = _sanitize_tool_categories(data.get("disabled_tool_categories"), allowed_tool_categories) else: diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index 6542b46..054e3e2 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -127,11 +127,98 @@ from .chat_flow_runtime import ( detect_malformed_tool_call, ) -from .chat_flow_task_support import process_sub_agent_updates +from .chat_flow_task_support import process_sub_agent_updates, process_background_command_updates from .chat_flow_tool_loop import execute_tool_calls from .chat_flow_stream_loop import run_streaming_attempts from .deep_compression import run_deep_compression +async def _dispatch_completion_user_notice( + *, + web_terminal, + workspace, + sender, + client_sid: str, + username: str, + conversation_id: str, + user_message: str, + extra_payload: Optional[Dict[str, Any]] = None, +): + """复用子智能体完成后的 user 代发机制。""" + extra_payload = extra_payload or {} + try: + from .tasks import task_manager + workspace_id = getattr(workspace, "workspace_id", None) or "default" + session_data = { + "username": username, + "role": getattr(web_terminal, "user_role", "user"), + "is_api_user": getattr(web_terminal, "user_role", "") == "api", + "workspace_id": workspace_id, + "run_mode": getattr(web_terminal, "run_mode", None), + "thinking_mode": getattr(web_terminal, "thinking_mode", None), + "model_key": getattr(web_terminal, "model_key", None), + } + # 关键：通知类后台任务需要把 user_message 写入任务事件流， + # 否则前端轮询只会看到 AI/tool 事件，看不到 user_message。 + session_data["auto_user_message_event"] = True + session_data["auto_user_message_payload"] = dict(extra_payload or {}) + rec = task_manager.create_chat_task( + username, + workspace_id, + user_message, + [], + conversation_id, + model_key=session_data.get("model_key"), + thinking_mode=session_data.get("thinking_mode"), + run_mode=session_data.get("run_mode"), + session_data=session_data, + ) + payload = { + 'message': user_message, + 'conversation_id': conversation_id, + 'task_id': rec.task_id, + } + payload.update(extra_payload) + sender('user_message', payload) + return + except Exception as e: + debug_log(f"[CompletionNotice] 创建后台消息任务失败，回退直接执行: {e}") + + payload = { + 'message': user_message, + 'conversation_id': conversation_id, + } + payload.update(extra_payload) + sender('user_message', payload) + try: + task_handle = asyncio.create_task(handle_task_with_sender( + terminal=web_terminal, + workspace=workspace, + message=user_message, + images=[], + sender=sender, + client_sid=client_sid, + username=username, + videos=[] + )) + await task_handle + except Exception as inner_exc: + debug_log(f"[CompletionNotice] 回退处理 user_message 失败: {inner_exc}") + + +def _build_shared_waiting_payload(items: List[Dict[str, Any]]) -> Dict[str, Any]: + """构建与子智能体一致的 waiting 事件载荷结构。""" + normalized = [] + for item in items or []: + summary = item.get('summary') or item.get('command') or '后台任务' + normalized.append({ + 'agent_id': item.get('agent_id') or item.get('command_id') or item.get('task_id'), + 'summary': summary, + }) + return { + 'count': len(normalized), + 'tasks': normalized, + } + async def poll_sub_agent_completion(*, web_terminal, workspace, conversation_id, client_sid, username): """后台轮询子智能体完成状态，完成后触发新一轮对话""" from .extensions import socketio @@ -251,65 +338,36 @@ async def poll_sub_agent_completion(*, web_terminal, workspace, conversation_id, ] remaining_count = len(running_tasks) + len(pending_notice_tasks) has_remaining = remaining_count > 0 - - # 注册为后台任务，确保刷新后可恢复轮询 - from .tasks import task_manager - workspace_id = getattr(workspace, "workspace_id", None) or "default" - session_data = { - "username": username, - "role": getattr(web_terminal, "user_role", "user"), - "is_api_user": getattr(web_terminal, "user_role", "") == "api", - "workspace_id": workspace_id, - "run_mode": getattr(web_terminal, "run_mode", None), - "thinking_mode": getattr(web_terminal, "thinking_mode", None), - "model_key": getattr(web_terminal, "model_key", None), - } - rec = task_manager.create_chat_task( - username, - workspace_id, - user_message, - [], - conversation_id, - model_key=session_data.get("model_key"), - thinking_mode=session_data.get("thinking_mode"), - run_mode=session_data.get("run_mode"), - session_data=session_data, + await _dispatch_completion_user_notice( + web_terminal=web_terminal, + workspace=workspace, + sender=sender, + client_sid=client_sid, + username=username, + conversation_id=conversation_id, + user_message=user_message, + extra_payload={ + 'sub_agent_notice': True, + 'has_running_sub_agents': has_remaining, + 'remaining_count': remaining_count, + }, ) - debug_log(f"[SubAgent] 已创建后台任务: task_id={rec.task_id}") - sender('user_message', { - 'message': user_message, - 'conversation_id': conversation_id, - 'task_id': rec.task_id, - 'sub_agent_notice': True, - 'has_running_sub_agents': has_remaining, - 'remaining_count': remaining_count - }) except Exception as e: debug_log(f"[SubAgent] 创建后台任务失败，回退直接执行: {e}") - sender('user_message', { - 'message': user_message, - 'conversation_id': conversation_id, - 'sub_agent_notice': True, - 'has_running_sub_agents': has_remaining, - 'remaining_count': remaining_count - }) - try: - task = asyncio.create_task(handle_task_with_sender( - terminal=web_terminal, - workspace=workspace, - message=user_message, - images=[], - sender=sender, - client_sid=client_sid, - username=username, - videos=[] - )) - await task - debug_log(f"[SubAgent] process_message_task 调用成功") - except Exception as inner_exc: - debug_log(f"[SubAgent] process_message_task 失败: {inner_exc}") - import traceback - debug_log(f"[SubAgent] 错误堆栈: {traceback.format_exc()}") + await _dispatch_completion_user_notice( + web_terminal=web_terminal, + workspace=workspace, + sender=sender, + client_sid=client_sid, + username=username, + conversation_id=conversation_id, + user_message=user_message, + extra_payload={ + 'sub_agent_notice': True, + 'has_running_sub_agents': has_remaining, + 'remaining_count': remaining_count, + }, + ) return # 只处理第一个完成的子智能体 @@ -420,6 +478,77 @@ async def poll_sub_agent_completion(*, web_terminal, workspace, conversation_id, debug_log("[SubAgent] 后台轮询结束") + +async def poll_background_command_completion(*, web_terminal, workspace, conversation_id: str, client_sid: str, username: str): + """后台轮询 run_command 后台任务并在主流程结束后代发 user 通知。""" + from .extensions import socketio + + manager = getattr(web_terminal, "background_command_manager", None) + if not manager: + debug_log("[BgCommand] poll_background_command_completion: manager 不存在") + return + + max_wait_time = 3600 + start_wait = time.time() + debug_log(f"[BgCommand] 开始后台轮询，conversation_id={conversation_id}, username={username}") + + def sender(event_type, data): + try: + socketio.emit(event_type, data, room=f"user_{username}") + except Exception as e: + debug_log(f"[BgCommand] 发送事件失败: {event_type}, 错误: {e}") + + while (time.time() - start_wait) < max_wait_time: + client_stop_info = get_stop_flag(client_sid, username) + if client_stop_info: + stop_requested = client_stop_info.get('stop', False) if isinstance(client_stop_info, dict) else client_stop_info + if stop_requested: + debug_log("[BgCommand] 用户请求停止，终止轮询") + break + + if getattr(web_terminal, "_tool_loop_active", False): + debug_log("[BgCmdDebug] tool_loop_active=True, 延迟 user 代发轮询") + await asyncio.sleep(1) + continue + + updates = manager.poll_updates(conversation_id=conversation_id) + debug_log(f"[BgCmdDebug] background polling updates={len(updates)} conv={conversation_id}") + for update in updates: + command_id = update.get("command_id") + output = update.get("output") or "" + content = "[后台 run_command 完成]\n" + (output if output else "[no_output]") + prefix = "这是一句系统自动发送的user消息，用于通知你后台run_command已经运行完成" + user_message = f"{prefix}\n\n{content}" + debug_log(f"[BgCmdDebug] preparing user notice command_id={command_id} output_len={len(output)}") + manager.mark_notified(str(command_id)) + has_remaining = manager.has_pending_for_conversation(conversation_id) + await _dispatch_completion_user_notice( + web_terminal=web_terminal, + workspace=workspace, + sender=sender, + client_sid=client_sid, + username=username, + conversation_id=conversation_id, + user_message=user_message, + extra_payload={ + # 与子智能体完成通知完全复用同一前端通道/处理逻辑 + 'sub_agent_notice': True, + 'remaining_count': 1 if has_remaining else 0, + 'has_running_sub_agents': has_remaining, + 'background_command_notice': True, + 'has_running_background_commands': has_remaining, + }, + ) + debug_log(f"[BgCmdDebug] user notice dispatched command_id={command_id} has_remaining={has_remaining}") + return + + if not manager.has_pending_for_conversation(conversation_id): + debug_log("[BgCmdDebug] no pending background commands, stop polling") + break + await asyncio.sleep(5) + + debug_log("[BgCommand] 后台轮询结束") + async def handle_task_with_sender(terminal: WebTerminal, workspace: UserWorkspace, message, images, sender, client_sid, username: str, videos=None): """处理任务并发送消息 - 集成token统计版本""" from .extensions import socketio @@ -962,6 +1091,7 @@ async def handle_task_with_sender(terminal: WebTerminal, workspace: UserWorkspac conversation_id=conversation_id, last_tool_call_time=last_tool_call_time, process_sub_agent_updates=process_sub_agent_updates, + process_background_command_updates=process_background_command_updates, maybe_mark_failure_from_message=maybe_mark_failure_from_message, mark_force_thinking=mark_force_thinking, get_stop_flag=get_stop_flag, @@ -1007,6 +1137,8 @@ async def handle_task_with_sender(terminal: WebTerminal, workspace: UserWorkspac # 检查是否有后台运行的子智能体或待通知的完成任务 manager = getattr(web_terminal, "sub_agent_manager", None) has_running_sub_agents = False + bg_manager = getattr(web_terminal, "background_command_manager", None) + has_running_background_commands = False if manager: if not hasattr(web_terminal, "_announced_sub_agent_tasks"): web_terminal._announced_sub_agent_tasks = set() @@ -1052,12 +1184,41 @@ async def handle_task_with_sender(terminal: WebTerminal, workspace: UserWorkspac socketio.start_background_task(run_poll) - # 发送完成事件（如果有子智能体在运行，前端会保持等待状态） - if not has_running_sub_agents: + # 检查是否有后台 run_command 或待通知任务 + if bg_manager and conversation_id: + waiting_items = bg_manager.list_waiting_items(conversation_id) + if waiting_items: + has_running_background_commands = True + # 与子智能体完全复用同一 waiting 事件（前端已有稳定处理链路） + sender('sub_agent_waiting', _build_shared_waiting_payload(waiting_items)) + + def run_bg_poll(): + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(poll_background_command_completion( + web_terminal=web_terminal, + workspace=workspace, + conversation_id=conversation_id, + client_sid=client_sid, + username=username + )) + finally: + loop.close() + + socketio.start_background_task(run_bg_poll) + + has_running_completion_jobs = has_running_sub_agents or has_running_background_commands + + # 发送完成事件（如果有后台完成任务在运行，前端会保持等待状态） + if not has_running_completion_jobs: finalize_user_work_timer() sender('task_complete', { 'total_iterations': total_iterations, 'total_tool_calls': total_tool_calls, 'auto_fix_attempts': auto_fix_attempts, - 'has_running_sub_agents': has_running_sub_agents + # 沿用子智能体字段，确保前端直接走已验证通路 + 'has_running_sub_agents': has_running_completion_jobs, + 'has_running_background_commands': has_running_background_commands, }) diff --git a/server/chat_flow_task_support.py b/server/chat_flow_task_support.py index fab8b79..8384419 100644 --- a/server/chat_flow_task_support.py +++ b/server/chat_flow_task_support.py @@ -7,6 +7,40 @@ from typing import Dict, List, Optional from modules.sub_agent_manager import TERMINAL_STATUSES +def _insert_completion_notice_message( + *, + messages: List[Dict], + after_tool_call_id: Optional[str], + message: str, + inline: bool, + sender, + metadata: Optional[Dict] = None, + sub_agent_notice: bool = False, +): + """统一插入完成通知（system/user）并发送事件。""" + insert_index = len(messages) + if after_tool_call_id: + for idx, msg in enumerate(messages): + if msg.get("role") == "tool" and msg.get("tool_call_id") == after_tool_call_id: + insert_index = idx + 1 + break + + insert_role = "system" if inline else "user" + meta = dict(metadata or {}) + meta["inline"] = inline + messages.insert(insert_index, { + "role": insert_role, + "content": message, + "metadata": meta + }) + + payload = {'content': message, 'inline': inline} + if sub_agent_notice: + payload['sub_agent_notice'] = True + payload.update({k: v for k, v in meta.items() if k in {"command_id", "task_id", "background_command_notice"}}) + sender('system_message', payload) + + async def process_sub_agent_updates(*, messages: List[Dict], inline: bool = False, after_tool_call_id: Optional[str] = None, web_terminal, sender, debug_log, maybe_mark_failure_from_message): """轮询子智能体任务并通知前端，并把结果插入当前对话上下文。""" manager = getattr(web_terminal, "sub_agent_manager", None) @@ -75,6 +109,9 @@ async def process_sub_agent_updates(*, messages: List[Dict], inline: bool = Fals if task_id and task_id in web_terminal._announced_sub_agent_tasks: debug_log(f"[SubAgent] 任务 {task_id} 已通知过，跳过") continue + if isinstance(task_info, dict) and task_info.get("notified"): + debug_log(f"[SubAgent] 任务 {task_id} notified=true，跳过") + continue message = update.get("system_message") if not message: @@ -107,34 +144,89 @@ async def process_sub_agent_updates(*, messages: List[Dict], inline: bool = Fals except Exception as exc: debug_log(f"[SubAgent] 保存通知状态失败: {exc}") - debug_log(f"[SubAgent] 计算插入位置") - - insert_index = len(messages) - if after_tool_call_id: - for idx, msg in enumerate(messages): - if msg.get("role") == "tool" and msg.get("tool_call_id") == after_tool_call_id: - insert_index = idx + 1 - break - # 运行中插入 system 消息，避免触发新的 user 轮次；非运行中保持 user 通知 - insert_role = "system" if inline else "user" if not inline: prefix = "这是一句系统自动发送的user消息，用于通知你子智能体已经运行完成" if not message.startswith(prefix): message = f"{prefix}\n\n{message}" - messages.insert(insert_index, { - "role": insert_role, - "content": message, - "metadata": {"sub_agent_notice": True, "inline": inline, "task_id": task_id} - }) + _insert_completion_notice_message( + messages=messages, + after_tool_call_id=after_tool_call_id, + message=message, + inline=inline, + sender=sender, + metadata={"sub_agent_notice": True, "task_id": task_id}, + sub_agent_notice=True, + ) if inline: web_terminal._inline_sub_agent_notified.add(inline_key) - debug_log(f"[SubAgent] 插入子智能体通知位置: {insert_index} role={insert_role} after_tool_call_id={after_tool_call_id}") - sender('system_message', { - 'content': message, - 'inline': inline, - 'sub_agent_notice': True - }) + debug_log(f"[SubAgent] 插入子智能体通知 after_tool_call_id={after_tool_call_id} inline={inline}") + maybe_mark_failure_from_message(web_terminal, message) + + +async def process_background_command_updates(*, messages: List[Dict], inline: bool = False, after_tool_call_id: Optional[str] = None, web_terminal, sender, debug_log, maybe_mark_failure_from_message): + """轮询后台 run_command 完成状态并发送 system 消息通知。""" + manager = getattr(web_terminal, "background_command_manager", None) + if not manager: + debug_log("[BgCmdDebug] process_background_command_updates skipped: manager 不存在") + return + + conversation_id = getattr(getattr(web_terminal, "context_manager", None), "current_conversation_id", None) + debug_log(f"[BgCmdDebug] process_background_command_updates start inline={inline} after_tool_call_id={after_tool_call_id} conv={conversation_id}") + try: + updates = manager.poll_updates(conversation_id=conversation_id) + debug_log(f"[BgCmdDebug] poll_updates returned {len(updates)} updates") + except Exception as exc: + debug_log(f"[BgCommand] poll 更新失败: {exc}") + return + + if not updates: + debug_log("[BgCmdDebug] no background command updates") + return + + for update in updates: + command_id = update.get("command_id") + output = update.get("output") or "" + message = "[后台 run_command 完成]\n" + (output if output else "[no_output]") + status = update.get("status") + debug_log( + f"[BgCmdDebug] emit inline notice command_id={command_id} status={status} " + f"output_len={len(output)} inline={inline}" + ) + + # 与子智能体 inline 通知一致：写入对话历史，确保前端/历史都可见 + try: + web_terminal.context_manager.add_conversation( + "system", + message, + metadata={ + "background_command_notice": True, + "inline": inline, + "command_id": command_id, + }, + ) + debug_log(f"[BgCmdDebug] persisted system notice command_id={command_id}") + except Exception as exc: + debug_log(f"[BgCmdDebug] persist system notice failed command_id={command_id}: {exc}") + + _insert_completion_notice_message( + messages=messages, + after_tool_call_id=after_tool_call_id, + message=message, + inline=inline, + sender=sender, + metadata={ + "background_command_notice": True, + "command_id": command_id, + }, + sub_agent_notice=False, + ) + try: + manager.mark_notified(str(command_id)) + debug_log(f"[BgCmdDebug] mark_notified success command_id={command_id}") + except Exception: + debug_log(f"[BgCmdDebug] mark_notified failed command_id={command_id}") + pass maybe_mark_failure_from_message(web_terminal, message) diff --git a/server/chat_flow_tool_loop.py b/server/chat_flow_tool_loop.py index 80c25cc..83be749 100644 --- a/server/chat_flow_tool_loop.py +++ b/server/chat_flow_tool_loop.py @@ -18,7 +18,7 @@ from modules.personalization_manager import load_personalization_config, resolve from .deep_compression import run_deep_compression -async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, maybe_mark_failure_from_message, mark_force_thinking, get_stop_flag, clear_stop_flag, workspace=None): +async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, process_background_command_updates, maybe_mark_failure_from_message, mark_force_thinking, get_stop_flag, clear_stop_flag, workspace=None): previous_tool_loop_active = getattr(web_terminal, "_tool_loop_active", False) web_terminal._tool_loop_active = True # 执行每个工具 @@ -337,6 +337,17 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie 'inline': False }) maybe_mark_failure_from_message(web_terminal, system_msg) + if function_name == "sleep": + try: + if isinstance(result_data, dict) and result_data.get("mode") == "wait_sub_agent_ids": + waited_task_ids = result_data.get("waited_task_ids") or [] + if not hasattr(web_terminal, "_announced_sub_agent_tasks"): + web_terminal._announced_sub_agent_tasks = set() + for tid in waited_task_ids: + if tid: + web_terminal._announced_sub_agent_tasks.add(str(tid)) + except Exception: + pass monitor_snapshot_after = None if function_name in MONITOR_FILE_TOOLS: result_path = None @@ -572,6 +583,8 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie debug_log(f"[SubAgent] after tool={function_name} call_id={tool_call_id} -> poll updates") await process_sub_agent_updates(messages=messages, inline=True, after_tool_call_id=tool_call_id, web_terminal=web_terminal, sender=sender, debug_log=debug_log, maybe_mark_failure_from_message=maybe_mark_failure_from_message) + debug_log(f"[BgCmdDebug] after tool={function_name} call_id={tool_call_id} -> poll background command updates (inline)") + await process_background_command_updates(messages=messages, inline=True, after_tool_call_id=tool_call_id, web_terminal=web_terminal, sender=sender, debug_log=debug_log, maybe_mark_failure_from_message=maybe_mark_failure_from_message) await asyncio.sleep(0.2) diff --git a/server/conversation.py b/server/conversation.py index 521de18..8dcd9f5 100644 --- a/server/conversation.py +++ b/server/conversation.py @@ -730,6 +730,81 @@ def get_sub_agent_activity(task_id: str, terminal: WebTerminal, workspace: UserW return jsonify({"success": False, "error": str(exc)}), 500 +@conversation_bp.route('/api/background_commands', methods=['GET']) +@api_login_required +@with_terminal +def list_background_commands(terminal: WebTerminal, workspace: UserWorkspace, username: str): + """返回当前对话的后台 run_command 列表。""" + manager = getattr(terminal, "background_command_manager", None) + if not manager: + return jsonify({"success": True, "data": []}) + try: + conversation_id = terminal.context_manager.current_conversation_id + limit_raw = request.args.get("limit", "200") + try: + limit_num = max(1, min(int(limit_raw), 1000)) + except Exception: + limit_num = 200 + + records = manager.list_records(conversation_id=conversation_id, limit=limit_num) + data = [] + for rec in records: + result = rec.get("result") if isinstance(rec.get("result"), dict) else {} + data.append({ + "command_id": rec.get("command_id"), + "status": rec.get("status"), + "command": rec.get("command"), + "conversation_id": rec.get("conversation_id"), + "created_at": rec.get("created_at"), + "updated_at": rec.get("updated_at"), + "finished_at": rec.get("finished_at"), + "timeout": rec.get("timeout"), + "return_code": result.get("return_code"), + "run_in_background": True, + }) + return jsonify({"success": True, "data": data}) + 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 +def get_background_command_detail(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_with_output(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 = rec.get("result") if isinstance(rec.get("result"), dict) else {} + payload = { + "command_id": rec.get("command_id"), + "status": rec.get("status"), + "command": rec.get("command"), + "conversation_id": rec.get("conversation_id"), + "created_at": rec.get("created_at"), + "updated_at": rec.get("updated_at"), + "finished_at": rec.get("finished_at"), + "timeout": rec.get("timeout"), + "return_code": result.get("return_code"), + "message": result.get("message"), + "output": rec.get("output") or "", + "run_in_background": True, + } + return jsonify({"success": True, "data": payload}) + 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/server/tasks.py b/server/tasks.py index 8cc65e8..f3e779d 100644 --- a/server/tasks.py +++ b/server/tasks.py @@ -252,6 +252,23 @@ class TaskManager: except Exception as exc: raise RuntimeError(f"对话加载失败: {exc}") from exc + # 仅对“后台通知触发的新任务”补发 user_message 事件到任务事件流。 + # 这样前端轮询能即时看到这条 user 消息，而不是刷新后才从历史中看到。 + try: + if bool((rec.session_data or {}).get("auto_user_message_event")): + extra_payload = (rec.session_data or {}).get("auto_user_message_payload") or {} + if not isinstance(extra_payload, dict): + extra_payload = {} + payload = { + "message": rec.message, + "conversation_id": rec.conversation_id, + "task_id": rec.task_id, + } + payload.update(extra_payload) + self._append_event(rec, "user_message", payload) + except Exception as exc: + debug_log(f"[Task] 注入 user_message 事件失败: {exc}") + def sender(event_type, data): # 记录事件 self._append_event(rec, event_type, data) diff --git a/static/src/App.vue b/static/src/App.vue index 720f47a..42e70b2 100644 --- a/static/src/App.vue +++ b/static/src/App.vue @@ -351,6 +351,7 @@ /> + { this.autoResizeInput(); @@ -104,6 +106,7 @@ export function beforeUnmount() { window.removeEventListener('keydown', this.handleMobileOverlayEscape); this.teardownMobileViewportWatcher(); this.subAgentStopPolling(); + this.backgroundCommandStopPolling(); this.resourceStopContainerStatsPolling(); this.resourceStopProjectStoragePolling(); this.resourceStopUsageQuotaPolling(); diff --git a/static/src/app/methods/conversation.ts b/static/src/app/methods/conversation.ts index cc66c4f..9840939 100644 --- a/static/src/app/methods/conversation.ts +++ b/static/src/app/methods/conversation.ts @@ -277,10 +277,11 @@ export const conversationMethods = { const taskStore = useTaskStore(); const hasActiveTask = taskStore.hasActiveTask || this.taskInProgress || this.waitingForSubAgent; if (hasActiveTask) { + const waitingLabel = this.waitingForBackgroundCommand ? '后台指令' : '后台子智能体'; const confirmed = await this.confirmAction({ title: '切换对话', message: this.waitingForSubAgent - ? '后台子智能体正在运行，切换对话后将不会自动接收完成提示。确定要切换吗？' + ? `${waitingLabel}正在运行，切换对话后将不会自动接收完成提示。确定要切换吗？` : '当前有任务正在执行，切换对话后任务会停止。确定要切换吗？', confirmText: '切换', cancelText: '取消' @@ -301,6 +302,7 @@ export const conversationMethods = { if (this.waitingForSubAgent && !taskStore.hasActiveTask) { this.waitingForSubAgent = false; + this.waitingForBackgroundCommand = false; } this.streamingMessage = false; @@ -455,12 +457,13 @@ export const conversationMethods = { if (taskStore.hasActiveTask || this.taskInProgress || this.waitingForSubAgent) { hasActiveTask = true; + const waitingLabel = this.waitingForBackgroundCommand ? '后台指令' : '后台子智能体'; // 显示提示 const confirmed = await this.confirmAction({ title: '创建新对话', message: this.waitingForSubAgent - ? '后台子智能体正在运行，创建新对话后将不会自动接收完成提示。确定要创建吗？' + ? `${waitingLabel}正在运行，创建新对话后将不会自动接收完成提示。确定要创建吗？` : '当前有任务正在执行，创建新对话后任务会停止。确定要创建吗？', confirmText: '创建', cancelText: '取消' @@ -487,6 +490,7 @@ export const conversationMethods = { this.stopRequested = false; if (this.waitingForSubAgent && !taskStore.hasActiveTask) { this.waitingForSubAgent = false; + this.waitingForBackgroundCommand = false; } debugLog('[创建新对话] 任务已停止'); diff --git a/static/src/app/methods/history.ts b/static/src/app/methods/history.ts index 16a113e..5547a43 100644 --- a/static/src/app/methods/history.ts +++ b/static/src/app/methods/history.ts @@ -2,6 +2,7 @@ import { debugLog } from './common'; const SUB_AGENT_DONE_PREFIX_RE = /^(?:✅\s*)?子智能体\s*#?\s*(\d+)\s*任务摘要[:：]/; +const BG_RUN_COMMAND_DONE_PREFIX_RE = /^\[后台\s*run_command\s*完成\]/; function parseSubAgentDoneLabel(rawContent: any): string | null { const content = (rawContent || '').toString().trim(); @@ -11,6 +12,17 @@ function parseSubAgentDoneLabel(rawContent: any): string | null { return `子智能体${match[1]} 任务完成`; } +function parseBackgroundRunCommandDoneLabel(rawContent: any): string | null { + const content = (rawContent || '').toString().trim(); + if (!content) return null; + if (!BG_RUN_COMMAND_DONE_PREFIX_RE.test(content)) return null; + return '后台 run_command 完成'; +} + +function parseSystemNoticeLabel(rawContent: any): string | null { + return parseSubAgentDoneLabel(rawContent) || parseBackgroundRunCommandDoneLabel(rawContent); +} + export const historyMethods = { // ========================================== // 关键功能：获取并显示历史对话内容 @@ -327,10 +339,10 @@ export const historyMethods = { } if (message.role === 'system') { const rawContent = message.content || ''; - const label = parseSubAgentDoneLabel(rawContent); + const label = parseSystemNoticeLabel(rawContent); console.log('[DEBUG_SYSTEM][history] 命中 role=system 历史消息', { rawContent, - matchedSubAgentDone: !!label + matchedDoneNotice: !!label }); if (label) { // 历史中的 system 通知转换为 assistant system action，避免被 role=system 过滤掉 diff --git a/static/src/app/methods/taskPolling.ts b/static/src/app/methods/taskPolling.ts index c6d1dc6..0407c50 100644 --- a/static/src/app/methods/taskPolling.ts +++ b/static/src/app/methods/taskPolling.ts @@ -1,11 +1,118 @@ // @ts-nocheck import { debugLog } from './common'; +const debugNotifyLog = (..._args: any[]) => {}; +const keyNotifyLog = (...args: any[]) => { + console.log(...args); +}; + /** * 任务轮询事件处理器 * 将从 REST API 轮询获取的事件转换为前端状态更新 */ export const taskPollingMethods = { + stopWaitingTaskProbe() { + if (this.waitingTaskProbeTimer) { + debugNotifyLog('[DEBUG_NOTIFY][ui] stopWaitingTaskProbe'); + clearInterval(this.waitingTaskProbeTimer); + this.waitingTaskProbeTimer = null; + } + }, + + async tryResumeWaitingNoticeTask() { + if (!this.waitingForSubAgent || !this.currentConversationId) { + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:skip', { + waitingForSubAgent: this.waitingForSubAgent, + currentConversationId: this.currentConversationId + }); + return false; + } + try { + const { useTaskStore } = await import('../../stores/task'); + const taskStore = useTaskStore(); + if (taskStore.hasActiveTask) { + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:already-active', { + taskId: taskStore.currentTaskId, + status: taskStore.taskStatus + }); + return true; + } + + const resp = await fetch('/api/tasks'); + if (!resp.ok) { + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:tasks-api-not-ok', { + status: resp.status + }); + return false; + } + const result = await resp.json().catch(() => ({})); + const tasks = Array.isArray(result?.data) ? result.data : []; + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:tasks', { + currentConversationId: this.currentConversationId, + count: tasks.length, + running: tasks + .filter((t: any) => t?.status === 'running') + .map((t: any) => ({ + task_id: t?.task_id, + conversation_id: t?.conversation_id, + status: t?.status + })) + }); + const runningTask = tasks.find((task: any) => + task?.status === 'running' && task?.conversation_id === this.currentConversationId + ); + if (!runningTask?.task_id) { + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:no-matched-task'); + return false; + } + + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:resume', { + task_id: runningTask.task_id + }); + keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] tryResumeWaitingNoticeTask:resume', { + task_id: runningTask.task_id, + conversationId: this.currentConversationId + }); + // 关键修复：切到新的通知任务前，清空旧任务事件去重索引。 + // 否则新任务 event idx 与旧任务重叠时会被误判为“重复事件”而丢失 user_message。 + if (typeof this.clearProcessedEvents === 'function') { + this.clearProcessedEvents(); + } + taskStore.resumeTask(runningTask.task_id, { + status: 'running', + resetOffset: true, + eventHandler: (event: any) => this.handleTaskEvent(event) + }); + return true; + } catch (error) { + console.warn('[TaskPolling] 接管等待任务失败:', error); + return false; + } + }, + + startWaitingTaskProbe() { + if (this.waitingTaskProbeTimer) { + debugNotifyLog('[DEBUG_NOTIFY][ui] startWaitingTaskProbe:already-started'); + return; + } + debugNotifyLog('[DEBUG_NOTIFY][ui] startWaitingTaskProbe'); + keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] startWaitingTaskProbe', { + conversationId: this.currentConversationId + }); + this.waitingTaskProbeTimer = setInterval(async () => { + if (!this.waitingForSubAgent) { + this.stopWaitingTaskProbe(); + return; + } + const resumed = await this.tryResumeWaitingNoticeTask(); + if (resumed) { + debugNotifyLog('[DEBUG_NOTIFY][ui] waitingTaskProbe:resumed-and-stop'); + keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] waitingTaskProbe:resumed-and-stop'); + this.stopWaitingTaskProbe(); + } + }, 1000); + }, + /** * 处理任务事件（从轮询获取） */ @@ -22,11 +129,19 @@ export const taskPollingMethods = { eventType === 'sub_agent_waiting' || (eventType === 'user_message' && eventData?.sub_agent_notice) ) { - console.log('[DEBUG_SYSTEM][event] 捕获关键事件', { + debugNotifyLog('[DEBUG_NOTIFY][event] 捕获关键事件', { eventType, eventIdx, eventData }); + keyNotifyLog('[DEBUG_NOTIFY_KEY][event] key-event', { + eventType, + idx: eventIdx, + task_id: eventData?.task_id, + sub_agent_notice: !!eventData?.sub_agent_notice, + has_running_sub_agents: eventData?.has_running_sub_agents, + has_running_background_commands: eventData?.has_running_background_commands + }); } // 事件去重检查 @@ -167,6 +282,10 @@ export const taskPollingMethods = { break; case 'user_message': + debugNotifyLog('[DEBUG_NOTIFY][event] dispatch:user_message', { + idx: eventIdx, + data: eventData + }); this.handleUserMessage(eventData, eventIdx); break; @@ -712,6 +831,7 @@ export const taskPollingMethods = { handleTaskComplete(data: any) { const hasRunningSubAgents = !!data?.has_running_sub_agents; + const hasRunningBackgroundCommands = !!data?.has_running_background_commands; if (hasRunningSubAgents) { debugLog('[TaskPolling] 任务完成，但仍有后台子智能体运行'); } else { @@ -728,9 +848,15 @@ export const taskPollingMethods = { if (hasRunningSubAgents) { this.taskInProgress = true; this.waitingForSubAgent = true; + this.waitingForBackgroundCommand = hasRunningBackgroundCommands; + // 关键修复：主任务结束后将切到新的通知任务，先清空旧任务事件索引去重表。 + this.clearProcessedEvents(); + this.startWaitingTaskProbe(); } else { this.taskInProgress = false; this.waitingForSubAgent = false; + this.waitingForBackgroundCommand = false; + this.stopWaitingTaskProbe(); this.clearTaskState(); // 清理任务状态 } @@ -753,6 +879,8 @@ export const taskPollingMethods = { this.taskInProgress = false; this.stopRequested = false; this.waitingForSubAgent = false; + this.waitingForBackgroundCommand = false; + this.stopWaitingTaskProbe(); if (typeof this.clearPendingTools === 'function') { this.clearPendingTools('task_stopped'); @@ -764,6 +892,7 @@ export const taskPollingMethods = { // 新增统一清理方法 clearTaskState() { + this.stopWaitingTaskProbe(); (async () => { const { useTaskStore } = await import('../../stores/task'); const taskStore = useTaskStore(); @@ -778,9 +907,24 @@ export const taskPollingMethods = { handleUserMessage(data: any) { const message = (data?.message || data?.content || '').trim(); if (!message) { + debugNotifyLog('[DEBUG_NOTIFY][ui] handleUserMessage:empty', { data }); return; } debugLog('[TaskPolling] 收到用户消息事件'); + debugNotifyLog('[DEBUG_NOTIFY][ui] handleUserMessage', { + messagePreview: message.slice(0, 120), + sub_agent_notice: !!data?.sub_agent_notice, + background_command_notice: !!data?.background_command_notice, + task_id: data?.task_id, + has_running_sub_agents: data?.has_running_sub_agents, + has_running_background_commands: data?.has_running_background_commands + }); + keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] handleUserMessage', { + messagePreview: message.slice(0, 80), + task_id: data?.task_id, + sub_agent_notice: !!data?.sub_agent_notice, + background_command_notice: !!data?.background_command_notice + }); this.chatAddUserMessage(message, data?.images || [], data?.videos || []); this.taskInProgress = true; this.streamingMessage = false; @@ -791,6 +935,16 @@ export const taskPollingMethods = { } else if (typeof data?.remaining_count === 'number') { this.waitingForSubAgent = data.remaining_count > 0; } + if (typeof data?.has_running_background_commands === 'boolean') { + this.waitingForBackgroundCommand = data.has_running_background_commands; + } else if (data?.background_command_notice) { + this.waitingForBackgroundCommand = this.waitingForSubAgent; + } + if (this.waitingForSubAgent) { + this.startWaitingTaskProbe(); + } else { + this.stopWaitingTaskProbe(); + } } this.$forceUpdate(); this.conditionalScrollToBottom(); @@ -1363,9 +1517,11 @@ export const taskPollingMethods = { pendingTasks: pendingNotice.map((task: any) => ({ task_id: task?.task_id, status: task?.status })) }); this.waitingForSubAgent = true; + this.waitingForBackgroundCommand = false; this.taskInProgress = true; this.streamingMessage = false; this.stopRequested = false; + this.startWaitingTaskProbe(); this.$forceUpdate(); } } catch (error) { diff --git a/static/src/app/methods/ui.ts b/static/src/app/methods/ui.ts index c067cc0..396d5d9 100644 --- a/static/src/app/methods/ui.ts +++ b/static/src/app/methods/ui.ts @@ -18,6 +18,7 @@ import { import { debugLog } from './common'; const SUB_AGENT_DONE_PREFIX_RE = /^(?:✅\s*)?子智能体\s*#?\s*(\d+)\s*任务摘要[:：]/; +const BG_RUN_COMMAND_DONE_PREFIX_RE = /^\[后台\s*run_command\s*完成\]/; function parseSubAgentDoneLabel(rawContent: any): string | null { const content = (rawContent || '').toString().trim(); @@ -41,6 +42,21 @@ function parseSubAgentDoneLabel(rawContent: any): string | null { return `子智能体${agentId} 任务完成`; } +function parseBackgroundRunCommandDoneLabel(rawContent: any): string | null { + const content = (rawContent || '').toString().trim(); + if (!content) { + return null; + } + if (!BG_RUN_COMMAND_DONE_PREFIX_RE.test(content)) { + return null; + } + return '后台 run_command 完成'; +} + +function parseSystemNoticeLabel(rawContent: any): string | null { + return parseSubAgentDoneLabel(rawContent) || parseBackgroundRunCommandDoneLabel(rawContent); +} + export const uiMethods = { ensureScrollListener() { if (this._scrollListenerReady) { @@ -248,6 +264,12 @@ export const uiMethods = { // 切换到子智能体面板时立即刷新，避免等待轮询间隔 this.subAgentFetch(); this.subAgentStartPolling(); + return; + } + if (mode === 'backgroundCommands') { + // 切换到后台指令面板时立即刷新，避免等待轮询间隔 + this.backgroundCommandFetch(); + this.backgroundCommandStartPolling(); } }, @@ -873,17 +895,17 @@ export const uiMethods = { addSystemMessage(content) { console.log('[DEBUG_SYSTEM][ui] addSystemMessage 入参', { content }); - const subAgentDoneLabel = parseSubAgentDoneLabel(content); - if (!subAgentDoneLabel) { + const systemNoticeLabel = parseSystemNoticeLabel(content); + if (!systemNoticeLabel) { // 其他 system 消息全部隐藏，且不进入渲染链路，避免出现空白块 - console.log('[DEBUG_SYSTEM][ui] addSystemMessage 被拦截（非子智能体完成）', { content }); + console.log('[DEBUG_SYSTEM][ui] addSystemMessage 被拦截（非完成通知）', { content }); return; } console.log('[DEBUG_SYSTEM][ui] addSystemMessage 通过，写入 action', { original: content, - normalizedLabel: subAgentDoneLabel + normalizedLabel: systemNoticeLabel }); - this.chatAddSystemMessage(subAgentDoneLabel, { variant: 'sub_agent_done' }); + this.chatAddSystemMessage(systemNoticeLabel, { variant: 'sub_agent_done' }); this.$forceUpdate(); this.conditionalScrollToBottom(); }, diff --git a/static/src/app/state.ts b/static/src/app/state.ts index 93f8c09..9635e5e 100644 --- a/static/src/app/state.ts +++ b/static/src/app/state.ts @@ -11,6 +11,8 @@ export function dataState() { usePollingMode: true, // 后台子智能体等待状态 waitingForSubAgent: false, + // 后台 run_command 等待状态（用于文案区分） + waitingForBackgroundCommand: false, // 是否在对话区展示 system 消息 hideSystemMessages: true, @@ -21,6 +23,8 @@ export function dataState() { toolStacks: new Map(), // 当前任务是否仍在进行中（用于保持输入区的"停止"状态） taskInProgress: false, + // 等待后台通知期间，用于发现并接管新建任务的轮询定时器 + waitingTaskProbeTimer: null, // 记录上一次成功加载历史的对话ID，防止初始化阶段重复加载导致动画播放两次 lastHistoryLoadedConversationId: null, diff --git a/static/src/components/chat/ChatArea.vue b/static/src/components/chat/ChatArea.vue index b07aefb..22fde1b 100644 --- a/static/src/components/chat/ChatArea.vue +++ b/static/src/components/chat/ChatArea.vue @@ -449,6 +449,7 @@ const filteredMessages = computed(() => { const latestMessageIndex = computed(() => filteredMessages.value.length - 1); const SUB_AGENT_DONE_LABEL_RE = /^子智能体\d+\s*任务完成$/; const SUB_AGENT_DONE_PREFIX_RE = /^(?:✅\s*)?子智能体\s*#?\s*(\d+)\s*任务摘要[:：]/; +const BG_RUN_COMMAND_DONE_LABEL_RE = /^(?:\[)?后台\s*run_command\s*完成(?:\])?$/; const RENDERABLE_ACTION_TYPES = new Set(['thinking', 'text', 'tool', 'append', 'append_payload', 'system']); const nowMs = ref(Date.now()); let timerHandle: number | null = null; @@ -524,7 +525,7 @@ function getSubAgentSystemNoticeLabel(action: any): string | null { } return null; } - if (action.variant === 'sub_agent_done' && SUB_AGENT_DONE_LABEL_RE.test(content)) { + if (action.variant === 'sub_agent_done' && (SUB_AGENT_DONE_LABEL_RE.test(content) || BG_RUN_COMMAND_DONE_LABEL_RE.test(content))) { if (!debugLoggedSystemActionKeys.has(`${key}-variant-pass`)) { console.log('[DEBUG_SYSTEM][render] 命中 variant 渲染', { action, label: content }); debugLoggedSystemActionKeys.add(`${key}-variant-pass`); diff --git a/static/src/components/overlay/BackgroundCommandDialog.vue b/static/src/components/overlay/BackgroundCommandDialog.vue new file mode 100644 index 0000000..9f61029 --- /dev/null +++ b/static/src/components/overlay/BackgroundCommandDialog.vue @@ -0,0 +1,47 @@ + + + diff --git a/static/src/components/panels/LeftPanel.vue b/static/src/components/panels/LeftPanel.vue index cdb28d6..e3943c0 100644 --- a/static/src/components/panels/LeftPanel.vue +++ b/static/src/components/panels/LeftPanel.vue @@ -71,6 +71,14 @@ > + @@ -91,8 +99,14 @@ 待办列表 - - 子智能体 + + @@ -123,6 +137,26 @@ +
+
暂无后台指令
+
+
+
+ {{ command.command_id }} + {{ command.status || 'running' }} +
+
{{ command.command || '（空命令）' }}
+
+ {{ formatBackgroundCommandMeta(command) }} +
+
+
+
{{ fileTreeMessage }}