1246 lines
51 KiB
Python
1246 lines
51 KiB
Python
"""简单任务 API：将聊天任务与 WebSocket 解耦，支持后台运行与轮询。"""
|
||
from __future__ import annotations
|
||
import mimetypes
|
||
import json
|
||
import time
|
||
import threading
|
||
import uuid
|
||
import re
|
||
from collections import deque
|
||
from pathlib import Path
|
||
from typing import Dict, Any, Optional, List
|
||
|
||
from flask import Blueprint, request, jsonify
|
||
from flask import current_app, session
|
||
|
||
from .auth_helpers import api_login_required, get_current_username
|
||
from .context import get_user_resources, ensure_conversation_loaded
|
||
from .chat_flow import run_chat_task_sync
|
||
from .state import stop_flags
|
||
from .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
|
||
|
||
|
||
SKILL_FRONTMATTER_RE = re.compile(r"^---\s*\n(?P<body>.*?)\n---\s*\n?", re.S)
|
||
SKILL_FIELD_RE = re.compile(r"^(?P<key>name|description)\s*:\s*(?P<value>.*)$")
|
||
|
||
|
||
def _parse_skill_metadata(content: str, fallback_name: str) -> Dict[str, str]:
|
||
metadata = {"name": fallback_name, "description": ""}
|
||
match = SKILL_FRONTMATTER_RE.match(content or "")
|
||
if not match:
|
||
return metadata
|
||
for raw_line in match.group("body").splitlines():
|
||
field = SKILL_FIELD_RE.match(raw_line.strip())
|
||
if not field:
|
||
continue
|
||
key = field.group("key")
|
||
value = field.group("value").strip().strip('"').strip("'")
|
||
metadata[key] = value
|
||
metadata["name"] = metadata.get("name") or fallback_name
|
||
return metadata
|
||
|
||
|
||
def _workspace_skills_dir(workspace) -> Path:
|
||
return (Path(workspace.project_path).expanduser().resolve() / WORKSPACE_SKILLS_DIRNAME).resolve()
|
||
|
||
|
||
def _resolve_workspace_skill_path(workspace, raw_path: str) -> Path:
|
||
skills_dir = _workspace_skills_dir(workspace)
|
||
target = Path(str(raw_path or "")).expanduser()
|
||
if not target.is_absolute():
|
||
target = Path(workspace.project_path).expanduser().resolve() / target
|
||
target = target.resolve()
|
||
if target.name != "SKILL.md":
|
||
raise ValueError("skill 路径必须指向 SKILL.md")
|
||
try:
|
||
target.relative_to(skills_dir)
|
||
except ValueError as exc:
|
||
raise ValueError("skill 路径必须位于当前工作区 .agents/skills/ 内") from exc
|
||
if not target.is_file():
|
||
raise ValueError("skill 文件不存在")
|
||
return target
|
||
|
||
|
||
def _list_workspace_skills(workspace) -> List[Dict[str, str]]:
|
||
skills_dir = _workspace_skills_dir(workspace)
|
||
if not skills_dir.is_dir():
|
||
return []
|
||
result: List[Dict[str, str]] = []
|
||
for skill_file in sorted(skills_dir.glob("*/SKILL.md"), key=lambda p: p.parent.name.lower()):
|
||
try:
|
||
content = skill_file.read_text(encoding="utf-8")
|
||
except Exception:
|
||
continue
|
||
metadata = _parse_skill_metadata(content, skill_file.parent.name)
|
||
result.append({
|
||
"name": metadata.get("name") or skill_file.parent.name,
|
||
"description": metadata.get("description") or "",
|
||
"path": str(skill_file.resolve()),
|
||
})
|
||
return result
|
||
|
||
|
||
def _build_skill_context_messages(workspace, raw_refs: Any) -> List[Dict[str, str]]:
|
||
if not isinstance(raw_refs, list):
|
||
return []
|
||
messages: List[Dict[str, str]] = []
|
||
seen_paths = set()
|
||
for item in raw_refs[:10]:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
raw_path = str(item.get("path") or "").strip()
|
||
if not raw_path:
|
||
continue
|
||
skill_file = _resolve_workspace_skill_path(workspace, raw_path)
|
||
path_key = str(skill_file)
|
||
if path_key in seen_paths:
|
||
continue
|
||
seen_paths.add(path_key)
|
||
content = skill_file.read_text(encoding="utf-8")
|
||
metadata = _parse_skill_metadata(content, skill_file.parent.name)
|
||
messages.append({
|
||
"name": metadata.get("name") or skill_file.parent.name,
|
||
"path": path_key,
|
||
"content": content,
|
||
})
|
||
return messages
|
||
|
||
|
||
class TaskRecord:
|
||
__slots__ = (
|
||
"task_id",
|
||
"username",
|
||
"workspace_id",
|
||
"status",
|
||
"created_at",
|
||
"updated_at",
|
||
"message",
|
||
"conversation_id",
|
||
"events",
|
||
"thread",
|
||
"error",
|
||
"model_key",
|
||
"thinking_mode",
|
||
"run_mode",
|
||
"max_iterations",
|
||
"session_data",
|
||
"stop_requested",
|
||
"next_event_idx",
|
||
"runtime_pending_queue",
|
||
"runtime_guidance_queue",
|
||
)
|
||
|
||
def __init__(
|
||
self,
|
||
task_id: str,
|
||
username: str,
|
||
workspace_id: str,
|
||
message: str,
|
||
conversation_id: Optional[str],
|
||
model_key: Optional[str],
|
||
thinking_mode: Optional[bool],
|
||
run_mode: Optional[str],
|
||
max_iterations: Optional[int],
|
||
):
|
||
self.task_id = task_id
|
||
self.username = username
|
||
self.workspace_id = workspace_id
|
||
self.status = "pending"
|
||
self.created_at = time.time()
|
||
self.updated_at = self.created_at
|
||
self.message = message
|
||
self.conversation_id = conversation_id
|
||
# 刷新恢复时前端会从事件流重建进行中的输出，1000 在长流式回复下会过早截断，
|
||
# 导致“只恢复最后几个字符”。这里提高缓冲上限，优先保证重建完整性。
|
||
self.events: deque[Dict[str, Any]] = deque(maxlen=20000)
|
||
self.thread: Optional[threading.Thread] = None
|
||
self.error: Optional[str] = None
|
||
self.model_key = model_key
|
||
self.thinking_mode = thinking_mode
|
||
self.run_mode = run_mode
|
||
self.max_iterations = max_iterations
|
||
self.session_data: Dict[str, Any] = {}
|
||
self.stop_requested: bool = False
|
||
self.next_event_idx: int = 0
|
||
self.runtime_pending_queue: List[Dict[str, Any]] = []
|
||
self.runtime_guidance_queue: List[str] = []
|
||
|
||
|
||
class TaskManager:
|
||
"""线程内存版任务管理器，后续可替换为 Redis/DB。"""
|
||
|
||
def __init__(self):
|
||
self._tasks: Dict[str, TaskRecord] = {}
|
||
self._lock = threading.Lock()
|
||
|
||
def cleanup_old_tasks(self, max_age_seconds: int = 3600) -> int:
|
||
"""清理超过指定时间的已完成任务"""
|
||
now = time.time()
|
||
with self._lock:
|
||
to_remove = []
|
||
for task_id, rec in self._tasks.items():
|
||
if rec.status in {"succeeded", "failed", "canceled"}:
|
||
age = now - rec.updated_at
|
||
if age > max_age_seconds:
|
||
to_remove.append(task_id)
|
||
|
||
for task_id in to_remove:
|
||
del self._tasks[task_id]
|
||
|
||
return len(to_remove)
|
||
|
||
# ---- public APIs ----
|
||
def create_chat_task(
|
||
self,
|
||
username: str,
|
||
workspace_id: str,
|
||
message: str,
|
||
images: List[Any],
|
||
conversation_id: Optional[str],
|
||
videos: Optional[List[Any]] = None,
|
||
model_key: Optional[str] = None,
|
||
thinking_mode: Optional[bool] = None,
|
||
run_mode: Optional[str] = None,
|
||
max_iterations: Optional[int] = None,
|
||
session_data: Optional[Dict[str, Any]] = None,
|
||
message_source: Optional[str] = None,
|
||
goal_mode: bool = False,
|
||
skill_context_messages: Optional[List[Dict[str, str]]] = None,
|
||
) -> TaskRecord:
|
||
if run_mode:
|
||
normalized = str(run_mode).lower()
|
||
if normalized not in {"fast", "thinking", "deep"}:
|
||
raise ValueError("run_mode 只支持 fast/thinking/deep")
|
||
run_mode = normalized
|
||
# 单工作区互斥：禁止同一用户同一工作区并发任务
|
||
existing = [t for t in self.list_tasks(username, workspace_id) if t.status in {"pending", "running"}]
|
||
if existing:
|
||
raise RuntimeError("当前工作区已有运行中的任务，请稍后再试。")
|
||
task_id = str(uuid.uuid4())
|
||
record = TaskRecord(task_id, username, workspace_id, message, conversation_id, model_key, thinking_mode, run_mode, max_iterations)
|
||
# 记录当前 session 快照，便于后台线程内使用
|
||
if session_data is not None:
|
||
snapshot = dict(session_data)
|
||
snapshot.setdefault("workspace_id", workspace_id)
|
||
if message_source is not None:
|
||
snapshot.setdefault("message_source", str(message_source))
|
||
snapshot["goal_mode"] = bool(goal_mode)
|
||
if skill_context_messages:
|
||
snapshot["skill_context_messages"] = list(skill_context_messages)
|
||
try:
|
||
snapshot.setdefault("host_mode", session.get("host_mode"))
|
||
if snapshot.get("host_mode"):
|
||
snapshot.setdefault("host_workspace_id", session.get("host_workspace_id") or workspace_id)
|
||
except Exception:
|
||
if snapshot.get("host_mode"):
|
||
snapshot.setdefault("host_workspace_id", workspace_id)
|
||
record.session_data = snapshot
|
||
else:
|
||
try:
|
||
record.session_data = {
|
||
"username": session.get("username"),
|
||
"role": session.get("role"),
|
||
"is_api_user": session.get("is_api_user"),
|
||
"host_mode": session.get("host_mode"),
|
||
"host_workspace_id": session.get("host_workspace_id") or workspace_id,
|
||
"workspace_id": workspace_id,
|
||
"run_mode": session.get("run_mode"),
|
||
"thinking_mode": session.get("thinking_mode"),
|
||
"model_key": session.get("model_key"),
|
||
"message_source": str(message_source) if message_source is not None else None,
|
||
"goal_mode": bool(goal_mode),
|
||
"skill_context_messages": list(skill_context_messages or []),
|
||
}
|
||
except Exception:
|
||
record.session_data = {}
|
||
with self._lock:
|
||
self._tasks[task_id] = record
|
||
thread = threading.Thread(target=self._run_chat_task, args=(record, images, videos or []), daemon=True)
|
||
record.thread = thread
|
||
record.status = "running"
|
||
record.updated_at = time.time()
|
||
thread.start()
|
||
return record
|
||
|
||
def get_task(self, username: str, task_id: str) -> Optional[TaskRecord]:
|
||
with self._lock:
|
||
rec = self._tasks.get(task_id)
|
||
if not rec or rec.username != username:
|
||
return None
|
||
return rec
|
||
|
||
def list_tasks(self, username: str, workspace_id: Optional[str] = None) -> List[TaskRecord]:
|
||
with self._lock:
|
||
return [
|
||
rec
|
||
for rec in self._tasks.values()
|
||
if rec.username == username and (workspace_id is None or rec.workspace_id == workspace_id)
|
||
]
|
||
|
||
def cancel_task(self, username: str, task_id: str) -> bool:
|
||
rec = self.get_task(username, task_id)
|
||
if not rec:
|
||
return False
|
||
rec.stop_requested = True
|
||
# 标记停止标志；chat_flow 会检测 stop_flags
|
||
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
|
||
with self._lock:
|
||
rec.status = "cancel_requested"
|
||
rec.updated_at = time.time()
|
||
# 清空事件队列，防止新任务读取到旧事件
|
||
rec.events.clear()
|
||
debug_log(f"[Task] 任务 {task_id} 已取消，事件队列已清空")
|
||
return True
|
||
|
||
@staticmethod
|
||
def _normalize_runtime_pending_queue(raw_queue: Any) -> List[Dict[str, Any]]:
|
||
now_ts = time.time()
|
||
normalized: List[Dict[str, Any]] = []
|
||
if not isinstance(raw_queue, list):
|
||
return normalized
|
||
for raw_item in raw_queue:
|
||
if isinstance(raw_item, dict):
|
||
item_id = str(raw_item.get("id") or "").strip()
|
||
text = str(raw_item.get("text") or "").strip()
|
||
created_at = raw_item.get("created_at")
|
||
else:
|
||
item_id = ""
|
||
text = str(raw_item or "").strip()
|
||
created_at = None
|
||
if not text:
|
||
continue
|
||
if not item_id:
|
||
item_id = str(uuid.uuid4())
|
||
try:
|
||
created_at_float = float(created_at)
|
||
except Exception:
|
||
created_at_float = now_ts
|
||
normalized.append(
|
||
{
|
||
"id": item_id,
|
||
"text": text,
|
||
"created_at": created_at_float,
|
||
}
|
||
)
|
||
return normalized
|
||
|
||
@staticmethod
|
||
def _runtime_pending_public(queue: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
result: List[Dict[str, Any]] = []
|
||
for item in queue or []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
text = str(item.get("text") or "").strip()
|
||
item_id = str(item.get("id") or "").strip()
|
||
if not text or not item_id:
|
||
continue
|
||
result.append(
|
||
{
|
||
"id": item_id,
|
||
"text": text,
|
||
"created_at": item.get("created_at"),
|
||
}
|
||
)
|
||
return result
|
||
|
||
def enqueue_runtime_pending_message(
|
||
self, username: str, task_id: str, message: str, max_queue_size: int = 5
|
||
) -> Dict[str, Any]:
|
||
text = str(message or "").strip()
|
||
if not text:
|
||
return {"success": False, "code": "empty_message", "error": "消息不能为空"}
|
||
limit = int(max(1, max_queue_size))
|
||
with self._lock:
|
||
rec = self._tasks.get(task_id)
|
||
if not rec or rec.username != username:
|
||
return {"success": False, "code": "task_not_found", "error": "任务不存在"}
|
||
if rec.status not in {"pending", "running", "cancel_requested"}:
|
||
return {"success": False, "code": "task_not_running", "error": "任务已结束，无法追加消息"}
|
||
queue = self._normalize_runtime_pending_queue(getattr(rec, "runtime_pending_queue", None))
|
||
if len(queue) >= limit:
|
||
return {
|
||
"success": False,
|
||
"code": "queue_full",
|
||
"error": f"堆积消息已满（最多 {limit} 条）",
|
||
}
|
||
item = {
|
||
"id": str(uuid.uuid4()),
|
||
"text": text,
|
||
"created_at": time.time(),
|
||
}
|
||
queue.append(item)
|
||
rec.runtime_pending_queue = queue
|
||
rec.updated_at = time.time()
|
||
return {
|
||
"success": True,
|
||
"task_id": rec.task_id,
|
||
"item": item,
|
||
"messages": self._runtime_pending_public(queue),
|
||
}
|
||
|
||
def remove_runtime_pending_message(
|
||
self, username: str, task_id: str, message_id: str
|
||
) -> Dict[str, Any]:
|
||
target_id = str(message_id or "").strip()
|
||
if not target_id:
|
||
return {"success": False, "code": "invalid_message_id", "error": "消息ID无效"}
|
||
with self._lock:
|
||
rec = self._tasks.get(task_id)
|
||
if not rec or rec.username != username:
|
||
return {"success": False, "code": "task_not_found", "error": "任务不存在"}
|
||
queue = self._normalize_runtime_pending_queue(getattr(rec, "runtime_pending_queue", None))
|
||
remove_idx = -1
|
||
for idx, item in enumerate(queue):
|
||
if str(item.get("id") or "") == target_id:
|
||
remove_idx = idx
|
||
break
|
||
if remove_idx < 0:
|
||
return {"success": False, "code": "message_not_found", "error": "消息不存在"}
|
||
queue.pop(remove_idx)
|
||
rec.runtime_pending_queue = queue
|
||
rec.updated_at = time.time()
|
||
return {
|
||
"success": True,
|
||
"task_id": rec.task_id,
|
||
"messages": self._runtime_pending_public(queue),
|
||
}
|
||
|
||
def promote_runtime_pending_to_guidance(
|
||
self,
|
||
username: str,
|
||
task_id: str,
|
||
message_id: str,
|
||
max_guidance_queue_size: int = 5,
|
||
) -> Dict[str, Any]:
|
||
target_id = str(message_id or "").strip()
|
||
if not target_id:
|
||
return {"success": False, "code": "invalid_message_id", "error": "消息ID无效"}
|
||
guidance_limit = int(max(1, max_guidance_queue_size))
|
||
with self._lock:
|
||
rec = self._tasks.get(task_id)
|
||
if not rec or rec.username != username:
|
||
return {"success": False, "code": "task_not_found", "error": "任务不存在"}
|
||
if rec.status not in {"pending", "running", "cancel_requested"}:
|
||
return {"success": False, "code": "task_not_running", "error": "任务已结束，无法引导"}
|
||
queue = self._normalize_runtime_pending_queue(getattr(rec, "runtime_pending_queue", None))
|
||
guidance_queue = getattr(rec, "runtime_guidance_queue", None)
|
||
if not isinstance(guidance_queue, list):
|
||
guidance_queue = []
|
||
if len(guidance_queue) >= guidance_limit:
|
||
return {
|
||
"success": False,
|
||
"code": "guidance_queue_full",
|
||
"error": f"引导队列已满（最多 {guidance_limit} 条）",
|
||
}
|
||
selected = None
|
||
remain_queue: List[Dict[str, Any]] = []
|
||
for item in queue:
|
||
if selected is None and str(item.get("id") or "") == target_id:
|
||
selected = item
|
||
continue
|
||
remain_queue.append(item)
|
||
if not selected:
|
||
return {"success": False, "code": "message_not_found", "error": "消息不存在"}
|
||
selected_text = str(selected.get("text") or "").strip()
|
||
if not selected_text:
|
||
return {"success": False, "code": "empty_message", "error": "消息内容为空"}
|
||
guidance_queue.append(selected_text)
|
||
rec.runtime_guidance_queue = guidance_queue
|
||
rec.runtime_pending_queue = remain_queue
|
||
rec.updated_at = time.time()
|
||
return {
|
||
"success": True,
|
||
"task_id": rec.task_id,
|
||
"queued_count": len(guidance_queue),
|
||
"messages": self._runtime_pending_public(remain_queue),
|
||
}
|
||
|
||
def get_runtime_pending_messages(self, username: str, task_id: str) -> List[Dict[str, Any]]:
|
||
with self._lock:
|
||
rec = self._tasks.get(task_id)
|
||
if not rec or rec.username != username:
|
||
return []
|
||
queue = self._normalize_runtime_pending_queue(getattr(rec, "runtime_pending_queue", None))
|
||
rec.runtime_pending_queue = queue
|
||
return self._runtime_pending_public(queue)
|
||
|
||
def enqueue_runtime_guidance(
|
||
self,
|
||
username: str,
|
||
task_id: str,
|
||
message: str,
|
||
max_queue_size: int = 5,
|
||
source: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
text = str(message or "").strip()
|
||
if not text:
|
||
return {"success": False, "code": "empty_message", "error": "引导内容不能为空"}
|
||
with self._lock:
|
||
rec = self._tasks.get(task_id)
|
||
if not rec or rec.username != username:
|
||
return {"success": False, "code": "task_not_found", "error": "任务不存在"}
|
||
if rec.status not in {"pending", "running", "cancel_requested"}:
|
||
return {"success": False, "code": "task_not_running", "error": "任务已结束，无法追加引导"}
|
||
queue = getattr(rec, "runtime_guidance_queue", None)
|
||
if not isinstance(queue, list):
|
||
queue = []
|
||
rec.runtime_guidance_queue = queue
|
||
if len(queue) >= int(max(1, max_queue_size)):
|
||
return {
|
||
"success": False,
|
||
"code": "queue_full",
|
||
"error": f"引导队列已满（最多 {int(max(1, max_queue_size))} 条）",
|
||
}
|
||
normalized_source = str(source or "").strip().lower()
|
||
if normalized_source:
|
||
queue.append({"text": text, "source": normalized_source})
|
||
else:
|
||
queue.append(text)
|
||
rec.updated_at = time.time()
|
||
return {
|
||
"success": True,
|
||
"queued_count": len(queue),
|
||
"task_id": rec.task_id,
|
||
}
|
||
|
||
def pop_runtime_guidance_for_injection(self, username: str, task_id: str) -> Optional[Any]:
|
||
with self._lock:
|
||
rec = self._tasks.get(task_id)
|
||
if not rec or rec.username != username:
|
||
return None
|
||
queue = getattr(rec, "runtime_guidance_queue", None)
|
||
if not isinstance(queue, list) or not queue:
|
||
return None
|
||
item = queue.pop(0)
|
||
rec.updated_at = time.time()
|
||
if isinstance(item, dict):
|
||
text = str(item.get("text") or "").strip()
|
||
if not text:
|
||
return None
|
||
src = str(item.get("source") or "").strip().lower()
|
||
return {"text": text, "source": src} if src else {"text": text}
|
||
text = str(item or "").strip()
|
||
return text or None
|
||
|
||
def consume_runtime_guidance_messages(self, username: str, task_id: str) -> List[str]:
|
||
with self._lock:
|
||
rec = self._tasks.get(task_id)
|
||
if not rec or rec.username != username:
|
||
return []
|
||
queue = getattr(rec, "runtime_guidance_queue", None)
|
||
if not isinstance(queue, list) or not queue:
|
||
return []
|
||
items: List[str] = []
|
||
for item in queue:
|
||
if isinstance(item, dict):
|
||
text = str(item.get("text") or "").strip()
|
||
else:
|
||
text = str(item or "").strip()
|
||
if text:
|
||
items.append(text)
|
||
rec.runtime_guidance_queue = []
|
||
rec.updated_at = time.time()
|
||
return items
|
||
|
||
def consume_runtime_guidance_for_injection(self, username: str, task_id: str) -> List[Any]:
|
||
"""按原始结构取出整批引导/通知消息（支持 str 与 {text,source}）。"""
|
||
with self._lock:
|
||
rec = self._tasks.get(task_id)
|
||
if not rec or rec.username != username:
|
||
return []
|
||
queue = getattr(rec, "runtime_guidance_queue", None)
|
||
if not isinstance(queue, list) or not queue:
|
||
return []
|
||
items: List[Any] = []
|
||
for item in queue:
|
||
if isinstance(item, dict):
|
||
text = str(item.get("text") or "").strip()
|
||
if not text:
|
||
continue
|
||
src = str(item.get("source") or "").strip().lower()
|
||
items.append({"text": text, "source": src} if src else {"text": text})
|
||
else:
|
||
text = str(item or "").strip()
|
||
if text:
|
||
items.append(text)
|
||
rec.runtime_guidance_queue = []
|
||
rec.updated_at = time.time()
|
||
return items
|
||
|
||
# ---- internal helpers ----
|
||
def _append_event(self, rec: TaskRecord, event_type: str, data: Dict[str, Any]):
|
||
if isinstance(data, dict):
|
||
data = dict(data)
|
||
data.setdefault("task_id", rec.task_id)
|
||
if rec.conversation_id:
|
||
data.setdefault("conversation_id", rec.conversation_id)
|
||
if rec.workspace_id:
|
||
data.setdefault("workspace_id", rec.workspace_id)
|
||
with self._lock:
|
||
if event_type in {"goal_progress", "goal_completed", "goal_stopped"} and isinstance(data, dict):
|
||
rec.session_data["goal_progress"] = dict(data)
|
||
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,
|
||
"data": data,
|
||
"ts": time.time(),
|
||
})
|
||
rec.updated_at = time.time()
|
||
|
||
def _run_chat_task(self, rec: TaskRecord, images: List[Any], videos: List[Any]):
|
||
username = rec.username
|
||
workspace_id = rec.workspace_id
|
||
terminal = None
|
||
workspace = None
|
||
stop_hint = False
|
||
try:
|
||
# 为后台线程构造最小请求上下文，填充 session
|
||
from server.app import app as flask_app
|
||
with flask_app.test_request_context():
|
||
try:
|
||
for k, v in (rec.session_data or {}).items():
|
||
if v is not None:
|
||
session[k] = v
|
||
if session.get("host_mode"):
|
||
session["workspace_id"] = workspace_id
|
||
session["host_workspace_id"] = session.get("host_workspace_id") or workspace_id
|
||
write_host_workspace_debug(
|
||
"tasks.run_chat_task.apply_host_session",
|
||
task_id=rec.task_id,
|
||
workspace_id=workspace_id,
|
||
host_workspace_id=session.get("host_workspace_id"),
|
||
)
|
||
except Exception:
|
||
pass
|
||
terminal, workspace = get_user_resources(username, workspace_id=workspace_id)
|
||
if not terminal or not workspace:
|
||
raise RuntimeError("系统未初始化")
|
||
stop_hint = bool(stop_flags.get(rec.task_id, {}).get("stop"))
|
||
|
||
def _apply_requested_model_mode():
|
||
# API 传入的模型/模式配置
|
||
if rec.model_key:
|
||
try:
|
||
terminal.set_model(rec.model_key)
|
||
except Exception as exc:
|
||
debug_log(f"[Task] 设置模型失败 {rec.model_key}: {exc}")
|
||
if rec.run_mode:
|
||
try:
|
||
terminal.set_run_mode(rec.run_mode)
|
||
except Exception as exc:
|
||
debug_log(f"[Task] 设置运行模式失败 {rec.run_mode}: {exc}")
|
||
elif rec.thinking_mode is not None:
|
||
try:
|
||
terminal.set_run_mode("thinking" if rec.thinking_mode else "fast")
|
||
except Exception as exc:
|
||
debug_log(f"[Task] 设置思考模式失败: {exc}")
|
||
|
||
_apply_requested_model_mode()
|
||
if rec.max_iterations:
|
||
try:
|
||
terminal.max_iterations_override = int(rec.max_iterations)
|
||
except Exception:
|
||
terminal.max_iterations_override = None
|
||
try:
|
||
debug_log(
|
||
"[Task] effective terminal state "
|
||
f"model_key={getattr(terminal, 'model_key', None)!r} "
|
||
f"run_mode={getattr(terminal, 'run_mode', None)!r} "
|
||
f"thinking_mode={getattr(terminal, 'thinking_mode', None)!r} "
|
||
f"deep_mode={getattr(terminal, 'deep_thinking_mode', None)!r}"
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
# 确保会话加载
|
||
conversation_id = rec.conversation_id
|
||
try:
|
||
conversation_id, _ = ensure_conversation_loaded(terminal, conversation_id, workspace=workspace)
|
||
rec.conversation_id = conversation_id
|
||
except Exception as exc:
|
||
raise RuntimeError(f"对话加载失败: {exc}") from exc
|
||
|
||
# 对话加载会按会话元数据恢复历史模型/模式，这里再覆盖一次用户本次请求参数
|
||
_apply_requested_model_mode()
|
||
try:
|
||
debug_log(
|
||
"[Task] post-conversation effective state "
|
||
f"model_key={getattr(terminal, 'model_key', None)!r} "
|
||
f"run_mode={getattr(terminal, 'run_mode', None)!r} "
|
||
f"thinking_mode={getattr(terminal, 'thinking_mode', None)!r} "
|
||
f"deep_mode={getattr(terminal, 'deep_thinking_mode', None)!r}"
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
# 仅对“后台通知触发的新任务”补发 user_message 事件到任务事件流。
|
||
# 这样前端轮询能即时看到这条 user 消息，而不是刷新后才从历史中看到。
|
||
try:
|
||
if bool((rec.session_data or {}).get("auto_user_message_event")):
|
||
extra_payload = (rec.session_data or {}).get("auto_user_message_payload") or {}
|
||
if not isinstance(extra_payload, dict):
|
||
extra_payload = {}
|
||
payload = {
|
||
"message": rec.message,
|
||
"conversation_id": rec.conversation_id,
|
||
"task_id": rec.task_id,
|
||
}
|
||
payload.update(extra_payload)
|
||
self._append_event(rec, "user_message", payload)
|
||
except Exception as exc:
|
||
debug_log(f"[Task] 注入 user_message 事件失败: {exc}")
|
||
|
||
def sender(event_type, data):
|
||
if isinstance(data, dict):
|
||
data = dict(data)
|
||
if event_type == "compression_finished":
|
||
migrated_conversation_id = str(data.get("conversation_id") or "").strip()
|
||
if migrated_conversation_id:
|
||
with self._lock:
|
||
rec.conversation_id = migrated_conversation_id
|
||
rec.updated_at = time.time()
|
||
data.setdefault("task_id", rec.task_id)
|
||
if rec.conversation_id:
|
||
data.setdefault("conversation_id", rec.conversation_id)
|
||
if rec.workspace_id:
|
||
data.setdefault("workspace_id", rec.workspace_id)
|
||
# 记录事件
|
||
self._append_event(rec, event_type, data)
|
||
# 在线用户仍然收到实时推送（房间 user_{username}）
|
||
try:
|
||
from .extensions import socketio
|
||
socketio.emit(event_type, data, room=f"user_{username}")
|
||
except Exception:
|
||
pass
|
||
|
||
# 轮询模式需要把 context_manager 的回调切到当前任务 sender，
|
||
# 否则 token_update 等事件只走 websocket，前端任务轮询拿不到实时更新。
|
||
previous_ctx_callback = None
|
||
try:
|
||
if terminal and getattr(terminal, "context_manager", None):
|
||
previous_ctx_callback = getattr(terminal.context_manager, "_web_terminal_callback", None)
|
||
terminal.context_manager.set_web_terminal_callback(sender)
|
||
except Exception as exc:
|
||
debug_log(f"[Task] 设置上下文回调失败: {exc}")
|
||
|
||
# 将 task_id 作为 client_sid，供 stop_flags 检测
|
||
previous_auto_user_event = None
|
||
previous_message_source = None
|
||
previous_auto_user_payload = None
|
||
previous_goal_mode_requested = None
|
||
previous_skill_context_messages = None
|
||
try:
|
||
previous_auto_user_event = getattr(terminal, "_auto_user_message_event", False)
|
||
previous_message_source = getattr(terminal, "_current_user_message_source", None)
|
||
previous_auto_user_payload = getattr(terminal, "_auto_user_message_payload", None)
|
||
previous_goal_mode_requested = getattr(terminal, "_goal_mode_requested", False)
|
||
previous_skill_context_messages = getattr(terminal, "_skill_context_messages", None)
|
||
setattr(
|
||
terminal,
|
||
"_auto_user_message_event",
|
||
bool((rec.session_data or {}).get("auto_user_message_event")),
|
||
)
|
||
setattr(
|
||
terminal,
|
||
"_auto_user_message_payload",
|
||
dict((rec.session_data or {}).get("auto_user_message_payload") or {}),
|
||
)
|
||
setattr(
|
||
terminal,
|
||
"_current_user_message_source",
|
||
str((rec.session_data or {}).get("message_source") or "user"),
|
||
)
|
||
setattr(
|
||
terminal,
|
||
"_goal_mode_requested",
|
||
bool((rec.session_data or {}).get("goal_mode")),
|
||
)
|
||
setattr(
|
||
terminal,
|
||
"_skill_context_messages",
|
||
list((rec.session_data or {}).get("skill_context_messages") or []),
|
||
)
|
||
except Exception:
|
||
previous_auto_user_event = None
|
||
previous_message_source = None
|
||
previous_auto_user_payload = None
|
||
previous_goal_mode_requested = None
|
||
previous_skill_context_messages = None
|
||
|
||
try:
|
||
run_chat_task_sync(
|
||
terminal=terminal,
|
||
message=rec.message,
|
||
images=images,
|
||
sender=sender,
|
||
client_sid=rec.task_id,
|
||
workspace=workspace,
|
||
username=username,
|
||
videos=videos,
|
||
)
|
||
finally:
|
||
try:
|
||
if previous_auto_user_event is not None:
|
||
setattr(terminal, "_auto_user_message_event", previous_auto_user_event)
|
||
if previous_auto_user_payload is not None:
|
||
setattr(terminal, "_auto_user_message_payload", previous_auto_user_payload)
|
||
if previous_message_source is not None:
|
||
setattr(terminal, "_current_user_message_source", previous_message_source)
|
||
if previous_goal_mode_requested is not None:
|
||
setattr(terminal, "_goal_mode_requested", previous_goal_mode_requested)
|
||
if previous_skill_context_messages is not None:
|
||
setattr(terminal, "_skill_context_messages", previous_skill_context_messages)
|
||
else:
|
||
setattr(terminal, "_skill_context_messages", [])
|
||
if terminal and getattr(terminal, "context_manager", None):
|
||
terminal.context_manager.set_web_terminal_callback(previous_ctx_callback)
|
||
except Exception as exc:
|
||
debug_log(f"[Task] 恢复上下文回调失败: {exc}")
|
||
|
||
# 结束状态
|
||
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()
|
||
except Exception as exc:
|
||
debug_log(f"[Task] 后台任务失败: {exc}")
|
||
self._append_event(rec, "error", {"message": str(exc)})
|
||
with self._lock:
|
||
rec.status = "failed"
|
||
rec.error = str(exc)
|
||
rec.updated_at = time.time()
|
||
finally:
|
||
# 清理 stop_flags
|
||
stop_flags.pop(rec.task_id, None)
|
||
# 清理一次性配置
|
||
if terminal and hasattr(terminal, "max_iterations_override"):
|
||
try:
|
||
delattr(terminal, "max_iterations_override")
|
||
except Exception:
|
||
terminal.max_iterations_override = None
|
||
|
||
|
||
task_manager = TaskManager()
|
||
tasks_bp = Blueprint("tasks", __name__)
|
||
|
||
|
||
def start_task_cleanup_scheduler():
|
||
"""启动任务清理定时器"""
|
||
def cleanup_loop():
|
||
while True:
|
||
try:
|
||
count = task_manager.cleanup_old_tasks(3600)
|
||
if count > 0:
|
||
debug_log(f"[Task] 清理了 {count} 个旧任务")
|
||
except Exception as e:
|
||
debug_log(f"[Task] 清理任务失败: {e}")
|
||
time.sleep(600) # 每 10 分钟
|
||
|
||
thread = threading.Thread(target=cleanup_loop, daemon=True, name="TaskCleanup")
|
||
thread.start()
|
||
debug_log("[Task] 任务清理定时器已启动")
|
||
|
||
|
||
# 启动清理定时器
|
||
start_task_cleanup_scheduler()
|
||
|
||
|
||
def _conversation_title_for_task(rec: TaskRecord) -> Optional[str]:
|
||
conv_id = (getattr(rec, "conversation_id", None) or "").strip()
|
||
if not conv_id:
|
||
return None
|
||
if not conv_id.startswith("conv_"):
|
||
conv_id = f"conv_{conv_id}"
|
||
try:
|
||
if rec.username == "host":
|
||
path = Path(DATA_DIR).expanduser().resolve() / "conversations" / rec.workspace_id / f"{conv_id}.json"
|
||
else:
|
||
from . import state
|
||
workspace = state.user_manager.ensure_user_workspace(rec.username, rec.workspace_id or "default")
|
||
path = Path(workspace.data_dir).expanduser().resolve() / "conversations" / f"{conv_id}.json"
|
||
if not path.exists():
|
||
return None
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
title = str(data.get("title") or "").strip()
|
||
return title or None
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _workspace_label_for_task(rec: TaskRecord) -> Optional[str]:
|
||
try:
|
||
if rec.username == "host":
|
||
from modules.host_workspace_manager import load_host_workspace_catalog
|
||
catalog = load_host_workspace_catalog()
|
||
for item in catalog.get("workspaces") or []:
|
||
if item.get("workspace_id") == rec.workspace_id:
|
||
return str(item.get("label") or rec.workspace_id)
|
||
else:
|
||
from . import state
|
||
item = state.user_manager.list_user_workspaces(rec.username).get(rec.workspace_id)
|
||
if item:
|
||
return str(item.get("label") or rec.workspace_id)
|
||
except Exception:
|
||
pass
|
||
return rec.workspace_id
|
||
|
||
|
||
def _task_public_payload(rec: TaskRecord, *, include_title: bool = True) -> Dict[str, Any]:
|
||
payload = {
|
||
"task_id": rec.task_id,
|
||
"username": rec.username,
|
||
"workspace_id": rec.workspace_id,
|
||
"workspace_label": _workspace_label_for_task(rec),
|
||
"status": rec.status,
|
||
"created_at": rec.created_at,
|
||
"updated_at": rec.updated_at,
|
||
"message": rec.message,
|
||
"conversation_id": rec.conversation_id,
|
||
"error": rec.error,
|
||
"message_source": (rec.session_data or {}).get("message_source"),
|
||
"goal_mode": bool((rec.session_data or {}).get("goal_mode")),
|
||
"goal_progress": (rec.session_data or {}).get("goal_progress"),
|
||
}
|
||
if include_title:
|
||
payload["conversation_title"] = _conversation_title_for_task(rec)
|
||
return payload
|
||
|
||
|
||
@tasks_bp.route("/api/tasks", methods=["GET"])
|
||
@api_login_required
|
||
def list_tasks_api():
|
||
username = get_current_username()
|
||
workspace_id = (request.args.get("workspace_id") or "").strip() or None
|
||
status_filter = (request.args.get("status") or "").strip().lower()
|
||
recs = task_manager.list_tasks(username, workspace_id)
|
||
if status_filter:
|
||
if status_filter == "active":
|
||
active = {"pending", "running", "cancel_requested"}
|
||
recs = [rec for rec in recs if rec.status in active]
|
||
else:
|
||
wanted = {part.strip() for part in status_filter.split(",") if part.strip()}
|
||
recs = [rec for rec in recs if rec.status in wanted]
|
||
return jsonify({
|
||
"success": True,
|
||
"data": [
|
||
_task_public_payload(r)
|
||
for r in sorted(recs, key=lambda x: x.created_at, reverse=True)
|
||
]
|
||
})
|
||
|
||
|
||
@tasks_bp.route("/api/skills", methods=["GET"])
|
||
@api_login_required
|
||
def list_skills_api():
|
||
username = get_current_username()
|
||
workspace_id = session.get("workspace_id") or "default"
|
||
try:
|
||
_terminal, workspace = get_user_resources(username, workspace_id)
|
||
if not workspace:
|
||
return jsonify({"success": False, "error": "工作区不可用"}), 400
|
||
return jsonify({
|
||
"success": True,
|
||
"skills": _list_workspace_skills(workspace),
|
||
})
|
||
except Exception as exc:
|
||
debug_log(f"[SkillsAPI] 列出技能失败: {exc}")
|
||
return jsonify({"success": False, "error": str(exc)}), 500
|
||
|
||
|
||
@tasks_bp.route("/api/tasks", methods=["POST"])
|
||
@api_login_required
|
||
def create_task_api():
|
||
username = get_current_username()
|
||
workspace_id = session.get("workspace_id") or "default"
|
||
payload = request.get_json() or {}
|
||
message = (payload.get("message") or "").strip()
|
||
images, videos = _normalize_media_payload(payload.get("images") or [], payload.get("videos") or [])
|
||
conversation_id = payload.get("conversation_id")
|
||
if not message and not images and not videos:
|
||
return jsonify({"success": False, "error": "消息不能为空"}), 400
|
||
model_key = payload.get("model_key")
|
||
thinking_mode = payload.get("thinking_mode")
|
||
run_mode = payload.get("run_mode")
|
||
message_source = payload.get("message_source")
|
||
max_iterations = payload.get("max_iterations")
|
||
goal_mode = bool(payload.get("goal_mode"))
|
||
skill_context_messages: List[Dict[str, str]] = []
|
||
raw_skill_refs = payload.get("skill_refs")
|
||
if raw_skill_refs:
|
||
try:
|
||
_terminal, workspace = get_user_resources(username, workspace_id)
|
||
if not workspace:
|
||
return jsonify({"success": False, "error": "工作区不可用"}), 400
|
||
skill_context_messages = _build_skill_context_messages(workspace, raw_skill_refs)
|
||
except ValueError as exc:
|
||
return jsonify({"success": False, "error": str(exc)}), 400
|
||
except Exception as exc:
|
||
debug_log(f"[SkillsAPI] 读取技能上下文失败: {exc}")
|
||
return jsonify({"success": False, "error": "读取 skill 失败"}), 500
|
||
try:
|
||
debug_log(
|
||
"[TaskAPI] create_task payload "
|
||
f"model_key={model_key!r} run_mode={run_mode!r} thinking_mode={thinking_mode!r} "
|
||
f"conversation_id={conversation_id!r} images={len(images)} videos={len(videos)}"
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
rec = task_manager.create_chat_task(
|
||
username,
|
||
workspace_id,
|
||
message,
|
||
images,
|
||
conversation_id,
|
||
videos=videos,
|
||
model_key=model_key,
|
||
thinking_mode=thinking_mode,
|
||
run_mode=run_mode,
|
||
max_iterations=max_iterations,
|
||
message_source=message_source,
|
||
goal_mode=goal_mode,
|
||
skill_context_messages=skill_context_messages,
|
||
)
|
||
except RuntimeError as exc:
|
||
return jsonify({"success": False, "error": str(exc)}), 409
|
||
return jsonify({
|
||
"success": True,
|
||
"data": {
|
||
"task_id": rec.task_id,
|
||
"workspace_id": rec.workspace_id,
|
||
"status": rec.status,
|
||
"created_at": rec.created_at,
|
||
"conversation_id": rec.conversation_id,
|
||
}
|
||
}), 202
|
||
|
||
|
||
def _media_path(item: Any) -> str:
|
||
if isinstance(item, dict):
|
||
return str(item.get("path") or "")
|
||
return str(item or "")
|
||
|
||
|
||
def _is_video_item(item: Any) -> bool:
|
||
path = _media_path(item)
|
||
if not path:
|
||
return False
|
||
mime, _ = mimetypes.guess_type(path)
|
||
return bool(mime and mime.startswith("video/"))
|
||
|
||
|
||
def _is_image_item(item: Any) -> bool:
|
||
path = _media_path(item)
|
||
if not path:
|
||
return False
|
||
mime, _ = mimetypes.guess_type(path)
|
||
return bool(mime and mime.startswith("image/"))
|
||
|
||
|
||
def _normalize_media_payload(images: List[Any], videos: List[Any]) -> tuple[List[Any], List[Any]]:
|
||
"""纠偏媒体字段：把误传到 images 的视频项自动归入 videos。"""
|
||
fixed_images: List[Any] = []
|
||
fixed_videos: List[Any] = []
|
||
|
||
for item in images or []:
|
||
if _is_video_item(item):
|
||
fixed_videos.append(item)
|
||
else:
|
||
fixed_images.append(item)
|
||
|
||
for item in videos or []:
|
||
if _is_image_item(item):
|
||
fixed_images.append(item)
|
||
else:
|
||
fixed_videos.append(item)
|
||
|
||
return fixed_images, fixed_videos
|
||
|
||
|
||
@tasks_bp.route("/api/tasks/<task_id>", methods=["GET"])
|
||
@api_login_required
|
||
def get_task_api(task_id: str):
|
||
started_at = time.time()
|
||
username = get_current_username()
|
||
poll_req_id = request.headers.get("X-Task-Poll", "-")
|
||
rec = task_manager.get_task(username, task_id)
|
||
if not rec:
|
||
log_conn_diag(
|
||
f"task-poll-missing req={poll_req_id} user={username} task_id={task_id}"
|
||
)
|
||
return jsonify({"success": False, "error": "任务不存在"}), 404
|
||
try:
|
||
offset = int(request.args.get("from", 0))
|
||
except Exception:
|
||
offset = 0
|
||
events = [e for e in rec.events if e["idx"] >= offset]
|
||
next_offset = events[-1]["idx"] + 1 if events else offset
|
||
elapsed_ms = (time.time() - started_at) * 1000.0
|
||
should_log = (
|
||
offset == 0
|
||
or len(events) > 0
|
||
or rec.status != "running"
|
||
or elapsed_ms >= 800
|
||
)
|
||
if should_log:
|
||
log_conn_diag(
|
||
"task-poll "
|
||
f"req={poll_req_id} user={username} task_id={task_id} status={rec.status} "
|
||
f"from={offset} events={len(events)} next_offset={next_offset} "
|
||
f"elapsed_ms={elapsed_ms:.1f} slow={elapsed_ms >= 800}"
|
||
)
|
||
return jsonify({
|
||
"success": True,
|
||
"data": {
|
||
"task_id": rec.task_id,
|
||
"workspace_id": rec.workspace_id,
|
||
"status": rec.status,
|
||
"created_at": rec.created_at,
|
||
"updated_at": rec.updated_at,
|
||
"message": rec.message,
|
||
"conversation_id": rec.conversation_id,
|
||
"error": rec.error,
|
||
"message_source": (rec.session_data or {}).get("message_source"),
|
||
"goal_mode": bool((rec.session_data or {}).get("goal_mode")),
|
||
"goal_progress": (rec.session_data or {}).get("goal_progress"),
|
||
"events": events,
|
||
"next_offset": next_offset,
|
||
"runtime_queued_messages": task_manager.get_runtime_pending_messages(
|
||
username, task_id
|
||
),
|
||
}
|
||
})
|
||
|
||
|
||
@tasks_bp.route("/api/tasks/<task_id>/cancel", methods=["POST"])
|
||
@api_login_required
|
||
def cancel_task_api(task_id: str):
|
||
username = get_current_username()
|
||
ok = task_manager.cancel_task(username, task_id)
|
||
if not ok:
|
||
return jsonify({"success": False, "error": "任务不存在"}), 404
|
||
return jsonify({"success": True})
|
||
|
||
|
||
@tasks_bp.route("/api/tasks/<task_id>/runtime_guidance", methods=["POST"])
|
||
@api_login_required
|
||
def enqueue_runtime_guidance_api(task_id: str):
|
||
username = get_current_username()
|
||
payload = request.get_json() or {}
|
||
message = (payload.get("message") or "").strip()
|
||
if not message:
|
||
return jsonify({"success": False, "error": "引导内容不能为空"}), 400
|
||
|
||
result = task_manager.enqueue_runtime_guidance(username, task_id, message)
|
||
if not result.get("success"):
|
||
code = result.get("code") or "runtime_guidance_failed"
|
||
if code == "task_not_found":
|
||
return jsonify({"success": False, "error": result.get("error") or "任务不存在"}), 404
|
||
if code in {"queue_full", "task_not_running"}:
|
||
return jsonify({"success": False, "error": result.get("error") or "任务状态不允许"}), 409
|
||
return jsonify({"success": False, "error": result.get("error") or "追加引导失败"}), 400
|
||
|
||
return jsonify(
|
||
{
|
||
"success": True,
|
||
"data": {
|
||
"task_id": result.get("task_id"),
|
||
"queued_count": result.get("queued_count", 0),
|
||
},
|
||
}
|
||
)
|
||
|
||
|
||
@tasks_bp.route("/api/tasks/<task_id>/runtime_queue", methods=["POST"])
|
||
@api_login_required
|
||
def enqueue_runtime_queue_message_api(task_id: str):
|
||
username = get_current_username()
|
||
payload = request.get_json() or {}
|
||
message = (payload.get("message") or "").strip()
|
||
if not message:
|
||
return jsonify({"success": False, "error": "消息不能为空"}), 400
|
||
result = task_manager.enqueue_runtime_pending_message(username, task_id, message)
|
||
if not result.get("success"):
|
||
code = result.get("code") or "runtime_queue_enqueue_failed"
|
||
if code == "task_not_found":
|
||
return jsonify({"success": False, "error": result.get("error") or "任务不存在"}), 404
|
||
if code in {"queue_full", "task_not_running"}:
|
||
return jsonify({"success": False, "error": result.get("error") or "任务状态不允许"}), 409
|
||
return jsonify({"success": False, "error": result.get("error") or "追加消息失败"}), 400
|
||
return jsonify(
|
||
{
|
||
"success": True,
|
||
"data": {
|
||
"task_id": result.get("task_id"),
|
||
"item": result.get("item"),
|
||
"messages": result.get("messages") or [],
|
||
},
|
||
}
|
||
)
|
||
|
||
|
||
@tasks_bp.route("/api/tasks/<task_id>/runtime_queue/<message_id>", methods=["DELETE"])
|
||
@api_login_required
|
||
def delete_runtime_queue_message_api(task_id: str, message_id: str):
|
||
username = get_current_username()
|
||
result = task_manager.remove_runtime_pending_message(username, task_id, message_id)
|
||
if not result.get("success"):
|
||
code = result.get("code") or "runtime_queue_delete_failed"
|
||
if code == "task_not_found":
|
||
return jsonify({"success": False, "error": result.get("error") or "任务不存在"}), 404
|
||
if code == "message_not_found":
|
||
return jsonify({"success": False, "error": result.get("error") or "消息不存在"}), 404
|
||
return jsonify({"success": False, "error": result.get("error") or "删除失败"}), 400
|
||
return jsonify(
|
||
{
|
||
"success": True,
|
||
"data": {
|
||
"task_id": result.get("task_id"),
|
||
"messages": result.get("messages") or [],
|
||
},
|
||
}
|
||
)
|
||
|
||
|
||
@tasks_bp.route("/api/tasks/<task_id>/runtime_queue/<message_id>/guide", methods=["POST"])
|
||
@api_login_required
|
||
def guide_runtime_queue_message_api(task_id: str, message_id: str):
|
||
username = get_current_username()
|
||
result = task_manager.promote_runtime_pending_to_guidance(username, task_id, message_id)
|
||
if not result.get("success"):
|
||
code = result.get("code") or "runtime_queue_guide_failed"
|
||
if code == "task_not_found":
|
||
return jsonify({"success": False, "error": result.get("error") or "任务不存在"}), 404
|
||
if code == "message_not_found":
|
||
return jsonify({"success": False, "error": result.get("error") or "消息不存在"}), 404
|
||
if code in {"guidance_queue_full", "task_not_running"}:
|
||
return jsonify({"success": False, "error": result.get("error") or "任务状态不允许"}), 409
|
||
return jsonify({"success": False, "error": result.get("error") or "引导失败"}), 400
|
||
return jsonify(
|
||
{
|
||
"success": True,
|
||
"data": {
|
||
"task_id": result.get("task_id"),
|
||
"queued_count": result.get("queued_count", 0),
|
||
"messages": result.get("messages") or [],
|
||
},
|
||
}
|
||
)
|