diff --git a/backend/app/apps/resolve.py b/backend/app/apps/resolve.py index 1f45c938..cd260aa5 100644 --- a/backend/app/apps/resolve.py +++ b/backend/app/apps/resolve.py @@ -1582,6 +1582,10 @@ async def _rejected_exits(session: AsyncSession, runs: list[Run], out: dict[str, names = {str(v.get("node_id")) for v in via or [] if isinstance(v, dict) and v.get("node_id")} if names: gates[run_id] = names + for v in via or []: + # 子工作流里的关卡(「sub/x」)不在运行的图上:名字照终态事件上记的(「子工作流 › x」),不写内部 id + if isinstance(v, dict) and v.get("node_id") and isinstance(v.get("label"), str) and v["label"]: + out[run_id].end_labels.setdefault(str(v["node_id"]), v["label"]) if not gates: return wanted = sorted({n for names in gates.values() for n in names}) diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index bfc48df8..25dd3c17 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -136,7 +136,8 @@ def _wrap( "attempt": attempt + 1, "ts": time.time(), } - return {**updates, "trail": [trail]} + # 节点自己交回的 trail(子工作流里走了驳回出口的那几步,multi._rejected_steps)排在这一步前面 + return {**updates, "trail": [*(updates.get("trail") or []), trail]} except GraphBubbleUp: # interrupt() / Command 靠异常向上传递控制流,不是执行失败: # 必须原样放行,否则人工介入会被当成节点报错吞掉。 @@ -230,7 +231,10 @@ async def _rejected_exit(node: GraphNode, ctx: NodeContext, error: NodeRejected, # 节点在驳回之前已经算出的(NodeRejected.exit_updates:双盲复算的出具缺口、台账条目)照样交出;驳回的那几个键在上面 payload = {**(extra.get("payload") or {}), **error.exit_payload()} nodes = run_ctx.spec.node_map() - to = [{"node_id": run_ctx.global_id(t), "label": nodes[t].title if t in nodes else t} for t in dict.fromkeys(targets)] + # 名字按这次运行的说法记(子工作流里的写「子工作流 › x」,RunContext.global_label):父运行的图上没有子工作流里的节点, + # 实例页、通知、待办、Webhook 都照这个名字说 + to = [{"node_id": run_ctx.global_id(t), "label": run_ctx.global_label(nodes[t].title if t in nodes else t)} + for t in dict.fromkeys(targets)] try: artifact_id: str | None = await put_json(payload, kind="node_output", run_id=run_ctx.run_id, node_id=run_ctx.global_id(node.id), @@ -244,7 +248,7 @@ async def _rejected_exit(node: GraphNode, ctx: NodeContext, error: NodeRejected, outcome="rejected", **said, to=to, **({"evidence": ledger} if ledger else {})) updates: dict[str, Any] = {"nodes": {node.id: payload}, "trail": [{ "node_id": node.id, "type": str(node.type), "duration_ms": elapsed, "attempt": attempt, "ts": time.time(), - "outcome": "rejected", "label": node.title, **said, "to": to}]} + "outcome": "rejected", "label": run_ctx.global_label(node.title), **said, "to": to}]} if ledger: updates["evidence"] = ledger if extra.get("usage"): diff --git a/backend/app/engine/context.py b/backend/app/engine/context.py index fb85a203..6aff018f 100644 --- a/backend/app/engine/context.py +++ b/backend/app/engine/context.py @@ -43,9 +43,16 @@ class RunContext: """这张图里的节点 id → 这次运行里的 id(事件、工件上记的那个):最外层原样,子工作流里加前缀。""" return f"{self.prefix}{node_id}" + def global_label(self, title: str) -> str: + """这张图里的节点标题 → 这次运行里给人看的名字:最外层原样,子工作流里接上子工作流的名字(「子工作流 › x」,嵌套的 + 一层层接上,同 multi.run_subgraph 记在异常上的 label)。父运行的图上没有子工作流里的节点,各处照这个名字说。""" + return f"{self.extra.get(LABEL_PREFIX_KEY) or ''}{title}" + #: 子工作流里的节点 id 前缀(RunContext.extra 里的键,engine/nodes/multi.run_subgraph 放) PREFIX_KEY = "node_prefix" +#: 子工作流里的节点名前缀(「子工作流 › 」,同上):RunContext.global_label +LABEL_PREFIX_KEY = "node_label_prefix" #: 父运行的事件流(RunContext.extra 里的键,同上)。子工作流在父节点里用 ainvoke 跑,LangGraph 不把它的 custom 事件交给 #: 父图的流(ainvoke 不开 custom 模式,get_stream_writer 拿到的是空的那个):子工作流里每一步以前在运行的事件里一条都 #: 没有,时间线看不到、封存也不含它查过的数。所以 run_subgraph 把父图那一刻的写入口交下来,子工作流的节点经它发 diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index 80b20064..17222b1d 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -14,7 +14,7 @@ from app.core.errors import describe_exception, raw_detail from app.core.events import EventType from app.db.base import SessionLocal from app.db.models import Workflow -from app.engine.context import PREFIX_KEY, STREAM_KEY, NodeContext, NodeError, stream_writer +from app.engine.context import LABEL_PREFIX_KEY, PREFIX_KEY, STREAM_KEY, NodeContext, NodeError, stream_writer from app.engine.lineage import whole_ref from app.engine.state import GraphState, message_text, template_context from app.engine.toolcalls import ( @@ -691,7 +691,8 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 恢复重放的标记和签批人只属于父图的节点,不能漏进子图(节点 id 可能重名)。前缀和写入口见 engine/context: # 子图里的节点经父运行的写入口发事件、节点 id 带前缀;嵌套的子图拿到的是同一个写入口、更长的前缀 extra={**{k: v for k, v in ctx.run.extra.items() if k not in ("resumed", "actors")}, - PREFIX_KEY: f"{ctx.run.global_id(ctx.node.id)}/", STREAM_KEY: stream_writer(ctx.run)}, + PREFIX_KEY: f"{ctx.run.global_id(ctx.node.id)}/", LABEL_PREFIX_KEY: f"{ctx.run.global_label(ctx.node.title)} › ", + STREAM_KEY: stream_writer(ctx.run)}, ) ctx.emit(EventType.LOG, level="info", @@ -753,6 +754,11 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: } if ledger := [_forwarded(e, ctx.node.id) for e in result_state.get("evidence") or [] if isinstance(e, dict)]: updates["evidence"] = ledger + if rejected := _rejected_steps(result_state, ctx.node.id): + # 子工作流里走了驳回出口的那几步(engine/compiler._rejected_exit 记的 outcome=rejected)交回父图的 trail:运行收尾 + # 据此在终态事件上照实写「途经驳回」(runner._rejected_via),实例页、通知、定时任务的 Webhook 说得出「子工作流 › 把关」 + # 被驳回、转到哪。名字、转到哪几步在子图里已经按这次运行的说法记好(global_label、global_id),这里只接上节点 id 的前缀 + updates["trail"] = rejected var_name = ctx.cfg("assign_to", "") if var_name: updates["vars"] = {var_name: output} @@ -763,6 +769,12 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: _LEDGER_NODE_KEYS = ("node_id", "card") +def _rejected_steps(sub_state: dict[str, Any], node_id: str) -> list[dict[str, Any]]: + """子工作流的 trail 里走了驳回出口的那几步,节点 id 换成父图这一层的(「子工作流节点 id/…」)。""" + return [{**t, "node_id": f"{node_id}/{t.get('node_id')}"} for t in sub_state.get("trail") or [] + if isinstance(t, dict) and t.get("outcome") == "rejected"] + + def _forwarded(entry: dict[str, Any], node_id: str) -> dict[str, Any]: """子图台账里的一条交回父图:节点 id 加上「子工作流节点 id/」前缀,别的照抄。""" out = dict(entry) diff --git a/backend/app/engine/replay.py b/backend/app/engine/replay.py index 3d4d1078..e95e2446 100644 --- a/backend/app/engine/replay.py +++ b/backend/app/engine/replay.py @@ -75,4 +75,32 @@ async def ask(ctx: NodeContext, payload: dict[str, Any], **request: Any) -> Any: 答复真正到达的那次执行里出现。 """ await once(ctx, EventType.HUMAN_REQUESTED, **request) - return interrupt(payload) + route = reject_route(ctx) if isinstance(payload, dict) else None + return interrupt({**payload, "reject_route": route} if route else payload) + + +def reject_route(ctx: NodeContext) -> dict[str, Any] | None: + """驳回之后去哪(总方案 §3 的驳回出口),放进中断内容:待办卡的提示、驳回之后的那句、处理过的结果照它说,不再一律写 + 「转入驳回分支」或「本次运行终止」。 + + - {"to": [{node_id, label}]}:接了驳回出口,驳回后转到这几步(节点 id、名字按这次运行的说法:子工作流里的写 + 「子工作流 › x」,同 compiler._rejected_exit 记在 node.finished 上的 to); + - {"ends": "run"}:运行就此结束、记为被驳回(没接驳回出口的关卡;审批设了驳回即终止时驳回出口走不到,也是这样); + - {"ends": "path"}:审批模式没接驳回出口、没设驳回即终止:这一路到审批为止,不再往下走(engine/nodes/human)。 + + 不是关卡的节点(agent 自己调工具时的审批:驳回了 agent 照常往下想)为 None,中断内容和以前一字不差。""" + from app.engine.schema import GATE_TYPES, approve_mode, reject_targets + + node, spec = ctx.node, ctx.run.spec + if node.type not in GATE_TYPES: + return None + approving = approve_mode(spec, node) + if approving and bool(ctx.cfg("stop_on_reject", False)): + return {"ends": "run"} + targets = reject_targets(spec, node) + if targets: + nodes = spec.node_map() + return {"to": [{"node_id": ctx.run.global_id(t), + "label": ctx.run.global_label(nodes[t].title if t in nodes else t)} + for t in dict.fromkeys(targets)]} + return {"ends": "path" if approving else "run"} diff --git a/backend/app/scheduler/webhook.py b/backend/app/scheduler/webhook.py index c911b118..7c32d96f 100644 --- a/backend/app/scheduler/webhook.py +++ b/backend/app/scheduler/webhook.py @@ -77,6 +77,8 @@ RERUN_ACTION = "如需重新发起,请在 AgentLab 的「定时任务」中点 REJECTED_HEADLINE = "未出具(被驳回)" REJECTED_STATUS = "被驳回" REJECTED_ACTION = "驳回的说明请在 AgentLab 的「记录」中查看" +#: 照驳回那条路成功结束的(驳回出口):标题、状态后面接这一句 +VIA_REJECTED = "(途经驳回)" # -------------------------------------------------------------------------- @@ -118,6 +120,9 @@ class Notice: error_node: str | None = None #: 运行结束在 cancelled 的原因(Run.end_reason):rejected 时按「未出具(被驳回)」说,error_node 是驳回的那一步 end_reason: str | None = None + #: 照驳回那条路成功结束的运行途经的驳回出口(run.finished 的 rejected_via,engine/runner._rejected_via):标题、状态写 + #: 途经驳回,另起一行写谁被驳回、转到哪一步(同实例页 apps/resolve.rejected_exit_text)。没途经驳回的为 None + rejected_via: list[dict[str, Any]] | None = None next_fire: bool = False #: 事件触发的那一次(fire_kind=event):事件的说法(scheduler/triggers.event_text),不含出口字段的值、变化的内容 trigger_text: str | None = None @@ -130,6 +135,11 @@ class Notice: def rejected(self) -> bool: return self.status == "cancelled" and self.end_reason == "rejected" + @property + def via_rejection(self) -> bool: + """成功结束、途经驳回出口(审批接了驳回出口、按驳回那条路走完):状态是 succeeded,但不能只说「运行完成」。""" + return self.status == "succeeded" and bool(self.rejected_via) + @property def never_started(self) -> bool: """失败的运行从没开始执行(没有开始时间):服务重启时还在排队、被改成失败的就是这种,没有断点可续。""" @@ -160,10 +170,18 @@ def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]: headline = REJECTED_HEADLINE else: headline = _HEADLINE.get(notice.event, notice.event) + if notice.via_rejection: + headline += VIA_REJECTED title = f"定时任务「{notice.schedule_name}」{headline}" lines: list[tuple[str, str]] = [] if notice.status: lines.append(("状态", _status_label(notice))) + if notice.via_rejection: + from app.apps.resolve import rejected_exit_text + + # 一个关卡一句:「「把关」被驳回,转到「退回申请人」」(名字是搭图时起的,不带查询到的数据) + lines.append(("驳回", ";".join(f"「{str(v.get('label') or v.get('node_id') or '')[:80]}」{rejected_exit_text(v)}" + for v in notice.rejected_via or [] if isinstance(v, dict)))) if notice.tier: gaps = f"(缺口 {notice.gaps} 条)" if notice.gaps is not None else "" lines.append(("出具档位", f"{ISSUANCE_LABEL.get(notice.tier, notice.tier)}{gaps}")) @@ -244,6 +262,8 @@ def _status_label(notice: Notice) -> str: return "已挂起,等待核实" if notice.action_pending else "已挂起,可继续运行" if notice.rejected: return REJECTED_STATUS + if notice.via_rejection: + return f"{_STATUS_LABEL['succeeded']}{VIA_REJECTED}" return _STATUS_LABEL.get(notice.status or "", notice.status or "") @@ -554,12 +574,23 @@ async def _notice(session: Any, schedule: Schedule, fire: ScheduleFire, run: Run 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, + rejected_via=await _rejected_via_of(session, run) if run is not None and run.status == "succeeded" else None, # 事件触发的任务没有下一次「到点」,启用着就是下一次事件来了照常发起 next_fire=bool(schedule.enabled and (schedule.next_run_at is not None or schedule.trigger_kind == "event")), trigger_text=_trigger_text(fire), ) +async def _rejected_via_of(session: Any, run: Run) -> list[dict[str, Any]] | None: + """这次运行途经的驳回出口:最后一条 run.finished 上的 rejected_via(子工作流里的那几步名字写「子工作流 › x」),没有为 None。""" + data = await session.scalar(select(RunEvent.data).where(RunEvent.run_id == run.id, RunEvent.type == "run.finished") + .order_by(RunEvent.seq.desc()).limit(1)) + via = data.get("rejected_via") if isinstance(data, dict) else None + if not isinstance(via, list): + return None + return [v for v in via if isinstance(v, dict)] or None + + def _trigger_text(fire: ScheduleFire) -> str | None: event = fire.inputs.get("event") if isinstance(fire.inputs, dict) else None text = event.get("text") if isinstance(event, dict) else None diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 01b207b7..f7f2d1f0 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -376,8 +376,10 @@ async def test_disagreement_pauses_with_the_agreed_payload_and_card(monkeypatch, approval = await pending(run.id) assert approval.mode == "input" and approval.node_id == "dual" payload = approval.payload - # 约定的六个键,另加 on_reject(实际生效的处置):审核人在卡上要看得出没采用原流程会不会让运行失败 - assert set(payload) == {"kind", "node_id", "mode", "title", "card_artifact", "summary", "on_reject"} + # 约定的六个键,另加 on_reject(实际生效的处置):审核人在卡上要看得出没采用原流程会不会让运行失败;reject_route + # (驳回之后去哪,engine/replay.reject_route):没接驳回出口,驳回后运行就此结束 + assert set(payload) == {"kind", "node_id", "mode", "title", "card_artifact", "summary", "on_reject", "reject_route"} + assert payload["reject_route"] == {"ends": "run"} assert payload["kind"] == "dual_check" and payload["node_id"] == "dual" and payload["mode"] == "input" assert payload["on_reject"] == "degrade" [item] = payload["summary"] diff --git a/backend/tests/test_reject_exit_wording.py b/backend/tests/test_reject_exit_wording.py new file mode 100644 index 00000000..5f072208 --- /dev/null +++ b/backend/tests/test_reject_exit_wording.py @@ -0,0 +1,197 @@ +"""驳回出口之后,待办、通知、定时任务的 Webhook 照实说「被驳回,转到 X」(docs/design/orchestration-round/00-总方案.md §3)。 + +实例页已经这么说(test_reject_exit.test_the_instance_says_rejected_and_where_it_went)。这里对的是另外几处: +- 待办:中断内容带着驳回之后去哪(reject_route):接了驳回出口的写转到哪几步,没接的写运行就此结束,审批模式没接、没设 + 驳回即终止的写这一路到此为止。待办卡的提示、驳回后的提示、处理过的结果照它说,不再一律写「转入驳回分支」「本次运行终止」; +- 子工作流里的关卡走了驳回出口:父运行的终态事件照实带上(「子工作流 › 把关」被驳回,转到「子工作流 › 退回」),实例页、 + 通知说这一句,不写内部 id、不写成「已完成」; +- 定时任务的 Webhook:途经驳回、照驳回那条路成功结束的,标题和状态写途经驳回,另起一行写谁被驳回、转到哪一步。 +图和数据都是虚构的。 +""" +from __future__ import annotations + +import asyncio + +import pytest +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent, Task +from app.engine.runner import run_manager +from tests import ledger_samples as S + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}} + + +START = node("start", "input", fields=[{"name": "q"}]) +DONE = node("done", "output", "办结", fields=[{"name": "结果", "value": "通过了"}]) +BACK = node("back", "output", "退回申请人", fields=[{"name": "结果", "value": "退回:{{ nodes.gate.note }}"}]) + + +def gated(gate: dict, *, exit_to: dict | None = BACK) -> dict: + normal = "approved" if gate["data"]["config"].get("mode") == "approve" else "out" + nodes = [START, gate, DONE] + ([exit_to] if exit_to else []) + edges = [{"source": "start", "target": "gate"}, {"source": "gate", "target": "done", "sourceHandle": normal}] + if exit_to: + edges.append({"source": "gate", "target": exit_to["id"], "sourceHandle": "rejected"}) + return {"nodes": nodes, "edges": edges} + + +async def _wait(run_id: str, statuses: tuple[str, ...], timeout: float = 20.0) -> Run: + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row is not None and row.status in statuses and not run_manager.is_active(run_id): + return row + await asyncio.sleep(0.05) + raise AssertionError(f"运行 {run_id} 没有进入 {statuses}") + + +async def _parked(graph: dict) -> tuple[Run, dict]: + """发起、停在关卡上,交回运行和那条待办的中断内容。""" + run = await run_manager.start(graph=graph, input_payload={"q": "x"}) + await _wait(run.id, ("interrupted",)) + await run_manager.wait_idle(run.id) + async with SessionLocal() as session: + [task] = list((await session.execute(select(Task).where(Task.run_id == run.id))).scalars()) + from app.db.models import Approval + + approval = await session.get(Approval, task.approval_id) + return run, dict(approval.payload or {}) + + +APPROVE = node("gate", "human", "把关", mode="approve", title="可以吗") +INPUT = node("gate", "human", "补充", mode="input", title="补一下") +TOOL = node("gate", "tool", "算一下", tool="calculator", args={"expression": "1+1"}, approval="always") + + +@pytest.mark.parametrize("gate", [APPROVE, INPUT, TOOL], ids=["approve", "input", "tool"]) +async def test_the_task_says_where_a_rejection_goes(gate): + _, payload = await _parked(gated(gate)) + assert payload["reject_route"] == {"to": [{"node_id": "back", "label": "退回申请人"}]} + + +@pytest.mark.parametrize("gate", [INPUT, TOOL], ids=["input", "tool"]) +async def test_without_the_exit_the_task_says_the_run_ends(gate): + _, payload = await _parked(gated(gate, exit_to=None)) + assert payload["reject_route"] == {"ends": "run"} + + +async def test_an_approval_without_the_exit_says_this_path_stops_here(): + _, payload = await _parked(gated(APPROVE, exit_to=None)) + assert payload["reject_route"] == {"ends": "path"} + stop = node("gate", "human", "把关", mode="approve", title="可以吗", stop_on_reject=True) + _, payload = await _parked(gated(stop)) + assert payload["reject_route"] == {"ends": "run"}, "驳回即终止优先:驳回出口那条线走不到" + + +async def test_an_agent_tool_approval_has_no_route(): + """agent 自己调工具的审批不是关卡:驳回了 agent 照常往下想,中断内容不带 reject_route。""" + from app.engine.replay import reject_route + from app.engine.schema import GraphSpec + + spec = GraphSpec.model_validate(gated(node("gate", "agent", "助手", prompt="x"))) + + class Ctx: + class run: # noqa: N801 - 只要 spec 和 global_id + pass + + ctx = Ctx() + ctx.node = spec.node_map()["gate"] + ctx.run.spec = spec + assert reject_route(ctx) is None + + +# -------------------------------------------------------------------------- +# 子工作流里的关卡走了驳回出口 +# -------------------------------------------------------------------------- + + +async def _sub_rejected() -> Run: + inner = {"nodes": [node("s", "input"), node("gate", "human", "把关", mode="approve", title="可以吗"), + node("ok", "output", "通过", fields=[{"name": "r", "value": "ok"}]), + node("back", "output", "退回", fields=[{"name": "r", "value": "退回:{{ nodes.gate.note }}"}])], + "edges": [{"source": "s", "target": "gate"}, + {"source": "gate", "target": "ok", "sourceHandle": "approved"}, + {"source": "gate", "target": "back", "sourceHandle": "rejected"}]} + sub_id = await S.publish("复核子流程", graph=inner, level="draft") + graph = S.chain(node("start", "input"), node("sub", "subgraph", "复核子流程", workflow_id=sub_id, workflow_version=1), + node("out", "output", "成果", fields=[{"name": "r", "value": "{{ nodes.sub.output.r }}"}])) + run = await run_manager.start(graph=graph, input_payload={}) + await _wait(run.id, ("interrupted",)) + await run_manager.wait_idle(run.id) + await run_manager.resume(run.id, {"approved": False, "note": "不对"}, actor="张工") + return await _wait(run.id, ("cancelled", "failed", "succeeded")) + + +async def test_a_reject_exit_inside_a_subworkflow_is_named_in_the_parent(): + from app.apps.resolve import _trails, instance_state + + final = await _sub_rejected() + assert final.status == "succeeded", final.error + assert final.output == {"r": "退回:不对"} + async with SessionLocal() as session: + [ended] = list((await session.execute(select(RunEvent).where( + RunEvent.run_id == final.id, RunEvent.type == "run.finished"))).scalars()) + run = await session.get(Run, final.id) + state = instance_state(run, (await _trails(session, [run]))[run.id]) + [via] = ended.data["rejected_via"] + assert (via["node_id"], via["label"]) == ("sub/gate", "复核子流程 › 把关") + assert via["to"] == [{"node_id": "sub/back", "label": "复核子流程 › 退回"}] + # 实例页、通知(notify_run 取的就是这一句):照实说被驳回、转到哪,不写「sub/gate」,不画成已完成 + assert state == {"code": "cancelled", "text": "「复核子流程 › 把关」被驳回,转到「复核子流程 › 退回」"} + + +async def test_the_inner_task_names_the_route_with_the_subworkflow(): + inner = {"nodes": [node("s", "input"), node("gate", "human", "补充", mode="input", title="补一下"), + node("back", "output", "退回", fields=[{"name": "r", "value": "x"}]), + node("ok", "output", "通过", fields=[{"name": "r", "value": "y"}])], + "edges": [{"source": "s", "target": "gate"}, {"source": "gate", "target": "ok"}, + {"source": "gate", "target": "back", "sourceHandle": "rejected"}]} + sub_id = await S.publish("补充子流程", graph=inner, level="draft") + graph = S.chain(node("start", "input"), node("sub", "subgraph", "补充子流程", workflow_id=sub_id, workflow_version=1), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.sub.output.r }}"}])) + _, payload = await _parked(graph) + assert payload["reject_route"] == {"to": [{"node_id": "sub/back", "label": "补充子流程 › 退回"}]} + + +# -------------------------------------------------------------------------- +# 定时任务的 Webhook +# -------------------------------------------------------------------------- + + +def test_the_webhook_says_a_run_went_through_a_rejection(): + from app.scheduler.webhook import Notice, message + + via = [{"node_id": "gate", "label": "把关", "kind": "approve", "decision": "reject", "actor": "张工", "note": "不对", + "to": [{"node_id": "back", "label": "退回申请人"}]}] + title, lines, _ = message(Notice(event="finished", schedule_id="s1", schedule_name="周报", timezone="UTC", + status="succeeded", rejected_via=via)) + assert "途经驳回" in title + said = dict(lines) + assert said["状态"] == "已完成(途经驳回)" + assert said["驳回"] == "「把关」被驳回,转到「退回申请人」" + # 没途经驳回的照旧 + title, lines, _ = message(Notice(event="finished", schedule_id="s1", schedule_name="周报", timezone="UTC", + status="succeeded")) + assert title == "定时任务「周报」运行完成" and dict(lines)["状态"] == "已完成" and "驳回" not in dict(lines) + + +async def test_the_webhook_notice_reads_the_route_from_the_end_event(): + from app.scheduler.webhook import _rejected_via_of + + final = await _sub_rejected() + async with SessionLocal() as session: + run = await session.get(Run, final.id) + via = await _rejected_via_of(session, run) + assert [v["label"] for v in via] == ["复核子流程 › 把关"] diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 4c4d5056..55bbbfe6 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -5075,6 +5075,8 @@ export const DUAL_CHECK_TEXT = { /** 认不出任何指标时留的出口:驳回后节点按人工驳回失败 */ reject: '驳回并终止', emptyReject: '驳回后节点失败,本次运行终止。可以在备注中说明原因', + /** 节点接了驳回出口(中断内容的 reject_route):驳回后转到那一步 */ + emptyRejectRoute: (goes: string) => `驳回后${goes}。可以在备注中说明原因`, rejected: '已驳回,本次运行终止', }, diff --git a/frontend/src/lib/termsApp.ts b/frontend/src/lib/termsApp.ts index 9f7fb8c4..d17b8762 100644 --- a/frontend/src/lib/termsApp.ts +++ b/frontend/src/lib/termsApp.ts @@ -110,7 +110,7 @@ export const TASK_TEXT = { group: (name: string) => `组「${name}」`, }, assignmentStatus: { pending: '待回复', responded: '已回复', withdrawn: '已撤回' }, - decision: { approve: '同意', reject: '驳回', submit: '已提交', confirmed: '已生效', not_applied_resend: '重新发送', + decision: { approve: '同意', reject: '驳回', expired_reject: '超时驳回', submit: '已提交', confirmed: '已生效', not_applied_resend: '重新发送', not_applied_abandon: '放弃' } as Record, verified: '已登录核实', unverified: '自报署名', @@ -125,9 +125,13 @@ export const TASK_TEXT = { editedContent: '修改后的内容', approveHint: '批准:继续运行;驳回:转入「驳回」分支', inputHint: '提交:以你填写的内容继续运行;驳回:终止本次运行', + /** 驳回之后去哪(中断内容的 reject_route,见 REJECT_ROUTE_TEXT):带了的按它说,老待办没带的照上面两句 */ + approveHintRoute: (goes: string) => `批准:继续运行;驳回:${goes}`, + inputHintRoute: (goes: string) => `提交:以你填写的内容继续运行;驳回:${goes}`, doneApprove: '已批准,运行继续', doneReject: '已驳回,转入驳回分支', doneRejectStop: '已驳回,本次运行终止', + doneRejectRoute: (goes: string) => `已驳回,${goes}`, doneSubmit: '已提交,运行继续', doneVote: '已记下你的意见,等待其他人处理', formInvalid: (n: number) => `有 ${n} 项需要修改`, @@ -154,6 +158,34 @@ export const TASK_TEXT = { code: '代码', } +/** + * 驳回之后去哪:待办的中断内容带着 reject_route(服务端 engine/replay.reject_route,总方案 §3 的驳回出口)。 + * - to:接了驳回出口,驳回后转到这几步(名字是这次运行里的说法:子工作流里的写「子工作流 › x」); + * - ends=run:没接驳回出口(审批设了驳回即终止的也是),驳回后运行就此结束、记为被驳回; + * - ends=path:审批没接驳回出口、没设驳回即终止,这一路到审批为止,不再往下走。 + * 老待办没带的为 null,各处照原来的说法 + */ +export type RejectRoute = { to: string[] } | { ends: 'run' | 'path' } + +export function rejectRouteOf(payload: Record | null | undefined): RejectRoute | null { + const raw = payload?.reject_route + if (!raw || typeof raw !== 'object') return null + const r = raw as { to?: unknown; ends?: unknown } + if (Array.isArray(r.to)) { + const to = r.to.map((t: any) => String(t?.label || t?.node_id || '')).filter(Boolean) + return to.length ? { to } : null + } + return r.ends === 'run' || r.ends === 'path' ? { ends: r.ends } : null +} + +export const REJECT_ROUTE_TEXT = { + /** 「转到「通知申请人」」/「本次运行终止」/「这一路到此为止,不再往下走」 */ + goes: (r: RejectRoute): string => ('to' in r ? `转到${r.to.map((t) => `「${t}」`).join('、')}` + : r.ends === 'run' ? '本次运行终止' : '这一路到此为止,不再往下走'), + /** 处理过的待办「结果」那一格:驳回之后去了哪 */ + outcome: (decision: string, r: RejectRoute): string => `${decision},${REJECT_ROUTE_TEXT.goes(r)}`, +} + /** 动作确认的哈希卡(§3.8:确认时看过的就是发出去的) */ export const ACTION_CONFIRM_TEXT = { head: '确认后发送', @@ -171,6 +203,8 @@ export const ACTION_CONFIRM_TEXT = { noHash: '缺少内容哈希,无法确认发送', done: '已确认,按此内容发送', rejected: '已驳回,不发送', + /** 驳回之后去哪(reject_route):「已驳回,不发送,转到「通知申请人」」 */ + rejectedRoute: (goes: string) => `已驳回,不发送,${goes}`, } /** 核实结果未知的动作:三个选项(§2.5 A5 第 4 条) */ diff --git a/frontend/src/run/DualCheckApproval.tsx b/frontend/src/run/DualCheckApproval.tsx index 7f25f936..37173bec 100644 --- a/frontend/src/run/DualCheckApproval.tsx +++ b/frontend/src/run/DualCheckApproval.tsx @@ -10,6 +10,7 @@ import { Spinner, useToast } from '../components/ui' import { segmentState } from '../lib/evidence' import { NONE } from '../lib/format' import { CATALOG_TABLE_FIELD_LABEL, DUAL_CHECK_TEXT } from '../lib/terms' +import { REJECT_ROUTE_TEXT, TASK_TEXT, rejectRouteOf } from '../lib/termsApp' import { patchValueText } from './catalogPatch' import { RationaleDoc } from './ChallengeSummary' import type { @@ -421,6 +422,9 @@ export function DualCheckApproval({ approval, onResolved, decide }: DualCheckApp const summary = useMemo(() => summaryOf(payload.summary), [payload.summary]) const cardId = str(payload.card_artifact) ?? '' const onReject = useOnReject(approval.run_id, str(payload.node_id) ?? str(approval.node_id), payload.on_reject) + // 驳回之后去哪(中断内容的 reject_route):接了驳回出口的,驳回后转到那一步,不是「本次运行终止」 + const route = rejectRouteOf(payload) + const routed = route && 'to' in route ? REJECT_ROUTE_TEXT.goes(route) : null const [card, setCard] = useState<{ id: string; data?: Card | null } | null>(null) useEffect(() => { @@ -468,7 +472,7 @@ export function DualCheckApproval({ approval, onResolved, decide }: DualCheckApp } // 采用盲算、两边都不对时,节点设为「判为失败」就会让运行终止:提示照节点的设置说,读不到设置时不替它下结论 const rejected = Object.values(value.choices).some((c) => c !== 'a') - toast(!approved ? T.rejected : !rejected || onReject === 'degrade' ? T.done : onReject === 'fail' ? T.doneFail : T.doneUnknown, 'ok') + toast(!approved ? (routed ? TASK_TEXT.doneRejectRoute(routed) : T.rejected) : !rejected || onReject === 'degrade' ? T.done : onReject === 'fail' ? T.doneFail : T.doneUnknown, 'ok') } catch (e) { toast.error(e) } finally { @@ -514,9 +518,11 @@ export function DualCheckApproval({ approval, onResolved, decide }: DualCheckApp {noteField} -
{T.emptyReject}
+
+ {routed ? T.emptyRejectRoute(routed) : T.emptyReject} +
) ) : ( diff --git a/frontend/src/run/TaskCard.tsx b/frontend/src/run/TaskCard.tsx index 78484f99..9eb99474 100644 --- a/frontend/src/run/TaskCard.tsx +++ b/frontend/src/run/TaskCard.tsx @@ -39,7 +39,7 @@ import { ACTION_CONFIRM_TEXT, ACTION_ENV_LABEL, INQUIRY_TEXT, RECONCILE_TEXT, SIGNAL_TEXT, TASK_KIND_LABEL, TASK_STATUS_LABEL, TASK_TEXT, TOOL_TRUST_TEXT, TRIAL_TEXT, runName, } from '../lib/terms' -import { REVIEW_TASK_TEXT } from '../lib/termsApp' +import { REJECT_ROUTE_TEXT, REVIEW_TASK_TEXT, rejectRouteOf } from '../lib/termsApp' import { RECALL_TEXT } from '../lib/termsRecall' import { blueprintPath } from '../pages/blueprints/shared' import type { Approval, ApprovalReport } from '../types' @@ -102,6 +102,9 @@ export function TaskCard({ task, onResolved, compact = false, send, showOwner = const runClass: 'formal' | 'exploratory' | null = task.run_class ?? (task.app_id ? 'formal' : payload.run_class === 'formal' || payload.run_class === 'exploratory' ? payload.run_class : null) const binary = kind === 'approve' || kind === 'tool' || kind === 'code' + // 驳回之后去哪(中断内容的 reject_route):带了的照实说转到哪一步、运行终止还是这一路到此为止;老待办没带的照原来的说法 + const route = rejectRouteOf(payload) + const stops = route ? 'ends' in route && route.ends === 'run' : !binary const trustKey: string | undefined = kind === 'tool' && runClass !== 'formal' ? (typeof payload.trust_key === 'string' && payload.trust_key) || undefined : undefined const reports: ApprovalReport[] = Array.isArray(payload.reports) @@ -140,7 +143,8 @@ export function TaskCard({ task, onResolved, compact = false, send, showOwner = if (binary) void respond({ decision: 'approve' }, TASK_TEXT.doneApprove) else void respond({ decision: 'submit', values: { value: text } }, TASK_TEXT.doneSubmit) } - const reject = () => void respond({ decision: 'reject' }, binary ? TASK_TEXT.doneReject : TASK_TEXT.doneRejectStop) + const reject = () => void respond({ decision: 'reject' }, route ? TASK_TEXT.doneRejectRoute(REJECT_ROUTE_TEXT.goes(route)) + : binary ? TASK_TEXT.doneReject : TASK_TEXT.doneRejectStop) /** 「始终允许」:批准这一次,并让服务端把这个工具设成门控把关(工具页的信任档跟着变,目录重拉一次) */ const always = async () => { @@ -209,7 +213,8 @@ export function TaskCard({ task, onResolved, compact = false, send, showOwner = {kind === 'action_confirm' && ( void respond({ decision: 'approve', payload_hash: hash }, ACTION_CONFIRM_TEXT.done)} - onReject={() => void respond({ decision: 'reject' }, ACTION_CONFIRM_TEXT.rejected)} /> + onReject={() => void respond({ decision: 'reject' }, route && 'to' in route + ? ACTION_CONFIRM_TEXT.rejectedRoute(REJECT_ROUTE_TEXT.goes(route)) : ACTION_CONFIRM_TEXT.rejected)} /> )} {kind === 'reconcile' && ( {' '} - {binary ? (trial ? TRIAL_TEXT.reject : TASK_TEXT.reject) : (trial ? TRIAL_TEXT.rejectStop : TASK_TEXT.rejectStop)} + {stops ? (trial ? TRIAL_TEXT.rejectStop : TASK_TEXT.rejectStop) : (trial ? TRIAL_TEXT.reject : TASK_TEXT.reject)} {inquiring && ( @@ -272,7 +277,8 @@ export function TaskCard({ task, onResolved, compact = false, send, showOwner = )} {!compact && (
- {binary ? TASK_TEXT.approveHint : TASK_TEXT.inputHint} + {route ? (binary ? TASK_TEXT.approveHintRoute : TASK_TEXT.inputHintRoute)(REJECT_ROUTE_TEXT.goes(route)) + : binary ? TASK_TEXT.approveHint : TASK_TEXT.inputHint}
)} {/* 「始终允许」会改工具的信任档、影响之后的运行:常显同一句旁注,不只藏在悬停里 */} @@ -429,8 +435,14 @@ export function Due({ dueAt, now }: { dueAt: string | null; now: number }) { } /** 处理过的:谁、什么时候、结果(实例页上不是处理人的人看这一步的状态也用它) */ -export function Resolved({ task }: { task: Pick }) { - const decision = typeof task.outcome?.decision === 'string' ? TASK_TEXT.decision[task.outcome.decision] ?? task.outcome.decision : null +export function Resolved({ task }: { + task: Pick & { payload?: Task['payload'] } +}) { + const raw = typeof task.outcome?.decision === 'string' ? task.outcome.decision : null + const said = raw ? TASK_TEXT.decision[raw] ?? raw : null + // 驳回了的写驳回之后去了哪(中断内容的 reject_route):「驳回,转到「通知申请人」」,不只写「驳回」 + const route = raw === 'reject' || raw === 'expired_reject' ? rejectRouteOf(task.payload) : null + const decision = said && route ? REJECT_ROUTE_TEXT.outcome(said, route) : said const at = task.resolved_at ? formatDateTime(task.resolved_at) : '' return (
diff --git a/scripts/check-inbox.mjs b/scripts/check-inbox.mjs index edf06083..f9d38ee0 100644 --- a/scripts/check-inbox.mjs +++ b/scripts/check-inbox.mjs @@ -13,6 +13,8 @@ // 待办箱一行写所属实例,同标题的分得开; // - 补充输入带表单的:本地校验,服务端 422 的错误挂回字段;表单里的选人控件按这条待办的候选池取候选(使用者读不到 // 整份名录),不在池里的服务端 422(code=principal)挂回那个字段(05-已知遗留 (b)); +// - 驳回之后去哪(中断内容的 reject_route):接了驳回出口的提示、驳回后的提示、处理过的结果写「转到「x」」,子工作流里的 +// 写「子工作流 › x」、不露内部 id;补充输入接了出口的按钮写「驳回」,没接出口的工具审批写「驳回并终止」; // - 关登录时签批人一行写本机署名;会签只记下一票(resolved 为假)时提示「等待其他人处理」; // - 审核个人应用(kind=review,payload 同服务端 blueprints/personal.review_payload):写提交人(显示名)、目的、说明;没有 // 「同意」——「取副本,去发布」调 POST /api/blueprints/{id}/copy 后进副本蓝图;取过的给「打开我的副本」(只认自己取的那 @@ -204,6 +206,24 @@ const TASKS = { reviewer: { type: 'string', title: '复核人', 'x-widget': 'principal' } } }, payload: { kind: 'human_node', message: '请指定复核人' }, }, + // 驳回之后去哪(中断内容的 reject_route,服务端 engine/replay.reject_route):接了驳回出口的写转到哪一步,没接的写运行终止 + routed: { + ...base, id: 'routed', approval_id: 'ap9', kind: 'approve', title: '主管审批(接了驳回出口)', + payload: { kind: 'human_node', message: '王五报销 800 元', reject_route: { to: [{ node_id: 'tell', label: '通知申请人' }] } }, + }, + routedInput: { + ...base, id: 'routedInput', approval_id: 'ap10', kind: 'input', title: '补填用途(接了驳回出口)', + payload: { kind: 'human_node', message: '请补填用途', reject_route: { to: [{ node_id: 'sub/back', label: '复核子流程 › 退回' }] } }, + }, + stopTool: { + ...base, id: 'stopTool', approval_id: 'ap11', kind: 'tool', title: '调用工具:erp_query', + payload: { kind: 'tool_approval', tool: 'erp_query', args: { q: 1 }, reject_route: { ends: 'run' } }, + }, + rejectedDone: { + ...base, id: 'rejectedDone', approval_id: 'ap12', kind: 'approve', title: '主管审批(已驳回)', status: 'resolved', + can_respond: false, outcome: { decision: 'reject' }, resolved_by: 'zhangsan', resolved_at: ago(300_000), + payload: { kind: 'human_node', message: '赵六报销 50 元', reject_route: { to: [{ node_id: 'tell', label: '通知申请人' }] } }, + }, input: { ...base, id: 'input', approval_id: 'ap5', kind: 'input', title: '补填发票信息', form: { format: 'agentlab.form/1', type: 'object', required: ['invoice'], properties: { @@ -612,6 +632,38 @@ await section('口径召回:后台的系统通知点开去待办页(不去 } }) +await section('驳回之后去哪:提示、按钮、驳回后的提示、处理过的结果', async () => { + const state = { writes: [] } + const page = await open(state) + await page.goto(`${WEB}/apps/inbox/routed`) + await card(page).waitFor({ timeout: 15000 }) + let text = await card(page).innerText() + check('接了驳回出口的审批:提示写驳回后转到哪一步', text.includes('驳回:转到「通知申请人」'), text.slice(-80)) + check('……不再写「转入「驳回」分支」', !text.includes('转入「驳回」分支')) + await page.locator('[data-task-reject]').click() + check('驳回之后提示转到哪一步', await page.getByText('已驳回,转到「通知申请人」').waitFor({ timeout: 3000 }).then(() => true, () => false)) + check('驳回照约定交给待办接口', state.writes.some((w) => w.path === '/api/tasks/routed/respond' && w.body?.decision === 'reject')) + + await page.goto(`${WEB}/apps/inbox/routedInput`) + await card(page).waitFor({ timeout: 15000 }) + text = await card(page).innerText() + check('补充输入接了驳回出口:按钮写「驳回」,不写「驳回并终止」', (await page.locator('[data-task-reject]').innerText()).trim() === '驳回') + check('……提示写子工作流里的那一步', text.includes('驳回:转到「复核子流程 › 退回」'), text.slice(-80)) + check('……不露内部 id', !text.includes('sub/back')) + + await page.goto(`${WEB}/apps/inbox/stopTool`) + await card(page).waitFor({ timeout: 15000 }) + text = await card(page).innerText() + check('没接驳回出口的工具审批:按钮写「驳回并终止」', (await page.locator('[data-task-reject]').innerText()).trim() === '驳回并终止') + check('……提示写本次运行终止', text.includes('驳回:本次运行终止'), text.slice(-80)) + + await page.goto(`${WEB}/apps/inbox/rejectedDone`) + await card(page).waitFor({ timeout: 15000 }) + const resolved = await page.locator('[data-task-resolved]').first().innerText() + check('处理过的:结果写驳回之后去了哪', resolved.includes('结果:驳回,转到「通知申请人」'), resolved) + check('没有未捕获的运行时错误', page.errors.length === 0, page.errors[0] ?? '') +}) + await section('待办箱:列表和徽标', async () => { const page = await open({ writes: [] }) await page.goto(`${WEB}/apps?tab=inbox`)