From 0b923d6c53b6be7f8148ef54e30a4ed7616389e2 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Sat, 4 Apr 2026 20:52:31 +0800 Subject: [PATCH] fix: harden polling task cancel lifecycle --- server/conversation.py | 57 +++++++++++++++++++++++++++ server/tasks.py | 7 +++- static/src/app/methods/message.ts | 12 +++++- static/src/app/methods/taskPolling.ts | 47 ++++++++++++++++++++++ static/src/stores/task.ts | 27 +++++++++++-- 5 files changed, 143 insertions(+), 7 deletions(-) diff --git a/server/conversation.py b/server/conversation.py index 3b3a2f2..ab884ae 100644 --- a/server/conversation.py +++ b/server/conversation.py @@ -110,6 +110,42 @@ def _terminate_running_sub_agents(terminal: WebTerminal, reason: str = "") -> in return stopped_count +def _cancel_running_tasks(username: str, workspace_id: str, timeout_seconds: float = 4.0) -> Tuple[int, bool]: + """取消当前工作区运行中的主任务，并等待其停止，避免切换对话后串写。""" + try: + from .tasks import task_manager + except Exception as exc: + debug_log(f"[TaskCancel] 导入 task_manager 失败: {exc}") + return 0, True + + active_statuses = {"pending", "running", "cancel_requested"} + + def _active_tasks(): + try: + recs = task_manager.list_tasks(username, workspace_id) + except Exception: + recs = task_manager.list_tasks(username) + return [rec for rec in recs if getattr(rec, "status", None) in active_statuses] + + running = _active_tasks() + if not running: + return 0, True + + canceled = 0 + for rec in running: + task_id = getattr(rec, "task_id", None) + if task_id and task_manager.cancel_task(username, task_id): + canceled += 1 + + deadline = time.time() + max(timeout_seconds, 0.5) + while time.time() < deadline: + if not _active_tasks(): + return canceled, True + time.sleep(0.1) + + return canceled, False + + # === 背景生成对话标题（从 app_legacy 拆分） === @conversation_bp.route('/api/conversations', methods=['GET']) @api_login_required @@ -160,6 +196,17 @@ def create_conversation(terminal: WebTerminal, workspace: UserWorkspace, usernam thinking_mode = data.get('thinking_mode') if preserve_mode and 'thinking_mode' in data else None run_mode = data.get('mode') if preserve_mode and 'mode' in data else None + workspace_id = session.get("workspace_id") or "default" + canceled_count, settled = _cancel_running_tasks(username=username, workspace_id=workspace_id) + if canceled_count: + debug_log(f"[TaskCancel] 创建新对话前取消任务: {canceled_count}") + if not settled: + return jsonify({ + "success": False, + "error": "任务正在停止中，请稍后再试", + "message": "检测到后台任务仍在停止，请稍后再创建新对话。" + }), 409 + _terminate_running_sub_agents(terminal, reason="用户创建新对话") result = terminal.create_new_conversation(thinking_mode=thinking_mode, run_mode=run_mode) @@ -237,6 +284,16 @@ def load_conversation(conversation_id, terminal: WebTerminal, workspace: UserWor try: current_id = getattr(getattr(terminal, "context_manager", None), "current_conversation_id", None) if current_id and current_id != conversation_id: + workspace_id = session.get("workspace_id") or "default" + canceled_count, settled = _cancel_running_tasks(username=username, workspace_id=workspace_id) + if canceled_count: + debug_log(f"[TaskCancel] 切换对话前取消任务: {canceled_count}") + if not settled: + return jsonify({ + "success": False, + "error": "任务正在停止中，请稍后再试", + "message": "检测到后台任务仍在停止，请稍后再切换对话。" + }), 409 _terminate_running_sub_agents(terminal, reason="用户切换对话") result = terminal.load_conversation(conversation_id) diff --git a/server/tasks.py b/server/tasks.py index 4d08ec5..bc6db9d 100644 --- a/server/tasks.py +++ b/server/tasks.py @@ -35,6 +35,7 @@ class TaskRecord: "max_iterations", "session_data", "stop_requested", + "next_event_idx", ) def __init__( @@ -66,6 +67,7 @@ class TaskRecord: self.max_iterations = max_iterations self.session_data: Dict[str, Any] = {} self.stop_requested: bool = False + self.next_event_idx: int = 0 class TaskManager: @@ -185,7 +187,10 @@ class TaskManager: # ---- internal helpers ---- def _append_event(self, rec: TaskRecord, event_type: str, data: Dict[str, Any]): with self._lock: - idx = rec.events[-1]["idx"] + 1 if rec.events else 0 + idx = getattr(rec, "next_event_idx", None) + if idx is None: + idx = rec.events[-1]["idx"] + 1 if rec.events else 0 + rec.next_event_idx = idx + 1 rec.events.append({ "idx": idx, "type": event_type, diff --git a/static/src/app/methods/message.ts b/static/src/app/methods/message.ts index f070e33..acb8043 100644 --- a/static/src/app/methods/message.ts +++ b/static/src/app/methods/message.ts @@ -278,11 +278,14 @@ export const messageMethods = { // 等待后端确认 await new Promise(resolve => setTimeout(resolve, 300)); + const shouldKeepBusy = Boolean(taskStore.currentTaskId) && + ['running', 'pending', 'cancel_requested'].includes(String(taskStore.taskStatus)); // 清理前端状态 this.clearPendingTools('user_stop'); this.streamingMessage = false; - this.taskInProgress = false; + // 若后台已回传停止事件，不要再次把输入区锁回“停止中” + this.taskInProgress = shouldKeepBusy; this.forceUnlockMonitor('user_stop'); if (typeof this.clearProcessedEvents === 'function') { this.clearProcessedEvents(); @@ -311,11 +314,16 @@ export const messageMethods = { }); } catch (error) { console.error('[Message] 取消任务失败:', error); + const { useTaskStore } = await import('../../stores/task'); + const taskStore = useTaskStore(); + const shouldKeepBusy = Boolean(taskStore.currentTaskId) && + ['running', 'pending', 'cancel_requested'].includes(String(taskStore.taskStatus)); // 即使失败也清理状态 this.clearPendingTools('user_stop'); this.streamingMessage = false; - this.taskInProgress = false; + // 如果任务其实已结束，允许按钮恢复发送态 + this.taskInProgress = shouldKeepBusy; this.forceUnlockMonitor('user_stop'); if (typeof this.clearProcessedEvents === 'function') { this.clearProcessedEvents(); diff --git a/static/src/app/methods/taskPolling.ts b/static/src/app/methods/taskPolling.ts index 97b95ca..dbd505e 100644 --- a/static/src/app/methods/taskPolling.ts +++ b/static/src/app/methods/taskPolling.ts @@ -121,6 +121,10 @@ export const taskPollingMethods = { this.handleTaskComplete(eventData, eventIdx); break; + case 'task_stopped': + this.handleTaskStopped(eventData, eventIdx); + break; + case 'error': this.handleTaskError(eventData, eventIdx); break; @@ -681,6 +685,22 @@ export const taskPollingMethods = { } }, + handleTaskStopped(data: any, eventIdx: number) { + debugLog('[TaskPolling] 任务已停止, idx:', eventIdx, data); + + this.streamingMessage = false; + this.taskInProgress = false; + this.stopRequested = false; + this.waitingForSubAgent = false; + + if (typeof this.clearPendingTools === 'function') { + this.clearPendingTools('task_stopped'); + } + this.$forceUpdate(); + + this.clearTaskState(); + }, + // 新增统一清理方法 clearTaskState() { (async () => { @@ -716,6 +736,33 @@ export const taskPollingMethods = { }, handleTaskError(data: any) { + const shouldRetry = Boolean(data?.retry); + if (shouldRetry) { + const retryIn = Number(data?.retry_in) || 5; + const attempt = Number(data?.attempt) || 1; + const maxAttempts = Number(data?.max_attempts) || attempt; + + debugLog('[TaskPolling] API错误，等待自动重试', { + retryIn, + attempt, + maxAttempts, + message: data?.message + }); + + this.stopRequested = false; + this.taskInProgress = true; + this.streamingMessage = true; + this.$forceUpdate(); + + this.uiPushToast({ + title: '即将重试', + message: `将在 ${retryIn} 秒后重试（第 ${attempt}/${maxAttempts} 次）`, + type: 'info', + duration: Math.max(retryIn, 1) * 1000 + }); + return; + } + const errorMessage = data.message || '未知错误'; const errorType = data.error_type || 'unknown'; diff --git a/static/src/stores/task.ts b/static/src/stores/task.ts index 812eeac..b68f487 100644 --- a/static/src/stores/task.ts +++ b/static/src/stores/task.ts @@ -93,12 +93,16 @@ export const useTaskStore = defineStore('task', { // 更新任务状态 this.taskStatus = data.status; this.taskUpdatedAt = data.updated_at; + let sawTaskStoppedEvent = false; // 处理新事件 if (data.events && data.events.length > 0) { debugLog(`[Task] 收到 ${data.events.length} 个新事件`); for (const event of data.events) { + if (event?.type === 'task_stopped') { + sawTaskStoppedEvent = true; + } try { eventHandler(event); } catch (err) { @@ -109,6 +113,24 @@ export const useTaskStore = defineStore('task', { this.lastEventIndex = data.next_offset; } + // 兜底：如果后端状态已经是 canceled 但未返回 task_stopped 事件， + // 补发一个本地停止事件，确保前端解除“停止中”状态。 + if (data.status === 'canceled' && !sawTaskStoppedEvent) { + debugLog('[Task] 状态已取消但未收到 task_stopped，补发本地事件'); + try { + eventHandler({ + type: 'task_stopped', + data: { + task_id: data.task_id || this.currentTaskId, + conversation_id: data.conversation_id || null, + synthetic: true + } + }); + } catch (err) { + console.error('[Task] 处理补发 task_stopped 失败:', err); + } + } + // 如果任务已完成，停止轮询 if (this.isTaskCompleted) { debugLog('[Task] 任务已完成，停止轮询:', this.taskStatus); @@ -219,10 +241,7 @@ export const useTaskStore = defineStore('task', { throw new Error(result.error || '取消任务失败'); } - this.taskStatus = 'canceled'; - this.stopPolling(); - - debugLog('[Task] 任务已取消'); + debugLog('[Task] 已发送取消请求，等待后端停止确认'); } catch (error) { console.error('[Task] 取消任务失败:', error); throw error;