diff --git a/backend/app/apps/resolve.py b/backend/app/apps/resolve.py index 9e01e068..1f45c938 100644 --- a/backend/app/apps/resolve.py +++ b/backend/app/apps/resolve.py @@ -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)说同一个名字。""" diff --git a/backend/app/blueprints/publish_plan.py b/backend/app/blueprints/publish_plan.py index b3c7a289..80438a16 100644 --- a/backend/app/blueprints/publish_plan.py +++ b/backend/app/blueprints/publish_plan.py @@ -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]: diff --git a/backend/app/scheduler/webhook.py b/backend/app/scheduler/webhook.py index 90dac23f..c911b118 100644 --- a/backend/app/scheduler/webhook.py +++ b/backend/app/scheduler/webhook.py @@ -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]: diff --git a/backend/tests/test_subgraph_end_labels.py b/backend/tests/test_subgraph_end_labels.py new file mode 100644 index 00000000..aca694ca --- /dev/null +++ b/backend/tests/test_subgraph_end_labels.py @@ -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"