From 53b3669f1d84384dece8a0ea34241040a56de657 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Tue, 23 Jun 2026 19:27:50 +0800 Subject: [PATCH] =?UTF-8?q?fix(stop):=20=E5=BC=BA=E5=81=9C=E6=AD=A2?= =?UTF-8?q?=E6=8C=89=E9=92=AE=E4=B8=A4=E6=AE=B5=E5=BC=8F=E9=80=BB=E8=BE=91?= =?UTF-8?q?=E4=B8=8E=20work=5Ftimer=20=E6=8C=81=E4=B9=85=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/chat_flow.py | 78 ++++- server/tasks/models.py | 279 ++++++++++++++++-- static/src/app/methods/message/send.ts | 48 ++- .../src/app/methods/taskPolling/lifecycle.ts | 76 ++++- static/src/composables/useLegacySocket.ts | 17 +- static/src/stores/task.ts | 11 +- utils/conversation_manager/crud_mixin.py | 47 +++ 7 files changed, 501 insertions(+), 55 deletions(-) diff --git a/server/chat_flow.py b/server/chat_flow.py index 411c619..5125426 100644 --- a/server/chat_flow.py +++ b/server/chat_flow.py @@ -46,6 +46,7 @@ from modules.personalization_manager import ( THINKING_INTERVAL_MIN, THINKING_INTERVAL_MAX, ) +from modules.sub_agent.state import TERMINAL_STATUSES as SUB_AGENT_TERMINAL_STATUSES from modules.upload_security import UploadSecurityError from modules.user_manager import UserWorkspace from modules.usage_tracker import QUOTA_DEFAULTS @@ -180,22 +181,85 @@ def process_message_task(terminal: WebTerminal, message: str, images, sender, cl entry = get_stop_flag(client_sid, username) if not isinstance(entry, dict): - entry = {'stop': False, 'task': None, 'terminal': None} + entry = {'stop': False, 'task': None, 'terminal': None, 'loop': None} entry['stop'] = False entry['task'] = task entry['terminal'] = terminal + entry['loop'] = loop set_stop_flag(client_sid, username, entry) try: loop.run_until_complete(task) except asyncio.CancelledError: - debug_log(f"任务 {client_sid} 被成功取消") - sender('task_stopped', { - 'message': '任务已停止', - 'reason': 'user_requested' - }) + debug_log(f"[ChatFlow] 任务被成功取消: client_sid={client_sid}") + # 检测是否仍有后台任务在跑，通知前端保持停止按钮 + has_running_sub_agents = False + has_running_background_commands = False + conversation_id = getattr(getattr(terminal, 'context_manager', None), 'current_conversation_id', None) + if conversation_id: + 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"}): + has_running_sub_agents = True + break + except Exception as exc: + debug_log(f"[Task] 取消时检查后台子智能体失败: {exc}") + bg_manager = getattr(terminal, 'background_command_manager', None) + if bg_manager: + try: + bg_manager.reconcile_stale_records(conversation_id=conversation_id) + waiting_items = bg_manager.list_waiting_items(conversation_id) + if waiting_items: + has_running_background_commands = True + except Exception as exc: + debug_log(f"[Task] 取消时检查后台命令失败: {exc}") + debug_log( + f"[ChatFlow] 任务取消，最终停止事件由 _run_chat_task 发送: client_sid={client_sid}, " + f"has_running_sub_agents={has_running_sub_agents}, " + f"has_running_background_commands={has_running_background_commands}" + ) + # 取消时立即把当前对话最后一条 working 的用户消息 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 + except Exception as exc: + debug_log(f"[ChatFlow] 取消时标记 work_timer 完成失败: {exc}") + # task_stopped 事件统一在 _run_chat_task finally 中发送，避免重复 reset_system_state(terminal) - + loop.close() except Exception as e: # 【新增】错误时确保对话状态不丢失 diff --git a/server/tasks/models.py b/server/tasks/models.py index cfccee5..b88688d 100644 --- a/server/tasks/models.py +++ b/server/tasks/models.py @@ -22,6 +22,8 @@ from server.utils_common import debug_log, log_conn_diag from utils.host_workspace_debug import write_host_workspace_debug from config import DATA_DIR, WORKSPACE_SKILLS_DIRNAME from modules.goal_state_manager import GoalStateManager, REASON_USER_CANCEL +from modules.sub_agent.state import TERMINAL_STATUSES as SUB_AGENT_TERMINAL_STATUSES +from modules.background_command_manager import BackgroundCommandManager, TERMINAL_STATUSES as BG_COMMAND_TERMINAL_STATUSES SKILL_FRONTMATTER_RE = re.compile(r"^---\s*\n(?P.*?)\n---\s*\n?", re.S) @@ -51,6 +53,7 @@ class TaskRecord: "next_event_idx", "runtime_pending_queue", "runtime_guidance_queue", + "last_cancel_at", ) def __init__( @@ -87,6 +90,7 @@ class TaskRecord: self.next_event_idx: int = 0 self.runtime_pending_queue: List[Dict[str, Any]] = [] self.runtime_guidance_queue: List[str] = [] + self.last_cancel_at: Optional[float] = None class TaskManager: """线程内存版任务管理器，后续可替换为 Redis/DB。""" @@ -202,26 +206,125 @@ class TaskManager: def cancel_task(self, username: str, task_id: str) -> bool: rec = self.get_task(username, task_id) if not rec: + debug_log(f"[TaskCancel] cancel_task 找不到任务: task_id={task_id}") return False - rec.stop_requested = True - # 标记停止标志；chat_flow 会检测 stop_flags + + # 立即快照进入时的状态，后续所有两段式判断都用它。 + # _run_chat_task finally 会在硬取消后并发修改 rec.status， + # 如果后面再读会被误判为第二下。 + status_at_entry = rec.status + + debug_log( + f"[TaskCancel] 入口: task_id={task_id}, status={status_at_entry}, " + f"conversation_id={rec.conversation_id}, workspace_id={rec.workspace_id}" + ) + + # 1. 硬取消主 asyncio task（如果引用还在） entry = stop_flags.get(task_id) - if not isinstance(entry, dict): - entry = {'stop': False, 'task': None, 'terminal': None} - stop_flags[task_id] = entry - entry['stop'] = True - # 注释掉直接取消任务，改为通过停止标志让任务内部处理 - # try: - # if entry.get('task') and hasattr(entry['task'], "cancel"): - # entry['task'].cancel() - # except Exception: - # pass + terminal = None + if isinstance(entry, dict): + terminal = entry.get('terminal') + loop = entry.get('loop') + task = entry.get('task') + debug_log( + f"[TaskCancel] stop_flags entry: loop_exists={loop is not None}, " + f"task_exists={task is not None}, task_done={task.done() if task else 'n/a'}" + ) + if loop and task and not task.done(): + try: + loop.call_soon_threadsafe(task.cancel) + debug_log(f"[TaskCancel] 已投递硬取消: task_id={task_id}") + except Exception as exc: + debug_log(f"[TaskCancel] 硬取消失败: task_id={task_id}, error={exc}") + entry['stop'] = True + else: + debug_log(f"[TaskCancel] stop_flags 无 entry，设软标志: task_id={task_id}") + stop_flags[task_id] = {'stop': True, 'task': None, 'terminal': None, 'loop': None} + entry = stop_flags[task_id] + + rec.stop_requested = True + + # 2. 停止目标模式 + workspace = None + if rec.workspace_id: + try: + _, workspace = get_user_resources(username, rec.workspace_id) + if workspace: + gsm = GoalStateManager(workspace.data_dir) + if gsm.is_active(): + gsm.mark_stopped(REASON_USER_CANCEL) + debug_log(f"[Goal] 用户取消任务 {task_id}，同步停止工作区目标模式") + except Exception as exc: + debug_log(f"[Goal] 取消任务时停止目标模式失败: {exc}") + + # 3. 丢弃已经引导的内容（预输入队列保持不变，正常结束后再插入） with self._lock: - rec.status = "cancel_requested" + rec.runtime_guidance_queue = [] rec.updated_at = time.time() - # 清空事件队列，防止新任务读取到旧事件 + + # 4. 两段式停止：严格按进入时的 rec.status 判断（快照），不依赖 task.done() 的异步时机。 + # 原因：投递硬取消后 _run_chat_task finally 可能并发把 status 改成 cancel_requested， + # 若此时再读 rec.status 会误判为第二下，导致第一下就误杀后台。 + # - 第一下（running/pending）：只硬取消主智能体并置为 cancel_requested，保留后台任务。 + # - 第二下（cancel_requested/stopped/canceled）：清理后台任务并置为 stopped。 + # - 额外保护：极短时间内（<500ms）的重复 cancel 请求视为抖动，不执行第二下。 + now = time.time() + is_first_press = status_at_entry in {"running", "pending"} + + if not terminal and rec.workspace_id: + try: + terminal, _ = get_user_resources(username, rec.workspace_id) + except Exception as exc: + debug_log(f"[TaskCancel] 取消任务时重新获取 terminal 失败: {exc}") + + if is_first_press: + # 第一下：主智能体仍在跑，只请求取消，保留后台任务。 + # 注意 1：不要清空 rec.events，否则 _run_chat_task finally 发送的 + # task_stopped 事件会被丢弃，前端无法感知并弹提示。 + # 注意 2：_run_chat_task finally 可能并发先执行并把 status 设为 stopped， + # 此时不要再覆盖回 cancel_requested。 + with self._lock: + if rec.status in {"running", "pending"}: + rec.status = "cancel_requested" + rec.updated_at = now + rec.last_cancel_at = now + debug_log( + f"[TaskCancel] 第一下：请求取消智能体，保留后台任务: " + f"task_id={task_id}, status_at_entry={status_at_entry}, " + f"current_status={rec.status}" + ) + return True + + # 第二下：清理后台任务（过滤 500ms 内的重复请求） + if rec.last_cancel_at is not None and (now - rec.last_cancel_at) < 0.5: + debug_log( + f"[TaskCancel] 忽略 500ms 内重复取消请求: task_id={task_id}, " + f"elapsed={now - rec.last_cancel_at:.3f}s" + ) + return True + + has_running_background = self._cleanup_background_tasks(rec, terminal) + with self._lock: + rec.status = "stopped" + rec.updated_at = time.time() + rec.last_cancel_at = time.time() rec.events.clear() - debug_log(f"[Task] 任务 {task_id} 已取消，事件队列已清空") + debug_log( + f"[TaskCancel] 第二下：清理后台并停止: task_id={task_id}, " + f"has_running_background={has_running_background}" + ) + try: + from server.extensions import socketio + socketio.emit('task_stopped', { + 'message': '任务已停止', + 'reason': 'user_requested', + 'task_id': task_id, + 'conversation_id': rec.conversation_id, + 'has_running_sub_agents': False, + 'has_running_background_commands': False, + }, room=f"user_{username}") + except Exception as exc: + debug_log(f"[TaskCancel] 发送 task_stopped 事件失败: {exc}") return True @staticmethod @@ -500,6 +603,89 @@ class TaskManager: return items # ---- internal helpers ---- + def _cleanup_background_tasks(self, rec: TaskRecord, terminal: Optional[Any]) -> bool: + """清理指定任务/对话下的所有后台子智能体和后台命令，返回是否清理到任何任务。""" + has_running_background = False + if not terminal or not rec.conversation_id: + return has_running_background + + sub_agent_manager = getattr(terminal, 'sub_agent_manager', None) + if sub_agent_manager: + try: + sub_agent_manager.reconcile_task_states(conversation_id=rec.conversation_id) + for task_info in list(sub_agent_manager.tasks.values()): + if task_info.get('conversation_id') != rec.conversation_id: + continue + status = task_info.get('status') + if status not in SUB_AGENT_TERMINAL_STATUSES.union({"terminated"}): + has_running_background = True + try: + sub_agent_manager.terminate_sub_agent(task_id=task_info.get('task_id')) + except Exception as exc: + debug_log(f"[TaskCancel] 终止子智能体失败: {exc}") + except Exception as exc: + debug_log(f"[TaskCancel] 检查后台子智能体失败: {exc}") + + bg_manager = getattr(terminal, 'background_command_manager', None) + if bg_manager: + try: + bg_manager.reconcile_stale_records(conversation_id=rec.conversation_id) + waiting_items = bg_manager.list_waiting_items(rec.conversation_id) + for item in waiting_items: + has_running_background = True + try: + bg_manager.cancel_command(item.get('command_id')) + except Exception as exc: + debug_log(f"[TaskCancel] 取消后台命令失败: {exc}") + # 把该对话下所有未通知的终态记录也标为已通知，避免后续幽灵通知 + try: + with bg_manager._lock: + for record in bg_manager._records.values(): + if record.get('conversation_id') != rec.conversation_id: + continue + if record.get('status') in BG_COMMAND_TERMINAL_STATUSES and not record.get('notified'): + record['notified'] = True + record['updated_at'] = time.time() + except Exception as exc: + debug_log(f"[TaskCancel] 标记后台通知状态失败: {exc}") + except Exception as exc: + debug_log(f"[TaskCancel] 检查后台命令失败: {exc}") + + return has_running_background + + @staticmethod + def _has_running_background(rec: TaskRecord, terminal: Optional[Any]) -> Dict[str, bool]: + """检测指定任务/对话是否还有运行中的后台子智能体或后台命令。""" + result = {"has_running_sub_agents": False, "has_running_background_commands": False} + if not terminal or not rec.conversation_id: + return result + + sub_agent_manager = getattr(terminal, 'sub_agent_manager', None) + if sub_agent_manager: + try: + sub_agent_manager.reconcile_task_states(conversation_id=rec.conversation_id) + for task_info in sub_agent_manager.tasks.values(): + if task_info.get('conversation_id') != rec.conversation_id: + continue + status = task_info.get('status') + if status not in SUB_AGENT_TERMINAL_STATUSES.union({"terminated"}): + result["has_running_sub_agents"] = True + break + except Exception as exc: + debug_log(f"[Task] 检查后台子智能体失败: {exc}") + + bg_manager = getattr(terminal, 'background_command_manager', None) + if bg_manager: + try: + bg_manager.reconcile_stale_records(conversation_id=rec.conversation_id) + waiting_items = bg_manager.list_waiting_items(rec.conversation_id) + if waiting_items: + result["has_running_background_commands"] = True + except Exception as exc: + debug_log(f"[Task] 检查后台命令失败: {exc}") + + return result + def _append_event(self, rec: TaskRecord, event_type: str, data: Dict[str, Any]): if isinstance(data, dict): data = dict(data) @@ -735,9 +921,66 @@ class TaskManager: # 结束状态 canceled_flag = rec.stop_requested or stop_hint or bool(stop_flags.get(rec.task_id, {}).get("stop")) - with self._lock: - rec.status = "canceled" if canceled_flag else "succeeded" - rec.updated_at = time.time() + if canceled_flag: + # 用户取消时，根据是否还有后台任务决定状态： + # - 有后台任务：保持 cancel_requested，等待用户第二下清理 + # - 无后台任务：直接 stopped + # 注意：如果 cancel_task 第二下已经先把 rec.status 设为 stopped，则保持 stopped。 + bg_state = self._has_running_background(rec, terminal) + has_bg = bg_state["has_running_sub_agents"] or bg_state["has_running_background_commands"] + with self._lock: + if rec.status == "stopped": + new_status = "stopped" + else: + new_status = "cancel_requested" if has_bg else "stopped" + rec.status = new_status + rec.updated_at = time.time() + debug_log( + f"[TaskRun] 任务线程结束: task_id={rec.task_id}, canceled_flag={canceled_flag}, " + f"new_status={new_status}, bg_state={bg_state}" + ) + # 统一发送 task_stopped，携带后台任务状态 + try: + from server.extensions import socketio + stopped_payload = { + 'message': '任务已停止', + 'reason': 'user_requested', + 'task_id': rec.task_id, + 'conversation_id': rec.conversation_id, + 'has_running_sub_agents': bg_state["has_running_sub_agents"], + 'has_running_background_commands': bg_state["has_running_background_commands"], + } + socketio.emit('task_stopped', stopped_payload, room=f"user_{rec.username}") + # 同时写入轮询事件队列，确保 websocket 丢失时前端仍能收到 + self._append_event(rec, "task_stopped", stopped_payload) + debug_log( + f"[TaskRun] 已发送 task_stopped: task_id={rec.task_id}, " + f"has_bg={has_bg}, room=user_{rec.username}" + ) + 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}") except Exception as exc: debug_log(f"[Task] 后台任务失败: {exc}") self._append_event(rec, "error", {"message": str(exc)}) diff --git a/static/src/app/methods/message/send.ts b/static/src/app/methods/message/send.ts index ce695d0..23450dc 100644 --- a/static/src/app/methods/message/send.ts +++ b/static/src/app/methods/message/send.ts @@ -1,5 +1,5 @@ // @ts-nocheck -import { debugLog } from '../common'; +import { debugLog, goalModeDebugLog } from '../common'; import { useTaskStore } from '../../../stores/task'; import { useModelStore } from '../../../stores/model'; import { @@ -335,7 +335,13 @@ export const sendMethods = { }, async stopTask() { console.log('[DEBUG_AWAITING] ===== stopTask ====='); + if (this._stopTaskRunning) { + goalModeDebugLog('stopTask:debounce-rejected', { stopRequested: this.stopRequested }); + return; + } + this._stopTaskRunning = true; if (this.compressionInProgress) { + this._stopTaskRunning = false; this.uiPushToast({ title: '对话自动压缩中', message: '压缩进行中，当前不可停止任务', @@ -345,7 +351,16 @@ export const sendMethods = { } const canStop = this.composerBusy && !this.stopRequested; + goalModeDebugLog('stopTask:entry', { + composerBusy: this.composerBusy, + stopRequested: this.stopRequested, + taskInProgress: this.taskInProgress, + streamingUi: this.streamingUi, + canStop, + currentTaskId: this.currentTaskId, + }); if (!canStop) { + this._stopTaskRunning = false; return; } @@ -368,15 +383,19 @@ export const sendMethods = { if (taskStore.currentTaskId) { await taskStore.cancelTask(); - // 立即停止轮询，防止旧任务的目标模式事件在切换对话后仍被处理 - taskStore.stopPolling('user_stop'); } - // 等待后端确认 + // 等待后端确认；轮询继续，task_stopped 事件到达后会由 taskStore 自动停止轮询 await new Promise((resolve) => setTimeout(resolve, 300)); const shouldKeepBusy = - Boolean(taskStore.currentTaskId) && - ['running', 'pending', 'cancel_requested'].includes(String(taskStore.taskStatus)); + ['running', 'pending', 'cancel_requested', 'canceled'].includes(String(taskStore.taskStatus)); + + goalModeDebugLog('stopTask:try-end', { + currentTaskId: taskStore.currentTaskId, + taskStatus: taskStore.taskStatus, + shouldKeepBusy, + streamingMessage: this.streamingMessage, + }); // 清理前端状态 this.clearPendingTools('user_stop'); @@ -414,8 +433,14 @@ export const sendMethods = { const { useTaskStore } = await import('../../../stores/task'); const taskStore = useTaskStore(); const shouldKeepBusy = - Boolean(taskStore.currentTaskId) && - ['running', 'pending', 'cancel_requested'].includes(String(taskStore.taskStatus)); + ['running', 'pending', 'cancel_requested', 'canceled'].includes(String(taskStore.taskStatus)); + + goalModeDebugLog('stopTask:catch', { + error: String(error), + currentTaskId: taskStore.currentTaskId, + taskStatus: taskStore.taskStatus, + shouldKeepBusy, + }); // 即使失败也清理状态 this.clearPendingTools('user_stop'); @@ -435,14 +460,15 @@ export const sendMethods = { } this.uiPushToast({ - title: '停止失败', - message: '任务可能仍在运行，请刷新页面', - type: 'warning' + title: '停止请求已发送', + message: '若任务未停止，请再次点击停止按钮', + type: 'info' }); } finally { // 确保清除 dropToolEvents 和 stopRequested 标志 this.dropToolEvents = false; this.stopRequested = false; + this._stopTaskRunning = false; } } }; diff --git a/static/src/app/methods/taskPolling/lifecycle.ts b/static/src/app/methods/taskPolling/lifecycle.ts index ca1e3bd..fb71c45 100644 --- a/static/src/app/methods/taskPolling/lifecycle.ts +++ b/static/src/app/methods/taskPolling/lifecycle.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 { @@ -106,10 +106,14 @@ export const lifecycleMethods = { } } // 检查事件 task_id 是否仍是当前正在接管的任务，防止切换/新建对话后旧轮询的在途响应串写。 + // task_stopped / task_complete / error 属于任务终态事件，即使 currentTaskId 已被清理 + //（例如用户点击停止后 stopPolling 清空了 ID），只要 conversation_id 匹配就应该处理。 + const isTerminalTaskEvent = ['task_complete', 'task_stopped', 'error'].includes(eventType); if ( !crossConversationAllowed.has(eventType) && eventData.task_id && - (!taskStore.currentTaskId || eventData.task_id !== taskStore.currentTaskId) + (!taskStore.currentTaskId || eventData.task_id !== taskStore.currentTaskId) && + !(isTerminalTaskEvent && eventData.conversation_id === this.currentConversationId) ) { restoreDebugLog('event:drop-task-mismatch', { eventType, @@ -441,6 +445,7 @@ export const lifecycleMethods = { }); }, handleTaskStopped(data: any, eventIdx: number) { + goalModeDebugLog('handleTaskStopped:entered', { eventIdx, data }); jsonDebug('handleTaskStopped:before', { eventIdx, data, @@ -450,28 +455,67 @@ export const lifecycleMethods = { }); debugLog('[TaskPolling] 任务已停止, idx:', eventIdx, data); - this.markLatestUserWorkCompleted(); + const hasRunningSubAgents = !!data?.has_running_sub_agents; + const hasRunningBackgroundCommands = !!data?.has_running_background_commands; + const hasRunningBackground = hasRunningSubAgents || hasRunningBackgroundCommands; + + goalModeDebugLog('handleTaskStopped', { + taskInProgress: this.taskInProgress, + streamingMessage: this.streamingMessage, + hasRunningSubAgents, + hasRunningBackgroundCommands, + data, + }); + this.cleanupTrailingEmptyAssistantPlaceholder('task_stopped'); this.streamingMessage = false; - this.taskInProgress = false; this.stopRequested = false; - this.waitingForSubAgent = false; - this.waitingForBackgroundCommand = false; - this.stopWaitingTaskProbe(); - if (typeof this.clearPendingTools === 'function') { - this.clearPendingTools('task_stopped'); + // 智能体工作本身已结束，无论是否还有后台任务，先把 work_timer 标记为完成 + this.markLatestUserWorkCompleted(); + + if (hasRunningBackground) { + // 智能体已停，但后台任务还在跑：保持停止按钮，等用户第二下清理 + this.taskInProgress = true; + this.waitingForSubAgent = hasRunningSubAgents; + this.waitingForBackgroundCommand = hasRunningBackgroundCommands; + debugLog('[TaskPolling] 任务已停止，仍有后台任务运行，保持停止态'); + goalModeDebugLog('handleTaskStopped:show-toast', { hasRunningSubAgents, hasRunningBackgroundCommands }); + try { + this.uiPushToast({ + title: '智能体已停止', + message: '再次按下停止按钮以结束后台任务', + type: 'info' + }); + } catch (err) { + console.warn('[TaskPolling] uiPushToast 调用失败:', err); + } + } else { + this.taskInProgress = false; + this.waitingForSubAgent = false; + this.waitingForBackgroundCommand = false; + this.stopWaitingTaskProbe(); + if (typeof this.clearPendingTools === 'function') { + this.clearPendingTools('task_stopped'); + } + this.clearTaskState(); + this.$nextTick(() => { + if (typeof this.tryAutoSendRuntimeQueuedMessages === 'function') { + this.tryAutoSendRuntimeQueuedMessages('task_stopped'); + } + }); } + + goalModeDebugLog('handleTaskStopped:after', { + taskInProgress: this.taskInProgress, + waitingForSubAgent: this.waitingForSubAgent, + waitingForBackgroundCommand: this.waitingForBackgroundCommand, + hasRunningBackground, + }); + this.scheduleTodoListRefresh(100); setTimeout(() => this.refreshRunningWorkspaceTasks?.(), 0); this.$forceUpdate(); - - this.clearTaskState(); - this.$nextTick(() => { - if (typeof this.tryAutoSendRuntimeQueuedMessages === 'function') { - this.tryAutoSendRuntimeQueuedMessages('task_stopped'); - } - }); jsonDebug('handleTaskStopped:after', { taskInProgress: this.taskInProgress, streamingMessage: this.streamingMessage, diff --git a/static/src/composables/useLegacySocket.ts b/static/src/composables/useLegacySocket.ts index 6139045..521b372 100644 --- a/static/src/composables/useLegacySocket.ts +++ b/static/src/composables/useLegacySocket.ts @@ -1573,7 +1573,22 @@ export async function initializeLegacySocket(ctx: any) { // 任务停止 ctx.socket.on('task_stopped', (data) => { socketLog('任务已停止:', data.message); - ctx.taskInProgress = false; + const hasRunningSubAgents = !!data?.has_running_sub_agents; + const hasRunningBackgroundCommands = !!data?.has_running_background_commands; + const hasRunningBackground = hasRunningSubAgents || hasRunningBackgroundCommands; + if (!hasRunningBackground) { + ctx.taskInProgress = false; + } else { + ctx.waitingForSubAgent = hasRunningSubAgents; + ctx.waitingForBackgroundCommand = hasRunningBackgroundCommands; + if (typeof ctx.uiPushToast === 'function') { + ctx.uiPushToast({ + title: '智能体已停止', + message: '再次按下停止按钮以结束后台任务', + type: 'info' + }); + } + } ctx.scheduleResetAfterTask('socket:task_stopped', { preserveMonitorWindows: true }); }); diff --git a/static/src/stores/task.ts b/static/src/stores/task.ts index b700b09..8edef7e 100644 --- a/static/src/stores/task.ts +++ b/static/src/stores/task.ts @@ -1,6 +1,6 @@ // @ts-nocheck import { defineStore } from 'pinia'; -import { debugLog } from '../app/methods/common'; +import { debugLog, goalModeDebugLog } from '../app/methods/common'; const debugNotifyLog = (...args: any[]) => { void args; @@ -55,7 +55,7 @@ export const useTaskStore = defineStore('task', { getters: { hasActiveTask: (state) => state.currentTaskId !== null && state.taskStatus === 'running', - isTaskCompleted: (state) => ['succeeded', 'failed', 'canceled'].includes(state.taskStatus) + isTaskCompleted: (state) => ['succeeded', 'failed', 'canceled', 'stopped'].includes(state.taskStatus) }, actions: { @@ -206,6 +206,12 @@ export const useTaskStore = defineStore('task', { } // 更新任务状态 + goalModeDebugLog('taskStore.poll:status', { + taskId, + from: fromOffset, + status: data.status, + isTaskCompleted: this.isTaskCompleted, + }); this.taskStatus = data.status; this.taskUpdatedAt = data.updated_at; const runtimeQueueMessages = Array.isArray(data?.runtime_queued_messages) @@ -506,6 +512,7 @@ export const useTaskStore = defineStore('task', { this.pollingInFlight = false; this.pollingWarned = false; this.runtimeQueueSnapshotKey = ''; + goalModeDebugLog('taskStore.stopPolling', { reason, taskStatus: this.taskStatus, currentTaskId: this.currentTaskId }); this.currentTaskId = null; // 清除任务 ID }, diff --git a/utils/conversation_manager/crud_mixin.py b/utils/conversation_manager/crud_mixin.py index 8287a96..e4abea5 100644 --- a/utils/conversation_manager/crud_mixin.py +++ b/utils/conversation_manager/crud_mixin.py @@ -346,6 +346,53 @@ class CrudMixin: print(f"⌘ 保存对话失败 {conversation_id}: {e}") return False + def mark_latest_user_work_completed(self, conversation_id: str, finished_at: Optional[str] = None) -> bool: + """将对话中最后一条 status 为 working 的用户消息的 work_timer 标记为完成。 + + 用于任务停止或正常结束后，刷新页面时不再恢复为"工作中"状态。 + """ + if not conversation_id: + return False + try: + data = self.load_conversation(conversation_id) + if not data: + return False + messages = data.get("messages") or [] + completed = False + now_iso = finished_at or datetime.now().isoformat() + for msg in reversed(messages): + 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): + continue + if timer.get("status") != "working": + continue + started_at = timer.get("started_at") or msg.get("timestamp") or now_iso + try: + start_dt = datetime.fromisoformat(str(started_at).replace("Z", "+00:00")) + end_dt = datetime.fromisoformat(now_iso.replace("Z", "+00:00")) + duration_ms = max(0, int((end_dt - start_dt).total_seconds() * 1000)) + except Exception: + duration_ms = timer.get("duration_ms", 0) or 0 + timer["status"] = "completed" + timer["started_at"] = started_at + timer["finished_at"] = now_iso + timer["duration_ms"] = duration_ms + msg["metadata"] = metadata + completed = True + break + if not completed: + return False + data["updated_at"] = datetime.now().isoformat() + self._save_conversation_file(conversation_id, data) + self._update_index(conversation_id, data) + return True + except Exception as exc: + print(f"⌘ 标记用户工作完成失败 {conversation_id}: {exc}") + return False + def update_project_snapshot( self, conversation_id: str,