子工作流里那一步结束了运行时各处同一个名字:新增 apps/resolve.end_label(终态事件上记的「子工作流 › x」),定时任务的 Webhook 不再在找不到节点时什么都不写、发布单不再写内部 id「sub/x」;补测试

This commit is contained in:
ThearchyHelios
2026-10-18 18:28:17 +08:00
parent 4ca7d0c768
commit ea28f02eeb
4 changed files with 77 additions and 9 deletions
+12
View File
@@ -1548,6 +1548,18 @@ async def _trails(session: AsyncSession, runs: list[Run]) -> dict[str, _Trail]:
return out
async def end_label(session: AsyncSession, run: Run) -> str | None:
"""运行结束在哪一步的名字,照终态事件(run.failed / run.cancelled)上记的 label:子工作流里的那一步写「子工作流 › x」。
图上找不到这一步(id 是「sub/x」)的地方都用它,和实例页、通知(run_node_label)同一个说法;没记的为 None。"""
if not run.error_node_id:
return None
data = await session.scalar(select(RunEvent.data).where(
RunEvent.run_id == run.id, RunEvent.type.in_(("run.failed", "run.cancelled"))).order_by(RunEvent.seq.desc()).limit(1))
if isinstance(data, dict) and data.get("node_id") == run.error_node_id and isinstance(data.get("label"), str):
return data["label"] or None
return None
def run_node_label(run: Run, trail: _Trail, node_id: str) -> str:
"""运行里一个节点给人看的名字:运行的图上的标题;图上没有的(子工作流里的那一步「sub/x」)用终态事件上记的名字
(「子工作流 › x」);都没有的照旧写 id。实例页、通知(notify/service)说同一个名字。"""
+7 -4
View File
@@ -380,7 +380,9 @@ async def _preflight(session: AsyncSession, bp: Blueprint, target: _Target, *, l
elif is_trial_of(trial, bp) and trial.status == "cancelled" and trial.end_reason == "rejected":
# 搭建者在试运行里驳回了写入、审批:那是他的决定,不是图出了错,不说成「试运行失败了」、也不挡发布;只照实说
# 没走完——驳回之后的步骤没用真数看过
where = _node_title(trial.graph, trial.error_node_id)
from app.apps.resolve import end_label
where = _node_title(trial.graph, trial.error_node_id, await end_label(session, trial))
out.append(_problem(TRIAL_REJECTED, f"最近一次试运行{f'在「{where}」' if where else ''}被驳回、没有走完",
"warning", source="plan"))
elif not is_trial_of(trial, bp) or trial.status != "succeeded":
@@ -390,15 +392,16 @@ async def _preflight(session: AsyncSession, bp: Blueprint, target: _Target, *, l
return [_with_who(p) for p in out]
def _node_title(graph: Any, node_id: str | None) -> str | None:
"""节点在图上的名字(没有名字的给 id);没有节点时为 None。"""
def _node_title(graph: Any, node_id: str | None, end_label: str | None = None) -> str | None:
"""节点在图上的名字(没有名字的给 id);没有节点时为 None。图上找不到的(子工作流里的那一步「sub/x」)用终态事件上记的
名字(「子工作流 › x」,apps/resolve.end_label),没记的照旧给 id。"""
if not node_id:
return None
for node in (graph or {}).get("nodes") or []:
if isinstance(node, dict) and node.get("id") == node_id:
data = node.get("data") if isinstance(node.get("data"), dict) else {}
return str(data.get("label") or node_id)
return node_id
return end_label or node_id
def _with_who(issue: dict[str, Any]) -> dict[str, Any]:
+14 -5
View File
@@ -551,8 +551,8 @@ async def _notice(session: Any, schedule: Schedule, fire: ScheduleFire, run: Run
pending_approvals=pending or None,
suspended=run is not None and run.status == "interrupted" and not pending,
action_pending=seen.action_pending,
error_node=_failed_node(run) if run is not None and (run.status == "failed" or run.end_reason == "rejected")
else None,
error_node=_failed_node(run, await _end_label(session, run))
if run is not None and (run.status == "failed" or run.end_reason == "rejected") else None,
end_reason=run.end_reason if run is not None else None,
# 事件触发的任务没有下一次「到点」,启用着就是下一次事件来了照常发起
next_fire=bool(schedule.enabled and (schedule.next_run_at is not None or schedule.trigger_kind == "event")),
@@ -566,15 +566,24 @@ def _trigger_text(fire: ScheduleFire) -> str | None:
return text if isinstance(text, str) and text else None
def _failed_node(run: Run) -> str | None:
"""失败节点在图上的名字(没有名字的给节点 id)。节点名是搭图时起的,不带查询到的数据。"""
def _failed_node(run: Run, end_label: str | None = None) -> str | None:
"""失败节点在图上的名字(没有名字的给节点 id)。节点名是搭图时起的,不带查询到的数据。图上找不到的(子工作流里的那一步,
id 是「sub/x」)用终态事件上记的名字(「子工作流 › x」,apps/resolve.end_label),和实例页、通知同一个说法。"""
if not run.error_node_id:
return None
for node in (run.graph or {}).get("nodes") or []:
if isinstance(node, dict) and node.get("id") == run.error_node_id:
label = (node.get("data") or {}).get("label") if isinstance(node.get("data"), dict) else None
return str(label or run.error_node_id)[:80]
return None
return end_label[:80] if end_label else None
async def _end_label(session: Any, run: Run | None) -> str | None:
if run is None:
return None
from app.apps.resolve import end_label
return await end_label(session, run)
async def _issuance_of(session: Any, run: Run) -> tuple[str | None, int | None]:
+44
View File
@@ -0,0 +1,44 @@
"""子工作流里那一步结束了运行时,各处写同一个名字(「子工作流 › x」,终态事件上记的 label):apps/resolve.end_label、
定时任务的 Webhook(scheduler/webhook._failed_node)、发布单的「最近一次试运行在「…」被驳回」(blueprints/publish_plan)。
图上找得到的节点照旧用图上的名字。运行和图都是虚构的。
"""
from __future__ import annotations
from app.apps.resolve import end_label
from app.blueprints.publish_plan import _node_title
from app.db.base import SessionLocal
from app.db.models import Run, RunEvent
from app.scheduler.webhook import _failed_node
GRAPH = {"nodes": [{"id": "sub", "type": "subgraph", "data": {"label": "核对子流程", "config": {}}}], "edges": []}
async def _run(error_node: str, label: str | None) -> Run:
async with SessionLocal() as session:
run = Run(graph=GRAPH, status="failed", error_node_id=error_node, run_class="exploratory")
session.add(run)
await session.flush()
data = {"node_id": error_node, **({"label": label} if label else {})}
session.add(RunEvent(run_id=run.id, seq=1, type="run.failed", node_id=error_node, ts=0.0, data=data))
await session.commit()
return run
async def test_a_step_inside_a_subworkflow_is_named_by_the_end_event_everywhere():
run = await _run("sub/fetch", "核对子流程 › 取数")
async with SessionLocal() as session:
said = await end_label(session, run)
assert said == "核对子流程 › 取数"
assert _failed_node(run, said) == "核对子流程 › 取数", "Webhook 不再在找不到节点时什么都不写"
assert _node_title(run.graph, run.error_node_id, said) == "核对子流程 › 取数", "发布单不再写内部 id「sub/fetch」"
async def test_nodes_on_the_graph_keep_their_graph_names_and_missing_labels_fall_back():
run = await _run("sub", "别的说法")
async with SessionLocal() as session:
said = await end_label(session, run)
assert _failed_node(run, said) == "核对子流程" and _node_title(run.graph, "sub", said) == "核对子流程"
bare = await _run("sub/x", None)
async with SessionLocal() as session:
assert await end_label(session, bare) is None
assert _failed_node(bare, None) is None and _node_title(bare.graph, "sub/x", None) == "sub/x"