同一个子工作流节点执行多次时子图里的节点分得开:子图这一次跑过的每一步交回父图的 trail(节点 id 接上前缀),下一次执行从那里接着数第几次执行(engine/evidence.next_exec),台账、证据面板不再几条都记第 1 次;写外部系统的幂等键按这次运行里的节点 id 算(「sub/pay」)。以前循环里第二轮的子图写入算出和第一轮同一个键,交回第一轮的回执、外部系统少收一笔,父图里同名的节点、套同一个工作流的两个子工作流节点也撞键。升级前停在子图里的运行恢复时照旧按子图里的 id 认出已经生效的那一笔、不重发(父图里有同名节点时不认)

This commit is contained in:
ThearchyHelios
2026-10-19 09:16:14 +08:00
parent d1ad670fb3
commit 8ba0b27097
5 changed files with 222 additions and 26 deletions
+8
View File
@@ -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 把父图那一刻的写入口交下来,子工作流的节点经它发
+22 -14
View File
@@ -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)
+9 -4
View File
@@ -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)
+25 -8
View File
@@ -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)
+158
View File
@@ -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