572 lines
23 KiB
Python
572 lines
23 KiB
Python
"""External-task multi-agent scheduled analysis API (任务分析定时).
|
||
|
||
Two audiences share this router:
|
||
|
||
- **对方系统（external）**：创建/删除任务（推送式接入）、时间线详情（他们详情页
|
||
拉取渲染）、问答历史查询、**定时数据交付**（6 块 JSON + 时间轴期刻度）。鉴权统一
|
||
走现有 Bearer token（对方后端/前端用我方
|
||
登录接口换 token），不建独立 API-Key 机制。
|
||
- **我方管理页（internal）**：任务列表 / 绑定 CRUD / 运行记录与席位过程消息 /
|
||
手动触发 / 数据块标题与排序编辑。与任务分析前端页（official-task-analysis）一一对应。
|
||
|
||
数据交付（第三期）：智能体在定时执行里经 ``POST /data/submit`` 入库（内部接口，同样
|
||
走 Bearer，技能脚本缺省经公共 token 端点现签）；提交落在「提交时刻的当前最新一期」，
|
||
首次提交无期自动开期。合并规则：顶层 key 合并、同 key 后写覆盖。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any, Literal
|
||
|
||
from fastapi import APIRouter, HTTPException, Request
|
||
from pydantic import BaseModel, Field
|
||
|
||
from deerflow.persistence.task_analysis import TaskAnalysisStore
|
||
from deerflow.runtime.task_analysis.data_delivery import (
|
||
block_defs_payload,
|
||
period_detail_payload,
|
||
periods_payload,
|
||
submit_block_data,
|
||
)
|
||
from deerflow.runtime.task_analysis.live_messages import (
|
||
live_rows_to_process_messages,
|
||
merge_process_messages,
|
||
)
|
||
from deerflow.runtime.task_analysis.service import TaskAnalysisService, next_run_at
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/task-analysis", tags=["task-analysis"])
|
||
|
||
|
||
def _store(request: Request) -> TaskAnalysisStore:
|
||
store = getattr(request.app.state, "task_analysis_store", None)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Task analysis store not available")
|
||
return store
|
||
|
||
|
||
def _service(request: Request) -> TaskAnalysisService:
|
||
service = getattr(request.app.state, "task_analysis_service", None)
|
||
if service is None:
|
||
raise HTTPException(status_code=503, detail="Task analysis service not running")
|
||
return service
|
||
|
||
|
||
def _user_id(request: Request) -> str:
|
||
user = getattr(request.state, "user", None)
|
||
if user is not None:
|
||
return str(user.id)
|
||
from deerflow.runtime.user_context import get_effective_user_id
|
||
|
||
return get_effective_user_id()
|
||
|
||
|
||
async def _task_or_404(store: TaskAnalysisStore, *, task_id: str | None = None, source: str | None = None) -> dict[str, Any]:
|
||
task = (
|
||
await store.get_task_by_source(source) if source else await store.get_task(task_id or "")
|
||
)
|
||
if task is None:
|
||
raise HTTPException(status_code=404, detail="任务不存在")
|
||
return task
|
||
|
||
|
||
# ── 对外接口（对方系统调用；Bearer token 鉴权与内部接口一致） ────────────────
|
||
|
||
|
||
class ExternalTaskCreateRequest(BaseModel):
|
||
source_task_id: str = Field(min_length=1, max_length=128)
|
||
name: str = Field(min_length=1, max_length=512)
|
||
description: str | None = Field(default=None, max_length=8192)
|
||
|
||
|
||
@router.post("/external/tasks")
|
||
async def external_create_task(request: Request, body: ExternalTaskCreateRequest) -> dict:
|
||
"""对方创建完任务后调用；响应带 default_chain_id（我方记录的最近一次绑定链）。"""
|
||
store = _store(request)
|
||
try:
|
||
task = await store.create_task(
|
||
{"source_task_id": body.source_task_id, "name": body.name, "description": body.description}
|
||
)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||
return {"task_id": task["id"], "default_chain_id": await store.last_bound_chain_id()}
|
||
|
||
|
||
@router.delete("/external/tasks/{source_task_id}")
|
||
async def external_delete_task(request: Request, source_task_id: str) -> dict:
|
||
"""对方删除任务时同步删除：任务 + 绑定 + 执行记录 + 过程消息 + 时间线版本。"""
|
||
store = _store(request)
|
||
task = await _task_or_404(store, source=source_task_id)
|
||
await store.delete_task_cascade(task["id"])
|
||
return {"deleted": True, "task_id": task["id"]}
|
||
|
||
|
||
@router.get("/external/tasks/{source_task_id}/detail")
|
||
async def external_task_detail(
|
||
request: Request, source_task_id: str, limit: int = 20, offset: int = 0
|
||
) -> dict:
|
||
"""时间线详情：对方详情页按任务号拉取（版本倒序，含各块内容）。"""
|
||
store = _store(request)
|
||
task = await _task_or_404(store, source=source_task_id)
|
||
binding = await store.get_binding(task["id"])
|
||
versions = await store.list_versions(task["id"], limit=limit, offset=offset)
|
||
return {
|
||
"task": {"task_id": task["id"], "source_task_id": task["source_task_id"], "name": task["name"]},
|
||
"binding": {
|
||
"chain_id": binding.get("chain_id"),
|
||
"cron_expr": binding.get("cron_expr"),
|
||
"schedule_type": binding.get("schedule_type") or "cron",
|
||
"enabled": binding.get("enabled"),
|
||
}
|
||
if binding
|
||
else None,
|
||
"timeline": versions,
|
||
}
|
||
|
||
|
||
@router.get("/external/runs/{run_id}/messages")
|
||
async def external_run_messages(
|
||
request: Request, run_id: str, seat: str | None = None, limit: int = 500
|
||
) -> dict:
|
||
"""问答历史 / 执行过程查询（seat 过滤可只看某席位或 __coordinator__ 总控）。"""
|
||
store = _store(request)
|
||
run = await store.get_run(run_id)
|
||
if run is None:
|
||
raise HTTPException(status_code=404, detail="运行记录不存在")
|
||
messages = await _messages_for_run(request, run, seat=seat, limit=limit)
|
||
return {"run": {k: run[k] for k in ("id", "trigger", "status", "started_at", "finished_at")}, "messages": messages}
|
||
|
||
|
||
# ── 定时数据交付（6 块 JSON + 时间轴；第三期） ───────────────────────────────
|
||
|
||
|
||
class DataSubmitRequest(BaseModel):
|
||
# 外部任务号（对方系统的任务 id，非 DeerFlow 内部 task_id）。
|
||
task_id: str = Field(min_length=1, max_length=128)
|
||
block_key: str = Field(min_length=1, max_length=32)
|
||
# JSON 对象即可，不做 schema 校验（schema 会变）；非对象由 pydantic 拒绝（422）。
|
||
data: dict[str, Any]
|
||
|
||
|
||
@router.post("/data/submit")
|
||
async def submit_data_block(request: Request, body: DataSubmitRequest) -> dict:
|
||
"""智能体数据入库：落「提交时刻的当前最新一期」并按流水合并（顶层 key 合并、同 key 后写覆盖）。
|
||
|
||
供 task-data-submit 技能（席位执行时 bash 调用）使用；首次提交无期自动开期
|
||
（trigger=external），保证写入永远成功。
|
||
"""
|
||
store = _store(request)
|
||
source = body.task_id.strip()
|
||
task = await store.get_task_by_source(source)
|
||
if task is None:
|
||
raise HTTPException(status_code=404, detail="任务不存在")
|
||
try:
|
||
result = await submit_block_data(
|
||
store,
|
||
task_id=task["id"],
|
||
block_key=body.block_key,
|
||
data=body.data,
|
||
submitted_by=_user_id(request),
|
||
)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||
return {"source_task_id": task["source_task_id"], "task_id": task["id"], **result}
|
||
|
||
|
||
@router.get("/external/tasks/{source_task_id}/blocks")
|
||
async def external_task_blocks(request: Request, source_task_id: str) -> dict:
|
||
"""数据块定义列表（块 key 固定 block_1~block_6，标题/排序可经管理端修改）。"""
|
||
task = await _task_or_404(_store(request), source=source_task_id)
|
||
return await block_defs_payload(_store(request), task)
|
||
|
||
|
||
@router.get("/external/tasks/{source_task_id}/periods")
|
||
async def external_task_periods(request: Request, source_task_id: str, limit: int = 500) -> dict:
|
||
"""时间轴刻度列表（期号升序；含是否当前期与各块更新时间摘要，进行中的刻度实时可见）。"""
|
||
task = await _task_or_404(_store(request), source=source_task_id)
|
||
return await periods_payload(_store(request), task, limit=max(1, min(limit, 2000)))
|
||
|
||
|
||
@router.get("/external/tasks/{source_task_id}/periods/{period}")
|
||
async def external_task_period_detail(request: Request, source_task_id: str, period: str) -> dict:
|
||
"""某期 6 块完整 JSON + 各块 updated_at；``period`` 传期号数字或 ``latest``。"""
|
||
store = _store(request)
|
||
task = await _task_or_404(store, source=source_task_id)
|
||
if period != "latest" and not period.isdigit():
|
||
raise HTTPException(status_code=400, detail="period 必须是期号数字或 latest")
|
||
payload = await period_detail_payload(store, task, period)
|
||
if payload is None:
|
||
raise HTTPException(status_code=404, detail="期不存在")
|
||
return payload
|
||
|
||
|
||
# ── 内部管理接口（任务分析页） ───────────────────────────────────────────────
|
||
|
||
|
||
class BindingUpsertRequest(BaseModel):
|
||
# 业务链；缺省时使用「默认业务链」配置（default-chain 接口维护），都没有则 400。
|
||
chain_id: str | None = Field(default=None, max_length=64)
|
||
cron: str = Field(default="0 9 * * *", max_length=64)
|
||
timezone: str = Field(default="Asia/Shanghai", max_length=64)
|
||
# 执行方式：cron = 按频率定时；once = 只执行一次（创建即立刻执行，之后不再自动）。
|
||
# None = 更新时保持原值（新建缺省 cron），避免局部更新（如启停）把 once 改回 cron。
|
||
schedule_type: Literal["cron", "once"] | None = None
|
||
enabled: bool = True
|
||
model: str | None = Field(default=None, max_length=128)
|
||
coordinator_prompt: str | None = Field(default=None, max_length=8192)
|
||
|
||
|
||
class TaskCreateRequest(BaseModel):
|
||
name: str = Field(min_length=1, max_length=512)
|
||
source_task_id: str | None = Field(default=None, max_length=128)
|
||
description: str | None = Field(default=None, max_length=8192)
|
||
binding: BindingUpsertRequest | None = None
|
||
|
||
|
||
@router.get("/tasks")
|
||
async def list_tasks(request: Request) -> dict:
|
||
store = _store(request)
|
||
return {"tasks": await store.list_tasks()}
|
||
|
||
|
||
@router.post("/tasks")
|
||
async def create_task(request: Request, body: TaskCreateRequest) -> dict:
|
||
"""页面一步式创建：任务信息 + 可选绑定（业务链 + 频率 + 启用 + 总控提示）。"""
|
||
store = _store(request)
|
||
user_id = _user_id(request)
|
||
source = body.source_task_id or f"TS-{_next_source_suffix()}"
|
||
try:
|
||
task = await store.create_task(
|
||
{"source_task_id": source, "name": body.name, "description": body.description}
|
||
)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||
|
||
binding = None
|
||
if body.binding is not None:
|
||
binding = await _upsert_binding(request, store, task["id"], body.binding, user_id)
|
||
# 只执行一次：创建即立刻执行（enabled 关闭则不执行，等手动「立即执行」）。
|
||
run = None
|
||
if binding is not None and binding.get("schedule_type") == "once" and binding.get("enabled"):
|
||
try:
|
||
run = await _service(request).launch_binding(binding["id"], manual=False)
|
||
except Exception:
|
||
logger.exception("task-analysis once binding %s immediate launch failed", binding.get("id"))
|
||
return {"task": task, "binding": binding, "run": run}
|
||
|
||
|
||
@router.delete("/tasks/{task_id}")
|
||
async def delete_task(request: Request, task_id: str) -> dict:
|
||
store = _store(request)
|
||
task = await _task_or_404(store, task_id=task_id)
|
||
await store.delete_task_cascade(task["id"])
|
||
return {"deleted": True}
|
||
|
||
|
||
@router.put("/tasks/{task_id}/binding")
|
||
async def upsert_binding(request: Request, task_id: str, body: BindingUpsertRequest) -> dict:
|
||
store = _store(request)
|
||
task = await _task_or_404(store, task_id=task_id)
|
||
return {"binding": await _upsert_binding(request, store, task["id"], body, _user_id(request))}
|
||
|
||
|
||
@router.delete("/tasks/{task_id}/binding")
|
||
async def delete_binding(request: Request, task_id: str) -> dict:
|
||
store = _store(request)
|
||
task = await _task_or_404(store, task_id=task_id)
|
||
return {"deleted": await store.delete_binding(task["id"])}
|
||
|
||
|
||
@router.post("/tasks/{task_id}/trigger")
|
||
async def trigger_run(request: Request, task_id: str) -> dict:
|
||
"""立即执行一次（手动触发的定时语义执行，非总控问答）。"""
|
||
task = await _task_or_404(_store(request), task_id=task_id)
|
||
binding = await _store(request).get_binding(task["id"])
|
||
if binding is None:
|
||
raise HTTPException(status_code=400, detail="任务未绑定分析方案")
|
||
run = await _service(request).launch_binding(binding["id"], manual=True)
|
||
if run is None:
|
||
raise HTTPException(status_code=409, detail="执行被跳过（业务链缺失或已停用）")
|
||
return {"run": run}
|
||
|
||
|
||
class BlockDefUpdate(BaseModel):
|
||
block_key: str = Field(min_length=1, max_length=32)
|
||
title: str | None = Field(default=None, max_length=128)
|
||
sort_order: int | None = Field(default=None, ge=0, le=9999)
|
||
|
||
|
||
class BlockDefsUpdateRequest(BaseModel):
|
||
blocks: list[BlockDefUpdate] = Field(min_length=1, max_length=32)
|
||
|
||
|
||
@router.put("/tasks/{task_id}/blocks")
|
||
async def update_task_blocks(request: Request, task_id: str, body: BlockDefsUpdateRequest) -> dict:
|
||
"""修改数据块标题/排序（管理端；块集合固定 6 条，不能增删）。"""
|
||
store = _store(request)
|
||
task = await _task_or_404(store, task_id=task_id)
|
||
try:
|
||
defs = await store.update_block_defs(
|
||
task["id"], [block.model_dump() for block in body.blocks]
|
||
)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||
return {"blocks": defs}
|
||
|
||
|
||
@router.get("/tasks/{task_id}/runs")
|
||
async def list_runs(request: Request, task_id: str, limit: int = 50) -> list[dict]:
|
||
store = _store(request)
|
||
task = await _task_or_404(store, task_id=task_id)
|
||
return await store.list_runs(task["id"], limit=limit)
|
||
|
||
|
||
@router.get("/runs/{run_id}")
|
||
async def get_run(request: Request, run_id: str) -> dict:
|
||
run = await _store(request).get_run(run_id)
|
||
if run is None:
|
||
raise HTTPException(status_code=404, detail="运行记录不存在")
|
||
return run
|
||
|
||
|
||
@router.get("/runs/{run_id}/messages")
|
||
async def get_run_messages(
|
||
request: Request, run_id: str, seat: str | None = None, limit: int = 500
|
||
) -> dict:
|
||
store = _store(request)
|
||
run = await store.get_run(run_id)
|
||
if run is None:
|
||
raise HTTPException(status_code=404, detail="运行记录不存在")
|
||
return {"messages": await _messages_for_run(request, run, seat=seat, limit=limit)}
|
||
|
||
|
||
# ── helpers ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
async def _messages_for_run(
|
||
request: Request,
|
||
run: dict[str, Any],
|
||
*,
|
||
seat: str | None,
|
||
limit: int,
|
||
) -> list[dict[str, Any]]:
|
||
"""Stored orchestration notes + live agent events (same source as scheduled tasks)."""
|
||
store = _store(request)
|
||
stored = await store.list_run_messages(run["id"], seat=seat, limit=limit)
|
||
seat_ids = [seat] if seat else list((run.get("seat_results") or {}).keys())
|
||
if not seat_ids:
|
||
return stored
|
||
merged: list[dict[str, Any]] = []
|
||
for seat_id in seat_ids:
|
||
seat_stored = [row for row in stored if row.get("seat_agent_id") == seat_id] if not seat else stored
|
||
if seat and seat_id != seat:
|
||
continue
|
||
live_rows = await _live_agent_event_rows(request, run, seat_id)
|
||
live = live_rows_to_process_messages(
|
||
live_rows,
|
||
seat=seat_id,
|
||
stage_index=int((run.get("seat_results") or {}).get(seat_id, {}).get("stage_index") or 0),
|
||
)
|
||
merged.extend(merge_process_messages(stored=seat_stored, live=live, limit=limit))
|
||
if seat:
|
||
return merged[:limit]
|
||
# Unfiltered: keep original stored order when there is no live overlay.
|
||
if not any((row.get("metadata") or {}).get("source") == "live" for row in merged):
|
||
return stored[:limit]
|
||
return merged[:limit]
|
||
|
||
|
||
async def _live_agent_event_rows(
|
||
request: Request, run: dict[str, Any], seat: str
|
||
) -> list[dict[str, Any]]:
|
||
info = ((run.get("seat_results") or {}).get(seat) or {})
|
||
thread_id = info.get("thread_id")
|
||
agent_run_id = info.get("agent_run_id")
|
||
if not thread_id:
|
||
return []
|
||
rows: list[dict[str, Any]] = []
|
||
try:
|
||
from app.gateway.deps import get_run_event_store
|
||
|
||
event_store = get_run_event_store(request)
|
||
if agent_run_id:
|
||
rows = await event_store.list_messages_by_run(
|
||
thread_id, agent_run_id, limit=500, user_id=None
|
||
)
|
||
if not rows:
|
||
rows = await event_store.list_messages(thread_id, limit=500)
|
||
except Exception:
|
||
logger.debug("task-analysis live event-store read failed for seat %s", seat, exc_info=True)
|
||
rows = []
|
||
if rows:
|
||
return rows
|
||
return await _checkpoint_event_rows(request, thread_id)
|
||
|
||
|
||
async def _checkpoint_event_rows(request: Request, thread_id: str) -> list[dict[str, Any]]:
|
||
"""Same fallback scheduled tasks use: project the live LangGraph checkpoint."""
|
||
try:
|
||
from app.gateway.deps import get_checkpointer
|
||
from deerflow.runtime.serialization import serialize_channel_values
|
||
|
||
checkpointer = get_checkpointer(request)
|
||
tup = await checkpointer.aget_tuple(
|
||
{"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
||
)
|
||
except Exception:
|
||
logger.debug("task-analysis checkpoint read failed for %s", thread_id, exc_info=True)
|
||
return []
|
||
if tup is None:
|
||
return []
|
||
checkpoint = getattr(tup, "checkpoint", {}) or {}
|
||
try:
|
||
values = serialize_channel_values(checkpoint.get("channel_values") or {})
|
||
except Exception:
|
||
return []
|
||
messages = values.get("messages") or []
|
||
rows: list[dict[str, Any]] = []
|
||
for index, message in enumerate(messages):
|
||
if not isinstance(message, dict):
|
||
continue
|
||
raw_type = message.get("type") or message.get("role") or ""
|
||
if raw_type in {"human", "user"}:
|
||
event_type = "llm.human.input"
|
||
elif raw_type in {"ai", "assistant"}:
|
||
event_type = "llm.ai.response"
|
||
elif raw_type == "tool":
|
||
event_type = "llm.tool.result"
|
||
else:
|
||
continue
|
||
rows.append(
|
||
{
|
||
"seq": index + 1,
|
||
"event_type": event_type,
|
||
"content": message,
|
||
"metadata": {"source": "checkpoint"},
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
async def _upsert_binding(
|
||
request: Request,
|
||
store: TaskAnalysisStore,
|
||
external_task_id: str,
|
||
body: BindingUpsertRequest,
|
||
user_id: str,
|
||
) -> dict[str, Any]:
|
||
# 业务链缺省：使用模块设置里的默认业务链。
|
||
chain_id = body.chain_id or await store.get_setting("default_chain_id")
|
||
if not chain_id:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="未指定业务链，且未配置默认业务链（可在任务分析页或 PUT /default-chain 配置）",
|
||
)
|
||
# 绑定时固化业务链快照（seats/stages/goals/总控提示），执行前由 service 再刷新。
|
||
chain_snapshot = await _load_chain_snapshot(request, chain_id, user_id)
|
||
# 执行方式：请求未带时沿用现值（新建缺省 cron）；once 永不排期（next_run_at 恒空）。
|
||
existing = await store.get_binding(external_task_id)
|
||
schedule_type = body.schedule_type or (existing or {}).get("schedule_type") or "cron"
|
||
next_at = (
|
||
None if schedule_type == "once" else next_run_at(body.cron, body.timezone).isoformat()
|
||
)
|
||
return await store.upsert_binding(
|
||
external_task_id,
|
||
{
|
||
"chain_id": chain_id,
|
||
"chain_snapshot": chain_snapshot,
|
||
"cron_expr": body.cron,
|
||
"timezone": body.timezone,
|
||
"schedule_type": schedule_type,
|
||
"enabled": body.enabled,
|
||
"next_run_at": next_at,
|
||
"model": body.model,
|
||
"coordinator_prompt": body.coordinator_prompt,
|
||
"created_by": user_id,
|
||
},
|
||
)
|
||
|
||
|
||
async def _load_chain_snapshot(request: Request, chain_id: str, user_id: str) -> dict[str, Any]:
|
||
from deerflow.runtime.task_analysis.service import build_chain_snapshot
|
||
|
||
chain_store = getattr(request.app.state, "roundtable_chain_store", None)
|
||
if chain_store is None:
|
||
return {}
|
||
try:
|
||
chain = await chain_store.get_chain(chain_id, user_id)
|
||
except Exception:
|
||
logger.exception("task-analysis load chain %s for snapshot failed", chain_id)
|
||
return {}
|
||
if chain is None:
|
||
raise HTTPException(status_code=400, detail=f"业务链不存在或不可见：{chain_id}")
|
||
return build_chain_snapshot(chain)
|
||
|
||
|
||
def _next_source_suffix() -> str:
|
||
import time
|
||
|
||
return f"LOCAL-{int(time.time() * 1000) % 100000000:08d}"
|
||
|
||
|
||
# ── 默认业务链 / 任务完成状态 ──────────────────────────────────────────────
|
||
|
||
|
||
class DefaultChainUpdateRequest(BaseModel):
|
||
chain_id: str | None = Field(default=None, max_length=64)
|
||
|
||
|
||
def _is_admin(request: Request) -> bool:
|
||
"""宽松管理员判定：无鉴权部署视为单管理空间（对齐 business_mapping）。"""
|
||
user = getattr(request.state, "user", None)
|
||
return user is None or getattr(user, "system_role", None) == "admin"
|
||
|
||
|
||
@router.get("/default-chain")
|
||
async def get_default_chain(request: Request) -> dict:
|
||
"""当前默认业务链（绑定不传 chain_id 时使用）。"""
|
||
store = _store(request)
|
||
chain_id = await store.get_setting("default_chain_id")
|
||
title = None
|
||
if chain_id:
|
||
snapshot = await _load_chain_snapshot(request, chain_id, _user_id(request))
|
||
title = snapshot.get("title")
|
||
return {"chain_id": chain_id, "chain_title": title}
|
||
|
||
|
||
@router.put("/default-chain")
|
||
async def set_default_chain(request: Request, body: DefaultChainUpdateRequest) -> dict:
|
||
"""配置默认业务链（仅管理员；chain_id 传空清除）。"""
|
||
if not _is_admin(request):
|
||
raise HTTPException(status_code=403, detail="仅管理员可配置默认业务链")
|
||
store = _store(request)
|
||
if body.chain_id:
|
||
# 配置前校验链存在（快照能加载即存在；不存在会 400）。
|
||
await _load_chain_snapshot(request, body.chain_id, _user_id(request))
|
||
await store.set_setting("default_chain_id", body.chain_id)
|
||
else:
|
||
await store.set_setting("default_chain_id", "")
|
||
return await get_default_chain(request)
|
||
|
||
|
||
@router.get("/external/tasks/{source_task_id}/status")
|
||
async def get_external_task_status(request: Request, source_task_id: str) -> dict:
|
||
"""任务完成状态：一次都没执行成功过 = pending（未完成）；成功过 = completed（已完成）。"""
|
||
store = _store(request)
|
||
task = await _task_or_404(store, source=source_task_id)
|
||
runs = await store.list_runs(task["id"], limit=200)
|
||
succeeded = [r for r in runs if r.get("status") == "succeeded"]
|
||
last_finished = max(
|
||
(r.get("finished_at") for r in succeeded if r.get("finished_at")), default=None
|
||
)
|
||
return {
|
||
"source_task_id": source_task_id,
|
||
"task_id": task["id"],
|
||
"status": "completed" if succeeded else "pending",
|
||
"succeeded_runs": len(succeeded),
|
||
"total_runs": len(runs),
|
||
"last_finished_at": last_finished,
|
||
}
|