mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 22:31:59 +00:00
发起运行的规则抽成 engine/launch.py 的 launch_run：以前全写在 api/runs.py 的 start_run 里、只认 HTTP 请求，后面的定时任务要用同一套规则发起正式运行，另抄一份迟早走样。正式运行只能从当前发布版本发起、口径升版未声明策略就拦、绑定的工具不存在就拦、发起后记受管级别那条日志和目录变化、口径升版处置，全部搬进服务；拒绝时抛 LaunchRefused(status_code, message, code)，start_run 只剩一层包装，把它换回和以前逐字相同的响应（带机器码的仍回 {detail, code}）。正式运行的快照图不成形，以前由 main.py 的 ValueError 兜底回 400，现在在服务里换成同样的 400 拒绝，调度器也拿得到明确的拒绝。node_setting、tool_bindings、missing_tools 跟着搬过去，runs.py 再导出，旧的导入路径照旧可用；runner 固定数据版本取绑定时改从 engine/launch 导入，不再依赖接口层。新测试先把 HTTP 的 17 种拒绝（含工具缺失的数据、工具、混合、超过 8 处四种说法）按原话写死、在旧代码上通过，再逐条比对直接调 launch_run 的结果；另测服务发起的正式运行照样写治理事件
This commit is contained in:
ThearchyHelios
+23
-220
@@ -2,7 +2,6 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
from urllib.parse import unquote
|
||||
@@ -12,24 +11,17 @@ from pydantic import BaseModel, Field
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from sqlalchemy import select as _select
|
||||
|
||||
from app.api.coded import CodedHTTPException
|
||||
from app.core.bus import bus
|
||||
from app.core import artifact_store
|
||||
from app.core.events import EventType
|
||||
from app.data.catalog_impact import catalog_drift
|
||||
from app.db.base import SessionLocal, get_session
|
||||
from app.db.models import Approval, Artifact, Run, RunEvent, Workflow, WorkflowVersion
|
||||
from app.engine.governance import governance_note, unresolved_caliber_upgrades
|
||||
from app.db.models import Approval, Artifact, Run, RunEvent
|
||||
# 发起规则和发起前检查绑定的工具搬到了 engine/launch.py。这几个名字以前从这里导入，照旧可用
|
||||
from app.engine.launch import ( # noqa: F401
|
||||
TOOL_MISSING, LaunchRefused, launch_run, missing_tools, node_setting, tool_bindings,
|
||||
)
|
||||
from app.engine.runner import run_manager
|
||||
from app.engine.schema import GraphSpec, NodeType, type_label
|
||||
|
||||
router = APIRouter(prefix="/api/runs", tags=["runs"])
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: 发起时发现绑定的工具在本机不存在（见 missing_tools）
|
||||
TOOL_MISSING = "run_tool_missing"
|
||||
|
||||
|
||||
def actor_of(header: str | None) -> str | None:
|
||||
@@ -90,218 +82,29 @@ async def start_run(
|
||||
session: AsyncSession = Depends(get_session),
|
||||
x_actor: str | None = Header(default=None),
|
||||
) -> Run:
|
||||
actor = actor_of(x_actor)
|
||||
# 没挂在工作流上（画布没保存就跑、问数据当场搭的）。老数据里是「临时图」，前端两种都认
|
||||
name = "未保存的工作流"
|
||||
version: int | None = None
|
||||
version_hash: str | None = None
|
||||
governed = False
|
||||
|
||||
if payload.run_class == "formal":
|
||||
# 正式运行的全部语义就这一条规则：必须引用一个不可变的已发布版本。
|
||||
# 传裸 graph、或工作流还没发布过，都进不了 formal。
|
||||
if not payload.workflow_id:
|
||||
raise HTTPException(400, "正式运行需要从已保存的工作流发起：请求中没有指定工作流")
|
||||
if payload.graph is not None:
|
||||
raise HTTPException(400, "正式运行只能使用已发布的版本，画布上未发布的修改不会生效。请先保存并发布")
|
||||
workflow = await session.get(Workflow, payload.workflow_id)
|
||||
if not workflow:
|
||||
raise HTTPException(404, "工作流不存在，可能已被删除")
|
||||
# 只看 published_version：status 反映的是当前画布（改过图就退回 draft），
|
||||
# 而 formal 跑的是已发布的那一版，两者本来就可以不一致
|
||||
if not workflow.published_version:
|
||||
raise HTTPException(409, f"「{workflow.name}」还没有发布版本，请先在编排页发布")
|
||||
version = payload.version or workflow.published_version
|
||||
# "不可变"和"过过闸"是两件事。快照确实不可变，但 PATCH 保存出来的
|
||||
# 草稿版本同样有快照——只认 version 存不存在的话，工作流只要发布过
|
||||
# 任意一个版本，之后所有草稿都能以 formal 身份启动，发布期的
|
||||
# validate_graph + lint_for_publish 全部绕过。
|
||||
if version != workflow.published_version:
|
||||
raise HTTPException(
|
||||
409,
|
||||
f"v{version} 不是当前发布版本（v{workflow.published_version}）。"
|
||||
"正式运行只能使用当前发布版本；如需运行这一版，请先发布它。",
|
||||
)
|
||||
snapshot = (
|
||||
await session.execute(
|
||||
_select(WorkflowVersion).where(
|
||||
WorkflowVersion.workflow_id == workflow.id,
|
||||
WorkflowVersion.version == version,
|
||||
)
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if not snapshot:
|
||||
raise HTTPException(404, f"版本 v{version} 不存在")
|
||||
graph = snapshot.graph
|
||||
version_hash = snapshot.graph_hash or artifact_store.graph_hash(graph)
|
||||
name = workflow.name
|
||||
# 按这一版发布时记下的级别；记级别之前发布的老版本没有，照旧看工作流现在的 status
|
||||
governed = (snapshot.level == "governed") if snapshot.level else workflow.status == "governed"
|
||||
|
||||
# 口径升版是被迫处置的事件：钉住的方法卡有新版本而未声明策略，拒绝启动
|
||||
spec = GraphSpec.model_validate(graph)
|
||||
errors, upgrade_events = await unresolved_caliber_upgrades(session, spec)
|
||||
if errors:
|
||||
raise HTTPException(409, "；".join(errors))
|
||||
# 数据目录和口径卡不同：目录是给人和助手看的说明，不是运行的输入。发布之后变了只提醒、不拦——
|
||||
# 拦下来等于让一次文字修订卡住正式出具
|
||||
catalog_changes = await catalog_drift(session, snapshot.catalog_versions)
|
||||
else:
|
||||
graph = payload.graph
|
||||
upgrade_events = []
|
||||
catalog_changes = []
|
||||
if payload.workflow_id:
|
||||
workflow = await session.get(Workflow, payload.workflow_id)
|
||||
if not workflow:
|
||||
raise HTTPException(404, "工作流不存在，可能已被删除")
|
||||
graph = graph or workflow.graph
|
||||
name = workflow.name
|
||||
if not graph:
|
||||
raise HTTPException(400, "没有要运行的内容：请求中既没有工作流 ID，也没有工作流内容")
|
||||
await _refuse_missing_tools(session, graph)
|
||||
|
||||
from app.api.settings import resolve_run_scope
|
||||
|
||||
memory_scope, collection = await resolve_run_scope(
|
||||
session, payload.memory_scope, payload.collection
|
||||
)
|
||||
"""发起规则全在 engine/launch.py（定时任务到点发起也用它），这里只把拒绝换回 HTTP 响应。"""
|
||||
try:
|
||||
run = await run_manager.start(
|
||||
graph=graph,
|
||||
input_payload=payload.input,
|
||||
return await launch_run(
|
||||
session,
|
||||
workflow_id=payload.workflow_id,
|
||||
workflow_name=name,
|
||||
memory_scope=memory_scope,
|
||||
collection=collection,
|
||||
run_class="formal" if payload.run_class == "formal" else "exploratory",
|
||||
version=version,
|
||||
version_hash=version_hash,
|
||||
started_by=actor,
|
||||
graph=payload.graph,
|
||||
input_payload=payload.input,
|
||||
memory_scope=payload.memory_scope,
|
||||
collection=payload.collection,
|
||||
run_class=payload.run_class,
|
||||
version=payload.version,
|
||||
actor=actor_of(x_actor),
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(400, str(e)) from e
|
||||
|
||||
if payload.run_class == "formal":
|
||||
# 按不按受管出具在发起这一刻定下来、记进封存范围：status 反映的是当前画布，中途改图、
|
||||
# 审批停下再恢复时再去读它，报告节点和出口复核可能各读到不同的值
|
||||
await run_manager.note(run.id, "log", **governance_note(
|
||||
governed=governed, workflow_name=name, version=version))
|
||||
if catalog_changes:
|
||||
# 落在封存范围内：之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表。
|
||||
# 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的（界面分开说）
|
||||
await run_manager.note(run.id, EventType.CATALOG_DRIFT, version=version, count=len(catalog_changes),
|
||||
tables=catalog_changes)
|
||||
# 已声明策略的升版记入事件流，出具物上能看到"这期换口径了、怎么处置的"
|
||||
for ev in upgrade_events:
|
||||
payload_ev = dict(ev)
|
||||
node = payload_ev.pop("node_id", None)
|
||||
await run_manager.note(run.id, "caliber.upgrade", node_id=node, **payload_ev)
|
||||
return run
|
||||
except LaunchRefused as e:
|
||||
raise _refused(e) from e
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 发起前检查绑定的工具
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def node_setting(spec: GraphSpec, node: Any, key: str) -> Any:
|
||||
"""节点配置的实际取值，和运行时 NodeContext.cfg 同一个规则：节点上没写（None、空串）就取图级 defaults。"""
|
||||
value = node.config.get(key)
|
||||
return spec.defaults.get(key) if value in (None, "") else value
|
||||
|
||||
|
||||
def tool_bindings(spec: GraphSpec) -> list[tuple[str, str]]:
|
||||
"""图里每一处绑定的工具：(谁绑的, 工具名)。调用工具节点的 tool、Agent 的 tools、协作成员的 tools。
|
||||
|
||||
取值和运行时一致（node_setting）：Agent 节点上没写 tools、协作节点上没写成员，运行时用的是
|
||||
图级 defaults 里的，这里也算它们——不然发起前检查放过了缺的工具，发起时固定数据版本也漏了源。
|
||||
|
||||
模板拼出来的名字（{{ vars.tool }}）也在内：发起前检查（_bindings）跳过它们，发起时固定数据
|
||||
版本（engine/runner.py）要知道有没有这种名字——有的话说不准会用到哪个数据源。
|
||||
"""
|
||||
out: list[tuple[str, str]] = []
|
||||
for node in spec.nodes:
|
||||
who = f"「{node.title}」（{type_label(node.type)}）"
|
||||
if node.type == NodeType.TOOL:
|
||||
names = [node_setting(spec, node, "tool")]
|
||||
out.extend((who, n) for n in names if isinstance(n, str) and n.strip())
|
||||
elif node.type == NodeType.AGENT:
|
||||
out.extend((who, n) for n in node_setting(spec, node, "tools") or []
|
||||
if isinstance(n, str) and n.strip())
|
||||
elif node.type == NodeType.SUPERVISOR:
|
||||
for member in node_setting(spec, node, "agents") or []:
|
||||
if not isinstance(member, dict):
|
||||
continue
|
||||
label = f"「{node.title}」的成员「{member.get('name') or '?'}」"
|
||||
out.extend((label, n) for n in member.get("tools") or [] if isinstance(n, str) and n.strip())
|
||||
return [(who, name.strip()) for who, name in out]
|
||||
|
||||
|
||||
def _bindings(spec: GraphSpec) -> list[tuple[str, str]]:
|
||||
"""发起前要检查的绑定。模板拼出来的工具名（{{ vars.tool }}）发起时还不知道是谁，交给运行时。"""
|
||||
return [(who, name) for who, name in tool_bindings(spec) if "{{" not in name]
|
||||
|
||||
|
||||
async def missing_tools(session: AsyncSession, spec: GraphSpec) -> list[tuple[str, str, str]]:
|
||||
"""本机不存在的绑定：(谁绑的, 工具名, 去哪接入：data / tools)。
|
||||
|
||||
解析和运行时 build_tools 一致：内置（静态注册）、mcp:<服务>/<工具>、db_query__ / db_schema__
|
||||
<数据源>、其余按自定义工具。数据源和自定义工具只认启用着的——停用的运行时同样拿不到。
|
||||
MCP 只看服务登记着、启用着没有：连不连得上、工具还在不在，要连一次才知道，发起时不去等它，
|
||||
上次没连上或者工具清单里没有的只在日志里提一句。
|
||||
"""
|
||||
from app.db.models import CustomTool, DataSource, McpServer
|
||||
from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX
|
||||
from app.tools.registry import all_specs
|
||||
|
||||
bindings = _bindings(spec)
|
||||
if not bindings:
|
||||
return []
|
||||
builtin = all_specs()
|
||||
sources = {n for (n,) in await session.execute(select(DataSource.name).where(DataSource.enabled.is_(True)))}
|
||||
custom = {n for (n,) in await session.execute(select(CustomTool.name).where(CustomTool.enabled.is_(True)))}
|
||||
servers = {r.name: r for r in (await session.execute(select(McpServer))).scalars()}
|
||||
missing: list[tuple[str, str, str]] = []
|
||||
for who, name in bindings:
|
||||
if name.startswith("mcp:"):
|
||||
server, _, tool = name[4:].partition("/")
|
||||
row = servers.get(server)
|
||||
if row is None or not row.enabled:
|
||||
missing.append((who, name, "tools"))
|
||||
elif row.status == "error":
|
||||
logger.warning("%s 绑的 %s：MCP 服务 %s 上次没连上，运行时可能拿不到这个工具", who, name, server)
|
||||
elif tool not in ("", "*") and row.tools_cache and tool not in row.tools_cache:
|
||||
logger.warning("%s 绑的 %s：MCP 服务 %s 上次交回的工具清单里没有 %s", who, name, server, tool)
|
||||
elif name in builtin:
|
||||
continue
|
||||
elif name.startswith((QUERY_PREFIX, SCHEMA_PREFIX)):
|
||||
source = name[len(QUERY_PREFIX):] if name.startswith(QUERY_PREFIX) else name[len(SCHEMA_PREFIX):]
|
||||
if source not in sources:
|
||||
missing.append((who, name, "data"))
|
||||
elif name not in custom:
|
||||
missing.append((who, name, "tools"))
|
||||
return missing
|
||||
|
||||
|
||||
async def _refuse_missing_tools(session: AsyncSession, graph: dict[str, Any]) -> None:
|
||||
"""绑定的工具在本机不存在就不发起：以前要跑到那一步才失败，前面的步骤白跑、钱白花。"""
|
||||
try:
|
||||
spec = GraphSpec.model_validate(graph)
|
||||
except Exception: # noqa: BLE001 - 图本身不成形交给 run_manager.start 报
|
||||
return
|
||||
missing = await missing_tools(session, spec)
|
||||
if not missing:
|
||||
return
|
||||
lines = "；".join(f"{who}绑定的 {name}" for who, name, _ in missing[:8])
|
||||
more = f"等 {len(missing)} 处" if len(missing) > 8 else ""
|
||||
places = {where for _, _, where in missing}
|
||||
if places == {"data"}:
|
||||
fix = "请到「数据」页接入，或在节点中重新选择"
|
||||
elif places == {"tools"}:
|
||||
fix = "请到「工具」页接入，或在节点中重新选择"
|
||||
else:
|
||||
fix = "数据源工具请到「数据」页接入，其他工具请到「工具」页接入，或在节点中重新选择"
|
||||
raise CodedHTTPException(422, f"绑定的工具不存在：{lines}{more}。{fix}", TOOL_MISSING)
|
||||
def _refused(e: LaunchRefused) -> HTTPException:
|
||||
"""发起被拒的 HTTP 响应，和规则搬去 engine/launch.py 之前逐字相同：带机器码的回 {detail, code}，
|
||||
不带的回 {detail}。"""
|
||||
if e.code:
|
||||
return CodedHTTPException(e.status_code, e.message, e.code)
|
||||
return HTTPException(e.status_code, e.message)
|
||||
|
||||
|
||||
#: workflow_id 取这个值时只要没挂在任何工作流上的运行：画布上没保存就跑的临时图，
|
||||
|
||||
@@ -0,0 +1,273 @@
|
||||
"""发起一次运行：HTTP 发起（api/runs.py 的 start_run）和定时任务到点发起共用的那套规则。
|
||||
|
||||
以前这些规则全写在 start_run 里、只认 HTTP 请求：定时任务要发起正式运行，要么另抄一份（两份规则迟早
|
||||
走样，正式运行「只能从当前发布版本发起」这条最要紧的规则就可能在一边被绕过），要么假装成一次 HTTP 请求。
|
||||
现在两边都调 launch_run。拒绝时抛 LaunchRefused：start_run 把它换回以前的 HTTP 响应（状态码、detail、
|
||||
code 逐字不变），调度器据此记一次未能发起。
|
||||
|
||||
发起前检查绑定的工具（missing_tools），以及它和 runner 固定数据版本共用的取绑定的办法（tool_bindings、
|
||||
node_setting）也放在这里：runner 以前要从接口层导入它们。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core import artifact_store
|
||||
from app.core.events import EventType
|
||||
from app.data.catalog_impact import catalog_drift
|
||||
from app.db.models import Run, Workflow, WorkflowVersion
|
||||
from app.engine.governance import governance_note, unresolved_caliber_upgrades
|
||||
from app.engine.runner import run_manager
|
||||
from app.engine.schema import GraphSpec, NodeType, type_label
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: 发起时发现绑定的工具在本机不存在（见 missing_tools）
|
||||
TOOL_MISSING = "run_tool_missing"
|
||||
|
||||
|
||||
class LaunchRefused(Exception):
|
||||
"""不发起这次运行。status_code 和 message 就是 HTTP 发起时回的状态码和 detail；code 是给程序认的机器码，
|
||||
没有的是 None（HTTP 响应里也就没有 code 字段）。"""
|
||||
|
||||
def __init__(self, status_code: int, message: str, code: str | None = None) -> None:
|
||||
super().__init__(message)
|
||||
self.status_code = status_code
|
||||
self.message = message
|
||||
self.code = code
|
||||
|
||||
|
||||
async def launch_run(
|
||||
session: AsyncSession,
|
||||
*,
|
||||
workflow_id: str | None = None,
|
||||
graph: dict[str, Any] | None = None,
|
||||
input_payload: dict[str, Any] | None = None,
|
||||
memory_scope: str | None = None,
|
||||
collection: str | None = None,
|
||||
run_class: str = "exploratory",
|
||||
version: int | None = None,
|
||||
actor: str | None,
|
||||
) -> Run:
|
||||
"""发起一次运行，返回刚建好的运行记录（执行已经交给 run_manager）。不发起就抛 LaunchRefused。
|
||||
|
||||
参数和 POST /api/runs 的请求体一一对应（input 叫 input_payload）；actor 是记在运行上的发起人。
|
||||
"""
|
||||
# 没挂在工作流上（画布没保存就发起、问数据当场搭的）。老数据里是「临时图」，前端两种都认
|
||||
name = "未保存的工作流"
|
||||
run_version: int | None = None
|
||||
version_hash: str | None = None
|
||||
governed = False
|
||||
formal = run_class == "formal"
|
||||
|
||||
if formal:
|
||||
# 正式运行的全部语义就这一条规则：必须引用一个不可变的已发布版本。
|
||||
# 传裸 graph、或工作流还没发布过，都进不了 formal。
|
||||
if not workflow_id:
|
||||
raise LaunchRefused(400, "正式运行需要从已保存的工作流发起：请求中没有指定工作流")
|
||||
if graph is not None:
|
||||
raise LaunchRefused(400, "正式运行只能使用已发布的版本，画布上未发布的修改不会生效。请先保存并发布")
|
||||
workflow = await session.get(Workflow, workflow_id)
|
||||
if not workflow:
|
||||
raise LaunchRefused(404, "工作流不存在，可能已被删除")
|
||||
# 只看 published_version：status 反映的是当前画布（改过图就退回 draft），
|
||||
# 而 formal 用的是已发布的那一版，两者本来就可以不一致
|
||||
if not workflow.published_version:
|
||||
raise LaunchRefused(409, f"「{workflow.name}」还没有发布版本，请先在编排页发布")
|
||||
run_version = version or workflow.published_version
|
||||
# "不可变"和"过过闸"是两件事。快照确实不可变，但 PATCH 保存出来的
|
||||
# 草稿版本同样有快照——只认 version 存不存在的话，工作流只要发布过
|
||||
# 任意一个版本，之后所有草稿都能以 formal 身份启动，发布期的
|
||||
# validate_graph + lint_for_publish 全部绕过。
|
||||
if run_version != workflow.published_version:
|
||||
raise LaunchRefused(
|
||||
409,
|
||||
f"v{run_version} 不是当前发布版本（v{workflow.published_version}）。"
|
||||
"正式运行只能使用当前发布版本；如需运行这一版，请先发布它。",
|
||||
)
|
||||
snapshot = (
|
||||
await session.execute(
|
||||
select(WorkflowVersion).where(
|
||||
WorkflowVersion.workflow_id == workflow.id,
|
||||
WorkflowVersion.version == run_version,
|
||||
)
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if not snapshot:
|
||||
raise LaunchRefused(404, f"版本 v{run_version} 不存在")
|
||||
graph = snapshot.graph
|
||||
version_hash = snapshot.graph_hash or artifact_store.graph_hash(graph)
|
||||
name = workflow.name
|
||||
# 按这一版发布时记下的级别；记级别之前发布的老版本没有，照旧看工作流现在的 status
|
||||
governed = (snapshot.level == "governed") if snapshot.level else workflow.status == "governed"
|
||||
|
||||
try:
|
||||
spec = GraphSpec.model_validate(graph)
|
||||
except ValueError as e:
|
||||
# 快照里的图不成形。以前这里没接，由 main.py 的 ValueError 兜底回 400、detail 是校验原话；
|
||||
# 原样换成拒绝，HTTP 响应不变，调度器也拿得到一个明确的拒绝
|
||||
raise LaunchRefused(400, str(e)) from e
|
||||
# 口径升版是被迫处置的事件：引用的方法卡有新版本而未声明策略，拒绝启动
|
||||
errors, upgrade_events = await unresolved_caliber_upgrades(session, spec)
|
||||
if errors:
|
||||
raise LaunchRefused(409, "；".join(errors))
|
||||
# 数据目录和口径卡不同：目录是给人和助手看的说明，不是运行的输入。发布之后变了只提醒、不拦——
|
||||
# 拦下来等于让一次文字修订卡住正式出具
|
||||
catalog_changes = await catalog_drift(session, snapshot.catalog_versions)
|
||||
else:
|
||||
upgrade_events = []
|
||||
catalog_changes = []
|
||||
if workflow_id:
|
||||
workflow = await session.get(Workflow, workflow_id)
|
||||
if not workflow:
|
||||
raise LaunchRefused(404, "工作流不存在，可能已被删除")
|
||||
graph = graph or workflow.graph
|
||||
name = workflow.name
|
||||
if not graph:
|
||||
raise LaunchRefused(400, "没有要运行的内容：请求中既没有工作流 ID，也没有工作流内容")
|
||||
await _refuse_missing_tools(session, graph)
|
||||
|
||||
from app.api.settings import resolve_run_scope
|
||||
|
||||
memory_scope, collection = await resolve_run_scope(session, memory_scope, collection)
|
||||
try:
|
||||
run = await run_manager.start(
|
||||
graph=graph,
|
||||
input_payload=input_payload if input_payload is not None else {},
|
||||
workflow_id=workflow_id,
|
||||
workflow_name=name,
|
||||
memory_scope=memory_scope,
|
||||
collection=collection,
|
||||
run_class="formal" if formal else "exploratory",
|
||||
version=run_version,
|
||||
version_hash=version_hash,
|
||||
started_by=actor,
|
||||
)
|
||||
except ValueError as e:
|
||||
raise LaunchRefused(400, str(e)) from e
|
||||
|
||||
if formal:
|
||||
# 按不按受管出具在发起这一刻定下来、记进封存范围：status 反映的是当前画布，中途改图、
|
||||
# 审批停下再恢复时再去读它，报告节点和出口复核可能各读到不同的值
|
||||
await run_manager.note(run.id, "log", **governance_note(
|
||||
governed=governed, workflow_name=name, version=run_version))
|
||||
if catalog_changes:
|
||||
# 落在封存范围内：之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表。
|
||||
# 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的（界面分开说）
|
||||
await run_manager.note(run.id, EventType.CATALOG_DRIFT, version=run_version, count=len(catalog_changes),
|
||||
tables=catalog_changes)
|
||||
# 已声明策略的升版记入事件流，出具物上能看到"这期换口径了、怎么处置的"
|
||||
for ev in upgrade_events:
|
||||
payload_ev = dict(ev)
|
||||
node = payload_ev.pop("node_id", None)
|
||||
await run_manager.note(run.id, "caliber.upgrade", node_id=node, **payload_ev)
|
||||
return run
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 发起前检查绑定的工具
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def node_setting(spec: GraphSpec, node: Any, key: str) -> Any:
|
||||
"""节点配置的实际取值，和运行时 NodeContext.cfg 同一个规则：节点上没写（None、空串）就取图级 defaults。"""
|
||||
value = node.config.get(key)
|
||||
return spec.defaults.get(key) if value in (None, "") else value
|
||||
|
||||
|
||||
def tool_bindings(spec: GraphSpec) -> list[tuple[str, str]]:
|
||||
"""图里每一处绑定的工具：(谁绑的, 工具名)。调用工具节点的 tool、Agent 的 tools、协作成员的 tools。
|
||||
|
||||
取值和运行时一致（node_setting）：Agent 节点上没写 tools、协作节点上没写成员，运行时用的是
|
||||
图级 defaults 里的，这里也算它们——不然发起前检查放过了缺的工具，发起时固定数据版本也漏了源。
|
||||
|
||||
模板拼出来的名字（{{ vars.tool }}）也在内：发起前检查（_bindings）跳过它们，发起时固定数据
|
||||
版本（engine/runner.py）要知道有没有这种名字——有的话说不准会用到哪个数据源。
|
||||
"""
|
||||
out: list[tuple[str, str]] = []
|
||||
for node in spec.nodes:
|
||||
who = f"「{node.title}」（{type_label(node.type)}）"
|
||||
if node.type == NodeType.TOOL:
|
||||
names = [node_setting(spec, node, "tool")]
|
||||
out.extend((who, n) for n in names if isinstance(n, str) and n.strip())
|
||||
elif node.type == NodeType.AGENT:
|
||||
out.extend((who, n) for n in node_setting(spec, node, "tools") or []
|
||||
if isinstance(n, str) and n.strip())
|
||||
elif node.type == NodeType.SUPERVISOR:
|
||||
for member in node_setting(spec, node, "agents") or []:
|
||||
if not isinstance(member, dict):
|
||||
continue
|
||||
label = f"「{node.title}」的成员「{member.get('name') or '?'}」"
|
||||
out.extend((label, n) for n in member.get("tools") or [] if isinstance(n, str) and n.strip())
|
||||
return [(who, name.strip()) for who, name in out]
|
||||
|
||||
|
||||
def _bindings(spec: GraphSpec) -> list[tuple[str, str]]:
|
||||
"""发起前要检查的绑定。模板拼出来的工具名（{{ vars.tool }}）发起时还不知道是谁，交给运行时。"""
|
||||
return [(who, name) for who, name in tool_bindings(spec) if "{{" not in name]
|
||||
|
||||
|
||||
async def missing_tools(session: AsyncSession, spec: GraphSpec) -> list[tuple[str, str, str]]:
|
||||
"""本机不存在的绑定：(谁绑的, 工具名, 去哪接入：data / tools)。
|
||||
|
||||
解析和运行时 build_tools 一致：内置（静态注册）、mcp:<服务>/<工具>、db_query__ / db_schema__
|
||||
<数据源>、其余按自定义工具。数据源和自定义工具只认启用着的——停用的运行时同样拿不到。
|
||||
MCP 只看服务登记着、启用着没有：连不连得上、工具还在不在，要连一次才知道，发起时不去等它，
|
||||
上次没连上或者工具清单里没有的只在日志里提一句。
|
||||
"""
|
||||
from app.db.models import CustomTool, DataSource, McpServer
|
||||
from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX
|
||||
from app.tools.registry import all_specs
|
||||
|
||||
bindings = _bindings(spec)
|
||||
if not bindings:
|
||||
return []
|
||||
builtin = all_specs()
|
||||
sources = {n for (n,) in await session.execute(select(DataSource.name).where(DataSource.enabled.is_(True)))}
|
||||
custom = {n for (n,) in await session.execute(select(CustomTool.name).where(CustomTool.enabled.is_(True)))}
|
||||
servers = {r.name: r for r in (await session.execute(select(McpServer))).scalars()}
|
||||
missing: list[tuple[str, str, str]] = []
|
||||
for who, name in bindings:
|
||||
if name.startswith("mcp:"):
|
||||
server, _, tool = name[4:].partition("/")
|
||||
row = servers.get(server)
|
||||
if row is None or not row.enabled:
|
||||
missing.append((who, name, "tools"))
|
||||
elif row.status == "error":
|
||||
logger.warning("%s 绑的 %s：MCP 服务 %s 上次没连上，运行时可能拿不到这个工具", who, name, server)
|
||||
elif tool not in ("", "*") and row.tools_cache and tool not in row.tools_cache:
|
||||
logger.warning("%s 绑的 %s：MCP 服务 %s 上次交回的工具清单里没有 %s", who, name, server, tool)
|
||||
elif name in builtin:
|
||||
continue
|
||||
elif name.startswith((QUERY_PREFIX, SCHEMA_PREFIX)):
|
||||
source = name[len(QUERY_PREFIX):] if name.startswith(QUERY_PREFIX) else name[len(SCHEMA_PREFIX):]
|
||||
if source not in sources:
|
||||
missing.append((who, name, "data"))
|
||||
elif name not in custom:
|
||||
missing.append((who, name, "tools"))
|
||||
return missing
|
||||
|
||||
|
||||
async def _refuse_missing_tools(session: AsyncSession, graph: dict[str, Any]) -> None:
|
||||
"""绑定的工具在本机不存在就不发起：以前要执行到那一步才失败，前面的步骤白做、钱白花。"""
|
||||
try:
|
||||
spec = GraphSpec.model_validate(graph)
|
||||
except Exception: # noqa: BLE001 - 图本身不成形交给 run_manager.start 报
|
||||
return
|
||||
missing = await missing_tools(session, spec)
|
||||
if not missing:
|
||||
return
|
||||
lines = "；".join(f"{who}绑定的 {name}" for who, name, _ in missing[:8])
|
||||
more = f"等 {len(missing)} 处" if len(missing) > 8 else ""
|
||||
places = {where for _, _, where in missing}
|
||||
if places == {"data"}:
|
||||
fix = "请到「数据」页接入，或在节点中重新选择"
|
||||
elif places == {"tools"}:
|
||||
fix = "请到「工具」页接入，或在节点中重新选择"
|
||||
else:
|
||||
fix = "数据源工具请到「数据」页接入，其他工具请到「工具」页接入，或在节点中重新选择"
|
||||
raise LaunchRefused(422, f"绑定的工具不存在：{lines}{more}。{fix}", TOOL_MISSING)
|
||||
@@ -1062,11 +1062,12 @@ async def _data_versions(session: Any, spec: GraphSpec) -> dict[str, Any]:
|
||||
async def _bound_sources(session: Any, spec: GraphSpec) -> tuple[set[str], bool]:
|
||||
"""图里（含子工作流）绑定的数据源工具指向的源名，以及有没有带模板、发起时定不下来的工具名。
|
||||
|
||||
绑定的取法和发起前检查同一个（api/runs.py 的 tool_bindings），子工作流照运行时的取法
|
||||
绑定的取法和发起前检查同一个（engine/launch.py 的 tool_bindings），子工作流照运行时的取法
|
||||
读出来接着找：固定了版本的读那一版，没固定的读工作流现在的图。读不到、图不成形的跳过——
|
||||
那个节点运行时自己会报错。同一个工作流（同一版）只读一次，互相引用也不会转圈。
|
||||
"""
|
||||
from app.api.runs import node_setting, tool_bindings
|
||||
# 在函数里导入：launch 在模块级导入了本模块（它要 run_manager）
|
||||
from app.engine.launch import node_setting, tool_bindings
|
||||
from app.db.models import WorkflowVersion
|
||||
from app.tools.datasource import source_of_tool
|
||||
|
||||
|
||||
@@ -0,0 +1,337 @@
|
||||
"""发起运行的规则搬到 engine/launch.py：HTTP 发起和定时任务到点发起共用同一个 launch_run。
|
||||
|
||||
- 每一种拒绝：POST /api/runs 的状态码、detail、code 先按搬家之前的原话写死（这一组先在旧代码上通过）；
|
||||
直接调 launch_run 抛 LaunchRefused，和 HTTP 的返回逐条一致
|
||||
- 发起之后的治理事件（受管级别那条日志、口径升版处置）在服务里写：不经 HTTP 发起的正式运行一样有
|
||||
- 旧的导入路径（app.api.runs 的 missing_tools、node_setting、tool_bindings、TOOL_MISSING）照旧可用；
|
||||
runner 不再从接口层导入
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import inspect
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Run, RunEvent, Workflow, WorkflowVersion
|
||||
from app.engine.runner import run_manager
|
||||
from app.engine.schema import GraphSpec
|
||||
from app.main import app
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def engine_up():
|
||||
await run_manager.setup()
|
||||
yield
|
||||
await run_manager.shutdown()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
def node(nid, ntype, label=None, **config):
|
||||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}}
|
||||
|
||||
|
||||
def chain(*nodes):
|
||||
return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
|
||||
|
||||
|
||||
SIMPLE = chain(node("start", "input"), node("out", "output", fields=[{"name": "r", "value": "完成"}]))
|
||||
|
||||
|
||||
def _tag() -> str:
|
||||
return uuid.uuid4().hex[:8]
|
||||
|
||||
|
||||
async def workflow(graph=SIMPLE, *, published=None, versions=(), snapshot=None, name=None) -> tuple[str, str]:
|
||||
"""建一个工作流。published 是当前发布版本号，versions 是要建快照的版本号，snapshot 是快照里的图（默认同 graph）。"""
|
||||
name = name or f"发起测试·周报-{_tag()}"
|
||||
async with SessionLocal() as session:
|
||||
wf = Workflow(name=name, graph=graph, status="published" if published else "draft",
|
||||
published_version=published, version=max(versions, default=1))
|
||||
session.add(wf)
|
||||
await session.flush()
|
||||
for v in versions:
|
||||
session.add(WorkflowVersion(workflow_id=wf.id, version=v,
|
||||
graph=snapshot if snapshot is not None else graph))
|
||||
await session.commit()
|
||||
return wf.id, name
|
||||
|
||||
|
||||
def tool_graph(*tools: tuple[str, str]) -> dict:
|
||||
"""每个 (节点名, 工具名) 一个调用工具节点。"""
|
||||
middle = [node(f"t{i}", "tool", label, tool=tool, args={}) for i, (label, tool) in enumerate(tools)]
|
||||
return chain(node("start", "input"), *middle, node("out", "output", fields=[{"name": "r", "value": "完成"}]))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 每一种拒绝：(请求体, (状态码, detail, code))。detail 是 None 时只比两边一致，另由 check 查要点
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def case_formal_without_workflow():
|
||||
return {"run_class": "formal"}, (400, "正式运行需要从已保存的工作流发起：请求中没有指定工作流", None)
|
||||
|
||||
|
||||
async def case_formal_with_canvas_graph():
|
||||
wf, _ = await workflow(published=1, versions=(1,))
|
||||
return ({"run_class": "formal", "workflow_id": wf, "graph": SIMPLE},
|
||||
(400, "正式运行只能使用已发布的版本，画布上未发布的修改不会生效。请先保存并发布", None))
|
||||
|
||||
|
||||
async def case_formal_workflow_gone():
|
||||
return {"run_class": "formal", "workflow_id": f"gone{_tag()}"}, (404, "工作流不存在，可能已被删除", None)
|
||||
|
||||
|
||||
async def case_formal_never_published():
|
||||
wf, name = await workflow()
|
||||
return {"run_class": "formal", "workflow_id": wf}, (409, f"「{name}」还没有发布版本，请先在编排页发布", None)
|
||||
|
||||
|
||||
async def case_formal_not_the_published_version():
|
||||
wf, _ = await workflow(published=2, versions=(1, 2))
|
||||
return ({"run_class": "formal", "workflow_id": wf, "version": 1},
|
||||
(409, "v1 不是当前发布版本（v2）。正式运行只能使用当前发布版本；如需运行这一版，请先发布它。", None))
|
||||
|
||||
|
||||
async def case_formal_snapshot_missing():
|
||||
wf, _ = await workflow(published=3, versions=(1, 2))
|
||||
return {"run_class": "formal", "workflow_id": wf}, (404, "版本 v3 不存在", None)
|
||||
|
||||
|
||||
async def case_formal_snapshot_malformed():
|
||||
"""快照里的图不成形：以前由 main.py 的 ValueError 兜底回 400，detail 是校验的原话。"""
|
||||
broken = {"nodes": "坏掉的图"}
|
||||
wf, _ = await workflow(published=1, versions=(1,), snapshot=broken)
|
||||
with pytest.raises(ValueError) as caught:
|
||||
GraphSpec.model_validate(broken)
|
||||
return {"run_class": "formal", "workflow_id": wf}, (400, str(caught.value), None)
|
||||
|
||||
|
||||
def library(version="v3"):
|
||||
return chain(node("start", "input"),
|
||||
node("caliber", "metrics", "周报口径卡", caliber="周报口径", caliber_version=version,
|
||||
metrics=[{"id": "gmv", "name": "销售额", "unit": "元", "expression": "vars.kpi.gmv"}]),
|
||||
node("out", "output", fields=[{"name": "r", "value": "{{ nodes.caliber.text }}"}]))
|
||||
|
||||
|
||||
def weekly(lib_id: str, **card) -> dict:
|
||||
return chain(node("start", "input"),
|
||||
node("fetch", "transform", mode="expression", expression="{'gmv': 4321.5}", assign_to="kpi"),
|
||||
node("card", "metrics", "本地口径卡",
|
||||
caliber_from={"workflow_id": lib_id, "workflow_version": 1, "node_id": "caliber"}, **card),
|
||||
node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}]))
|
||||
|
||||
|
||||
async def upstream_with_two_versions() -> str:
|
||||
async with SessionLocal() as session:
|
||||
lib = Workflow(name=f"口径库-{_tag()}", graph=library(), status="published", published_version=2, version=2)
|
||||
session.add(lib)
|
||||
await session.flush()
|
||||
session.add(WorkflowVersion(workflow_id=lib.id, version=1, graph=library()))
|
||||
session.add(WorkflowVersion(workflow_id=lib.id, version=2, graph=library(version="v4")))
|
||||
await session.commit()
|
||||
return lib.id
|
||||
|
||||
|
||||
async def case_formal_caliber_upgrade_undeclared():
|
||||
lib = await upstream_with_two_versions()
|
||||
graph = weekly(lib)
|
||||
wf, _ = await workflow(graph, published=1, versions=(1,))
|
||||
return {"run_class": "formal", "workflow_id": wf}, (409, None, None)
|
||||
|
||||
|
||||
def check_formal_caliber_upgrade_undeclared(detail: str) -> None:
|
||||
assert "「本地口径卡」" in detail and "v1" in detail and "v2" in detail and "「上游发布新版本时」" in detail
|
||||
|
||||
|
||||
async def case_saved_workflow_gone():
|
||||
return {"workflow_id": f"gone{_tag()}"}, (404, "工作流不存在，可能已被删除", None)
|
||||
|
||||
|
||||
async def case_nothing_to_run():
|
||||
return {}, (400, "没有要运行的内容：请求中既没有工作流 ID，也没有工作流内容", None)
|
||||
|
||||
|
||||
async def case_saved_workflow_is_empty():
|
||||
wf, _ = await workflow({})
|
||||
return {"workflow_id": wf}, (400, "没有要运行的内容：请求中既没有工作流 ID，也没有工作流内容", None)
|
||||
|
||||
|
||||
async def case_missing_datasource_tool():
|
||||
source = f"gone{_tag()}"
|
||||
return ({"graph": tool_graph(("取数", f"db_query__{source}"))},
|
||||
(422, f"绑定的工具不存在：「取数」（调用工具）绑定的 db_query__{source}。请到「数据」页接入，或在节点中重新选择",
|
||||
"run_tool_missing"))
|
||||
|
||||
|
||||
async def case_missing_custom_tool():
|
||||
tool = f"ghost_{_tag()}"
|
||||
return ({"graph": tool_graph(("查编码", tool))},
|
||||
(422, f"绑定的工具不存在：「查编码」（调用工具）绑定的 {tool}。请到「工具」页接入，或在节点中重新选择",
|
||||
"run_tool_missing"))
|
||||
|
||||
|
||||
async def case_missing_tools_of_both_kinds():
|
||||
source, tool = f"gone{_tag()}", f"ghost_{_tag()}"
|
||||
return ({"graph": tool_graph(("取数", f"db_schema__{source}"), ("查编码", tool))},
|
||||
(422, f"绑定的工具不存在：「取数」（调用工具）绑定的 db_schema__{source}；「查编码」（调用工具）绑定的 {tool}。"
|
||||
"数据源工具请到「数据」页接入，其他工具请到「工具」页接入，或在节点中重新选择", "run_tool_missing"))
|
||||
|
||||
|
||||
async def case_more_than_eight_missing_tools():
|
||||
tag = _tag()
|
||||
tools = [(f"第{i}步", f"ghost_{tag}_{i}") for i in range(10)]
|
||||
listed = "；".join(f"「{label}」（调用工具）绑定的 {tool}" for label, tool in tools[:8])
|
||||
return ({"graph": tool_graph(*tools)},
|
||||
(422, f"绑定的工具不存在：{listed}等 10 处。请到「工具」页接入，或在节点中重新选择", "run_tool_missing"))
|
||||
|
||||
|
||||
async def case_formal_missing_tool():
|
||||
source = f"gone{_tag()}"
|
||||
wf, _ = await workflow(tool_graph(("取数", f"db_query__{source}")), published=1, versions=(1,))
|
||||
return ({"run_class": "formal", "workflow_id": wf},
|
||||
(422, f"绑定的工具不存在：「取数」（调用工具）绑定的 db_query__{source}。请到「数据」页接入，或在节点中重新选择",
|
||||
"run_tool_missing"))
|
||||
|
||||
|
||||
async def case_graph_fails_validation():
|
||||
return ({"graph": {"nodes": [], "edges": []}},
|
||||
(400, "工作流未通过校验，请先在画布上修正标红的节点再运行：画布上还没有节点，请先从节点库拖入一个节点", None))
|
||||
|
||||
|
||||
#: 「不存在」的工作流、数据源、自定义工具都带随机后缀，库里不会恰好有同名的
|
||||
CASES = {name[len("case_"):]: fn for name, fn in dict(globals()).items() if name.startswith("case_")}
|
||||
|
||||
|
||||
def _body(status: int, detail: str, code: str | None) -> dict:
|
||||
return {"detail": detail, **({"code": code} if code else {})}
|
||||
|
||||
|
||||
async def _run_count() -> int:
|
||||
async with SessionLocal() as session:
|
||||
return len((await session.execute(select(Run.id))).all())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("case", list(CASES))
|
||||
async def test_http_refusals_are_unchanged(client, case):
|
||||
"""HTTP 这一侧逐字写死：搬家前后都得是这些状态码和原话。"""
|
||||
body, (status, detail, code) = await CASES[case]()
|
||||
before = await _run_count()
|
||||
res = await client.post("/api/runs", json=body)
|
||||
assert res.status_code == status, res.text
|
||||
got = res.json()
|
||||
if detail is None:
|
||||
globals()[f"check_{case}"](got["detail"])
|
||||
assert set(got) == ({"detail", "code"} if code else {"detail"}), got
|
||||
else:
|
||||
assert got == _body(status, detail, code)
|
||||
assert await _run_count() == before, "被拒的请求不该留下运行记录"
|
||||
|
||||
|
||||
def _launch_kwargs(body: dict) -> dict:
|
||||
"""请求体换成 launch_run 的参数：input 对应 input_payload，其余同名。"""
|
||||
out = {k: v for k, v in body.items() if k != "input"}
|
||||
if "input" in body:
|
||||
out["input_payload"] = body["input"]
|
||||
return out
|
||||
|
||||
|
||||
@pytest.mark.parametrize("case", list(CASES))
|
||||
async def test_launch_run_refuses_exactly_like_http(client, case):
|
||||
from app.engine.launch import LaunchRefused, launch_run
|
||||
|
||||
body, (status, detail, code) = await CASES[case]()
|
||||
before = await _run_count()
|
||||
async with SessionLocal() as session:
|
||||
with pytest.raises(LaunchRefused) as caught:
|
||||
await launch_run(session, actor="定时任务", **_launch_kwargs(body))
|
||||
refused = caught.value
|
||||
assert (refused.status_code, refused.code) == (status, code)
|
||||
if detail is not None:
|
||||
assert refused.message == detail
|
||||
res = await client.post("/api/runs", json=body)
|
||||
assert res.status_code == refused.status_code
|
||||
assert res.json() == _body(refused.status_code, refused.message, refused.code)
|
||||
assert await _run_count() == before
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 发起成功：参数照转、治理事件在服务里写
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _settle(run_id: str) -> Run:
|
||||
for _ in range(400):
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(Run, run_id)
|
||||
if row.status in ("succeeded", "failed", "cancelled") and row.manifest_seq is not None:
|
||||
return row
|
||||
await asyncio.sleep(0.03)
|
||||
raise AssertionError(f"运行 {run_id} 没有结束：{row.status}")
|
||||
|
||||
|
||||
async def _events(run_id: str, etype: str) -> list[RunEvent]:
|
||||
async with SessionLocal() as session:
|
||||
return list((await session.execute(
|
||||
select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype).order_by(RunEvent.seq)
|
||||
)).scalars())
|
||||
|
||||
|
||||
async def test_a_formal_launch_records_the_governance_notes():
|
||||
"""不经 HTTP 发起的正式运行（定时任务）也记下按什么级别出具、口径升版怎么处置的。"""
|
||||
from app.engine.launch import launch_run
|
||||
|
||||
lib = await upstream_with_two_versions()
|
||||
wf, name = await workflow(weekly(lib, upgrade_policy="dual"), published=1, versions=(1,))
|
||||
async with SessionLocal() as session:
|
||||
run = await launch_run(session, workflow_id=wf, run_class="formal", input_payload={"week": "第 40 周"},
|
||||
actor="定时任务")
|
||||
assert (run.run_class, run.version, run.workflow_name, run.started_by) == ("formal", 1, name, "定时任务")
|
||||
assert run.version_hash and run.input == {"week": "第 40 周"}
|
||||
row = await _settle(run.id)
|
||||
assert row.status == "succeeded", row.error
|
||||
notes = [e.data for e in await _events(run.id, "log") if e.data.get("code") == "run_governance"]
|
||||
assert len(notes) == 1 and notes[0]["governed"] is False and "v1" in notes[0]["message"]
|
||||
[upgrade] = await _events(run.id, "caliber.upgrade")
|
||||
assert upgrade.node_id == "card" and upgrade.data["policy"] == "dual" and upgrade.data["latest"] == 2
|
||||
|
||||
|
||||
async def test_an_exploratory_launch_takes_the_saved_graph_and_the_run_defaults():
|
||||
from app.engine.launch import launch_run
|
||||
|
||||
wf, name = await workflow()
|
||||
async with SessionLocal() as session:
|
||||
run = await launch_run(session, workflow_id=wf, actor=None)
|
||||
assert (run.run_class, run.workflow_id, run.workflow_name, run.version) == ("exploratory", wf, name, None)
|
||||
assert (run.memory_scope, run.collection) == ("default", "default") and run.input == {}
|
||||
row = await _settle(run.id)
|
||||
assert row.status == "succeeded", row.error
|
||||
assert [e for e in await _events(run.id, "log") if e.data.get("code") == "run_governance"] == []
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 旧的导入路径
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_the_old_import_paths_still_work():
|
||||
from app.api import runs as runs_api
|
||||
from app.engine import launch
|
||||
|
||||
for name in ("missing_tools", "node_setting", "tool_bindings", "TOOL_MISSING"):
|
||||
assert getattr(runs_api, name) is getattr(launch, name), name
|
||||
|
||||
|
||||
def test_the_runner_no_longer_imports_from_the_api_layer():
|
||||
from app.engine import runner
|
||||
|
||||
assert "app.api.runs" not in inspect.getsource(runner)
|
||||
Reference in New Issue
Block a user