diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index 08a8954..2ca7b53 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -1704,8 +1704,26 @@ class MainTerminalToolsExecutionMixin: task_id=task_id, timeout_seconds=arguments.get("timeout_seconds") ) - # 合并结果 - result.update(wait_result) + # 合并结果：保留创建元数据，使用执行结果作为主体展示， + # 避免 wait_result 里的 success=False + 旧 message 导致 + # 「create_sub_agent 失败：子智能体 X 已创建」的误导性文案。 + creation_meta = { + "agent_id": result.get("agent_id"), + "task_id": result.get("task_id"), + "deliverables_dir": result.get("deliverables_dir"), + "run_in_background": False, + } + execution_message = ( + wait_result.get("message") + or wait_result.get("system_message") + or result.get("message") + ) + result = { + **result, + **wait_result, + **creation_meta, + "message": execution_message, + } # 阻塞式执行不需要额外插入 system 消息 result.pop("system_message", None) # 标记已通知，避免后续轮询再插入 system 消息 diff --git a/modules/personalization_manager.py b/modules/personalization_manager.py index 9161c7a..62e04b4 100644 --- a/modules/personalization_manager.py +++ b/modules/personalization_manager.py @@ -91,7 +91,6 @@ DEFAULT_PERSONALIZATION_CONFIG: Dict[str, Any] = { "shallow_compress_keep_user_turn_tools": None, # 保留最近N次用户输入后的工具不压缩（默认3） "deep_compress_trigger_tokens": None, "deep_compress_form": "file", # 深压缩形式：file-生成文件 / inject-直接注入文件全文 - "deep_compress_behavior": "continue", # 手动压缩行为：continue-注入并触发请求 / wait-仅插入等待用户 "silent_tool_disable": True, # 禁用工具时不向模型插入提示（默认开启） "enhanced_tool_display": True, # 增强工具显示 "compact_message_display": "full", # 简略消息显示：full-完整原始内容 / brief-一行概要 @@ -396,13 +395,6 @@ def sanitize_personalization_payload( else: base["deep_compress_form"] = "file" - # 手动压缩行为：continue（注入并触发请求）/ wait（仅插入等待用户） - deep_behavior = data.get("deep_compress_behavior", base.get("deep_compress_behavior")) - if isinstance(deep_behavior, str) and deep_behavior.strip().lower() in ("continue", "wait"): - base["deep_compress_behavior"] = deep_behavior.strip().lower() - else: - base["deep_compress_behavior"] = "continue" - if "shallow_compress_trigger_tool_calls_interval" in data: base["shallow_compress_trigger_tool_calls_interval"] = _sanitize_optional_int( data.get("shallow_compress_trigger_tool_calls_interval"), diff --git a/modules/sub_agent/manager.py b/modules/sub_agent/manager.py index bb68db9..865912c 100644 --- a/modules/sub_agent/manager.py +++ b/modules/sub_agent/manager.py @@ -269,11 +269,28 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi while time.time() < deadline: self.reconcile_task_states() - if task.get("status") in TERMINAL_STATUSES: - return task.get("final_result") or {"success": False, "status": task.get("status")} + status = task.get("status") + + # 已到达终态：返回最终结果（持续 reconcile 直到 final_result 就绪） + if status in TERMINAL_STATUSES or status == "terminated": + if task.get("final_result"): + return task["final_result"] + # 终态但 final_result 尚未写入，短暂等待后重试 + time.sleep(SUB_AGENT_STATUS_POLL_INTERVAL) + self.reconcile_task_states() + if task.get("final_result"): + return task["final_result"] + return {"success": False, "status": status, "message": "子智能体已结束，但未获取到结果。"} + + # asyncio Task 已结束但状态可能还没同步：等待 final_result 就绪 if running_task and running_task.done(): self.reconcile_task_states() - return task.get("final_result") or {"success": False, "status": task.get("status")} + if task.get("final_result"): + return task["final_result"] + # 结果尚未落盘，继续轮询，避免把「已创建」误判为失败 + time.sleep(SUB_AGENT_STATUS_POLL_INTERVAL) + continue + time.sleep(SUB_AGENT_STATUS_POLL_INTERVAL) return self._handle_timeout(task) @@ -317,23 +334,44 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi *, agent_ids: Optional[List[int]] = None, ) -> Dict: - """获取指定子智能体的详细状态。""" + """获取指定子智能体的详细状态。 + + 对于已结束（completed/failed/timeout/terminated）的子智能体，同样返回其 + 最终状态，而不是返回「不存在」。 + """ if not agent_ids: return {"success": False, "error": "必须指定至少一个agent_id"} + def _find_task_by_agent_id(aid: int): + # 先查运行中/待运行的任务 + task = self._select_task(None, aid) + if task: + return task + # 再查已结束的任务（按创建时间取最新一条） + candidates = [ + t for t in self.tasks.values() + if t.get("agent_id") == aid + ] + if not candidates: + return None + candidates.sort(key=lambda item: item.get("created_at", 0), reverse=True) + return candidates[0] + results = [] for agent_id in agent_ids: - task = self._select_task(None, agent_id) + task = _find_task_by_agent_id(agent_id) if not task: results.append({ "agent_id": agent_id, "found": False, - "error": "未找到对应的子智能体任务" + "error": "子智能体不存在", }) continue - if task.get("status") not in TERMINAL_STATUSES.union({"terminated"}): + status = task.get("status") + if status not in TERMINAL_STATUSES.union({"terminated"}): self._check_task_status(task) + status = task.get("status") stats = {} stats_file = Path(task.get("stats_file", "")) @@ -348,7 +386,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi "agent_id": agent_id, "found": True, "task_id": task["task_id"], - "status": task["status"], + "status": status, "summary": task.get("summary"), "created_at": task.get("created_at"), "updated_at": task.get("updated_at"), diff --git a/server/chat_flow.py b/server/chat_flow.py index 5125426..170bdc7 100644 --- a/server/chat_flow.py +++ b/server/chat_flow.py @@ -57,6 +57,7 @@ from config.model_profiles import get_model_context_window, get_model_profile from .auth_helpers import api_login_required, resolve_admin_policy, get_current_user_record, get_current_username from .context import with_terminal, get_gui_manager, get_upload_guard, build_upload_error_response, ensure_conversation_loaded, reset_system_state, get_user_resources, get_or_create_usage_tracker +from .work_timer import finalize_conversation_work_timer from .utils_common import ( build_review_lines, debug_log, @@ -224,37 +225,15 @@ def process_message_task(terminal: WebTerminal, message: str, images, sender, cl f"has_running_sub_agents={has_running_sub_agents}, " f"has_running_background_commands={has_running_background_commands}" ) - # 取消时立即把当前对话最后一条 working 的用户消息 work_timer 标记为完成， - # 避免刷新页面后仍显示"工作中"。 + # 取消时：仅当对话真正空闲（没有后台子智能体/后台命令还在跑） + # 才把 work_timer 标记为完成，否则后台工作继续期间计时器应保持运行。 try: - if terminal and getattr(terminal, "context_manager", None): - cm = terminal.context_manager - history = getattr(cm, "conversation_history", None) or [] - for msg in reversed(history): - if not isinstance(msg, dict) or msg.get("role") != "user": - continue - metadata = msg.get("metadata") or {} - timer = metadata.get("work_timer") - if not isinstance(timer, dict) or timer.get("status") != "working": - continue - started_at = timer.get("started_at") or msg.get("timestamp") or datetime.now().isoformat() - try: - start_ts = datetime.fromisoformat(str(started_at).replace("Z", "+00:00")).timestamp() - duration_ms = int(max(0.0, (time.time() - start_ts) * 1000.0)) - except Exception: - duration_ms = timer.get("duration_ms", 0) or 0 - timer.update({ - "status": "completed", - "started_at": started_at, - "finished_at": datetime.now().isoformat(), - "duration_ms": duration_ms, - }) - cm.auto_save_conversation(force=True) - debug_log( - f"[ChatFlow] 取消时已将 work_timer 标记为完成: " - f"conversation_id={getattr(cm, 'current_conversation_id', None)}, started_at={started_at}" - ) - break + if terminal and conversation_id: + finalized = finalize_conversation_work_timer( + terminal, + conversation_id, + finished_at=datetime.now().isoformat(), + ) except Exception as exc: debug_log(f"[ChatFlow] 取消时标记 work_timer 完成失败: {exc}") # task_stopped 事件统一在 _run_chat_task finally 中发送，避免重复 diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index be6fe9e..eb7372d 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -53,6 +53,7 @@ from config.model_profiles import get_model_context_window, get_model_profile from .auth_helpers import api_login_required, resolve_admin_policy, get_current_user_record, get_current_username from .context import with_terminal, get_gui_manager, get_upload_guard, build_upload_error_response, ensure_conversation_loaded, reset_system_state, get_user_resources, get_or_create_usage_tracker +from .work_timer import finalize_conversation_work_timer from .utils_common import ( build_review_lines, debug_log, @@ -165,7 +166,12 @@ def _user_message_ui_defaults(source: str, *, auto_user_message_event: bool = Fa return {"visibility": "hidden", "starts_work": False} if normalized in {"guidance", "notify"}: return {"visibility": "compact", "starts_work": False} - if normalized in {"goal_review", "compression", "compression_handoff", "sub_agent", "background_command"}: + # 后台完成通知 / 运行期压缩续接：本质属于当前这轮工作的延续， + # 不应暂停上一轮计时器、也不应开启新的工作头与计时器。 + if normalized in {"sub_agent", "background_command", "compression", "compression_handoff"}: + return {"visibility": "compact", "starts_work": False} + # 目标审核续命：确实开启新一轮工作，保持 starts_work=True。 + if normalized == "goal_review": return {"visibility": "compact", "starts_work": True} if normalized == "presend": return {"visibility": "chat", "starts_work": True} @@ -342,6 +348,52 @@ def _record_hidden_versioning_checkpoint_after_run( debug_log(f"[Versioning] 记录快照异常: {exc}") +def _persist_and_echo_preceding_notice( + *, + web_terminal, + sender, + conversation_id: str, + message: str, + payload: Dict[str, Any], +) -> None: + """预写一条「前置完成通知」：写入对话历史 + socketio 回显（不触发新一轮工作）。 + + 用于「通知池」一次取出多条时，前 N-1 条随同一后续任务一起呈现。 + 这些通知必须落历史，否则模型看不到、刷新后也会丢失。 + """ + message = (message or "").strip() + if not message: + return + src = str(payload.get("message_source") or "sub_agent").strip().lower() + if src not in _VALID_USER_MESSAGE_SOURCES: + src = "sub_agent" + ui_defaults = _user_message_ui_defaults(src, auto_user_message_event=True) + # 前置通知渲染为完成态气泡：不开启独立工作计时器 + ui_defaults = {**ui_defaults, "starts_work": False} + metadata = { + "message_source": src, + "is_auto_generated": True, + "auto_message_type": "completion_notice", + **ui_defaults, + } + try: + cm = getattr(web_terminal, "context_manager", None) + if cm is not None: + cm.add_conversation("user", message, metadata=metadata) + except Exception as exc: + echo_payload = { + "message": message, + "conversation_id": conversation_id, + "message_source": src, + **ui_defaults, + **{k: v for k, v in (payload or {}).items() if k not in {"message", "conversation_id"}}, + } + echo_payload["starts_work"] = False + echo_payload["metadata"] = {**metadata} + sender("user_message", echo_payload) + except Exception as exc: + pass + async def _dispatch_completion_user_notice( *, web_terminal, @@ -352,9 +404,16 @@ async def _dispatch_completion_user_notice( conversation_id: str, user_message: str, extra_payload: Optional[Dict[str, Any]] = None, + preceding_notices: Optional[List[Dict[str, Any]]] = None, ): - """复用子智能体完成后的 user 代发机制。""" + """复用子智能体完成后的 user 代发机制。 + + preceding_notices：本批「通知池」里除最后一条以外的前置通知列表， + 每项为 {"message": str, "payload": dict}。这些会先于触发消息写入历史/事件流， + 但不各自触发新一轮工作；仅最后一条 user_message 作为后续任务的触发消息。 + """ extra_payload = extra_payload or {} + preceding_notices = preceding_notices or [] message_source = str(extra_payload.get("message_source") or "").strip().lower() if not message_source: if extra_payload.get("runtime_mode_notice"): @@ -367,6 +426,18 @@ async def _dispatch_completion_user_notice( message_source = "sub_agent" if message_source not in _VALID_USER_MESSAGE_SOURCES: message_source = "user" + + # 先把前置通知写入历史并 socketio 回显（在线客户端即时可见）。 + # 轮询客户端则通过后续任务事件流回放（见 _run_chat_task 的 preceding_user_notices 注入）。 + for item in preceding_notices: + _persist_and_echo_preceding_notice( + web_terminal=web_terminal, + sender=sender, + conversation_id=conversation_id, + message=str(item.get("message") or ""), + payload=dict(item.get("payload") or {}), + ) + try: from .tasks import task_manager workspace_id = getattr(workspace, "workspace_id", None) or "default" @@ -394,6 +465,16 @@ async def _dispatch_completion_user_notice( **dict(extra_payload or {}), **ui_defaults, } + # 轮询客户端通过后续任务事件流回放前置通知（在线客户端已由上面的 socketio 回显覆盖）。 + if preceding_notices: + session_data["preceding_user_notices"] = [ + { + "message": str(item.get("message") or ""), + "payload": dict(item.get("payload") or {}), + } + for item in preceding_notices + if str(item.get("message") or "").strip() + ] rec = task_manager.create_chat_task( username, workspace_id, @@ -468,347 +549,271 @@ def _build_shared_waiting_payload(items: List[Dict[str, Any]]) -> Dict[str, Any] 'tasks': normalized, } -async def poll_sub_agent_completion(*, web_terminal, workspace, conversation_id, client_sid, username): - """后台轮询子智能体完成状态，完成后触发新一轮对话""" +def _build_sub_agent_notice_text(update: Dict[str, Any], *, summary_override: Optional[str] = None) -> str: + """根据子智能体完成结果构建一条 user 通知文本（统一 [系统通知|sub_agent] 前缀）。 + + update 通常是 task 的 final_result，其中 system_message 已经是一条完整的格式化通知 + （含 ✅/❌ 前缀、统计、运行秒数、summary、交付目录）。这里直接复用它，只补统一前缀， + 不再在外层重复拼 agent/运行秒数/交付目录，避免内容出现两遍。 + """ + system_message = (update.get("system_message") or "").strip() + if not system_message: + # 兜底：极少数情况下 final_result 没有 system_message，用最小信息拼一条 + agent_id = update.get("agent_id") + summary = summary_override or update.get("summary") or update.get("message") or "" + deliverables_dir = update.get("deliverables_dir", "") or "" + lines = [f"子智能体{agent_id} ({summary}) 已完成任务。"] + msg = update.get("message") + if msg: + lines.append(str(msg)) + if deliverables_dir: + lines.append(f"交付目录：{deliverables_dir}") + system_message = "\n\n".join(lines) + return f"[系统通知|sub_agent]\n{system_message}" + + +def _collect_pending_completion_notices(*, web_terminal, conversation_id: str) -> List[Dict[str, Any]]: + """从子智能体 + 后台 run_command 两路统一取出所有待通知项（已按时间排序）。 + + 返回的每一项形如： + { + "kind": "sub_agent" | "background_command", + "message": , + "payload": <前端事件 extra payload>, + "sort_key": <用于跨两路排序的时间戳>, + } + 取出时即就地标记 notified，确保不会被下一轮重复消费。 + """ + notices: List[Dict[str, Any]] = [] + + # 1) 子智能体 + sub_manager = getattr(web_terminal, "sub_agent_manager", None) + if sub_manager: + if not hasattr(web_terminal, "_announced_sub_agent_tasks"): + web_terminal._announced_sub_agent_tasks = set() + # 关键：先 reconcile 刷新真实状态，然后**直接遍历 tasks** 找终态+未通知的任务。 + # 不依赖 poll_updates() —— 它只返回「本次调用里恰好 running→terminal」的跃迁项， + # 任务一旦被提前 reconcile 成 completed，跃迁就丢了，poll_updates 永远返回空。 + try: + sub_manager.reconcile_task_states(conversation_id=conversation_id) + except Exception: + pass + candidates = [ + t for t in sub_manager.tasks.values() + if isinstance(t, dict) + and t.get("conversation_id") == conversation_id + and t.get("status") in TERMINAL_STATUSES # completed/failed/timeout（不含 terminated） + and not t.get("notified") + and t.get("task_id") not in web_terminal._announced_sub_agent_tasks + ] + candidates.sort(key=lambda t: t.get("updated_at") or t.get("created_at") or 0) + for task_info in candidates: + task_id = task_info.get("task_id") + final_result = task_info.get("final_result") + if not isinstance(final_result, dict): + # 兜底：状态已是终态但 final_result 缺失，主动取一次 + try: + final_result = sub_manager._check_task_status(task_info) + except Exception: + pass + if not isinstance(final_result, dict): + final_result = {} + # 就地标记已通知 + if task_id: + web_terminal._announced_sub_agent_tasks.add(task_id) + task_info["notified"] = True + task_info["updated_at"] = time.time() + + agent_id = task_info.get("agent_id") + summary = task_info.get("summary") or "" + notice_text = _build_sub_agent_notice_text( + final_result, + summary_override=summary, + ) + notices.append({ + "kind": "sub_agent", + "message": notice_text, + "payload": { + "sub_agent_notice": True, + "message_source": "sub_agent", + "agent_id": agent_id, + "task_id": task_id, + }, + "sort_key": task_info.get("updated_at") or task_info.get("created_at") or time.time(), + }) + try: + sub_manager._save_state() + except Exception: + pass + + # 2) 后台 run_command + bg_manager = getattr(web_terminal, "background_command_manager", None) + if bg_manager: + try: + bg_updates = bg_manager.poll_updates(conversation_id=conversation_id) + except Exception: + bg_updates = [] + for update in bg_updates or []: + command_id = update.get("command_id") + command = update.get("command") or "" + output = update.get("output") or "" + return_code = update.get("return_code") + try: + bg_manager.mark_notified(str(command_id)) + except Exception: + pass + # 统一 [系统通知|background_command] 前缀；正文给出命令、退出码、输出 + header = "后台指令已完成。" + if command: + header += f"\n\n命令：{command}" + if return_code is not None: + header += f"\n退出码：{return_code}" + body = output if output else "[no_output]" + message_text = f"[系统通知|background_command]\n{header}\n\n输出：\n{body}" + notices.append({ + "kind": "background_command", + "message": message_text, + "payload": { + # 与子智能体完成通知完全复用同一前端通道/处理逻辑 + "sub_agent_notice": True, + "message_source": "background_command", + "background_command_notice": True, + }, + "sort_key": update.get("updated_at") or time.time(), + }) + + notices.sort(key=lambda item: item.get("sort_key") or 0) + + +def _has_pending_completion_work(*, web_terminal, conversation_id: str) -> bool: + """是否还有运行中或待通知的后台任务（用于决定是否继续轮询）。""" + sub_manager = getattr(web_terminal, "sub_agent_manager", None) + if sub_manager: + announced = getattr(web_terminal, "_announced_sub_agent_tasks", set()) + for task in sub_manager.tasks.values(): + if not isinstance(task, dict): + continue + if not task.get("run_in_background"): + continue + if task.get("conversation_id") != conversation_id: + continue + status = task.get("status") + if status not in TERMINAL_STATUSES.union({"terminated"}): + return True # 仍在运行 + if status != "terminated" and (task.get("task_id") not in announced) and not task.get("notified"): + return True # 已完成但未通知 + bg_manager = getattr(web_terminal, "background_command_manager", None) + if bg_manager: + try: + if bg_manager.has_pending_for_conversation(conversation_id): + return True + except Exception: + pass + return False + + +async def poll_completion_notifications(*, web_terminal, workspace, conversation_id, client_sid, username): + """统一的“通知池”轮询器。 + + 把子智能体完成通知与后台 run_command 完成通知合并到同一条链路： + - 每轮把两路所有待通知项一次性取出（池化），按时间排序； + - 这一批里前 N-1 条作为「预写通知」随同一个后续任务一起注入（写历史 + 事件流 + socketio）， + 不各自触发新一轮工作； + - 仅最后 1 条作为后续 chat task 的触发消息，启动一次「工作 → 停止」循环； + - 该后续任务结束后回到 handle_task_with_sender 结尾重新 spawn 本轮询器，继续消费剩余通知。 + + 这样多个后台任务同时完成时，会被合并成一次（而非每条都触发一轮停止-再工作）。 + """ from .extensions import socketio - manager = getattr(web_terminal, "sub_agent_manager", None) - if not manager: - debug_log("[SubAgent] poll_sub_agent_completion: manager 不存在") + sub_manager = getattr(web_terminal, "sub_agent_manager", None) + bg_manager = getattr(web_terminal, "background_command_manager", None) + if not sub_manager and not bg_manager: return if not hasattr(web_terminal, "_announced_sub_agent_tasks"): web_terminal._announced_sub_agent_tasks = set() - max_wait_time = 3600 # 最多等待1小时 + max_wait_time = 3600 # 最多等待 1 小时 start_wait = time.time() - debug_log(f"[SubAgent] 开始后台轮询，conversation_id={conversation_id}, username={username}") - - # 创建 sender 函数，用于发送 socket 事件 def sender(event_type, data): try: socketio.emit(event_type, data, room=f"user_{username}") - debug_log(f"[SubAgent] 发送事件: {event_type}") - except Exception as e: - debug_log(f"[SubAgent] 发送事件失败: {event_type}, 错误: {e}") + except Exception: + pass + loop_count = 0 while (time.time() - start_wait) < max_wait_time: - debug_log(f"[SubAgent] 轮询检查...") - + loop_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: - debug_log("[SubAgent] 用户请求停止，终止轮询") break # 若主对话仍在工具循环中，暂不消费完成事件，避免抢占 system 消息插入 if getattr(web_terminal, "_tool_loop_active", False): - debug_log("[SubAgent] 主对话工具循环中，延迟后台轮询发送 user 消息") await asyncio.sleep(1) continue - updates = manager.poll_updates() - debug_log(f"[SubAgent] poll_updates 返回 {len(updates)} 个更新") + notices = _collect_pending_completion_notices( + web_terminal=web_terminal, + conversation_id=conversation_id, + ) - for update in updates: - agent_id = update.get("agent_id") - summary = update.get("summary") - result_summary = update.get("result_summary") or update.get("message", "") - deliverables_dir = update.get("deliverables_dir", "") - status = update.get("status") - task_id = update.get("task_id") - task_info = manager.tasks.get(task_id) if task_id else None - task_conv_id = task_info.get("conversation_id") if isinstance(task_info, dict) else None - if task_conv_id and task_conv_id != conversation_id: - debug_log(f"[SubAgent] 跳过非当前对话任务: task={task_id} conv={task_conv_id} current={conversation_id}") - continue - if task_id and task_info is None: - debug_log(f"[SubAgent] 找不到任务详情，跳过: task={task_id}") - continue - if status == "terminated" or (isinstance(task_info, dict) and task_info.get("notified")): - debug_log(f"[SubAgent] 跳过已终止/已通知任务: task={task_id} status={status}") - continue - - debug_log(f"[SubAgent] 子智能体{agent_id}完成，状态: {status}") - - # 构建 user 消息（后台完成时才发送） - prefix = "这是一句系统自动发送的user消息，用于通知你子智能体已经运行完成" - runtime_line = "" - elapsed_seconds = update.get("runtime_seconds") - if elapsed_seconds is None: - elapsed_seconds = update.get("elapsed_seconds") - if status == "completed" and isinstance(elapsed_seconds, (int, float)): - runtime_line = f"\n\n运行了{int(round(elapsed_seconds))}秒" - user_message = f"""{prefix} - -子智能体{agent_id} ({summary}) 已完成任务。 - -{result_summary} -{runtime_line} - -交付目录：{deliverables_dir}""" - - debug_log(f"[SubAgent] 准备发送 user_message: {user_message[:100]}...") - - has_remaining = False - remaining_count = 0 - try: - if task_id: - web_terminal._announced_sub_agent_tasks.add(task_id) - if isinstance(task_info, dict): - task_info["notified"] = True - task_info["updated_at"] = time.time() - try: - manager._save_state() - except Exception as exc: - debug_log(f"[SubAgent] 保存通知状态失败: {exc}") - - # 计算剩余子智能体状态（用于前端清理等待标记） - if not hasattr(web_terminal, "_announced_sub_agent_tasks"): - web_terminal._announced_sub_agent_tasks = set() - announced = web_terminal._announced_sub_agent_tasks - running_tasks = [ - task for task in manager.tasks.values() - if isinstance(task, dict) - and task.get("status") not in TERMINAL_STATUSES.union({"terminated"}) - and task.get("run_in_background") - and task.get("conversation_id") == conversation_id - ] - pending_notice_tasks = [ - task for task in manager.tasks.values() - if isinstance(task, dict) - and task.get("status") in TERMINAL_STATUSES.union({"terminated"}) - and task.get("run_in_background") - and task.get("conversation_id") == conversation_id - and task.get("task_id") not in announced - and not task.get("notified") - ] - remaining_count = len(running_tasks) + len(pending_notice_tasks) - has_remaining = remaining_count > 0 - 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, - 'message_source': 'sub_agent', - 'has_running_sub_agents': has_remaining, - 'remaining_count': remaining_count, - }, - ) - except Exception as e: - debug_log(f"[SubAgent] 创建后台任务失败，回退直接执行: {e}") - 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, - 'message_source': 'sub_agent', - 'has_running_sub_agents': has_remaining, - 'remaining_count': remaining_count, - }, - ) - - return # 只处理第一个完成的子智能体 - - # 检查是否还有运行中的任务 - running_tasks = [ - task for task in manager.tasks.values() - if task.get("status") not in {"completed", "failed", "timeout", "terminated"} - and task.get("run_in_background") - and task.get("conversation_id") == conversation_id - ] - - debug_log(f"[SubAgent] 当前还有 {len(running_tasks)} 个运行中的任务") - - if not running_tasks: - debug_log("[SubAgent] 所有子智能体已完成") - # 若状态已提前被更新为终态（poll_updates 返回空），补发完成提示 - completed_tasks = [ - task for task in manager.tasks.values() - if task.get("status") in {"completed", "failed", "timeout"} - and task.get("run_in_background") - and task.get("conversation_id") == conversation_id - and not task.get("notified") - ] - if completed_tasks: - completed_tasks.sort( - key=lambda item: item.get("updated_at") or item.get("created_at") or 0, - reverse=True - ) - task = completed_tasks[0] - agent_id = task.get("agent_id") - summary = task.get("summary") or "" - final_result = task.get("final_result") or {} - result_summary = ( - final_result.get("message") - or final_result.get("result_summary") - or final_result.get("system_message") - or "" - ) - deliverables_dir = final_result.get("deliverables_dir") or task.get("deliverables_dir") or "" - status = final_result.get("status") or task.get("status") - debug_log(f"[SubAgent] 补发完成提示: task={task.get('task_id')} status={status}") - - user_message = f"""子智能体{agent_id} ({summary}) 已完成任务。 - -{result_summary} - -交付目录：{deliverables_dir}""" - - try: - task_id = task.get("task_id") - if task_id: - web_terminal._announced_sub_agent_tasks.add(task_id) - if isinstance(task, dict): - task["notified"] = True - task["updated_at"] = time.time() - try: - manager._save_state() - except Exception as exc: - debug_log(f"[SubAgent] 保存通知状态失败: {exc}") - sender('user_message', { - 'message': user_message, - 'conversation_id': conversation_id - }) - from .tasks import task_manager - 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), - } - # 标记为自动发送的user消息（子智能体完成通知） - session_data["auto_user_message_event"] = True - session_data["auto_user_message_payload"] = { - "sub_agent_notice": True, - "message_source": "sub_agent", - } - 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, - ) - debug_log(f"[SubAgent] 补发通知创建后台任务: task_id={rec.task_id}") - except Exception as e: - debug_log(f"[SubAgent] 补发通知创建后台任务失败，回退直接执行: {e}") - 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"[SubAgent] 补发完成提示失败: {inner_exc}") - import traceback - debug_log(f"[SubAgent] 错误堆栈: {traceback.format_exc()}") - break - - await asyncio.sleep(5) - - 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( + if notices: + # 取出后剩余是否还有未完成/未通知的后台工作（决定前端是否保持等待态） + has_remaining = _has_pending_completion_work( 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, - 'message_source': 'background_command', - '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}") + + preceding = notices[:-1] + last_notice = notices[-1] + + # 前 N-1 条预写通知：随后续任务注入历史/事件流，不单独触发工作 + preceding_payload: List[Dict[str, Any]] = [] + for item in preceding: + payload = dict(item.get("payload") or {}) + payload["starts_work"] = False # 预写通知渲染为完成态气泡，不起独立计时器 + preceding_payload.append({ + "message": item.get("message") or "", + "payload": payload, + }) + + last_payload = dict(last_notice.get("payload") or {}) + last_payload.update({ + "has_running_sub_agents": has_remaining, + "has_running_background_commands": has_remaining, + "remaining_count": 1 if has_remaining else 0, + }) + + try: + 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=last_notice.get("message") or "", + extra_payload=last_payload, + preceding_notices=preceding_payload, + ) + except Exception: + pass + + # 本轮已派发一个后续任务，由该任务结束后重新 spawn 本轮询器继续消费剩余通知 return - if not manager.has_pending_for_conversation(conversation_id): - debug_log("[BgCmdDebug] no pending background commands, stop polling") + # 没有待通知项：若也没有运行中的后台工作，结束轮询 + pending = _has_pending_completion_work(web_terminal=web_terminal, conversation_id=conversation_id) + if not pending: break + await asyncio.sleep(5) - debug_log("[BgCommand] 后台轮询结束") async def handle_task_with_sender( terminal: WebTerminal, @@ -1685,24 +1690,6 @@ async def handle_task_with_sender( 'tasks': [{'agent_id': t.get('agent_id'), 'summary': t.get('summary')} for t in notify_tasks] }) - # 启动后台任务来轮询/补发子智能体完成 - def run_poll(): - import asyncio - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - loop.run_until_complete(poll_sub_agent_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_poll) - # 检查是否有后台 run_command 或待通知任务 if bg_manager and conversation_id: try: @@ -1715,22 +1702,27 @@ async def handle_task_with_sender( # 与子智能体完全复用同一 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() + # 统一「通知池」轮询器：子智能体 + 后台 run_command 合并为单一轮询链路， + # 每轮一次性取出两路所有待通知项，避免逐条触发「工作 → 停止 → 再工作」循环。 + # 只 spawn 一个轮询器（无论是否同时存在子智能体与后台命令），从根本上消除 + # 两个轮询器同时 create_chat_task 撞「单工作区互斥」的问题。 + if has_running_sub_agents or has_running_background_commands: + def run_completion_poll(): + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(poll_completion_notifications( + 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) + socketio.start_background_task(run_completion_poll) has_running_completion_jobs = has_running_sub_agents or has_running_background_commands @@ -1781,6 +1773,16 @@ async def handle_task_with_sender( # 发送完成事件（如果有后台完成任务在运行，前端会保持等待状态） if not has_running_completion_jobs: + # 对话真正空闲：把 work_timer 持久化到后端，避免刷新后恢复成「工作中」 + try: + if web_terminal and conversation_id: + finalize_conversation_work_timer( + web_terminal, + conversation_id, + finished_at=datetime.now().isoformat(), + ) + except Exception: + pass finalize_user_work_timer() finalize_run_versioning_checkpoint("completed") else: diff --git a/server/chat_flow_task_support.py b/server/chat_flow_task_support.py index a007ce9..5b9c31d 100644 --- a/server/chat_flow_task_support.py +++ b/server/chat_flow_task_support.py @@ -33,7 +33,15 @@ def _runtime_message_ui_defaults(src: str, *, inline: bool = False) -> Dict[str, return {"visibility": "hidden", "starts_work": False} if normalized in {"guidance", "notify"}: return {"visibility": "compact", "starts_work": False} - if normalized in {"goal_review", "compression", "compression_handoff", "sub_agent", "background_command", "presend"}: + # 后台完成通知 / 运行期压缩续接：属于当前这轮工作的延续，不开启新的工作头与计时器， + # 也不暂停上一轮计时器。与 task_main 的 _user_message_ui_defaults 保持一致， + # 确保运行期直接渲染与刷新后从历史加载行为相同。 + if normalized in {"sub_agent", "background_command", "compression", "compression_handoff"}: + return {"visibility": "compact", "starts_work": False} + # 目标审核续命：开启新一轮工作。 + if normalized == "goal_review": + return {"visibility": "compact", "starts_work": True} + if normalized == "presend": return {"visibility": "compact", "starts_work": True} if normalized == "goal": # Backward-compatible fallback for old callers. Inline goal injections are diff --git a/server/conversation.py b/server/conversation.py index fa94f06..0be40cb 100644 --- a/server/conversation.py +++ b/server/conversation.py @@ -1207,7 +1207,6 @@ def compress_conversation(conversation_id, terminal: WebTerminal, workspace: Use 'clear_ui': True }, room=f"user_{username}") - compress_behavior = str(result.get("compress_behavior") or "continue").strip().lower() response_payload = { "success": True, "in_place": True, @@ -1216,35 +1215,14 @@ def compress_conversation(conversation_id, terminal: WebTerminal, workspace: Use "summary_failed": result.get("summary_failed", False), "guide_message": result.get("guide_message"), "compress_form": result.get("compress_form"), - "compress_behavior": compress_behavior, + # 手动压缩只有一种行为：生成压缩消息（引导语），不自动续接，等待用户继续发送消息才工作。 + "compress_behavior": "wait", "load_result": load_result } guide_message = (result.get("guide_message") or "").strip() - if guide_message and compress_behavior == "continue": - # 继续工作：创建续接任务，自动发送引导语并触发请求（同一对话）。 - try: - from .tasks import task_manager - workspace_id = session.get("workspace_id") or "default" - task_rec = task_manager.create_chat_task( - username, - workspace_id, - guide_message, - images=[], - conversation_id=normalized_id, - videos=[], - message_source="compression_handoff", - ) - response_payload["auto_task_started"] = True - response_payload["auto_task_id"] = task_rec.task_id - response_payload["auto_task_status"] = task_rec.status - except Exception as exc: - # 任务启动失败时前端会回退到手动发送 guide_message - debug_log(f"[Compression] 自动发送引导消息失败: {exc}") - response_payload["auto_task_started"] = False - response_payload["auto_task_error"] = str(exc) - elif guide_message and compress_behavior == "wait": - # 等待用户：只把引导语作为 user 消息追加进历史，不触发请求。 + if guide_message: + # 只把引导语作为 user 消息追加进历史，不触发请求。 try: terminal.context_manager.add_conversation( role="user", @@ -1264,11 +1242,11 @@ def compress_conversation(conversation_id, terminal: WebTerminal, workspace: Use "media_refs": [], "message_source": "compression_handoff", "visibility": "compact", - "starts_work": True, + "starts_work": False, "metadata": { "message_source": "compression_handoff", "visibility": "compact", - "starts_work": True, + "starts_work": False, }, "conversation_id": normalized_id, }, diff --git a/server/deep_compression.py b/server/deep_compression.py index c80691d..5d99ada 100644 --- a/server/deep_compression.py +++ b/server/deep_compression.py @@ -322,8 +322,6 @@ async def run_deep_compression( # 读取个性化压缩设置： # - compress_form: file（生成文件，引导语提示位置） / inject（把历次压缩内容注入引导语） - # - compress_behavior: continue（注入引导语并触发请求） / wait（仅注入引导语，等待用户） - # compress_behavior 仅作用于手动压缩；自动深压缩永远继续工作。 try: from modules.personalization_manager import load_personalization_config pconfig = load_personalization_config(workspace.data_dir) or {} @@ -332,11 +330,10 @@ async def run_deep_compression( compress_form = str(pconfig.get("deep_compress_form") or "file").strip().lower() if compress_form not in ("file", "inject"): compress_form = "file" - compress_behavior = str(pconfig.get("deep_compress_behavior") or "continue").strip().lower() - if compress_behavior not in ("continue", "wait"): - compress_behavior = "continue" - if mode != "manual": - compress_behavior = "continue" + # compress_behavior 固定规则（无个性化开关）： + # - 手动压缩：只生成压缩消息（引导语），不自动续接，等待用户继续发送消息才工作。 + # - 自动深压缩：当前任务尚未完成才触发，压缩后必须继续工作。 + compress_behavior = "wait" if mode == "manual" else "continue" # in-place 压缩全程操作"当前对话"的内存历史：总结、标记、状态写入都依赖 # cm.conversation_history / cm.conversation_metadata。若当前对话不是目标对话， diff --git a/server/tasks/models.py b/server/tasks/models.py index b88688d..c720780 100644 --- a/server/tasks/models.py +++ b/server/tasks/models.py @@ -17,6 +17,7 @@ from flask import current_app, session from server.auth_helpers import api_login_required, get_current_username from server.context import get_user_resources, ensure_conversation_loaded from server.chat_flow import run_chat_task_sync +from server.work_timer import finalize_conversation_work_timer from server.state import stop_flags from server.utils_common import debug_log, log_conn_diag from utils.host_workspace_debug import write_host_workspace_debug @@ -799,6 +800,25 @@ class TaskManager: # 这样前端轮询能即时看到这条 user 消息，而不是刷新后才从历史中看到。 try: if bool((rec.session_data or {}).get("auto_user_message_event")): + # 先回放本批「通知池」里的前置完成通知（除触发消息外的 N-1 条）， + # 保证轮询客户端按时间顺序看到所有完成通知，且不各自触发新一轮工作。 + preceding_notices = (rec.session_data or {}).get("preceding_user_notices") or [] + if isinstance(preceding_notices, list): + for item in preceding_notices: + if not isinstance(item, dict): + continue + notice_msg = str(item.get("message") or "").strip() + if not notice_msg: + continue + notice_payload = dict(item.get("payload") or {}) + notice_payload["starts_work"] = False + notice_event = { + "message": notice_msg, + "conversation_id": rec.conversation_id, + "task_id": rec.task_id, + } + notice_event.update(notice_payload) + self._append_event(rec, "user_message", notice_event) extra_payload = (rec.session_data or {}).get("auto_user_message_payload") or {} if not isinstance(extra_payload, dict): extra_payload = {} @@ -959,28 +979,28 @@ class TaskManager: ) except Exception as exc: debug_log(f"[TaskRun] 发送 task_stopped 失败: {exc}") - # 持久化：将最后一条用户消息的 work_timer 标记为完成，避免刷新后仍显示工作中 - try: - if terminal and getattr(terminal, "context_manager", None) and rec.conversation_id: - cm = getattr(terminal.context_manager, "conversation_manager", None) - if cm: - cm.mark_latest_user_work_completed(rec.conversation_id) - debug_log(f"[TaskRun] 已持久化 work_timer 完成: task_id={rec.task_id}, conversation_id={rec.conversation_id}") - except Exception as exc: - debug_log(f"[TaskRun] 持久化 work_timer 完成失败: {exc}") else: with self._lock: rec.status = "succeeded" rec.updated_at = time.time() - # 正常结束时同样持久化 work_timer，保证刷新后状态一致 - try: - if terminal and getattr(terminal, "context_manager", None) and rec.conversation_id: - cm = getattr(terminal.context_manager, "conversation_manager", None) - if cm: - cm.mark_latest_user_work_completed(rec.conversation_id) - debug_log(f"[TaskRun] 正常结束已持久化 work_timer 完成: task_id={rec.task_id}, conversation_id={rec.conversation_id}") - except Exception as exc: - debug_log(f"[TaskRun] 正常结束持久化 work_timer 完成失败: {exc}") + + # 任务线程结束：仅当对话真正空闲（无其它前台/后台任务）时才把 work_timer 标记为完成， + # 避免智能体已停、但后台子智能体/后台命令/压缩仍在进行时提前停止计时。 + try: + if terminal and rec.conversation_id: + active_task_ids = { + t.task_id + for t in self.list_tasks(username, rec.workspace_id) + if t.status in {"pending", "running"} + } + finalized = finalize_conversation_work_timer( + terminal, + rec.conversation_id, + exclude_task_id=rec.task_id, + active_task_ids=active_task_ids, + ) + except Exception as exc: + pass except Exception as exc: debug_log(f"[Task] 后台任务失败: {exc}") self._append_event(rec, "error", {"message": str(exc)}) diff --git a/server/work_timer.py b/server/work_timer.py new file mode 100644 index 0000000..71357fa --- /dev/null +++ b/server/work_timer.py @@ -0,0 +1,95 @@ +"""对话工作计时器持久化与空闲判定辅助函数。 + +把「对话真正停止运行（没有其它前台任务、也没有后台子智能体/后台命令/压缩）」 +这一时刻的 work_timer 同步到磁盘，并刷新当前 ContextManager 内存中的副本， +避免刷新页面后计时器又恢复成「工作中」。 +""" +from __future__ import annotations + +from datetime import datetime +from typing import Any, Optional, Set + +from modules.sub_agent import TERMINAL_STATUSES as SUB_AGENT_TERMINAL_STATUSES + + +def has_running_background_work(terminal: Any, conversation_id: Optional[str]) -> bool: + """检测指定对话是否还有未完成的后台工作（子智能体、后台命令、待通知完成项）。""" + if not terminal or not conversation_id: + return False + + sub_agent_manager = getattr(terminal, "sub_agent_manager", None) + if sub_agent_manager: + try: + sub_agent_manager.reconcile_task_states(conversation_id=conversation_id) + for task_info in sub_agent_manager.tasks.values(): + if task_info.get("conversation_id") != conversation_id: + continue + status = task_info.get("status") + if status not in SUB_AGENT_TERMINAL_STATUSES.union({"terminated"}): + return True + # 已完成但尚未发出通知，对话仍应视为运行中 + if status in SUB_AGENT_TERMINAL_STATUSES and not task_info.get("notified"): + return True + except Exception: + pass + + bg_manager = getattr(terminal, "background_command_manager", None) + if bg_manager: + try: + bg_manager.reconcile_stale_records(conversation_id=conversation_id) + if bg_manager.has_pending_for_conversation(conversation_id): + return True + except Exception: + pass + + return False + + +def finalize_conversation_work_timer( + terminal: Any, + conversation_id: Optional[str], + *, + finished_at: Optional[str] = None, + exclude_task_id: Optional[str] = None, + active_task_ids: Optional[Set[str]] = None, +) -> bool: + """当对话真正停止运行时，把 work_timer 持久化到后端并同步内存。 + + 参数： + terminal: 当前终端实例，需携带 context_manager。 + conversation_id: 要处理的对话 ID。 + finished_at: 完成时间 ISO 字符串，默认当前时间。 + exclude_task_id: 在判断「是否还有其它前台任务」时要排除的任务 ID。 + active_task_ids: 当前仍活跃的前台任务 ID 集合。若提供且除 exclude_task_id + 外仍有其它任务，则视为对话仍在运行，不做持久化。 + """ + if not terminal or not conversation_id: + return False + + # 1. 空闲判定：还有其它前台任务在跑？ + if active_task_ids: + others = {tid for tid in active_task_ids if tid and tid != exclude_task_id} + if others: + return False + + # 2. 空闲判定：还有后台工作？ + if has_running_background_work(terminal, conversation_id): + return False + + context_manager = getattr(terminal, "context_manager", None) + if not context_manager: + return False + + cm = getattr(context_manager, "conversation_manager", None) + if not cm: + return False + + now_iso = finished_at or datetime.now().isoformat() + + # 3. 持久化到磁盘并同步内存副本 + try: + return cm.mark_latest_user_work_completed( + conversation_id, now_iso, context_manager=context_manager + ) + except Exception: + return False diff --git a/static/src/app/methods/message/chat.ts b/static/src/app/methods/message/chat.ts index 90d6bc3..64b0c60 100644 --- a/static/src/app/methods/message/chat.ts +++ b/static/src/app/methods/message/chat.ts @@ -76,34 +76,14 @@ export const chatMethods = { await this.loadConversation(newId, { force: true }); } const guideMessage = (result.guide_message || '').trim(); - const autoTaskStarted = !!result.auto_task_started; - const autoTaskId = result.auto_task_id; - const compressBehavior = String(result.compress_behavior || 'continue'); - if (autoTaskStarted && autoTaskId) { - const { useTaskStore } = await import('../../../stores/task'); - const taskStore = useTaskStore(); - if (typeof this.clearProcessedEvents === 'function') { - this.clearProcessedEvents(); - } - taskStore.resumeTask(autoTaskId, { - status: result.auto_task_status || 'running', - resetOffset: true, - eventHandler: (event) => this.handleTaskEvent(event) - }); - this.taskInProgress = true; - if (typeof this.monitorShowPendingReply === 'function') { - this.monitorShowPendingReply(); - } - } else if (compressBehavior === 'wait') { - // 等待用户：后端已把引导语作为 user 消息追加进历史，这里刷新展示，不触发请求。 - if (newId && !isInPlace) { - await this.loadConversation(newId, { force: true }); - } else if (isInPlace) { - await this.loadConversation(this.currentConversationId, { force: true }); - } - } else if (guideMessage) { - await this.sendAutoUserMessage(guideMessage); + // 手动压缩只有一种行为：后端已把引导语作为 user 消息追加进历史， + // 这里刷新展示即可，不触发新一轮请求（等待用户继续发送消息才工作）。 + if (newId && !isInPlace) { + await this.loadConversation(newId, { force: true }); + } else if (isInPlace) { + await this.loadConversation(this.currentConversationId, { force: true }); } + void guideMessage; if (!isInPlace) { await this.loadConversationsList(); diff --git a/static/src/app/methods/taskPolling/lifecycle.ts b/static/src/app/methods/taskPolling/lifecycle.ts index fb71c45..b98af6e 100644 --- a/static/src/app/methods/taskPolling/lifecycle.ts +++ b/static/src/app/methods/taskPolling/lifecycle.ts @@ -471,11 +471,9 @@ export const lifecycleMethods = { this.streamingMessage = false; this.stopRequested = false; - // 智能体工作本身已结束，无论是否还有后台任务，先把 work_timer 标记为完成 - this.markLatestUserWorkCompleted(); - if (hasRunningBackground) { // 智能体已停，但后台任务还在跑：保持停止按钮，等用户第二下清理 + // 此时对话并未真正停止，work_timer 保持运行，不提前标记完成 this.taskInProgress = true; this.waitingForSubAgent = hasRunningSubAgents; this.waitingForBackgroundCommand = hasRunningBackgroundCommands; @@ -491,6 +489,8 @@ export const lifecycleMethods = { console.warn('[TaskPolling] uiPushToast 调用失败:', err); } } else { + // 对话真正停止：停止计时器并持久化 + this.markLatestUserWorkCompleted(); this.taskInProgress = false; this.waitingForSubAgent = false; this.waitingForBackgroundCommand = false; diff --git a/static/src/components/chat/actions/ToolAction.vue b/static/src/components/chat/actions/ToolAction.vue index 301ef59..532c658 100644 --- a/static/src/components/chat/actions/ToolAction.vue +++ b/static/src/components/chat/actions/ToolAction.vue @@ -194,6 +194,15 @@ function renderEnhancedToolResult(): string { return renderEasterEgg(result, args); } + // 子智能体类 + else if (name === 'create_sub_agent') { + return renderCreateSubAgent(result, args); + } else if (name === 'terminate_sub_agent') { + return renderTerminateSubAgent(result, args); + } else if (name === 'get_sub_agent_status') { + return renderGetSubAgentStatus(result, args); + } + // 默认显示 JSON return `
${escapeHtml(JSON.stringify(result || args, null, 2))}
`; } @@ -1080,6 +1089,125 @@ function renderEasterEgg(result: any, args: any): string { return html; } + +// ===== 子智能体类 ===== +function renderCreateSubAgent(result: any, args: any): string { + const status = formatToolStatusLabel(result, '✓ 已创建', '✗ 创建失败'); + const agentId = result.agent_id ?? args.agent_id ?? ''; + const taskId = result.task_id ?? ''; + const deliverablesDir = result.deliverables_dir ?? ''; + const taskDescription = args.task ?? ''; + const message = result.message ?? result.summary ?? ''; + + let html = '
'; + html += `
状态：${status}
`; + if (agentId !== '') { + html += `
子智能体 ID：${escapeHtml(String(agentId))}
`; + } + if (taskId !== '') { + html += `
任务 ID：${escapeHtml(String(taskId))}
`; + } + if (deliverablesDir !== '') { + html += `
交付目录：${escapeHtml(String(deliverablesDir))}
`; + } + if (taskDescription) { + const preview = String(taskDescription).slice(0, 200); + const suffix = String(taskDescription).length > 200 ? '…' : ''; + html += `
任务：${escapeHtml(preview + suffix)}
`; + } + html += '
'; + + if (message) { + html += '
'; + html += `
${escapeHtml(String(message))}
`; + html += '
'; + } + + return html; +} + +function renderTerminateSubAgent(result: any, args: any): string { + const status = formatToolStatusLabel(result, '✓ 已关闭', '✗ 关闭失败'); + const agentId = result.agent_id ?? args.agent_id ?? ''; + const taskId = result.task_id ?? ''; + const message = result.message ?? result.system_message ?? ''; + + let html = '
'; + html += `
状态：${status}
`; + if (agentId !== '') { + html += `
子智能体 ID：${escapeHtml(String(agentId))}
`; + } + if (taskId !== '') { + html += `
任务 ID：${escapeHtml(String(taskId))}
`; + } + html += '
'; + + if (message) { + html += '
'; + html += `
${escapeHtml(String(message))}
`; + html += '
'; + } + + return html; +} + +function renderGetSubAgentStatus(result: any, args: any): string { + if (!result.success) { + const error = result.error ?? '查询失败'; + return `
⚠️ ${escapeHtml(String(error))}
`; + } + + const results = Array.isArray(result.results) ? result.results : []; + if (results.length === 0) { + return '
未返回子智能体状态。
'; + } + + let html = '
'; + results.forEach((item: any) => { + const agentId = item.agent_id ?? ''; + const found = item.found !== false; + const status = String(item.status || ''); + const taskId = item.task_id ?? ''; + const summary = item.summary ?? (item.final_result?.message) ?? (item.final_result?.summary) ?? ''; + + html += '
'; + html += `
子智能体 #${escapeHtml(String(agentId))}
`; + + if (!found) { + html += '
'; + html += '
状态：不存在
'; + html += '
'; + } else { + html += '
'; + if (status === 'completed') { + html += '
状态：已完成
'; + } else if (status === 'terminated') { + html += '
状态：已终止
'; + } else if (status === 'running') { + html += '
状态：运行中
'; + } else if (status === 'pending') { + html += '
状态：等待中
'; + } else { + html += `
状态：${escapeHtml(status || '未知')}
`; + } + if (taskId !== '') { + html += `
任务 ID：${escapeHtml(String(taskId))}
`; + } + html += '
'; + + if (summary) { + html += '
'; + html += `
${escapeHtml(String(summary))}
`; + html += '
'; + } + } + + html += '
'; + }); + html += '
'; + + return html; +}