JOJO
2bfbddb0dd
- 补全从 api.py 拆到 skills.py 时遗漏的 _build_skill_context_messages 导入 - 捕获 skill 文件读取时的 UnicodeDecodeError/OSError，转为 400 错误 - 在 skill 路径解析与读取流程中增加详细 debug_log
323 lines
14 KiB
Python
323 lines
14 KiB
Python
"""简单任务 API：将聊天任务与 WebSocket 解耦，支持后台运行与轮询。"""
|
||
from __future__ import annotations
|
||
from server.tasks import tasks_bp
|
||
import mimetypes
|
||
import json
|
||
import time
|
||
import threading
|
||
import uuid
|
||
import traceback
|
||
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 server.auth_helpers import api_login_required, get_current_username
|
||
from server.context import get_user_resources, ensure_conversation_loaded
|
||
from server.chat_flow import run_chat_task_sync
|
||
from server.state import stop_flags
|
||
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 server.tasks import task_manager
|
||
from server.tasks.skills import _build_skill_context_messages
|
||
from server.tasks.helpers import _task_public_payload
|
||
from server.tasks.media import _normalize_media_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/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"))
|
||
# 用户显式发送非目标模式消息时，若工作区仍有上一轮残留的活动目标，先清理，
|
||
# 避免新对话错误继承旧目标状态。
|
||
if not goal_mode:
|
||
try:
|
||
_terminal, workspace = get_user_resources(username, workspace_id)
|
||
if workspace:
|
||
gsm = GoalStateManager(workspace.data_dir)
|
||
if gsm.is_active():
|
||
gsm.mark_stopped("new_conversation")
|
||
debug_log(f"[Goal] 新任务未开启目标模式，清理工作区残留目标状态")
|
||
except Exception as exc:
|
||
debug_log(f"[Goal] 新任务清理残留目标状态失败: {exc}")
|
||
skill_context_messages: List[Dict[str, str]] = []
|
||
raw_skill_refs = payload.get("skill_refs")
|
||
if raw_skill_refs:
|
||
try:
|
||
debug_log(
|
||
f"[SkillsAPI] create_task skill_refs type={type(raw_skill_refs).__name__} "
|
||
f"count={len(raw_skill_refs) if isinstance(raw_skill_refs, list) else 'N/A'} "
|
||
f"refs={raw_skill_refs!r}"
|
||
)
|
||
_terminal, workspace = get_user_resources(username, workspace_id)
|
||
if not workspace:
|
||
debug_log(f"[SkillsAPI] create_task workspace unavailable user={username} ws={workspace_id}")
|
||
return jsonify({"success": False, "error": "工作区不可用"}), 400
|
||
debug_log(
|
||
f"[SkillsAPI] create_task workspace OK project_path={getattr(workspace, 'project_path', None)!r} "
|
||
f"data_dir={getattr(workspace, 'data_dir', None)!r}"
|
||
)
|
||
skill_context_messages = _build_skill_context_messages(workspace, raw_skill_refs)
|
||
debug_log(f"[SkillsAPI] create_task built {len(skill_context_messages)} skill context messages")
|
||
except ValueError as exc:
|
||
debug_log(f"[SkillsAPI] create_task ValueError: {exc}")
|
||
return jsonify({"success": False, "error": str(exc)}), 400
|
||
except Exception as exc:
|
||
debug_log(f"[SkillsAPI] 读取技能上下文失败: {exc}")
|
||
debug_log(f"[SkillsAPI] 读取技能上下文失败 traceback:\n{traceback.format_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
|
||
|
||
@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()
|
||
rec = task_manager.get_task(username, task_id)
|
||
if not rec:
|
||
return jsonify({"success": False, "error": "任务不存在"}), 404
|
||
ok = task_manager.cancel_task(username, task_id)
|
||
# 用户取消任务时，一并停止该工作区的目标模式，避免后续新对话继承旧目标。
|
||
try:
|
||
if ok and rec.workspace_id:
|
||
_, 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}")
|
||
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 [],
|
||
},
|
||
}
|
||
)
|