fix(stop): 强停止按钮两段式逻辑与 work_timer 持久化
This commit is contained in:
parent
e90610ed05
commit
53b3669f1d
@ -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:
|
||||
# 【新增】错误时确保对话状态不丢失
|
||||
|
||||
@ -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<body>.*?)\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)})
|
||||
|
||||
@ -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;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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 });
|
||||
});
|
||||
|
||||
|
||||
@ -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
|
||||
},
|
||||
|
||||
|
||||
@ -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,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user