diff --git a/backend/app/engine/context.py b/backend/app/engine/context.py index 6aff018f..24bbac71 100644 --- a/backend/app/engine/context.py +++ b/backend/app/engine/context.py @@ -43,6 +43,12 @@ 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）。父运行的图上没有子工作流里的节点，各处照这个名字说。""" @@ -53,6 +59,8 @@ 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 17222b1d..c2d5a152 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -14,7 +14,9 @@ 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 LABEL_PREFIX_KEY, PREFIX_KEY, STREAM_KEY, NodeContext, NodeError, stream_writer +from app.engine.context import ( + LABEL_PREFIX_KEY, PREFIX_KEY, ROOT_IDS_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 ( @@ -692,8 +694,16 @@ 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)}, + 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)}, ) + # 这个子工作流节点之前几次执行里，子图各节点跑过的记录（交回父图 trail 的，见下）：子图从它们接着数第几次执行 + # （engine/evidence.next_exec）。以前每次都从空的 trail 起步，循环里第二轮的子图节点又记成第 1 次：台账分不出是哪一次， + # 写外部系统的幂等键（按第几次执行算）和第一轮撞上，第二轮的写入交回第一轮的回执、不发送 + prefix = f"{ctx.node.id}/" + seed = [{**t, "node_id": str(t.get("node_id"))[len(prefix):]} for t in state.get("trail") or [] + if isinstance(t, dict) and str(t.get("node_id") or "").startswith(prefix)] ctx.emit(EventType.LOG, level="info", message=f"进入子工作流「{workflow.name}」（{len(sub_spec.nodes)} 个节点）") @@ -711,7 +721,7 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: "output": {}, "messages": [], "usage": {}, - "trail": [], + "trail": seed, "loops": {}, }, {"recursion_limit": limit}, @@ -732,6 +742,9 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: raise output = result_state.get("output") or {} + # 这一次执行新跑的那几步（trail 只追加：前面是交进去的 seed） + ran = [t for t in (result_state.get("trail") or [])[len(seed):] if isinstance(t, dict)] + result_state = {**result_state, "trail": ran} result = { "output": output, "workflow": workflow.name, @@ -754,11 +767,12 @@ 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 + if ran: + # 子图这一次跑过的每一步交回父图的 trail，节点 id 接上前缀（「子工作流节点 id/…」）：同一个子工作流节点下一次执行时 + # 从这里接着数（上面的 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] var_name = ctx.cfg("assign_to", "") if var_name: updates["vars"] = {var_name: output} @@ -769,12 +783,6 @@ 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/nodes/tools.py b/backend/app/engine/nodes/tools.py index 7c299fcb..f2f61386 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -274,9 +274,14 @@ async def _run_write(state: GraphState, ctx: NodeContext, name: str, args: dict[ from app.process import actions 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 try: - prep = await actions.prepare(run_id=ctx.run.run_id, node_id=ctx.node.id, exec_no=exec_no, - capability=capability, args=args) + 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) except actions.ActionFailed as e: raise NodeError(ctx.node.id, str(e)) from e if prep.receipt is not None: @@ -317,8 +322,8 @@ async def _run_write(state: GraphState, ctx: NodeContext, name: str, args: dict[ # 外部系统会把审批人当成申请人；无人值守的写入没有人确认，不写 act on_behalf = {"sub": prep.run.get("started_by"), "act": confirmer} try: - result = await actions.send(prep, capability, run_id=ctx.run.run_id, node_id=ctx.node.id, - exec_no=exec_no, emit=ctx.emit, on_behalf=on_behalf) + result = await actions.send(prep, capability, run_id=ctx.run.run_id, node_id=node_id, + exec_no=exec_no, emit=ctx.emit, on_behalf=on_behalf, confirm_node=ctx.node.id) except actions.ActionFailed as e: raise NodeError(ctx.node.id, str(e)) from e return _write_updates(ctx, result) diff --git a/backend/app/process/actions.py b/backend/app/process/actions.py index 87ab7fca..14b4658b 100644 --- a/backend/app/process/actions.py +++ b/backend/app/process/actions.py @@ -236,6 +236,8 @@ class Prepared: run: dict[str, Any] = field(default_factory=dict) #: 按参数定义收好类型的参数（typed_args）：哈希、预览按它算，send 发的也是它 args: dict[str, Any] = field(default_factory=dict) + #: 记录上写的节点 id：认出升级前那种键（legacy_node_id）时照它记，同一个键下的记录 node_id 一致，下一次还认得出 + node_id: str | None = None @property def idem_key(self) -> str: @@ -247,8 +249,11 @@ async def latest(session: AsyncSession, key: str) -> ActionRecord | None: .order_by(ActionRecord.attempt.desc()).limit(1)) -async def prepare(*, run_id: str, node_id: str, exec_no: int, capability: Any, args: dict[str, Any]) -> Prepared: - """节点每次执行都先走这里：按 action_key 查记录，决定发不发、第几次发、走哪个环境。 +async def prepare(*, run_id: str, node_id: str, exec_no: int, capability: Any, args: dict[str, Any], + legacy_node_id: str | None = None) -> Prepared: + """节点每次执行都先走这里：按 action_key 查记录，决定发不发、第几次发、走哪个环境。node_id 是这次运行里的节点 id + （子工作流里的带前缀）；legacy_node_id 是升级前子工作流里按子图里的 id 记的那种（tools._run_write 交进来），按它也没查到 + 才算头一次。 sent / unknown 抛 ActionPending（不发）；failed 且没有人核实为「重新发送」的抛 ActionFailed。先查记录再收类型：已生效 （演练过）的交回存下的回执、已发出 / 结果未知的挂起，都和这一次渲染出来的参数无关——发出之后能力的参数定义改了、手上的 @@ -263,13 +268,21 @@ 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 None and legacy_node_id: + # 升级前子工作流里的写入按子图里的节点 id 记幂等键（node_id 也记成它）：恢复时整个子图重跑，照那个键认出已经 + # 记下的那一笔，接着用它的键（不重发、第几次发送接着数）。新记的子图写入 node_id 带前缀，认不到这里来 + 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: + 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 {} endpoint_name = endpoint.name if endpoint is not None else None if prev is not None and prev.state in ("confirmed", "simulated"): # 交回的是生效的那一次：指纹、预览、环境、第几次都照记录，不按这一次的参数重算 return Prepared(key=key, attempt=prev.attempt, payload_hash=prev.payload_hash, preview=prev.preview, - environment=prev.environment, endpoint=endpoint_name, receipt=_receipt_of(prev), run=facts) + environment=prev.environment, endpoint=endpoint_name, receipt=_receipt_of(prev), run=facts, + node_id=node_id) if prev is not None and prev.state in ("sent", "unknown"): raise ActionPending(prev.id) formal = run is not None and run.run_class == "formal" @@ -290,15 +303,18 @@ async def prepare(*, run_id: str, node_id: str, exec_no: int, capability: Any, a args = typed_args(capability, args) return Prepared(key=key, attempt=attempt, payload_hash=payload_hash(name, args), preview=masked(args, masked_paths(capability)), environment=environment, endpoint=endpoint_name, - run=facts, args=args) + run=facts, args=args, node_id=node_id) async def send(prep: Prepared, capability: Any, *, run_id: str, node_id: str, exec_no: int, - emit: Callable[..., None], on_behalf: str | dict[str, Any] | None = None) -> dict[str, Any]: + emit: Callable[..., None], on_behalf: str | dict[str, Any] | None = None, + confirm_node: str | None = None) -> dict[str, Any]: """先写记录再发送。返回节点的产出；结果未知抛 ActionPending，确定失败抛 ActionFailed。发的是 prepare 收好类型的 那一份参数（prep.args），和确认卡上的哈希是同一份。 - on_behalf 原样交给平台执行器写身份断言：{"sub": 运行的发起人, "act": 确认人}（capabilities.endpoints.on_behalf_parts）。""" + on_behalf 原样交给平台执行器写身份断言：{"sub": 运行的发起人, "act": 确认人}（capabilities.endpoints.on_behalf_parts）。 + 记录上的节点 id 照 prepare 认定的（prep.node_id）；confirm_node 是动作确认待办上记的节点 id（确认卡上的，子工作流里是 + 子图里的 id），不给就按 node_id 找。""" from app.capabilities.registry import endpoint_request, masked_paths name = capability.name @@ -306,11 +322,12 @@ async def send(prep: Prepared, capability: Any, *, run_id: str, node_id: str, ex rec = await session.scalar(select(ActionRecord).where(ActionRecord.idem_key == prep.idem_key)) if rec is None: rec = ActionRecord( - action_key=prep.key, attempt=prep.attempt, idem_key=prep.idem_key, run_id=run_id, node_id=node_id, + action_key=prep.key, attempt=prep.attempt, idem_key=prep.idem_key, run_id=run_id, + node_id=prep.node_id or node_id, exec_no=exec_no, capability=name, capability_version=getattr(capability, "version", None), endpoint_id=getattr(capability, "endpoint_id", None), environment=prep.environment, payload_hash=prep.payload_hash, preview=prep.preview, state="pending", - confirm_task_id=await _confirm_task(session, run_id, node_id), + confirm_task_id=await _confirm_task(session, run_id, confirm_node or node_id), app_id=prep.run.get("app_id"), workflow_id=prep.run.get("workflow_id"), workflow_version=prep.run.get("workflow_version"), graph_hash=prep.run.get("graph_hash")) session.add(rec) diff --git a/backend/tests/test_subgraph_repeat.py b/backend/tests/test_subgraph_repeat.py new file mode 100644 index 00000000..e7ebac88 --- /dev/null +++ b/backend/tests/test_subgraph_repeat.py @@ -0,0 +1,158 @@ +"""同一个子工作流节点在一次运行里执行多次（循环体里、退回修改绕回来）：子图里的节点每次执行都要分得开。 + +以前子图每次都从空状态起步，里面的节点每次都记成「第 1 次执行」： +- 台账里同一个节点（「sub/fetch」）几条查询都记 exec 1，分不出是第几次； +- 写外部系统的幂等键按 (运行, 节点 id, 第几次执行) 算，子图里的节点 id 还是子图里的（pay，不带前缀）：循环第二轮的 + 写入算出同一个键，找到第一轮生效的记录，交回第一轮的回执、这一轮不发送——外部系统少收一笔，界面上只有一句「已经生效」； + 父图里同名的节点、两个子工作流节点套同一个工作流时也撞同一个键。 +现在：子图里的节点接着父运行里同一个子工作流节点之前几次数（exec 1、2…），幂等键按这次运行里的节点 id（「sub/pay」）算； +升级前停下的运行（子图里的写入按子图里的 id 记的），恢复时照旧认得出已经生效的那一笔，不重发。 +图、能力、外部系统都是虚构的。 +""" +from __future__ import annotations + +import asyncio + +import pytest +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import ActionRecord, Run, RunEvent +from app.engine.runner import run_manager +from app.process import actions +from tests import ledger_samples as S +from tests.fakes.fake_erp import PAY, FakeErp, endpoint_row, install + + +@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}} + + +def edge(source, target, handle=None): + return {"source": source, "target": target, **({"sourceHandle": handle} if handle else {})} + + +async def _finish(graph: dict, **kwargs) -> Run: + run = await run_manager.start(graph=graph, input_payload={}, **kwargs) + 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 ("succeeded", "failed", "interrupted") and not run_manager.is_active(run.id): + return row + raise AssertionError(f"运行 {run.id} 没跑完：{row.status}") + + +async def _events(run_id: str, kind: str, node_id: str | None = None) -> list[RunEvent]: + async with SessionLocal() as session: + q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == kind) + if node_id: + q = q.where(RunEvent.node_id == node_id) + return list((await session.execute(q.order_by(RunEvent.seq))).scalars()) + + +def looped(sub_id: str, *, before: list[dict] | None = None) -> dict: + """开始 → (before…) → 循环两轮，每轮跑一次子工作流 → 成果。""" + nodes = [node("start", "input"), *(before or []), + 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("out", "output", fields=[{"name": "r", "value": "完成"}])] + chain = ["start", *[b["id"] for b in before or []], "each"] + edges = [edge(a, b) for a, b in zip(chain, chain[1:])] + edges += [edge("each", "sub", "body"), edge("sub", "each"), edge("each", "out", "done")] + return {"nodes": nodes, "edges": edges} + + +async def test_inner_steps_count_on_across_invocations(tmp_path): + await S.fresh_source(tmp_path / "shop.db") + inner = S.chain(node("s", "input"), node("fetch", "tool", tool=S.tool(), args={"sql": "SELECT COUNT(*) AS n FROM orders"}), + node("o", "output", fields=[{"name": "n", "value": "{{ nodes.fetch.text }}"}])) + sub_id = await S.publish("取数子流程", graph=inner, level="draft") + row = await _finish(looped(sub_id)) + assert row.status == "succeeded", row.error + finished = await _events(row.id, "node.finished", "sub/fetch") + execs = [e["exec"] for f in finished for e in f.data.get("evidence") or [] if e.get("kind") == "query"] + assert execs == [1, 2], "第二轮是子图里那一步的第 2 次执行，不再又记成第 1 次" + + +@pytest.fixture +def erp() -> FakeErp: + return FakeErp() + + +async def _unattended(monkeypatch, erp) -> None: + """付款能力设成可以无人值守写入、节点不要确认：两轮都直接发，不停下等人（和确认无关，只看幂等键分不分得开）。""" + caps = install(monkeypatch, erp, await endpoint_row(test=True), lookup=False) + caps[PAY].unattended_write = True + + +def pay(nid: str = "pay") -> dict: + return node(nid, "tool", "付款", tool=PAY, approval="never", args={"amount": "{{ input.n }}", "card": "6222020200112233"}) + + +async def test_a_write_inside_a_loop_is_sent_on_every_round(monkeypatch, erp): + await _unattended(monkeypatch, erp) + inner = S.chain(node("s", "input", fields=[{"name": "n"}]), pay(), + node("o", "output", fields=[{"name": "单号", "value": "{{ nodes.pay.receipt.id }}"}])) + sub_id = await S.publish("付款子流程", graph=inner, level="draft") + row = await _finish(looped(sub_id)) + assert row.status == "succeeded", row.error + assert len(erp.writes("test")) == 2, "第二轮的写入照发，不交回第一轮的回执" + async with SessionLocal() as session: + records = list((await session.execute(select(ActionRecord).where(ActionRecord.run_id == row.id) + .order_by(ActionRecord.created_at))).scalars()) + assert [(r.node_id, r.exec_no) for r in records] == [("sub/pay", 1), ("sub/pay", 2)] + assert len({r.action_key for r in records}) == 2 + assert not [e for e in await _events(row.id, "log") if e.data.get("code") == "action_already_applied"] + + +async def test_a_namesake_in_the_parent_does_not_share_the_key(monkeypatch, erp): + """父图里也有一个 pay：子图里的 pay 不再和它撞同一个幂等键。""" + 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"}]), pay(), + node("sub", "subgraph", "子流程", workflow_id=sub_id, workflow_version=1, input={"n": "7"}), + node("out", "output", fields=[{"name": "r", "value": "完成"}])) + run = await run_manager.start(graph=graph, input_payload={"n": 5}) + 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 ("succeeded", "failed") and not run_manager.is_active(run.id): + break + assert row.status == "succeeded", row.error + assert len(erp.writes("test")) == 2 + async with SessionLocal() as session: + ids = sorted(r.node_id for r in (await session.execute( + select(ActionRecord).where(ActionRecord.run_id == row.id))).scalars()) + assert ids == ["pay", "sub/pay"] + + +async def test_a_write_recorded_before_the_upgrade_is_still_found(monkeypatch, erp): + """升级前子图里的写入按子图里的 id（pay）记幂等键。停在子图里的运行恢复时整个子图重跑：照旧按那个键找到已经生效的 + 记录、交回回执，不重发。父图里有同名节点时不这样认（那一条可能是父图那个节点的）。""" + 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"} + 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