mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
子工作流里的写入跨升级认旧键时按事件和内容认是谁的、是哪一笔(复核 L1、L2):升级前子图里的写入按子图里的 id 记键,和父图同名节点的键一样。以前父图有同名节点时一律不认旧键:子图里结果未知的那一笔核实为已生效后接着跑,用新键又发一笔;父图那个节点按自己的键找到子图那一笔、交回它的回执,自己那笔一直发不出去。现在一条记录是哪个节点写的按它的 action.* 事件上记的节点 id 认(认不出的按能力和内容指纹):按自己的键找到别人那一笔的换一个只属于自己的键;认旧键时还要内容指纹是这一次的——跨升级的循环第二轮按老算法同样是第 1 次执行,找到的是第一轮那一笔,金额不同就照常发,不再记「已经生效」。去掉不再用的最外层节点 id 传递;改对两处过时的说明(审批恢复时子图从自己的断点接着跑;终态事件的途经驳回含子工作流里的)。补跨版本测试:同一个进程里先装成升级前的代码跑到挂起、再换回现在的代码接着跑
This commit is contained in:
@@ -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 把父图那一刻的写入口交下来,子工作流的节点经它发
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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() 已经不是确认的那次恢复,
|
||||
按待办上记的处理人认(会签时是凑齐的那一票)。"""
|
||||
|
||||
@@ -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 {}
|
||||
|
||||
@@ -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 {}
|
||||
|
||||
@@ -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"]
|
||||
|
||||
Reference in New Issue
Block a user