JOJO/agent-Specialization
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: harden polling task cancel lifecycle

Browse Source
This commit is contained in:
JOJO 2026-04-04 20:52:31 +08:00
parent ddc34093de
commit 0b923d6c53
5 changed files with 143 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
server/conversation.py
View File

@ -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)

7
server/tasks.py
View File

@ -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,

12
static/src/app/methods/message.ts
View File

@ -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();

47
static/src/app/methods/taskPolling.ts
View File

@ -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';

27
static/src/stores/task.ts
View File

@ -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;