From e29ccb318edaec9e6db8fe629a992f45f7cbe68c Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Mon, 13 Jul 2026 20:05:02 +0800 Subject: [PATCH] =?UTF-8?q?fix(multi-agent):=20=E4=BF=AE=E5=A4=8D=E6=83=85?= =?UTF-8?q?=E5=86=B52=E4=B8=BB=E6=B6=88=E6=81=AF=E6=B1=A0=E6=B4=BE?= =?UTF-8?q?=E5=8F=91=E9=93=BE=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复多智能体模式下主智能体空闲时收不到子智能体输出推送的一系列 bug： 1. poll_multi_agent_notifications 死锁：原实现先等所有 running 实例 退出再 drain 消息池，导致 ask_master 在 await 期间永远等不到主对话 回答。改为池优先：pool 有消息立即派发，不管 running 状态。 2. _dispatch_multi_agent_idle_messages 缺 import：调用 inject_multi_agent_master_message 但文件顶部从未导入，NameError 被外层 except 吞掉，task 永远建不起来。 3. dispatch 内调试日志引用 rec 错位：dispatch_ma_idle_sender_user_message 被放到 create_chat_task 之前，触发 UnboundLocalError，task 同样建不起来。 4. session_data['auto_user_message_payload'] / preceding_user_notices payload 漏写 auto_message_type：前端 isMultiAgentMessage() 只认 auto_message_type.startsWith('multi_agent_')，空字符串走 fallback 通知渲染。 5. dispatch 第①步重复持久化：对包含 last 的全部消息都调 inject_multi_agent_master_message 落盘，之后 task 又在 handle_task_with_sender 再 add_conversation，导致历史里出现两条相同 user 消息（前一条多智能体渲染， 后一条通知渲染）。前置 N-1 条只持久化一次，最后一条交给后续 task 自己持久化。 6. last 赋值时机错位：last_emit_payload 在 last=parsed_messages[-1] 之前引用， UnboundLocalError 再次吃掉后续链路。 7. handle_task_with_sender 多智能体分支漏写 visibility='chat'： _user_message_ui_defaults('sub_agent') 默认 visibility='compact'， 透传到落盘 metadata 后，前端从后端加载历史时走通知渲染分支。显式 user_message_metadata['visibility']='chat' 强制走多智能体专用渲染。 --- core/main_terminal_parts/context/messages.py | 8 + modules/multi_agent/state.py | 14 + modules/sub_agent/manager.py | 102 +++- modules/sub_agent/state.py | 15 + modules/sub_agent/task.py | 29 + prompts/multi_agent/available_agents.txt | 2 + server/chat_flow_task_main.py | 530 ++++++++++++++++-- server/chat_flow_task_support.py | 32 +- server/tasks/helpers.py | 1 + server/tasks/models.py | 17 +- static/src/app/methods/common.ts | 2 + .../src/app/methods/taskPolling/lifecycle.ts | 22 +- static/src/app/methods/taskPolling/probe.ts | 128 ++++- static/src/composables/useLegacySocket.ts | 67 ++- static/src/stores/task.ts | 33 ++ 15 files changed, 911 insertions(+), 91 deletions(-) diff --git a/core/main_terminal_parts/context/messages.py b/core/main_terminal_parts/context/messages.py index 8605c46..cb06a7a 100644 --- a/core/main_terminal_parts/context/messages.py +++ b/core/main_terminal_parts/context/messages.py @@ -386,6 +386,14 @@ class MessagesMixin: # 调试：记录所有 system 消息 if conv["role"] == "system": logger.info(f"[DEBUG build_messages] 添加 system 消息: content前50字={conv['content'][:50]}") + # 调试：记录多智能体子智能体输出消息是否进入上下文 + if metadata.get("multi_agent_message"): + logger.info( + f"[DEBUG build_messages] 添加多智能体子智能体消息到上下文: " + f"role={conv['role']}, subtype={metadata.get('multi_agent_subtype')}, " + f"display_name={metadata.get('multi_agent_display_name')}, " + f"content前80字={str(conv.get('content', ''))[:80]}" + ) messages.append({ "role": conv["role"], "content": content_payload diff --git a/modules/multi_agent/state.py b/modules/multi_agent/state.py index 67a1b96..57a5ac9 100644 --- a/modules/multi_agent/state.py +++ b/modules/multi_agent/state.py @@ -328,12 +328,26 @@ class MultiAgentState: # ----- 主对话注入 ----- def push_master_message(self, message_text: str) -> None: """把一条 user 消息追加到主对话待插入队列。""" + ma_debug( + "state_push_master_message", + conversation_id=self.conversation_id, + state_id=id(self), + queue_len_before=len(self.pending_master_messages), + msg_preview=str(message_text)[:200], + ) self.pending_master_messages.append(message_text) def drain_master_messages(self) -> List[str]: """取出（清空）所有待插入主对话的消息。""" msgs = self.pending_master_messages self.pending_master_messages = [] + ma_debug( + "state_drain_master_messages", + conversation_id=self.conversation_id, + state_id=id(self), + drained_count=len(msgs), + previews=[str(m)[:150] for m in msgs], + ) return msgs def has_pending_master_messages(self) -> bool: diff --git a/modules/sub_agent/manager.py b/modules/sub_agent/manager.py index 7a53697..551a790 100644 --- a/modules/sub_agent/manager.py +++ b/modules/sub_agent/manager.py @@ -275,6 +275,13 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi multi_agent_state=multi_agent_state, display_name=display_name, ) + ma_debug( + "manager_create_sub_agent_state", + task_id=task_id, + agent_id=agent_id, + conversation_id=conversation_id, + state_id=id(multi_agent_state) if multi_agent_state else None, + ) task_coro = sub_agent.run() asyncio_task = self._run_coro(task_coro) sub_agent._task = asyncio_task @@ -283,12 +290,16 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi self._sub_agent_instances[agent_id] = sub_agent def _on_done(fut): - self._running_tasks.pop(task_id, None) - self._sub_agent_instances.pop(agent_id, None) - self.reconcile_task_states(conversation_id=conversation_id) - # 多智能体模式：结束时把状态写回 MultiAgentState - if multi_agent_mode and multi_agent_state: - self._on_multi_agent_task_done(task_id, agent_id, multi_agent_state, sub_agent) + try: + self._running_tasks.pop(task_id, None) + self._sub_agent_instances.pop(agent_id, None) + self.reconcile_task_states(conversation_id=conversation_id) + # 多智能体模式：结束时把状态写回 MultiAgentState + if multi_agent_mode and multi_agent_state: + self._on_multi_agent_task_done(task_id, agent_id, multi_agent_state, sub_agent) + except Exception as exc: + logger.exception(f"[SubAgent] task {task_id} 完成回调异常: {exc}") + ma_debug("manager_on_done_exception", task_id=task_id, agent_id=agent_id, error=str(exc)) asyncio_task.add_done_callback(_on_done) @@ -704,11 +715,15 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi self._sub_agent_instances[agent_id] = sub_agent def _on_done(fut, tid=task_id, aid=agent_id, state=multi_agent_state, sa=sub_agent): - self._running_tasks.pop(tid, None) - self._sub_agent_instances.pop(aid, None) - self.reconcile_task_states(conversation_id=conversation_id) - if multi_agent_mode and state: - self._on_multi_agent_task_done(tid, aid, state, sa) + try: + self._running_tasks.pop(tid, None) + self._sub_agent_instances.pop(aid, None) + self.reconcile_task_states(conversation_id=conversation_id) + if multi_agent_mode and state: + self._on_multi_agent_task_done(tid, aid, state, sa) + except Exception as exc: + logger.exception(f"[SubAgent] restored task {tid} 完成回调异常: {exc}") + ma_debug("manager_restore_on_done_exception", task_id=tid, agent_id=aid, error=str(exc)) asyncio_task.add_done_callback(_on_done) restored += 1 @@ -734,14 +749,34 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi from modules.multi_agent.state import MultiAgentState state = self.multi_agent_states.get(conversation_id) if state: + ma_debug( + "manager_get_or_create_ma_state_reuse", + conversation_id=conversation_id, + state_id=id(state), + manager_id=id(self), + ) return state state = MultiAgentState(conversation_id=conversation_id) self.multi_agent_states[conversation_id] = state + ma_debug( + "manager_get_or_create_ma_state_create", + conversation_id=conversation_id, + state_id=id(state), + manager_id=id(self), + ) return state def get_multi_agent_state(self, conversation_id: str): """获取该会话的多智能体状态。""" - return self.multi_agent_states.get(conversation_id) + state = self.multi_agent_states.get(conversation_id) + ma_debug( + "manager_get_multi_agent_state", + conversation_id=conversation_id, + found=bool(state), + state_id=id(state) if state else None, + manager_id=id(self), + ) + return state def drop_multi_agent_state(self, conversation_id: str) -> None: """删除会话状态（会话结束时调用）。""" @@ -765,7 +800,47 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi ): task["multi_agent_mode"] = True task["updated_at"] = time.time() - return super().reconcile_task_states(conversation_id=conversation_id) + changed = super().reconcile_task_states(conversation_id=conversation_id) + + # 多智能体模式下，子智能体进入 idle 后底层 asyncio.Task 仍在等待唤醒， + # 父类 reconcile 会据此把任务标回 running。这里根据内存中的 SubAgentTask + # 实例重新把 idle 状态写回任务记录，使运行态与 MultiAgentState 保持一致。 + extra_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 + if not task.get("multi_agent_mode"): + continue + agent_id = task.get("agent_id") + inst = self._sub_agent_instances.get(agent_id) if agent_id else None + if inst: + if getattr(inst, "_idle", False): + if task.get("status") != "idle": + task["status"] = "idle" + task["updated_at"] = time.time() + ma_debug( + "reconcile_task_runtime_state_idle_fix", + task_id=task.get("task_id"), + agent_id=agent_id, + ) + extra_changed += 1 + elif task.get("status") == "idle": + # 子智能体已被唤醒且 _idle=false，但 output 文件或父类 reconcile + # 可能仍把任务标为 idle。这里强制同步回 running。 + task["status"] = "running" + task["updated_at"] = time.time() + ma_debug( + "reconcile_task_runtime_state_running_fix", + task_id=task.get("task_id"), + agent_id=agent_id, + ) + extra_changed += 1 + if extra_changed: + self._save_state() + changed += extra_changed + return changed def inject_message_to_sub_agent(self, agent_id: int, message_text: str) -> bool: """同事件循环中向子智能体上下文插入 user 消息。 @@ -828,6 +903,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi return # 兜底：取出当前 task status（由 _finalize_task 设置） + final_status = final_task.get("status") if final_status in TERMINAL_STATUSES: state.mark_status(agent_id, final_status, last_output=str(final_task.get("final_result") or "")) ma_debug("manager_ma_state_set", agent_id=agent_id, status=final_status, reason="task_terminal_status") diff --git a/modules/sub_agent/state.py b/modules/sub_agent/state.py index 1cb351c..de168bf 100644 --- a/modules/sub_agent/state.py +++ b/modules/sub_agent/state.py @@ -62,7 +62,22 @@ class SubAgentStateMixin: for conv_id, snapshot in loaded_ma_states.items(): try: if isinstance(snapshot, dict): + # 关键：不要覆盖内存中已存在的 MultiAgentState， + # 否则 SubAgentTask 持有的旧引用上的 pending_master_messages + # 会被新的空 state 覆盖，导致子智能体输出丢失。 + if conv_id in multi_agent_states: + ma_debug( + "load_state_skip_existing_ma_state", + conversation_id=conv_id, + existing_state_id=id(multi_agent_states[conv_id]), + ) + continue multi_agent_states[conv_id] = MultiAgentState.from_snapshot(snapshot) + ma_debug( + "load_state_restore_ma_state", + conversation_id=conv_id, + state_id=id(multi_agent_states[conv_id]), + ) except Exception as exc: logger.warning(f"恢复多智能体状态失败 {conv_id}: {exc}") except Exception as exc: diff --git a/modules/sub_agent/task.py b/modules/sub_agent/task.py index 779a71e..a0e52c3 100644 --- a/modules/sub_agent/task.py +++ b/modules/sub_agent/task.py @@ -180,6 +180,16 @@ class SubAgentTask: pending_messages=[m.get("role") for m in self.messages[-3:]], ) self._idle = False + # 关键修复：从 idle 唤醒后必须同步更新 MultiAgentState 状态为 running， + # 否则 has_running_multi_agent 会误判为 false，导致主对话提前进入空闲、 + # 后续子智能体输出无法推送到主智能体。 + if self.multi_agent_state: + self.multi_agent_state.mark_status(self.agent_id, "running") + ma_debug( + "sub_agent_idle_wake_mark_running", + task_id=self.task_id, + agent_id=self.agent_id, + ) continue turn += 1 @@ -272,7 +282,19 @@ class SubAgentTask: def _forward_output_to_master(self, output_text: str, *, is_final: bool = False) -> None: """把子智能体的 assistant 文本输出转发成主对话的 user 消息，并写入进度文件供前端查看。""" + ma_debug( + "sub_agent_forward_output_enter", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + multi_agent_mode=self.multi_agent_mode, + has_state=bool(self.multi_agent_state), + state_id=id(self.multi_agent_state) if self.multi_agent_state else None, + output_len=len(output_text), + is_final=is_final, + ) if not self.multi_agent_state: + ma_debug("sub_agent_forward_no_state", task_id=self.task_id, agent_id=self.agent_id) return # 如果这是对 pending 提问的回答，不走主对话转发，而是返回到 ask 工具结果 if self._provide_answer(output_text): @@ -281,6 +303,13 @@ class SubAgentTask: from modules.multi_agent.state import build_sub_agent_output_text msg = build_sub_agent_output_text(self.display_name, output_text.strip(), is_final=is_final) self.multi_agent_state.push_master_message(msg) + ma_debug( + "sub_agent_forward_pushed", + task_id=self.task_id, + agent_id=self.agent_id, + state_id=id(self.multi_agent_state) if self.multi_agent_state else None, + msg_preview=msg[:200], + ) # 同时记录到实例状态，供 list_active_sub_agents 使用 inst = self.multi_agent_state.get_instance(self.agent_id) if inst: diff --git a/prompts/multi_agent/available_agents.txt b/prompts/multi_agent/available_agents.txt index fed0804..5688a6f 100644 --- a/prompts/multi_agent/available_agents.txt +++ b/prompts/multi_agent/available_agents.txt @@ -5,3 +5,5 @@ {agents_list} 创建子智能体时请使用 `create_sub_agent` 工具，并传入合适的 `role_id`。 + +如果没有创建新智能体，无需使用list_agents工具，直接根据现有的子智能体来创建即可。 \ No newline at end of file diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index 22df386..187ab10 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -131,7 +131,13 @@ from .chat_flow_runtime import ( detect_malformed_tool_call, ) -from .chat_flow_task_support import process_sub_agent_updates, process_background_command_updates, process_multi_agent_master_messages +from .chat_flow_task_support import ( + process_sub_agent_updates, + process_background_command_updates, + process_multi_agent_master_messages, + inject_multi_agent_master_message, + _auto_message_type_for_multi_agent_subtype, +) 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 @@ -791,43 +797,21 @@ def _collect_pending_completion_notices(*, web_terminal, conversation_id: str) - "sort_key": update.get("updated_at") or time.time(), }) - # 3) 多智能体模式：把子智能体转发到主对话的 pending 消息也作为通知池项消费 - if getattr(web_terminal, "multi_agent_mode", False): - sub_manager = getattr(web_terminal, "sub_agent_manager", None) - if sub_manager: - state = sub_manager.get_multi_agent_state(conversation_id) - if state: - ma_messages = state.drain_master_messages() - ma_debug( - "collect_notices_drain_ma_messages", - conversation_id=conversation_id, - count=len(ma_messages), - previews=[str(m)[:200] for m in ma_messages], - ) - for msg_text in ma_messages: - notices.append({ - "kind": "multi_agent", - "message": msg_text, - "payload": { - "sub_agent_notice": True, - "message_source": "sub_agent", - "multi_agent_output": True, - }, - "sort_key": time.time(), - }) - notices.sort(key=lambda item: item.get("sort_key") or 0) return notices def _has_pending_completion_work(*, web_terminal, conversation_id: str) -> bool: - """是否还有运行中或待通知的后台任务（用于决定是否继续轮询）。""" + """是否还有运行中或待通知的传统后台任务（子智能体/后台 run_command）。 + + 多智能体模式下的 pending 消息由独立的 poll_multi_agent_notifications 处理， + 不再混入本函数。 + """ sub_manager = getattr(web_terminal, "sub_agent_manager", None) if sub_manager: announced = getattr(web_terminal, "_announced_sub_agent_tasks", set()) has_running_non_ma = False has_unnotified_non_ma = False - has_running_ma = False for task in sub_manager.tasks.values(): if not isinstance(task, dict): continue @@ -836,26 +820,13 @@ def _has_pending_completion_work(*, web_terminal, conversation_id: str) -> bool: status = task.get("status") multi_agent_flag = task.get("multi_agent_mode") or False if status not in TERMINAL_STATUSES.union({"terminated"}): - if multi_agent_flag: - has_running_ma = True - elif task.get("run_in_background"): + if not multi_agent_flag and task.get("run_in_background"): has_running_non_ma = True continue if not multi_agent_flag and task.get("run_in_background") and (task.get("task_id") not in announced) and not task.get("notified"): has_unnotified_non_ma = True if has_running_non_ma or has_unnotified_non_ma: return True - # 多智能体模式：只要有未消费的主对话消息，或还有未结束（非 idle/terminated）的实例，就继续轮询。 - # 即使全部实例都已 idle，只要 pending 消息尚未消费，也要继续，避免主任务结束后通知丢失。 - if getattr(web_terminal, "multi_agent_mode", False): - state = sub_manager.get_multi_agent_state(conversation_id) - if state: - if state.has_pending_master_messages(): - return True - if any(a.status not in {"idle", "terminated"} for a in state.list_all()): - return True - return False - return True bg_manager = getattr(web_terminal, "background_command_manager", None) if bg_manager: try: @@ -980,6 +951,364 @@ async def poll_completion_notifications(*, web_terminal, workspace, conversation await asyncio.sleep(5) +async def poll_multi_agent_notifications(*, web_terminal, workspace, conversation_id, client_sid, username): + """多智能体模式专用通知池。 + + 只在主任务结束后启动，负责两件事： + 1. 若主任务已空闲且 MultiAgentState 中有 pending_master_messages， + 一次性取出并触发新一轮主任务工作——即使仍有 running 子智能体也立即派发 + （主对话空闲时消息池有消息就该全部插入，设计中属情况2）。 + 2. 若 pending 为空且仍有 running 实例，循环等待下一次 push 新输出。 + + 重要：旧实现要求所有 running 实例退出再 drain，导致 ask_master 阻塞中的 + 子智能体永远等不到主对话回答而超时——现已修复为池优先。 + + 与 poll_completion_notifications 完全分离，避免多智能体消息和传统后台 + 通知竞争 task_manager 的单工作区互斥。 + """ + from .extensions import socketio + + max_wait_time = 3600 + start_wait = time.time() + + def sender(event_type, data): + try: + socketio.emit(event_type, data, room=f"user_{username}") + except Exception: + pass + + import threading as _threading + ma_debug( + "poll_ma_notifications_start", + conversation_id=conversation_id, + thread_id=_threading.get_ident(), + multi_agent_mode=getattr(web_terminal, "multi_agent_mode", False), + main_task_active_at_start=getattr(web_terminal, "_multi_agent_main_task_active", False), + ) + + tick_count = 0 + while (time.time() - start_wait) < max_wait_time: + tick_count += 1 + # 检查停止标志 + 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: + ma_debug("poll_ma_notifications_stop_requested", conversation_id=conversation_id, tick=tick_count) + break + + manager = getattr(web_terminal, "sub_agent_manager", None) + if not manager: + ma_debug("poll_ma_no_manager", conversation_id=conversation_id, tick=tick_count) + break + + state = manager.get_multi_agent_state(conversation_id) + if not state: + ma_debug("poll_ma_no_state", conversation_id=conversation_id, tick=tick_count) + break + + # 主任务又被唤醒了（用户发送新消息等），让主循环自己处理，本通知池退出 + if getattr(web_terminal, "_multi_agent_main_task_active", False): + ma_debug("poll_ma_notifications_main_task_active", conversation_id=conversation_id, tick=tick_count) + break + + # 状态快照 + all_insts = state.list_all() + statuses = [a.status for a in all_insts] + pending_count_now = len(state.pending_master_messages) + main_active = getattr(web_terminal, "_multi_agent_main_task_active", False) + ma_debug( + "poll_ma_tick", + conversation_id=conversation_id, + tick=tick_count, + instance_count=len(all_insts), + statuses=statuses, + pending_count=pending_count_now, + main_active=main_active, + ) + + # 1) 先检查 pending 池——主对话空闲时只要池里有消息就立即派发（情况2）。 + if state.has_pending_master_messages(): + pending = state.drain_master_messages() + ma_debug( + "poll_ma_notifications_dispatch", + conversation_id=conversation_id, + count=len(pending), + running_count=len([a for a in all_insts if a.status == "running"]), + previews=[str(m)[:200] for m in pending], + tick=tick_count, + ) + try: + await _dispatch_multi_agent_idle_messages( + web_terminal=web_terminal, + workspace=workspace, + sender=sender, + client_sid=client_sid, + username=username, + conversation_id=conversation_id, + messages=pending, + ) + ma_debug("poll_ma_dispatch_success", conversation_id=conversation_id, count=len(pending), tick=tick_count) + except Exception as exc: + debug_log(f"[MultiAgent] 分发 idle 消息失败: {exc}") + ma_debug( + "poll_ma_notifications_dispatch_error", + conversation_id=conversation_id, + error=str(exc), + error_type=type(exc).__name__, + tick=tick_count, + ) + # 分发后退出，由新任务结束后再决定是否继续轮询 + return + + # 2) 没有 pending 消息：检查还有哪些子智能体在运行，决定退出还是继续等 + running_agents = [a for a in all_insts if a.status == "running"] + if not running_agents: + ma_debug("poll_ma_notifications_no_work_end", conversation_id=conversation_id, tick=tick_count) + break + + ma_debug( + "poll_ma_notifications_wait_running", + conversation_id=conversation_id, + running_count=len(running_agents), + tick=tick_count, + ) + await asyncio.sleep(0.5) + + ma_debug("poll_ma_notifications_end", conversation_id=conversation_id, tick=tick_count) + + +async def _dispatch_multi_agent_idle_messages( + *, + web_terminal, + workspace, + sender, + client_sid, + username, + conversation_id, + messages: List[str], +): + """主智能体空闲时，把多智能体 pending 消息作为新一轮用户消息分发。 + + 所有消息先持久化到对话历史；前 N-1 条作为前置通知通过 socketio 推送； + 最后一条创建 task_type="notice" 任务触发 Team Leader 新一轮工作。 + """ + ma_debug( + "dispatch_ma_idle_enter", + conversation_id=conversation_id, + count=len(messages) if messages else 0, + previews=[str(m)[:150] for m in (messages or [])], + ) + if not messages: + ma_debug("dispatch_ma_idle_no_messages", conversation_id=conversation_id) + return + + from modules.multi_agent.state import parse_multi_agent_message + from .tasks import task_manager + + parsed_messages = [] + for msg_text in messages: + parsed = parse_multi_agent_message(msg_text) or {} + parsed_messages.append({ + "text": msg_text, + "display_name": parsed.get("display_name"), + "subtype": parsed.get("subtype"), + }) + + # 1) 前置通知全部写入历史 + emit给在线客户端（未后一条会由后续任务自己持久化， + # 避免重复产生两条相同 user 消息导致刷新后被渲染两遍）。 + ma_debug( + "dispatch_ma_idle_preceding", + conversation_id=conversation_id, + preceding_count=max(0, len(parsed_messages) - 1), + ) + for item in parsed_messages[:-1]: + inject_multi_agent_master_message( + web_terminal=web_terminal, + messages=None, + text=item["text"], + sender=sender, + conversation_id=conversation_id, + inline=False, + ) + + # 1.5) 最后一条只 emit 给在线客户端，不在这里持久化（后续 task handle_task_with_sender + # 会以这条文本作为 message 创建任务，并写入历史，从而避免重复写入。） + last = parsed_messages[-1] + message_source = "sub_agent" + last_emit_payload = { + "message": last["text"], + "content": last["text"], + "conversation_id": conversation_id, + "message_source": message_source, + "sub_agent_notice": True, + "multi_agent_message": True, + "multi_agent_display_name": last["display_name"], + "multi_agent_subtype": last["subtype"], + "auto_message_type": _auto_message_type_for_multi_agent_subtype(last["subtype"]), + "visibility": "chat", + "starts_work": False, + "metadata": { + "message_source": message_source, + "auto_message_type": _auto_message_type_for_multi_agent_subtype(last["subtype"]), + "multi_agent_message": True, + "multi_agent_display_name": last["display_name"], + "multi_agent_subtype": last["subtype"], + "visibility": "chat", + "starts_work": False, + }, + "timestamp": datetime.now().isoformat(), + } + try: + sender("user_message", last_emit_payload) + ma_debug( + "dispatch_ma_idle_last_emit", + conversation_id=conversation_id, + message_preview=last["text"][:200], + ) + except Exception as exc: + ma_debug( + "dispatch_ma_idle_last_emit_error", + conversation_id=conversation_id, + error=str(exc), + ) + + # 2) 前置通知的 socketio emit 已在上述注入 inject_multi_agent_master_message 内 + # 同时完成，避免在多智能体消息场景重复发送相同事件。 + + # 3) 最后一条触发新一轮工作 + ui_defaults = dict(_user_message_ui_defaults(message_source, auto_user_message_event=True)) + ui_defaults["starts_work"] = False + + workspace_id = getattr(workspace, "workspace_id", None) or "default" + host_mode = bool(getattr(workspace, "username", None) == "host") + session_data = { + "username": username, + "role": getattr(web_terminal, "user_role", "user"), + "is_api_user": getattr(web_terminal, "user_role", "") == "api", + "host_mode": host_mode, + "host_workspace_id": workspace_id if host_mode else None, + "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), + "message_source": message_source, + } + ma_auto_type = _auto_message_type_for_multi_agent_subtype(last["subtype"]) + # 重要：ui_defaults 默认会给出 visibility="compact"/starts_work=False， + # 多智能体主消息是正常聊天消息，必须在 **ui_defaults 之后覆写为 chat， + # 否则后端历史 metadata.visibility=compact 会让加载的消息走通知渲染。 + session_data["auto_user_message_event"] = True + session_data["auto_user_message_payload"] = { + **ui_defaults, + "message_source": message_source, + "sub_agent_notice": True, + "multi_agent_message": True, + "multi_agent_display_name": last["display_name"], + "multi_agent_subtype": last["subtype"], + "auto_message_type": ma_auto_type, + "starts_work": False, + "visibility": "chat", + "timestamp": datetime.now().isoformat(), + } + ma_debug( + "dispatch_ma_idle_before_create_task", + conversation_id=conversation_id, + last_text_preview=last["text"][:300], + workspace_id=workspace_id, + username=username, + ) + session_data["auto_user_message_payload"].setdefault("metadata", { + **ui_defaults, + "message_source": message_source, + "auto_message_type": ma_auto_type, + "multi_agent_message": True, + "multi_agent_display_name": last["display_name"], + "multi_agent_subtype": last["subtype"], + "starts_work": False, + "visibility": "chat", + }) + if len(parsed_messages) > 1: + session_data["preceding_user_notices"] = [ + { + "message": item["text"], + "payload": { + "message_source": "sub_agent", + "sub_agent_notice": True, + "multi_agent_message": True, + "multi_agent_display_name": item["display_name"], + "multi_agent_subtype": item["subtype"], + "auto_message_type": _auto_message_type_for_multi_agent_subtype(item["subtype"]), + "visibility": "chat", + "starts_work": False, + }, + } + for item in parsed_messages[:-1] + ] + + try: + rec = task_manager.create_chat_task( + username, + workspace_id, + last["text"], + [], + 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, + task_type="notice", + ) + except Exception as exc: + ma_debug( + "dispatch_ma_idle_create_task_exception", + conversation_id=conversation_id, + error=str(exc), + error_type=type(exc).__name__, + ) + raise + task_id_created = getattr(rec, "task_id", None) + ma_debug("dispatch_ma_idle_task_created", conversation_id=conversation_id, task_id=task_id_created) + + payload = { + "message": last["text"], + "content": last["text"], + "conversation_id": conversation_id, + "task_id": getattr(rec, "task_id", None), + "message_source": message_source, + "sub_agent_notice": True, + "multi_agent_message": True, + "multi_agent_display_name": last["display_name"], + "multi_agent_subtype": last["subtype"], + "visibility": "chat", + "starts_work": False, + "auto_message_type": _auto_message_type_for_multi_agent_subtype(last["subtype"]), + "metadata": { + "message_source": message_source, + "auto_message_type": _auto_message_type_for_multi_agent_subtype(last["subtype"]), + "multi_agent_message": True, + "multi_agent_display_name": last["display_name"], + "multi_agent_subtype": last["subtype"], + "visibility": "chat", + "starts_work": False, + }, + "timestamp": datetime.now().isoformat(), + } + sender("user_message", payload) + ma_debug( + "dispatch_ma_idle_sender_user_message", + conversation_id=conversation_id, + task_id=getattr(rec, "task_id", None), + payload_message_preview=last["text"][:200], + ) + ma_debug( + "dispatch_ma_idle_exit_ok", + conversation_id=conversation_id, + count=len(parsed_messages), + task_id=getattr(rec, "task_id", None), + ) + + async def handle_task_with_sender( terminal: WebTerminal, workspace: UserWorkspace, @@ -1008,6 +1337,8 @@ async def handle_task_with_sender( ma_debug( "handle_task_with_sender_start", conversation_id=conversation_id, + terminal_id=id(web_terminal), + manager_id=id(manager) if manager else None, message_preview=str(message)[:300] if message else None, pending_master_messages_count=pending_count, ) @@ -1058,6 +1389,7 @@ async def handle_task_with_sender( # 构建 user 消息来源与 metadata source_from_terminal = str(getattr(web_terminal, "_current_user_message_source", "user") or "user").strip().lower() user_message_source = source_from_terminal or "user" + multi_agent_meta: Dict[str, Any] = {} if auto_user_message_event: auto_payload = getattr(web_terminal, "_auto_user_message_payload", None) if isinstance(auto_payload, dict): @@ -1072,6 +1404,11 @@ async def handle_task_with_sender( user_message_source = "background_command" elif auto_payload.get("sub_agent_notice"): user_message_source = "sub_agent" + # 保留多智能体元数据，避免前端把子智能体输出渲染成普通 user 消息 + if auto_payload.get("multi_agent_message"): + multi_agent_meta["multi_agent_message"] = True + multi_agent_meta["multi_agent_display_name"] = auto_payload.get("multi_agent_display_name") + multi_agent_meta["multi_agent_subtype"] = auto_payload.get("multi_agent_subtype") if user_message_source not in _VALID_USER_MESSAGE_SOURCES: user_message_source = "user" @@ -1093,7 +1430,21 @@ async def handle_task_with_sender( # 如果是自动发送的user消息（子智能体/后台命令完成通知），添加标记 if auto_user_message_event: user_message_metadata["is_auto_generated"] = True - user_message_metadata["auto_message_type"] = "completion_notice" + if multi_agent_meta.get("multi_agent_message"): + # 多智能体消息：恢复专用 auto_message_type 与显示字段 + user_message_metadata["auto_message_type"] = _auto_message_type_for_multi_agent_subtype( + multi_agent_meta.get("multi_agent_subtype") + ) + user_message_metadata.update(multi_agent_meta) + # 多智能体消息是正常聊天消息，走多智能体动态专用渲染， + # 不应走 _user_message_ui_defaults 默认给出的 visibility="compact"（通知渲染）。 + user_message_metadata["visibility"] = "chat" + # 多智能体消息触发的是主智能体对子智能体输出的续答，不应显示新的 + # assistant 回复头部（Astrion/工作时间），因此 starts_work=false。 + # 前端通过单独的 multi_agent_message 标记恢复轮询。 + user_message_metadata["starts_work"] = False + else: + user_message_metadata["auto_message_type"] = "completion_notice" saved_user_message = web_terminal.context_manager.add_conversation( "user", message, @@ -1892,19 +2243,15 @@ async def handle_task_with_sender( if not hasattr(web_terminal, "_announced_sub_agent_tasks"): web_terminal._announced_sub_agent_tasks = set() - # 多智能体模式：检测是否还有运行中/idle 的多智能体实例。 - # 它们不是传统后台任务，不阻塞前端输入区，但需要通知池在主任务结束后 - # 继续消费 pending 消息并触发 Team Leader 响应。 + # 多智能体模式：检测是否还有真正运行中的实例，以及是否有尚未消费的 pending 消息。 + # idle 实例已完成当前输出、保留上下文等待后续指令；pending 消息会在主任务 + # 结束后由 poll_multi_agent_notifications 继续消费并触发新一轮工作。 + has_pending_ma_messages = False if getattr(web_terminal, "multi_agent_mode", False): - for task in manager.tasks.values(): - if ( - isinstance(task, dict) - and task.get("conversation_id") == conversation_id - and task.get("multi_agent_mode") - and task.get("status") not in TERMINAL_STATUSES.union({"terminated"}) - ): - has_running_multi_agent = True - break + state = manager.get_multi_agent_state(conversation_id) + if state: + has_running_multi_agent = any(a.status == "running" for a in state.list_all()) + has_pending_ma_messages = state.has_pending_master_messages() running_tasks = [ task for task in manager.tasks.values() @@ -1944,17 +2291,20 @@ async def handle_task_with_sender( # 与子智能体完全复用同一 waiting 事件（前端已有稳定处理链路） sender('sub_agent_waiting', _build_shared_waiting_payload(waiting_items)) - # 统一「通知池」轮询器：子智能体 + 后台 run_command + 多智能体 pending 消息 - # 合并为单一轮询链路，每轮一次性取出所有待通知项，避免逐条触发「工作 → 停止 → 再工作」循环。 - # 只 spawn 一个轮询器（无论是否同时存在子智能体与后台命令），从根本上消除 - # 两个轮询器同时 create_chat_task 撞「单工作区互斥」的问题。 - if has_running_sub_agents or has_running_background_commands or has_running_multi_agent: + # 传统后台任务通知池：只处理 run_in_background=True 的子智能体 / 后台 run_command。 + needs_completion_poll = ( + has_running_sub_agents + or has_running_background_commands + or _has_pending_completion_work( + web_terminal=web_terminal, conversation_id=conversation_id + ) + ) + if needs_completion_poll: ma_debug( "completion_poll_start", conversation_id=conversation_id, has_running_sub_agents=has_running_sub_agents, has_running_background_commands=has_running_background_commands, - has_running_multi_agent=has_running_multi_agent, ) def run_completion_poll(): import asyncio @@ -1972,7 +2322,60 @@ async def handle_task_with_sender( loop.close() socketio.start_background_task(run_completion_poll) - else: + + # 多智能体模式独立通知池：处理 running 实例 / idle 实例待消费的 pending 消息。 + # 与传统后台任务完全分离，避免两者竞争 create_chat_task 的单工作区互斥。 + ma_mode_flag = bool(getattr(web_terminal, "multi_agent_mode", False)) + needs_ma_poll = ma_mode_flag and (has_running_multi_agent or has_pending_ma_messages) + ma_debug( + "ma_poll_decision", + conversation_id=conversation_id, + multi_agent_mode=ma_mode_flag, + has_running_multi_agent=has_running_multi_agent, + has_pending_ma_messages=has_pending_ma_messages, + needs_ma_poll=needs_ma_poll, + ) + if needs_ma_poll: + ma_debug( + "ma_poll_start", + conversation_id=conversation_id, + has_running_multi_agent=has_running_multi_agent, + has_pending_ma_messages=has_pending_ma_messages, + ) + def run_ma_poll(): + import asyncio + import threading as _t + ma_debug( + "ma_poll_thread_enter", + conversation_id=conversation_id, + thread_id=_t.get_ident(), + ) + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(poll_multi_agent_notifications( + web_terminal=web_terminal, + workspace=workspace, + conversation_id=conversation_id, + client_sid=client_sid, + username=username + )) + except Exception as exc: + ma_debug( + "ma_poll_thread_exception", + conversation_id=conversation_id, + error=str(exc), + error_type=type(exc).__name__, + ) + else: + ma_debug("ma_poll_thread_exit_ok", conversation_id=conversation_id) + finally: + loop.close() + ma_debug("ma_poll_thread_finally", conversation_id=conversation_id) + + socketio.start_background_task(run_ma_poll) + + if not needs_completion_poll and not needs_ma_poll: ma_debug( "completion_poll_skip", conversation_id=conversation_id, @@ -2049,5 +2452,8 @@ async def handle_task_with_sender( # 沿用子智能体字段，确保前端直接走已验证通路 'has_running_sub_agents': has_running_completion_jobs, 'has_running_background_commands': has_running_background_commands, + # 多智能体模式：只要还有 running 实例或尚未消费的 pending 消息， + # 前端就应保持运行态并继续轮询。 + 'has_running_multi_agent': has_running_multi_agent or has_pending_ma_messages, 'pending_runtime_guidance_messages': pending_runtime_guidance_messages, }) diff --git a/server/chat_flow_task_support.py b/server/chat_flow_task_support.py index a13ecdc..38eb979 100644 --- a/server/chat_flow_task_support.py +++ b/server/chat_flow_task_support.py @@ -378,8 +378,13 @@ def inject_multi_agent_master_message( ctx_manager = getattr(web_terminal, "context_manager", None) if ctx_manager is not None: ctx_manager.add_conversation("user", raw, metadata=metadata) - except Exception: - pass + ma_debug( + "inject_ma_message_saved", + conversation_id=conversation_id, + text_preview=raw[:200], + ) + except Exception as exc: + ma_debug("inject_ma_message_save_error", error=str(exc)) if messages is not None: insert_index = len(messages) @@ -392,6 +397,13 @@ def inject_multi_agent_master_message( insert_index = end break messages.insert(insert_index, {"role": "user", "content": raw}) + ma_debug( + "inject_ma_message_inserted", + conversation_id=conversation_id, + insert_index=insert_index, + messages_len_after=len(messages), + text_preview=raw[:200], + ) if callable(sender): payload = { @@ -428,19 +440,35 @@ async def process_multi_agent_master_messages( after_tool_call_id: Optional[str] = None, ) -> int: """从 MultiAgentState 取出待插入主对话的消息并注入。返回注入条数。""" + manager = getattr(web_terminal, "sub_agent_manager", None) + ma_debug( + "process_ma_messages_enter", + multi_agent_mode=getattr(web_terminal, "multi_agent_mode", False), + has_manager=bool(manager), + manager_id=id(manager) if manager else None, + conversation_id=getattr(getattr(web_terminal, "context_manager", None), "current_conversation_id", None), + inline=inline, + after_tool_call_id=after_tool_call_id, + messages_len_before=len(messages) if messages else 0, + ) if not getattr(web_terminal, "multi_agent_mode", False): return 0 manager = getattr(web_terminal, "sub_agent_manager", None) if not manager: + ma_debug("process_ma_messages_no_manager") return 0 conversation_id = getattr(getattr(web_terminal, "context_manager", None), "current_conversation_id", None) if not conversation_id: + ma_debug("process_ma_messages_no_conversation_id") return 0 state = manager.get_multi_agent_state(conversation_id) if not state: + ma_debug("process_ma_messages_no_state", conversation_id=conversation_id) return 0 + ma_debug("process_ma_messages_got_state", conversation_id=conversation_id, state_id=id(state)) pending = state.drain_master_messages() if not pending: + ma_debug("process_ma_messages_empty", conversation_id=conversation_id, state_id=id(state)) return 0 debug_log(f"[MultiAgent] draining {len(pending)} pending master messages") ma_debug( diff --git a/server/tasks/helpers.py b/server/tasks/helpers.py index 0621a00..ef5f624 100644 --- a/server/tasks/helpers.py +++ b/server/tasks/helpers.py @@ -83,6 +83,7 @@ def _task_public_payload(rec: TaskRecord, *, include_title: bool = True) -> Dict "message_source": (rec.session_data or {}).get("message_source"), "goal_mode": bool((rec.session_data or {}).get("goal_mode")), "goal_progress": (rec.session_data or {}).get("goal_progress"), + "task_type": getattr(rec, "task_type", "chat"), } if include_title: payload["conversation_title"] = _conversation_title_for_task(rec) diff --git a/server/tasks/models.py b/server/tasks/models.py index c720780..b2374f0 100644 --- a/server/tasks/models.py +++ b/server/tasks/models.py @@ -55,6 +55,7 @@ class TaskRecord: "runtime_pending_queue", "runtime_guidance_queue", "last_cancel_at", + "task_type", ) def __init__( @@ -68,6 +69,7 @@ class TaskRecord: thinking_mode: Optional[bool], run_mode: Optional[str], max_iterations: Optional[int], + task_type: str = "chat", ): self.task_id = task_id self.username = username @@ -92,6 +94,7 @@ class TaskRecord: self.runtime_pending_queue: List[Dict[str, Any]] = [] self.runtime_guidance_queue: List[str] = [] self.last_cancel_at: Optional[float] = None + self.task_type = task_type class TaskManager: """线程内存版任务管理器，后续可替换为 Redis/DB。""" @@ -133,18 +136,22 @@ class TaskManager: message_source: Optional[str] = None, goal_mode: bool = False, skill_context_messages: Optional[List[Dict[str, str]]] = None, + task_type: str = "chat", ) -> TaskRecord: if run_mode: normalized = str(run_mode).lower() if normalized not in {"fast", "thinking", "deep"}: raise ValueError("run_mode 只支持 fast/thinking/deep") run_mode = normalized - # 单工作区互斥：禁止同一用户同一工作区并发任务 - existing = [t for t in self.list_tasks(username, workspace_id) if t.status in {"pending", "running"}] - if existing: - raise RuntimeError("当前工作区已有运行中的任务，请稍后再试。") + normalized_task_type = str(task_type or "chat").strip().lower() or "chat" + # 单工作区互斥：普通 chat 任务禁止同一用户同一工作区并发； + # notice（通知触发）任务允许与已完成的 chat 任务共存，用于后台通知重入。 + if normalized_task_type == "chat": + existing = [t for t in self.list_tasks(username, workspace_id) if t.status in {"pending", "running"} and getattr(t, "task_type", "chat") == "chat"] + if existing: + raise RuntimeError("当前工作区已有运行中的任务，请稍后再试。") task_id = str(uuid.uuid4()) - record = TaskRecord(task_id, username, workspace_id, message, conversation_id, model_key, thinking_mode, run_mode, max_iterations) + record = TaskRecord(task_id, username, workspace_id, message, conversation_id, model_key, thinking_mode, run_mode, max_iterations, task_type=normalized_task_type) # 记录当前 session 快照，便于后台线程内使用 if session_data is not None: snapshot = dict(session_data) diff --git a/static/src/app/methods/common.ts b/static/src/app/methods/common.ts index 868ebb5..d40403e 100644 --- a/static/src/app/methods/common.ts +++ b/static/src/app/methods/common.ts @@ -25,6 +25,8 @@ export const traceLog = (...args) => { function isGoalModeDebugEnabled() { if (typeof window === 'undefined') return false; try { + // 临时强制开启，用于调试多智能体消息轮询问题 + return true; const explicit = (window as any).__GOAL_MODE_DEBUG__; if (explicit === true || explicit === '1') return true; if (explicit === false || explicit === '0') return false; diff --git a/static/src/app/methods/taskPolling/lifecycle.ts b/static/src/app/methods/taskPolling/lifecycle.ts index b98af6e..bc6011d 100644 --- a/static/src/app/methods/taskPolling/lifecycle.ts +++ b/static/src/app/methods/taskPolling/lifecycle.ts @@ -372,8 +372,9 @@ export const lifecycleMethods = { }); const hasRunningSubAgents = !!data?.has_running_sub_agents; const hasRunningBackgroundCommands = !!data?.has_running_background_commands; - if (hasRunningSubAgents) { - debugLog('[TaskPolling] 任务完成，但仍有后台子智能体运行'); + const hasRunningMultiAgent = !!data?.has_running_multi_agent; + if (hasRunningSubAgents || hasRunningMultiAgent) { + debugLog('[TaskPolling] 任务完成，但仍有后台子智能体/多智能体运行'); } else { debugLog('[TaskPolling] 任务完成'); } @@ -381,11 +382,23 @@ export const lifecycleMethods = { // 同步处理状态更新 this.streamingMessage = false; this.stopRequested = false; - if (!hasRunningSubAgents) { + if (!hasRunningSubAgents && !hasRunningMultiAgent) { this.markLatestUserWorkCompleted(); } - if (hasRunningSubAgents) { + if (hasRunningMultiAgent) { + // 多智能体模式下主智能体已空闲，但仍有实例在运行，保持对话运行态继续接收后续消息； + // 不启动后台等待轮询，也不阻塞输入区（waitingForSubAgent=false）。 + // 但必须启动运行中任务探测，否则子智能体后续输出触发的新主任务无法被前端发现。 + jsonDebug('handleTaskComplete:hasRunningMultiAgent', { + taskInProgress: this.taskInProgress, + currentConversationId: this.currentConversationId + }); + this.taskInProgress = true; + this.waitingForSubAgent = false; + this.waitingForBackgroundCommand = hasRunningBackgroundCommands; + this.startMultiAgentTaskProbe(); + } else if (hasRunningSubAgents) { this.taskInProgress = true; this.waitingForSubAgent = true; this.waitingForBackgroundCommand = hasRunningBackgroundCommands; @@ -403,6 +416,7 @@ export const lifecycleMethods = { this.waitingForSubAgent = false; this.waitingForBackgroundCommand = false; this.stopWaitingTaskProbe(); + this.stopMultiAgentTaskProbe(); this.clearTaskState(); // 清理任务状态 this.$nextTick(() => { if (typeof this.tryAutoSendRuntimeQueuedMessages === 'function') { diff --git a/static/src/app/methods/taskPolling/probe.ts b/static/src/app/methods/taskPolling/probe.ts index 2a017e9..ca26b6b 100644 --- a/static/src/app/methods/taskPolling/probe.ts +++ b/static/src/app/methods/taskPolling/probe.ts @@ -1,5 +1,5 @@ // @ts-nocheck -import { debugLog } from '../common'; +import { debugLog, goalModeDebugLog } from '../common'; import { useTaskStore } from '../../../stores/task'; import { getMessageVisibility, messageStartsWork } from '../../../utils/messageVisibility'; import { @@ -86,6 +86,132 @@ export const probeMethods = { } }, 1000); }, + + // ---------- 多智能体任务探测 ---------- + stopMultiAgentTaskProbe() { + if (this.multiAgentTaskProbeTimer) { + debugNotifyLog('[DEBUG_NOTIFY][ui] stopMultiAgentTaskProbe'); + goalModeDebugLog('probe.stopMultiAgentTaskProbe', { + conversationId: this.currentConversationId, + taskInProgress: this.taskInProgress + }); + clearInterval(this.multiAgentTaskProbeTimer); + this.multiAgentTaskProbeTimer = null; + } + }, + startMultiAgentTaskProbe() { + if (this.multiAgentTaskProbeTimer) { + debugNotifyLog('[DEBUG_NOTIFY][ui] startMultiAgentTaskProbe:already-started'); + return; + } + debugNotifyLog('[DEBUG_NOTIFY][ui] startMultiAgentTaskProbe'); + keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] startMultiAgentTaskProbe', { + conversationId: this.currentConversationId + }); + goalModeDebugLog('probe.startMultiAgentTaskProbe', { + conversationId: this.currentConversationId, + taskInProgress: this.taskInProgress + }); + this.multiAgentTaskProbeTimer = setInterval(async () => { + if (!this.taskInProgress || !this.currentConversationId) { + debugNotifyLog('[DEBUG_NOTIFY][ui] multiAgentTaskProbe:stop-no-task-in-progress'); + goalModeDebugLog('probe.multiAgentTaskProbe_stop', { + reason: 'no_task_in_progress', + taskInProgress: this.taskInProgress, + currentConversationId: this.currentConversationId + }); + this.stopMultiAgentTaskProbe(); + return; + } + goalModeDebugLog('probe.multiAgentTaskProbe_tick', { + currentConversationId: this.currentConversationId + }); + const resumed = await this.tryResumeRunningTask(); + if (resumed) { + debugNotifyLog('[DEBUG_NOTIFY][ui] multiAgentTaskProbe:resumed-and-stop'); + keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] multiAgentTaskProbe:resumed-and-stop'); + this.stopMultiAgentTaskProbe(); + } + }, 1000); + }, + async tryResumeRunningTask() { + if (!this.currentConversationId) { + return false; + } + try { + const { useTaskStore } = await import('../../../stores/task'); + const taskStore = useTaskStore(); + if (taskStore.hasActiveTask) { + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeRunningTask:already-active', { + taskId: taskStore.currentTaskId, + status: taskStore.taskStatus + }); + return true; + } + + const resp = await fetch('/api/tasks'); + if (!resp.ok) { + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeRunningTask: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] tryResumeRunningTask: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?.status === 'pending') && + task?.conversation_id === this.currentConversationId && + (!(this.versioningHostMode || this.dockerProjectMode) || + !this.currentHostWorkspaceId || + task?.workspace_id === this.currentHostWorkspaceId) + ); + if (!runningTask?.task_id) { + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeRunningTask:no-matched-task'); + goalModeDebugLog('probe.tryResumeRunningTask_no_match', { + currentConversationId: this.currentConversationId, + tasksCount: tasks.length + }); + return false; + } + + debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeRunningTask:resume', { + task_id: runningTask.task_id + }); + goalModeDebugLog('probe.tryResumeRunningTask_resume', { + taskId: runningTask.task_id, + conversationId: this.currentConversationId + }); + keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] tryResumeRunningTask:resume', { + task_id: runningTask.task_id, + conversationId: this.currentConversationId + }); + 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; + } + }, + async inspectWaitingCompletionState() { const terminalStatuses = new Set(['completed', 'failed', 'timeout', 'terminated', 'cancelled']); let hasSubAgentPending = false; diff --git a/static/src/composables/useLegacySocket.ts b/static/src/composables/useLegacySocket.ts index a001294..da40cc2 100644 --- a/static/src/composables/useLegacySocket.ts +++ b/static/src/composables/useLegacySocket.ts @@ -19,6 +19,7 @@ import { io as createSocketClient } from 'socket.io-client'; import { renderLatexInRealtime } from './useMarkdownRenderer'; import { getMessageVisibility, messageStartsWork } from '../utils/messageVisibility'; +import { goalModeDebugLog } from '../app/methods/common'; export async function initializeLegacySocket(ctx: any) { try { @@ -907,10 +908,31 @@ export async function initializeLegacySocket(ctx: any) { // 用户消息（后台子智能体完成后自动触发） ctx.socket.on('user_message', (data) => { - if (ctx.usePollingMode && !ctx.waitingForSubAgent) { + const isMultiAgentMessage = !!( + data?.multi_agent_message || + data?.metadata?.multi_agent_message || + data?.multi_agent_output || + data?.metadata?.multi_agent_output + ); + goalModeDebugLog('legacy_socket_user_message_entry', { + usePollingMode: ctx.usePollingMode, + waitingForSubAgent: ctx.waitingForSubAgent, + isMultiAgentMessage, + currentConversationId: ctx.currentConversationId, + dataConversationId: data?.conversation_id, + message_source: data?.message_source || data?.metadata?.message_source, + starts_work: data?.starts_work, + auto_message_type: data?.auto_message_type || data?.metadata?.auto_message_type, + taskInProgress: ctx.taskInProgress + }); + // 多智能体消息在轮询模式下也需要处理，否则主智能体空闲时子智能体 + // 推送的消息无法触发前端恢复轮询，必须刷新页面才能看到。 + if (ctx.usePollingMode && !ctx.waitingForSubAgent && !isMultiAgentMessage) { + goalModeDebugLog('legacy_socket_user_message_early_return', { reason: 'polling_mode_no_waiting' }); return; } if (!data) { + goalModeDebugLog('legacy_socket_user_message_early_return', { reason: 'no_data' }); return; } if (data?.conversation_id && data.conversation_id !== ctx.currentConversationId) { @@ -1002,6 +1024,39 @@ export async function initializeLegacySocket(ctx: any) { } })(); } + // 多智能体消息没有 task_id，需要主动查找当前对话的运行中任务并恢复轮询。 + if (isMultiAgentMessage && ctx.usePollingMode && !data?.task_id) { + goalModeDebugLog('legacy_socket_user_message_ma_polling_start', { + currentConversationId: ctx.currentConversationId + }); + (async () => { + try { + const { useTaskStore } = await import('../stores/task'); + const taskStore = useTaskStore(); + const runningTask = await taskStore.loadRunningTask(ctx.currentConversationId); + goalModeDebugLog('legacy_socket_user_message_ma_polling_loaded', { + found: !!runningTask, + taskId: runningTask?.task_id, + taskStatus: runningTask?.status + }); + if (runningTask?.task_id) { + taskStore.resumeTask(runningTask.task_id); + goalModeDebugLog('legacy_socket_user_message_ma_polling_resumed', { + taskId: runningTask.task_id + }); + } + } catch (error) { + console.warn('恢复多智能体任务轮询失败', error); + goalModeDebugLog('legacy_socket_user_message_ma_polling_error', { + error: String(error) + }); + } + })(); + } + goalModeDebugLog('legacy_socket_user_message_end', { + taskInProgress: ctx.taskInProgress, + streamingMessage: ctx.streamingMessage + }); ctx.$forceUpdate(); ctx.conditionalScrollToBottom(); }); @@ -1617,9 +1672,13 @@ export async function initializeLegacySocket(ctx: any) { }); socketLog('任务完成', data); - // 多智能体模式下，主任务完成即视为可发送状态，后台子智能体是否 idle/running 不影响输入区 - if (ctx.multiAgentMode || !data.has_running_sub_agents) { - console.log('[DEBUG] 没有运行中的子智能体，重置任务状态'); + const hasRunningMultiAgent = !!data?.has_running_multi_agent; + // 多智能体模式下，若仍有运行中的多智能体实例，保持对话运行态，继续接收后续消息； + // 但此时主智能体已空闲，不应阻塞输入区（waitingForSubAgent=false）。 + if (hasRunningMultiAgent) { + console.log('[DEBUG] 多智能体实例仍在运行，保持任务状态但不阻塞输入'); + } else if (ctx.multiAgentMode || !data.has_running_sub_agents) { + console.log('[DEBUG] 没有运行中的子智能体/多智能体，重置任务状态'); if (ctx.waitingForSubAgent) { ctx.waitingForSubAgent = false; } diff --git a/static/src/stores/task.ts b/static/src/stores/task.ts index 8edef7e..609dae3 100644 --- a/static/src/stores/task.ts +++ b/static/src/stores/task.ts @@ -411,6 +411,10 @@ export const useTaskStore = defineStore('task', { startPolling(eventHandler?: (event: any) => void) { if (this.isPolling) { debugLog('[Task] 轮询已在运行'); + goalModeDebugLog('taskStore.startPolling_already_running', { + taskId: this.currentTaskId, + lastEventIndex: this.lastEventIndex + }); debugNotifyLog('[DEBUG_NOTIFY][task] startPolling:already-running', { taskId: this.currentTaskId, lastEventIndex: this.lastEventIndex @@ -420,10 +424,16 @@ export const useTaskStore = defineStore('task', { if (!this.currentTaskId) { debugLog('[Task] 没有任务ID，无法启动轮询'); + goalModeDebugLog('taskStore.startPolling_no_task_id'); return; } debugLog('[Task] 启动轮询:', this.currentTaskId); + goalModeDebugLog('taskStore.startPolling', { + taskId: this.currentTaskId, + lastEventIndex: this.lastEventIndex, + hasHandler: !!(eventHandler || (window as any).__taskEventHandler) + }); debugNotifyLog('[DEBUG_NOTIFY][task] startPolling', { taskId: this.currentTaskId, lastEventIndex: this.lastEventIndex @@ -470,8 +480,15 @@ export const useTaskStore = defineStore('task', { } = {} ) { if (!taskId) { + goalModeDebugLog('taskStore.resumeTask_no_task_id'); return; } + goalModeDebugLog('taskStore.resumeTask', { + incomingTaskId: taskId, + currentTaskId: this.currentTaskId, + isPolling: this.isPolling, + resetOffset: options?.resetOffset !== false + }); debugNotifyLog('[DEBUG_NOTIFY][task] resumeTask', { incomingTaskId: taskId, currentTaskId: this.currentTaskId, @@ -493,6 +510,11 @@ export const useTaskStore = defineStore('task', { this.pollingError = null; this.runtimeQueueSnapshotKey = ''; this.startPolling(options.eventHandler); + goalModeDebugLog('taskStore.resumeTask_started', { + taskId, + isPolling: this.isPolling, + lastEventIndex: this.lastEventIndex + }); }, stopPolling(reason = 'manual') { @@ -547,6 +569,7 @@ export const useTaskStore = defineStore('task', { async loadRunningTask(conversationId: string | null = null) { try { debugLog('[Task] 查找运行中的任务'); + goalModeDebugLog('taskStore.loadRunningTask_start', { conversationId }); const response = await fetch('/api/tasks'); if (!response.ok) { @@ -568,6 +591,11 @@ export const useTaskStore = defineStore('task', { if (runningTask) { debugLog('[Task] 发现运行中的任务:', runningTask.task_id); + goalModeDebugLog('taskStore.loadRunningTask_found', { + taskId: runningTask.task_id, + status: runningTask.status, + conversationId: runningTask.conversation_id + }); this.currentTaskId = runningTask.task_id; this.taskStatus = runningTask.status; @@ -603,9 +631,14 @@ export const useTaskStore = defineStore('task', { } debugLog('[Task] 没有运行中的任务'); + goalModeDebugLog('taskStore.loadRunningTask_not_found', { conversationId }); return null; } catch (error) { console.error('[Task] 加载运行中任务失败:', error); + goalModeDebugLog('taskStore.loadRunningTask_error', { + conversationId, + error: String(error) + }); return null; } },