From 4ca7d0c768ecfeb07900fbe77373b1a8840a95da Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 18 Oct 2026 18:25:24 +0800 Subject: [PATCH] =?UTF-8?q?=E9=A9=B3=E5=9B=9E=E5=87=BA=E5=8F=A3=E6=8E=A5?= =?UTF-8?q?=E4=B8=8A=E6=95=B0=E6=8D=AE=E4=BA=8B=E4=BB=B6=E8=A7=A6=E5=8F=91?= =?UTF-8?q?=EF=BC=9A=E6=AD=A3=E5=BC=8F=E8=BF=90=E8=A1=8C=E6=94=B6=E5=B0=BE?= =?UTF-8?q?=E6=97=B6=E6=8A=8A=E9=80=94=E7=BB=8F=E7=9A=84=E9=A9=B3=E5=9B=9E?= =?UTF-8?q?=EF=BC=88=5Frejected=5Fvia=EF=BC=89=E4=BA=A4=E7=BB=99=20outbox.?= =?UTF-8?q?stage=5Frun=5Ffinished=EF=BC=8Cworkflow.finished=20=E7=85=A7?= =?UTF-8?q?=E5=8F=91=E5=B9=B6=E5=B8=A6=E4=B8=8A=20rejected=5Fvia=EF=BC=8C?= =?UTF-8?q?=E7=BC=BA=E7=9C=81=E4=B8=8D=E8=AE=A4=E9=80=94=E7=BB=8F=E9=A9=B3?= =?UTF-8?q?=E5=9B=9E=E7=9A=84=E8=A7=A6=E5=8F=91=E4=B8=8D=E6=B4=BE=EF=BC=9B?= =?UTF-8?q?=E8=A1=A5=E4=B8=80=E6=9D=A1=E8=B5=B0=E9=A9=B3=E5=9B=9E=E5=87=BA?= =?UTF-8?q?=E5=8F=A3=E7=9A=84=E6=AD=A3=E5=BC=8F=E8=BF=90=E8=A1=8C=E4=B8=8E?= =?UTF-8?q?=E6=89=B9=E5=87=86=E8=B5=B0=E5=AE=8C=E7=9A=84=E5=AF=B9=E7=85=A7?= =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/runner.py | 3 +- backend/tests/test_reject_exit_triggers.py | 55 ++++++++++++++++++++++ 2 files changed, 57 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_reject_exit_triggers.py diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 35e67e5e..1b3d020c 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -1064,7 +1064,8 @@ class RunManager: if event is not None: await _stage(session, event, run) from app.scheduler import outbox # 数据事件触发：正式运行成功结束，同一个事务记一条待派发事件 - outbox.stage_run_finished(session, run) + # 途经驳回出口、按驳回那条路走完的也是 succeeded：照发，事件带上途经了哪几个驳回（缺省不认途经驳回的触发） + outbox.stage_run_finished(session, run, rejected_via=_rejected_via(state).get("rejected_via")) await session.commit() if event is not None: await bus.publish(event) diff --git a/backend/tests/test_reject_exit_triggers.py b/backend/tests/test_reject_exit_triggers.py new file mode 100644 index 00000000..a0ba8bc1 --- /dev/null +++ b/backend/tests/test_reject_exit_triggers.py @@ -0,0 +1,55 @@ +"""驳回出口和数据事件触发接在一起（docs/design/orchestration-round/00-总方案.md §3、§4；合并时接上的那一下：runner._finalize +把 _rejected_via(state) 交给 outbox.stage_run_finished）。 + +一次正式运行在审批那一步被驳回、走驳回出口跑完（状态 succeeded）：写下的 workflow.finished 带上途经了哪个关卡的驳回， +缺省不认途经驳回的下游不派；同一张图批准走完的那次，事件里照实说没途经驳回。工作流是虚构的。 +""" +from __future__ import annotations + +import uuid + +from app.db.base import SessionLocal +from app.db.models import Workflow, WorkflowVersion +from app.engine.launch import launch_run +from app.engine.runner import run_manager +from tests.test_reject_exit import GATES, _wait, gated +from tests.test_schedule_events import engine_up, events, settle # noqa: F401 （engine_up：清掉定时任务和事件） + + +async def _published(graph: dict) -> str: + async with SessionLocal() as session: + wf = Workflow(name=f"报销审批-{uuid.uuid4().hex[:6]}", graph=graph, status="published", published_version=1) + session.add(wf) + await session.flush() + session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=graph)) + await session.commit() + return wf.id + + +async def _formal_answered(workflow_id: str, reply: dict): + async with SessionLocal() as session: + run = await launch_run(session, workflow_id=workflow_id, run_class="formal", input_payload={"q": "x"}, + actor="检查员", origin_kind="launcher") + await _wait(run.id, ("interrupted",)) + await run_manager.wait_idle(run.id) + await run_manager.resume(run.id, reply, actor="张工") + return await settle(run.id) + + +async def test_a_formal_run_that_went_through_the_reject_exit_stages_an_event_marked_rejected(): + gate, _ = GATES["approve"] + wf = await _published(gated(gate)) + final = await _formal_answered(wf, {"approved": False, "note": "不对"}) + assert final.status == "succeeded" and final.output["结果"] == "退回：不对" + [ev] = await events("workflow.finished", wf) + assert ev.payload["rejected"] is True + assert [(v["node_id"], v["label"], v["decision"]) for v in ev.payload["rejected_via"]] == [("gate", "把关", "reject")] + + +async def test_the_same_workflow_approved_stages_an_event_not_marked_rejected(): + gate, _ = GATES["approve"] + wf = await _published(gated(gate)) + final = await _formal_answered(wf, {"approved": True}) + assert final.status == "succeeded" and final.output == {"结果": "通过了"} + [ev] = await events("workflow.finished", wf) + assert ev.payload["rejected"] is False and not ev.payload.get("rejected_via")