From c59b08592234c3aaba3a5d952fa1da5d680c585c Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 11:43:41 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AD=90=E5=B7=A5=E4=BD=9C=E6=B5=81=E9=87=8C?= =?UTF-8?q?=E7=9A=84=E5=86=99=E5=85=A5=E8=B7=A8=E5=8D=87=E7=BA=A7=E8=AE=A4?= =?UTF-8?q?=E6=97=A7=E9=94=AE=E6=97=B6=E6=8C=89=E4=BA=8B=E4=BB=B6=E5=92=8C?= =?UTF-8?q?=E5=86=85=E5=AE=B9=E8=AE=A4=E6=98=AF=E8=B0=81=E7=9A=84=E3=80=81?= =?UTF-8?q?=E6=98=AF=E5=93=AA=E4=B8=80=E7=AC=94=EF=BC=88=E5=A4=8D=E6=A0=B8?= =?UTF-8?q?=20L1=E3=80=81L2=EF=BC=89=EF=BC=9A=E5=8D=87=E7=BA=A7=E5=89=8D?= =?UTF-8?q?=E5=AD=90=E5=9B=BE=E9=87=8C=E7=9A=84=E5=86=99=E5=85=A5=E6=8C=89?= =?UTF-8?q?=E5=AD=90=E5=9B=BE=E9=87=8C=E7=9A=84=20id=20=E8=AE=B0=E9=94=AE?= =?UTF-8?q?=EF=BC=8C=E5=92=8C=E7=88=B6=E5=9B=BE=E5=90=8C=E5=90=8D=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E7=9A=84=E9=94=AE=E4=B8=80=E6=A0=B7=E3=80=82=E4=BB=A5?= =?UTF-8?q?=E5=89=8D=E7=88=B6=E5=9B=BE=E6=9C=89=E5=90=8C=E5=90=8D=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E6=97=B6=E4=B8=80=E5=BE=8B=E4=B8=8D=E8=AE=A4=E6=97=A7?= =?UTF-8?q?=E9=94=AE=EF=BC=9A=E5=AD=90=E5=9B=BE=E9=87=8C=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=E6=9C=AA=E7=9F=A5=E7=9A=84=E9=82=A3=E4=B8=80=E7=AC=94=E6=A0=B8?= =?UTF-8?q?=E5=AE=9E=E4=B8=BA=E5=B7=B2=E7=94=9F=E6=95=88=E5=90=8E=E6=8E=A5?= =?UTF-8?q?=E7=9D=80=E8=B7=91=EF=BC=8C=E7=94=A8=E6=96=B0=E9=94=AE=E5=8F=88?= =?UTF-8?q?=E5=8F=91=E4=B8=80=E7=AC=94=EF=BC=9B=E7=88=B6=E5=9B=BE=E9=82=A3?= =?UTF-8?q?=E4=B8=AA=E8=8A=82=E7=82=B9=E6=8C=89=E8=87=AA=E5=B7=B1=E7=9A=84?= =?UTF-8?q?=E9=94=AE=E6=89=BE=E5=88=B0=E5=AD=90=E5=9B=BE=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E7=AC=94=E3=80=81=E4=BA=A4=E5=9B=9E=E5=AE=83=E7=9A=84=E5=9B=9E?= =?UTF-8?q?=E6=89=A7=EF=BC=8C=E8=87=AA=E5=B7=B1=E9=82=A3=E7=AC=94=E4=B8=80?= =?UTF-8?q?=E7=9B=B4=E5=8F=91=E4=B8=8D=E5=87=BA=E5=8E=BB=E3=80=82=E7=8E=B0?= =?UTF-8?q?=E5=9C=A8=E4=B8=80=E6=9D=A1=E8=AE=B0=E5=BD=95=E6=98=AF=E5=93=AA?= =?UTF-8?q?=E4=B8=AA=E8=8A=82=E7=82=B9=E5=86=99=E7=9A=84=E6=8C=89=E5=AE=83?= =?UTF-8?q?=E7=9A=84=20action.*=20=E4=BA=8B=E4=BB=B6=E4=B8=8A=E8=AE=B0?= =?UTF-8?q?=E7=9A=84=E8=8A=82=E7=82=B9=20id=20=E8=AE=A4=EF=BC=88=E8=AE=A4?= =?UTF-8?q?=E4=B8=8D=E5=87=BA=E7=9A=84=E6=8C=89=E8=83=BD=E5=8A=9B=E5=92=8C?= =?UTF-8?q?=E5=86=85=E5=AE=B9=E6=8C=87=E7=BA=B9=EF=BC=89=EF=BC=9A=E6=8C=89?= =?UTF-8?q?=E8=87=AA=E5=B7=B1=E7=9A=84=E9=94=AE=E6=89=BE=E5=88=B0=E5=88=AB?= =?UTF-8?q?=E4=BA=BA=E9=82=A3=E4=B8=80=E7=AC=94=E7=9A=84=E6=8D=A2=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=E5=8F=AA=E5=B1=9E=E4=BA=8E=E8=87=AA=E5=B7=B1=E7=9A=84?= =?UTF-8?q?=E9=94=AE=EF=BC=9B=E8=AE=A4=E6=97=A7=E9=94=AE=E6=97=B6=E8=BF=98?= =?UTF-8?q?=E8=A6=81=E5=86=85=E5=AE=B9=E6=8C=87=E7=BA=B9=E6=98=AF=E8=BF=99?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=E7=9A=84=E2=80=94=E2=80=94=E8=B7=A8=E5=8D=87?= =?UTF-8?q?=E7=BA=A7=E7=9A=84=E5=BE=AA=E7=8E=AF=E7=AC=AC=E4=BA=8C=E8=BD=AE?= =?UTF-8?q?=E6=8C=89=E8=80=81=E7=AE=97=E6=B3=95=E5=90=8C=E6=A0=B7=E6=98=AF?= =?UTF-8?q?=E7=AC=AC=201=20=E6=AC=A1=E6=89=A7=E8=A1=8C=EF=BC=8C=E6=89=BE?= =?UTF-8?q?=E5=88=B0=E7=9A=84=E6=98=AF=E7=AC=AC=E4=B8=80=E8=BD=AE=E9=82=A3?= =?UTF-8?q?=E4=B8=80=E7=AC=94=EF=BC=8C=E9=87=91=E9=A2=9D=E4=B8=8D=E5=90=8C?= =?UTF-8?q?=E5=B0=B1=E7=85=A7=E5=B8=B8=E5=8F=91=EF=BC=8C=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E8=AE=B0=E3=80=8C=E5=B7=B2=E7=BB=8F=E7=94=9F=E6=95=88=E3=80=8D?= =?UTF-8?q?=E3=80=82=E5=8E=BB=E6=8E=89=E4=B8=8D=E5=86=8D=E7=94=A8=E7=9A=84?= =?UTF-8?q?=E6=9C=80=E5=A4=96=E5=B1=82=E8=8A=82=E7=82=B9=20id=20=E4=BC=A0?= =?UTF-8?q?=E9=80=92=EF=BC=9B=E6=94=B9=E5=AF=B9=E4=B8=A4=E5=A4=84=E8=BF=87?= =?UTF-8?q?=E6=97=B6=E7=9A=84=E8=AF=B4=E6=98=8E=EF=BC=88=E5=AE=A1=E6=89=B9?= =?UTF-8?q?=E6=81=A2=E5=A4=8D=E6=97=B6=E5=AD=90=E5=9B=BE=E4=BB=8E=E8=87=AA?= =?UTF-8?q?=E5=B7=B1=E7=9A=84=E6=96=AD=E7=82=B9=E6=8E=A5=E7=9D=80=E8=B7=91?= =?UTF-8?q?=EF=BC=9B=E7=BB=88=E6=80=81=E4=BA=8B=E4=BB=B6=E7=9A=84=E9=80=94?= =?UTF-8?q?=E7=BB=8F=E9=A9=B3=E5=9B=9E=E5=90=AB=E5=AD=90=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E9=87=8C=E7=9A=84=EF=BC=89=E3=80=82=E8=A1=A5=E8=B7=A8?= =?UTF-8?q?=E7=89=88=E6=9C=AC=E6=B5=8B=E8=AF=95=EF=BC=9A=E5=90=8C=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=E8=BF=9B=E7=A8=8B=E9=87=8C=E5=85=88=E8=A3=85=E6=88=90?= =?UTF-8?q?=E5=8D=87=E7=BA=A7=E5=89=8D=E7=9A=84=E4=BB=A3=E7=A0=81=E8=B7=91?= =?UTF-8?q?=E5=88=B0=E6=8C=82=E8=B5=B7=E3=80=81=E5=86=8D=E6=8D=A2=E5=9B=9E?= =?UTF-8?q?=E7=8E=B0=E5=9C=A8=E7=9A=84=E4=BB=A3=E7=A0=81=E6=8E=A5=E7=9D=80?= =?UTF-8?q?=E8=B7=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/context.py | 8 -- backend/app/engine/nodes/multi.py | 13 ++- backend/app/engine/nodes/tools.py | 14 ++- backend/app/engine/runner.py | 3 +- backend/app/process/actions.py | 59 ++++++++++- backend/tests/test_subgraph_repeat.py | 146 +++++++++++++++++++++++--- 6 files changed, 210 insertions(+), 33 deletions(-) diff --git a/backend/app/engine/context.py b/backend/app/engine/context.py index 24bbac71..6aff018f 100644 --- a/backend/app/engine/context.py +++ b/backend/app/engine/context.py @@ -43,12 +43,6 @@ class RunContext: """这张图里的节点 id → 这次运行里的 id（事件、工件上记的那个）：最外层原样，子工作流里加前缀。""" return f"{self.prefix}{node_id}" - @property - def root_ids(self) -> frozenset[str]: - """这次运行最外层图上的节点 id：最外层就是这张图的，子工作流里是 multi.run_subgraph 交下来的。""" - given = self.extra.get(ROOT_IDS_KEY) - return frozenset(given) if given is not None else frozenset(n.id for n in self.spec.nodes) - def global_label(self, title: str) -> str: """这张图里的节点标题 → 这次运行里给人看的名字：最外层原样，子工作流里接上子工作流的名字（「子工作流 › x」，嵌套的 一层层接上，同 multi.run_subgraph 记在异常上的 label）。父运行的图上没有子工作流里的节点，各处照这个名字说。""" @@ -59,8 +53,6 @@ class RunContext: PREFIX_KEY = "node_prefix" #: 子工作流里的节点名前缀（「子工作流 › 」，同上）：RunContext.global_label LABEL_PREFIX_KEY = "node_label_prefix" -#: 最外层图上的节点 id（同上，只在子工作流里有）：RunContext.root_ids -ROOT_IDS_KEY = "root_node_ids" #: 父运行的事件流（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 c2d5a152..f3ff3002 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -15,7 +15,7 @@ from app.core.events import EventType from app.db.base import SessionLocal from app.db.models import Workflow from app.engine.context import ( - LABEL_PREFIX_KEY, PREFIX_KEY, ROOT_IDS_KEY, STREAM_KEY, NodeContext, NodeError, stream_writer, + 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 @@ -694,9 +694,7 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 子图里的节点经父运行的写入口发事件、节点 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)}/", LABEL_PREFIX_KEY: f"{ctx.run.global_label(ctx.node.title)} › ", - STREAM_KEY: stream_writer(ctx.run), - # 最外层图上的节点 id：子图里的写入认升级前按子图里的 id 记的幂等键时，排除父图同名节点记的那一条（tools._run_write） - ROOT_IDS_KEY: ctx.run.extra.get(ROOT_IDS_KEY) or frozenset(n.id for n in ctx.run.spec.nodes)}, + STREAM_KEY: stream_writer(ctx.run)}, ) # 这个子工作流节点之前几次执行里，子图各节点跑过的记录（交回父图 trail 的，见下）：子图从它们接着数第几次执行 # （engine/evidence.next_exec）。以前每次都从空的 trail 起步，循环里第二轮的子图节点又记成第 1 次：台账分不出是哪一次， @@ -772,7 +770,7 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 从这里接着数（上面的 seed）；走了驳回出口的那几步（outcome=rejected）运行收尾据此在终态事件上照实写「途经驳回」 # （runner._rejected_via），实例页、通知、定时任务的 Webhook 说得出「子工作流 › 把关」被驳回、转到哪（名字、转到哪几步 # 在子图里已经按这次运行的说法记好：global_label、global_id）。父图自己的节点按自己的 id 数，不受影响 - updates["trail"] = [{**t, "node_id": f"{prefix}{t.get('node_id')}"} for t in ran] + updates["trail"] = _forwarded_trail(ran, prefix) var_name = ctx.cfg("assign_to", "") if var_name: updates["vars"] = {var_name: output} @@ -783,6 +781,11 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: _LEDGER_NODE_KEYS = ("node_id", "card") +def _forwarded_trail(ran: list[dict[str, Any]], prefix: str) -> list[dict[str, Any]]: + """子图这一次跑过的步骤交回父图 trail 的样子：节点 id 接上「子工作流节点 id/」前缀。""" + return [{**t, "node_id": f"{prefix}{t.get('node_id')}"} for t in ran] + + def _forwarded(entry: dict[str, Any], node_id: str) -> dict[str, Any]: """子图台账里的一条交回父图：节点 id 加上「子工作流节点 id/」前缀，别的照抄。""" out = dict(entry) diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index f2f61386..3f15785b 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -275,10 +275,10 @@ async def _run_write(state: GraphState, ctx: NodeContext, name: str, args: dict[ exec_no = next_exec(state, ctx.node.id) # 幂等键按这次运行里的节点 id 算（子工作流里的写「sub/pay」）：以前按子图里的 id（pay）算，父图里同名的节点、套同一个 - # 工作流的两个子工作流节点撞同一个键，后发的那一笔交回先发的回执、不发送。升级前停在子图里的运行按子图里的 id 记过， - # 恢复时整个子图重跑：照旧认那个键（legacy_node_id），已经生效的不重发；父图里有同名节点时不认（那一条可能是它的） - node_id = ctx.run.global_id(ctx.node.id) - legacy = ctx.node.id if node_id != ctx.node.id and ctx.node.id not in ctx.run.root_ids else None + # 工作流的两个子工作流节点撞同一个键，后发的那一笔交回先发的回执、不发送。升级前子工作流里的写入按子图里的 id 记过键： + # 写入节点本身重新执行时（结果未知核实之后、失败后接着跑；审批恢复时子图从自己的断点接着跑，不走到这里）照旧认那个键 + # （legacy_node_id），是不是这个节点的那一笔、是不是这一次的内容，由 actions.prepare 按那条记录的事件和内容指纹认 + node_id, legacy = _action_ids(ctx) try: prep = await actions.prepare(run_id=ctx.run.run_id, node_id=node_id, exec_no=exec_no, capability=capability, args=args, legacy_node_id=legacy) @@ -329,6 +329,12 @@ async def _run_write(state: GraphState, ctx: NodeContext, name: str, args: dict[ return _write_updates(ctx, result) +def _action_ids(ctx: NodeContext) -> tuple[str, str | None]: + """写入的幂等键按哪个节点 id 算：(这次运行里的 id, 升级前按子图里的 id 记过的那种；最外层的为 None)。""" + node_id = ctx.run.global_id(ctx.node.id) + return node_id, ctx.node.id if node_id != ctx.node.id else None + + async def _confirmed_by(run_id: str, node_id: str) -> str | None: """确认这次发送的人：节点重新执行（结果未知核实后接着跑、重启后恢复）时 ctx.actor() 已经不是确认的那次恢复， 按待办上记的处理人认（会签时是凑齐的那一票）。""" diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 1b3d020c..411d0ecf 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -1353,7 +1353,8 @@ def _rejected_via(state: dict[str, Any]) -> dict[str, Any]: """这次运行途经的驳回出口（engine/compiler._rejected_exit 记在 trail 上的 outcome=rejected）：{rejected_via: [{node_id, label, kind, decision, actor, note, to}]}，按先后。终态事件带上它，运行照它的路径结束（成功结束也一样）照实说「途经 驳回」；没有途经的不带这个键，终态事件和以前一字不差。读的是收尾时的状态：节点出错、被驳回结束的那几种收尾拿不到 - 状态（final_state 为空），不带；子工作流里的驳回出口不在这里（它的 trail 是子工作流自己的状态）。""" + 状态（final_state 为空），不带。子工作流里走了驳回出口的那几步也在这里：子图跑过的步骤交回父图的 trail（节点 id 带 + 「子工作流节点 id/」前缀、名字写「子工作流 › x」，engine/nodes/multi.run_subgraph）。""" via = [{k: t.get(k) for k in ("node_id", "label", "kind", "decision", "actor", "note", "to")} for t in state.get("trail") or [] if isinstance(t, dict) and t.get("outcome") == "rejected"] return {"rejected_via": via} if via else {} diff --git a/backend/app/process/actions.py b/backend/app/process/actions.py index 14b4658b..fa2de0fc 100644 --- a/backend/app/process/actions.py +++ b/backend/app/process/actions.py @@ -244,6 +244,51 @@ class Prepared: return f"{self.key}:{self.attempt}" +#: 按自己的键找到别人的那一笔时换用的键的前缀（prepare）：节点 id 里不会有这个字符 +OWN_KEY_MARK = "\x1e" +#: 带 action_id 的那几种事件：认一条记录是哪个节点写的（_owner） +_ACTION_EVENTS = (EventType.ACTION_PREPARED, EventType.ACTION_SENT, EventType.ACTION_CONFIRMED, EventType.ACTION_FAILED, + EventType.ACTION_UNKNOWN, EventType.ACTION_SIMULATED) + + +async def _owner(session: AsyncSession, rec: ActionRecord) -> str | None: + """这条记录是哪个节点写的（这次运行里的 id，子工作流里的带前缀）：按它的 action.* 事件上记的 node_id。记录上的 node_id + 升级前子工作流里记的是子图里的 id，和父图同名的节点分不开；事件一向按这次运行里的 id 记。认不出（事件没写上）为 None。""" + from app.db.models import RunEvent + + rows = (await session.execute(select(RunEvent.node_id, RunEvent.data).where( + RunEvent.run_id == rec.run_id, RunEvent.type.in_([str(t) for t in _ACTION_EVENTS])))).all() + return next((str(node) for node, data in rows if node and isinstance(data, dict) and data.get("action_id") == rec.id), + None) + + +def _same_payload(rec: ActionRecord, capability: Any, args: dict[str, Any]) -> bool | None: + """这条记录是不是这一次要发的内容：能力相同、内容指纹等于这一次参数收好类型之后的指纹。收不成类型（参数定义改了）的 + 比不了，为 None。""" + name = capability.name + if rec.capability != name: + return False + try: + typed = typed_args(capability, args) + except ActionFailed: + return None + return rec.payload_hash == payload_hash(name, typed) + + +async def _belongs(session: AsyncSession, rec: ActionRecord, node_id: str, capability: Any, args: dict[str, Any], *, + strict: bool = False) -> bool: + """这条记录是不是这个节点（node_id，这次运行里的 id）的。事件认得出是谁写的就按事件；认不出的按内容（能力、内容指纹）， + 比不了的当是（以前就是这样认的）。strict（认升级前的旧键时）：事件说是这个节点的，内容也要对得上——循环跨了升级时， + 同一个节点第二轮按老算法同样是第 1 次执行，找到的是第一轮那一笔；比不了内容（参数定义改了）的照事件认。""" + owner = await _owner(session, rec) + same = _same_payload(rec, capability, args) + if owner is not None and owner != node_id: + return False + if owner is not None and not strict: + return True + return same is not False + + async def latest(session: AsyncSession, key: str) -> ActionRecord | None: return await session.scalar(select(ActionRecord).where(ActionRecord.action_key == key) .order_by(ActionRecord.attempt.desc()).limit(1)) @@ -268,12 +313,20 @@ async def prepare(*, run_id: str, node_id: str, exec_no: int, capability: Any, a run = await session.get(Run, run_id) endpoint = await _endpoint(session, capability) prev = await latest(session, key) + if prev is not None and not await _belongs(session, prev, node_id, capability, args): + # 按自己的键找到的是别的节点的那一笔：升级前子工作流里同名的节点按子图里的 id 记的键，和父图这个节点的一样。 + # 那一笔不是这个节点的，换一个只属于这个节点的键（同样的输入永远算出同一个） + key = action_key(run_id, f"{OWN_KEY_MARK}{node_id}", exec_no) + prev = await latest(session, key) if prev is None and legacy_node_id: - # 升级前子工作流里的写入按子图里的节点 id 记幂等键（node_id 也记成它）：恢复时整个子图重跑，照那个键认出已经 - # 记下的那一笔，接着用它的键（不重发、第几次发送接着数）。新记的子图写入 node_id 带前缀，认不到这里来 + # 升级前子工作流里的写入按子图里的节点 id 记幂等键（node_id 也记成它）：写入节点本身重新执行时（结果未知核实之后、 + # 失败后接着跑）照那个键认出已经记下的那一笔，接着用它的键（不重发、第几次发送接着数）。只认这个节点的、这一次的： + # 那一笔的事件记着是这个节点写的（父图里同名的节点按同一个键写过），内容指纹也和这一次的一样（循环跨了升级时， + # 第二轮按老算法同样是第 1 次执行，找到的是第一轮那一笔） old_key = action_key(run_id, legacy_node_id, exec_no) old = await latest(session, old_key) - if old is not None and old.node_id == legacy_node_id: + if old is not None and old.node_id == legacy_node_id \ + and await _belongs(session, old, node_id, capability, args, strict=True): key, prev, node_id = old_key, old, legacy_node_id facts = {"app_id": run.app_id, "workflow_id": run.workflow_id, "workflow_version": run.version, "graph_hash": run.version_hash, "started_by": run.started_by} if run is not None else {} diff --git a/backend/tests/test_subgraph_repeat.py b/backend/tests/test_subgraph_repeat.py index e7ebac88..31e30365 100644 --- a/backend/tests/test_subgraph_repeat.py +++ b/backend/tests/test_subgraph_repeat.py @@ -137,22 +137,144 @@ async def test_a_namesake_in_the_parent_does_not_share_the_key(monkeypatch, erp) async def test_a_write_recorded_before_the_upgrade_is_still_found(monkeypatch, erp): - """升级前子图里的写入按子图里的 id（pay）记幂等键。停在子图里的运行恢复时整个子图重跑：照旧按那个键找到已经生效的 - 记录、交回回执，不重发。父图里有同名节点时不这样认（那一条可能是父图那个节点的）。""" + """升级前子图里的写入按子图里的 id（pay）记幂等键、记录上的 node_id 也是它；事件一向按这次运行里的 id（sub/pay）记。 + 写入节点本身重新执行时（结果未知核实之后接着跑）：事件说是这个节点的、内容指纹也是这一次的，照旧按那个键交回回执、不重发； + 事件说是父图同名节点的、或者内容不一样（跨升级的循环，第二轮按老算法也是第 1 次执行）的不认。""" + from app.core.events import EventType + from app.db.models import RunEvent + caps = install(monkeypatch, erp, await endpoint_row(test=True), lookup=False) - async with SessionLocal() as session: - run = Run(graph={"nodes": [node("sub", "subgraph")], "edges": []}, status="running", run_class="exploratory") - session.add(run) - await session.flush() - legacy = actions.action_key(run.id, "pay", 1) - session.add(ActionRecord(action_key=legacy, attempt=1, idem_key=f"{legacy}:1", run_id=run.id, node_id="pay", - exec_no=1, capability=PAY, environment="test", payload_hash="0" * 64, state="confirmed", - receipt={"summary": {"id": "PAY-OLD"}})) - await session.commit() - run_id = run.id args = {"amount": "5", "card": "6222020200112233"} + digest = actions.payload_hash(PAY, actions.typed_args(caps[PAY], args)) + + async def old_record(writer: str, payload: str) -> tuple[str, str]: + async with SessionLocal() as session: + run = Run(graph={"nodes": [node("sub", "subgraph")], "edges": []}, status="running", run_class="exploratory") + session.add(run) + await session.flush() + legacy = actions.action_key(run.id, "pay", 1) + rec = ActionRecord(action_key=legacy, attempt=1, idem_key=f"{legacy}:1", run_id=run.id, node_id="pay", + exec_no=1, capability=PAY, environment="test", payload_hash=payload, state="confirmed", + receipt={"summary": {"id": "PAY-OLD"}}) + session.add(rec) + await session.flush() + session.add(RunEvent(run_id=run.id, seq=1, type=str(EventType.ACTION_PREPARED), node_id=writer, ts=0.0, + data={"action_id": rec.id, "capability": PAY, "payload_hash": payload})) + await session.commit() + return run.id, legacy + + run_id, legacy = await old_record("sub/pay", digest) found = await actions.prepare(run_id=run_id, node_id="sub/pay", exec_no=1, capability=caps[PAY], args=args, legacy_node_id="pay") assert found.key == legacy and found.receipt is not None fresh = await actions.prepare(run_id=run_id, node_id="sub/pay", exec_no=1, capability=caps[PAY], args=args) assert fresh.key != legacy and fresh.receipt is None + # 父图里同名的 pay 按自己的键（和旧键一样）找到这一笔：事件说是 sub/pay 的，不是自己的，换一个只属于自己的键 + mine = await actions.prepare(run_id=run_id, node_id="pay", exec_no=1, capability=caps[PAY], args=args) + assert mine.key != legacy and mine.receipt is None + # 内容不一样（这一次是 7）：不当成这一次已经生效 + other = await actions.prepare(run_id=run_id, node_id="sub/pay", exec_no=1, capability=caps[PAY], + args={**args, "amount": "7"}, legacy_node_id="pay") + assert other.key != legacy and other.receipt is None + # 那一笔是父图 pay 写的（事件说 pay）：子图里的 pay 不认 + run_id, legacy = await old_record("pay", digest) + theirs = await actions.prepare(run_id=run_id, node_id="sub/pay", exec_no=1, capability=caps[PAY], args=args, + legacy_node_id="pay") + assert theirs.key != legacy and theirs.receipt is None + own = await actions.prepare(run_id=run_id, node_id="pay", exec_no=1, capability=caps[PAY], args=args) + assert own.key == legacy and own.receipt is not None + + +# -------------------------------------------------------------------------- +# 跨升级：升级前的运行按老算法记的键，升级后接着跑 +# -------------------------------------------------------------------------- + + +def _old_code(monkeypatch): + """装成升级前的代码：子工作流里的写入按子图里的 id 算幂等键、不认旧键，子图跑过的步骤不交回父图（exec 每次从 1 数）。 + 交回一个函数，调了就换回现在的代码（同一个进程里接力跑，等于升级后重启接着跑；能力登记的替身要重新装）。""" + from app.engine.nodes import multi, tools + + monkeypatch.setattr(tools, "_action_ids", lambda ctx: (ctx.node.id, None)) + monkeypatch.setattr(multi, "_forwarded_trail", lambda ran, prefix: []) + return monkeypatch.undo + + +def _amounts(erp) -> list[str]: + return [str((w.get("body") or {}).get("amount")) for w in erp.writes("test")] + + +async def _until_status(run_id: str, statuses: tuple[str, ...]) -> Run: + for _ in range(600): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in statuses and not run_manager.is_active(run_id): + return row + raise AssertionError(f"运行 {run_id} 停在 {row.status}") + + +async def test_an_unknown_write_inside_a_subworkflow_with_a_namesake_in_the_parent_survives_the_upgrade(monkeypatch, erp): + """升级前：子工作流里的 pay（1）发出去、结果未知、运行挂起，键按子图里的 id（pay）记——和父图里同名的 pay 一样。升级后 + 核实为已生效、接着跑：子图里的 pay 认出自己那一笔（事件记着是「sub/pay」写的），不再发；父图的 pay（99）按自己的键找到 + 的是子图那一笔、不是自己的，换用自己的键照常发。1 和 99 各只发一次。""" + from app.db.models import Task + from app.process import tasks + from tests.test_process_actions import ANN + + await _unattended(monkeypatch, erp) + inner = S.chain(node("s", "input", fields=[{"name": "n"}]), pay(), + node("o", "output", fields=[{"name": "r", "value": "x"}])) + sub_id = await S.publish("付款子流程", graph=inner, level="draft") + graph = S.chain(node("start", "input", fields=[{"name": "n"}]), + node("sub", "subgraph", "子流程", workflow_id=sub_id, workflow_version=1, input={"n": "1"}), + pay(), node("out", "output", fields=[{"name": "r", "value": "完成"}])) + upgrade = _old_code(monkeypatch) + erp.mode = "timeout_after_write" + run = await run_manager.start(graph=graph, input_payload={"n": 99}) + parked = await _until_status(run.id, ("interrupted", "failed", "succeeded")) + assert parked.status == "interrupted", parked.error + assert _amounts(erp) == ["1"] + upgrade() + await _unattended(monkeypatch, erp) # undo 连能力登记的替身一起撤了：重新装上 + erp.mode = "ok" + async with SessionLocal() as session: + task = (await session.execute(select(Task).where(Task.run_id == run.id, Task.kind == "reconcile", + Task.status == "pending"))).scalars().first() + assert task is not None + await tasks.respond(task.id, ANN, decision="confirmed", values={"external_id": "PAY-1"}) + final = await _until_status(run.id, ("succeeded", "failed")) + assert final.status == "succeeded", final.error + assert sorted(_amounts(erp)) == ["1", "99"], _amounts(erp) + + +async def test_a_loop_across_the_upgrade_sends_the_second_round(monkeypatch, erp): + """升级前跑完第一轮（子图里 pay 发了 1）、停在父图的审批上；升级后批准，第二轮按老算法同样是第 1 次执行，找到第一轮那一笔： + 内容指纹不一样（这一轮是 2），不当成已经生效，照常发；也不记「已经生效」的提示。""" + await _unattended(monkeypatch, erp) + inner = S.chain(node("s", "input", fields=[{"name": "n"}]), pay(), + node("o", "output", fields=[{"name": "r", "value": "x"}])) + sub_id = await S.publish("付款子流程", graph=inner, level="draft") + nodes = [node("start", "input"), + node("each", "loop", "逐项", mode="foreach", items="[1, 2]", item_var="it", max_iterations=5), + node("sub", "subgraph", "子流程", workflow_id=sub_id, workflow_version=1, input={"n": "{{ vars.it }}"}), + node("gate", "human", "把关", mode="approve", title="下一轮？"), + node("out", "output", fields=[{"name": "r", "value": "完成"}])] + edges = [edge("start", "each"), edge("each", "sub", "body"), edge("sub", "gate"), edge("gate", "each", "approved"), + edge("each", "out", "done")] + upgrade = _old_code(monkeypatch) + run = await run_manager.start(graph={"nodes": nodes, "edges": edges}, input_payload={}) + parked = await _until_status(run.id, ("interrupted", "failed", "succeeded")) + assert parked.status == "interrupted", parked.error + assert _amounts(erp) == ["1"] + upgrade() + await _unattended(monkeypatch, erp) + await run_manager.resume(run.id, {"approved": True}) + parked = await _until_status(run.id, ("interrupted", "failed", "succeeded")) + assert parked.status == "interrupted", parked.error + assert _amounts(erp) == ["1", "2"], "第二轮照发，不交回第一轮的回执" + assert not [e for e in await _events(run.id, "log") if e.data.get("code") == "action_already_applied"] + await run_manager.resume(run.id, {"approved": True}) + final = await _until_status(run.id, ("succeeded", "failed")) + assert final.status == "succeeded", final.error + assert _amounts(erp) == ["1", "2"]