From 7820556ab6ed5612b26ebddde9b229ce3516c230 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 00:03:24 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=91=E8=B5=B7=E8=BF=90=E8=A1=8C=E7=9A=84?= =?UTF-8?q?=E8=A7=84=E5=88=99=E6=8A=BD=E6=88=90=20engine/launch.py=20?= =?UTF-8?q?=E7=9A=84=20launch=5Frun=EF=BC=9A=E4=BB=A5=E5=89=8D=E5=85=A8?= =?UTF-8?q?=E5=86=99=E5=9C=A8=20api/runs.py=20=E7=9A=84=20start=5Frun=20?= =?UTF-8?q?=E9=87=8C=E3=80=81=E5=8F=AA=E8=AE=A4=20HTTP=20=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=EF=BC=8C=E5=90=8E=E9=9D=A2=E7=9A=84=E5=AE=9A=E6=97=B6?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E8=A6=81=E7=94=A8=E5=90=8C=E4=B8=80=E5=A5=97?= =?UTF-8?q?=E8=A7=84=E5=88=99=E5=8F=91=E8=B5=B7=E6=AD=A3=E5=BC=8F=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=EF=BC=8C=E5=8F=A6=E6=8A=84=E4=B8=80=E4=BB=BD=E8=BF=9F?= =?UTF-8?q?=E6=97=A9=E8=B5=B0=E6=A0=B7=E3=80=82=E6=AD=A3=E5=BC=8F=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E5=8F=AA=E8=83=BD=E4=BB=8E=E5=BD=93=E5=89=8D=E5=8F=91?= =?UTF-8?q?=E5=B8=83=E7=89=88=E6=9C=AC=E5=8F=91=E8=B5=B7=E3=80=81=E5=8F=A3?= =?UTF-8?q?=E5=BE=84=E5=8D=87=E7=89=88=E6=9C=AA=E5=A3=B0=E6=98=8E=E7=AD=96?= =?UTF-8?q?=E7=95=A5=E5=B0=B1=E6=8B=A6=E3=80=81=E7=BB=91=E5=AE=9A=E7=9A=84?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E4=B8=8D=E5=AD=98=E5=9C=A8=E5=B0=B1=E6=8B=A6?= =?UTF-8?q?=E3=80=81=E5=8F=91=E8=B5=B7=E5=90=8E=E8=AE=B0=E5=8F=97=E7=AE=A1?= =?UTF-8?q?=E7=BA=A7=E5=88=AB=E9=82=A3=E6=9D=A1=E6=97=A5=E5=BF=97=E5=92=8C?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=E5=8F=98=E5=8C=96=E3=80=81=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E5=8D=87=E7=89=88=E5=A4=84=E7=BD=AE=EF=BC=8C=E5=85=A8=E9=83=A8?= =?UTF-8?q?=E6=90=AC=E8=BF=9B=E6=9C=8D=E5=8A=A1=EF=BC=9B=E6=8B=92=E7=BB=9D?= =?UTF-8?q?=E6=97=B6=E6=8A=9B=20LaunchRefused(status=5Fcode,=20message,=20?= =?UTF-8?q?code)=EF=BC=8Cstart=5Frun=20=E5=8F=AA=E5=89=A9=E4=B8=80?= =?UTF-8?q?=E5=B1=82=E5=8C=85=E8=A3=85=EF=BC=8C=E6=8A=8A=E5=AE=83=E6=8D=A2?= =?UTF-8?q?=E5=9B=9E=E5=92=8C=E4=BB=A5=E5=89=8D=E9=80=90=E5=AD=97=E7=9B=B8?= =?UTF-8?q?=E5=90=8C=E7=9A=84=E5=93=8D=E5=BA=94=EF=BC=88=E5=B8=A6=E6=9C=BA?= =?UTF-8?q?=E5=99=A8=E7=A0=81=E7=9A=84=E4=BB=8D=E5=9B=9E=20{detail,=20code?= =?UTF-8?q?}=EF=BC=89=E3=80=82=E6=AD=A3=E5=BC=8F=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E7=9A=84=E5=BF=AB=E7=85=A7=E5=9B=BE=E4=B8=8D=E6=88=90=E5=BD=A2?= =?UTF-8?q?=EF=BC=8C=E4=BB=A5=E5=89=8D=E7=94=B1=20main.py=20=E7=9A=84=20Va?= =?UTF-8?q?lueError=20=E5=85=9C=E5=BA=95=E5=9B=9E=20400=EF=BC=8C=E7=8E=B0?= =?UTF-8?q?=E5=9C=A8=E5=9C=A8=E6=9C=8D=E5=8A=A1=E9=87=8C=E6=8D=A2=E6=88=90?= =?UTF-8?q?=E5=90=8C=E6=A0=B7=E7=9A=84=20400=20=E6=8B=92=E7=BB=9D=EF=BC=8C?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E5=99=A8=E4=B9=9F=E6=8B=BF=E5=BE=97=E5=88=B0?= =?UTF-8?q?=E6=98=8E=E7=A1=AE=E7=9A=84=E6=8B=92=E7=BB=9D=E3=80=82node=5Fse?= =?UTF-8?q?tting=E3=80=81tool=5Fbindings=E3=80=81missing=5Ftools=20?= =?UTF-8?q?=E8=B7=9F=E7=9D=80=E6=90=AC=E8=BF=87=E5=8E=BB=EF=BC=8Cruns.py?= =?UTF-8?q?=20=E5=86=8D=E5=AF=BC=E5=87=BA=EF=BC=8C=E6=97=A7=E7=9A=84?= =?UTF-8?q?=E5=AF=BC=E5=85=A5=E8=B7=AF=E5=BE=84=E7=85=A7=E6=97=A7=E5=8F=AF?= =?UTF-8?q?=E7=94=A8=EF=BC=9Brunner=20=E5=9B=BA=E5=AE=9A=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E7=89=88=E6=9C=AC=E5=8F=96=E7=BB=91=E5=AE=9A=E6=97=B6=E6=94=B9?= =?UTF-8?q?=E4=BB=8E=20engine/launch=20=E5=AF=BC=E5=85=A5=EF=BC=8C?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E4=BE=9D=E8=B5=96=E6=8E=A5=E5=8F=A3=E5=B1=82?= =?UTF-8?q?=E3=80=82=E6=96=B0=E6=B5=8B=E8=AF=95=E5=85=88=E6=8A=8A=20HTTP?= =?UTF-8?q?=20=E7=9A=84=2017=20=E7=A7=8D=E6=8B=92=E7=BB=9D=EF=BC=88?= =?UTF-8?q?=E5=90=AB=E5=B7=A5=E5=85=B7=E7=BC=BA=E5=A4=B1=E7=9A=84=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E3=80=81=E5=B7=A5=E5=85=B7=E3=80=81=E6=B7=B7=E5=90=88?= =?UTF-8?q?=E3=80=81=E8=B6=85=E8=BF=87=208=20=E5=A4=84=E5=9B=9B=E7=A7=8D?= =?UTF-8?q?=E8=AF=B4=E6=B3=95=EF=BC=89=E6=8C=89=E5=8E=9F=E8=AF=9D=E5=86=99?= =?UTF-8?q?=E6=AD=BB=E3=80=81=E5=9C=A8=E6=97=A7=E4=BB=A3=E7=A0=81=E4=B8=8A?= =?UTF-8?q?=E9=80=9A=E8=BF=87=EF=BC=8C=E5=86=8D=E9=80=90=E6=9D=A1=E6=AF=94?= =?UTF-8?q?=E5=AF=B9=E7=9B=B4=E6=8E=A5=E8=B0=83=20launch=5Frun=20=E7=9A=84?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=EF=BC=9B=E5=8F=A6=E6=B5=8B=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E5=8F=91=E8=B5=B7=E7=9A=84=E6=AD=A3=E5=BC=8F=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E7=85=A7=E6=A0=B7=E5=86=99=E6=B2=BB=E7=90=86=E4=BA=8B=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/runs.py | 243 +++------------------- backend/app/engine/launch.py | 273 +++++++++++++++++++++++++ backend/app/engine/runner.py | 5 +- backend/tests/test_run_launch.py | 337 +++++++++++++++++++++++++++++++ 4 files changed, 636 insertions(+), 222 deletions(-) create mode 100644 backend/app/engine/launch.py create mode 100644 backend/tests/test_run_launch.py diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index d61bd8a0..058ad518 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -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 取这个值时只要没挂在任何工作流上的运行：画布上没保存就跑的临时图， diff --git a/backend/app/engine/launch.py b/backend/app/engine/launch.py new file mode 100644 index 00000000..3186a690 --- /dev/null +++ b/backend/app/engine/launch.py @@ -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) diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 1be770d4..74151f86 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -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 diff --git a/backend/tests/test_run_launch.py b/backend/tests/test_run_launch.py new file mode 100644 index 00000000..4fb698a1 --- /dev/null +++ b/backend/tests/test_run_launch.py @@ -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)