合入遗留收口分支 nr8/l 的复核修复（变化记录先占住那一行再读改写、召回里行序不定的指标、事件触发列表按任务各取上限、选对象按名字查用 ilike、子工作流写入认旧键、几处测试的竞态）

# Conflicts:
#	backend/app/ledger/recall.py

backend/app/api/schedules.py

@@ -618,7 +618,10 @@ async def trigger_options(event: str | None = None, q: str | None = None, select
     async def page(model: Any, *conds: Any) -> tuple[list[Any], bool]:
         query = select(model).where(*conds)
         if words:
-            query = query.where(model.name.contains(words, autoescape=True))
+            # 名字里含这几个字，不分大小写：PostgreSQL 上 LIKE 分 ASCII 大小写、SQLite 不分，一律 ilike（转义同 runs._like）
+            from app.api.runs import _like
+
+            query = query.where(model.name.ilike(f"%{_like(words)}%", escape="\\"))
         rows = list((await session.execute(query.order_by(model.name, model.id).limit(limit + 1))).scalars())
         more = len(rows) > limit
         rows = rows[:limit]

backend/app/data/source_changes.py

@@ -29,7 +29,7 @@ from dataclasses import dataclass
 from datetime import datetime
 from typing import Any
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import func, or_, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -151,11 +151,16 @@ async def record(session: AsyncSession, source: Any, change: NewChange, detected
         raise ValueError(f"不认识的严重程度：{change.severity}")
     fp = fingerprint(change.kind, change.table, change.column, change.key)
     now = utcnow()
+    # 已经有了的：先按条件 UPDATE 记下又见到一次（同时占住那一行：SQLite 上拿到写锁、PostgreSQL 上锁住这一行），再读出来改
+    # 影响面——口径试算同时往 affects 里记没试算的那几版（ledger/recall._probe_skipped，同样先占住再读改写），谁也不盖掉谁
+    seen = await session.execute(update(SourceChange).where(
+        SourceChange.source_id == source.id, SourceChange.fingerprint == fp)
+        .values(last_seen_at=now, seen_count=func.coalesce(SourceChange.seen_count, 0) + 1)
+        .execution_options(synchronize_session=False))
     found = (await session.execute(select(SourceChange).where(
-        SourceChange.source_id == source.id, SourceChange.fingerprint == fp))).scalar_one_or_none()
+        SourceChange.source_id == source.id, SourceChange.fingerprint == fp)
+        .execution_options(populate_existing=True))).scalar_one_or_none() if seen.rowcount else None
     if found is not None:
-        found.last_seen_at = now
-        found.seen_count = int(found.seen_count or 0) + 1
         if found.status == "open":
             # 影响按这一次刷新；表结构变了之后口径试算的情况（ledger/recall 记的 probes）不是探查算的，留着
             probes = (found.affects or {}).get("probes") if isinstance(found.affects, dict) else None

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 把父图那一刻的写入口交下来，子工作流的节点经它发

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)

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() 已经不是确认的那次恢复，
     按待办上记的处理人认（会签时是凑齐的那一票）。"""

backend/app/engine/runner.py

@@ -1448,7 +1448,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 {}

backend/app/ledger/recall.py

@@ -66,7 +66,7 @@ from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from typing import Any
 
-from sqlalchemy import func, not_, select
+from sqlalchemy import func, not_, select, update
 
 from app.core.events import EventType
 from app.db.base import SessionLocal, utcnow
@@ -83,7 +83,7 @@ REASONS = ("caliber", "catalog", "structure")
 #: 发起的批次：发布后自动的、版本历史里手动接着算的
 BATCHES = ("auto", "manual")
 #: 一组的结论
-OUTCOMES = ("changed", "unchanged", "failed", "running", "pending", "not_recomputable")
+OUTCOMES = ("changed", "unchanged", "row_order", "failed", "running", "pending", "not_recomputable")
 #: 启动时接着做最近几天发布的
 RESUME_DAYS = 7
 #: 目录、表结构变了时最多看多少次探索运行（新的在前）
@@ -110,6 +110,8 @@ SUPERSEDED = "已有更新的发布版 v{latest}，召回在 v{latest} 那一版
 NOT_CURRENT = "v{version} 已不是当前发布版，召回不再接着算"
 #: 收尾
 CHANGED_TITLE = "口径「{name}」升到 v{version}，答案的数变了"
+#: 全部指标行序不定、核对不了时的待办标题（settle 的 row_order）
+ROW_ORDER_TITLE = "口径「{name}」升到 v{version}，答案里的数行序不定、没能核对，请人工核对"
 CATALOG_TITLE = "答案依据的数据目录有变：按新目录检查出了新的问题"
 STRUCTURE_TITLE = "口径「{name}」v{version} 依据的表结构变了，试算没有算出来"
 #: 依据有变的四种（ledger.basis_changed 的 kind）
@@ -533,11 +535,12 @@ def _reasons(plan: Plan) -> list[dict[str, Any]]:
 
 
 async def outcome_of(session: Any, run: Run) -> tuple[str, dict[str, Any]]:
-    """一次召回运行的结论：(changed / unchanged / failed / running, 收尾时记下的那一条 ledger.recalled 的内容)。"""
+    """一次召回运行的结论：(changed / unchanged / row_order（全部指标行序不定、核对不了）/ failed / running, 收尾时记下的那一条
+    ledger.recalled 的内容)。"""
     data = await _settled(session, run.id)
     if data is not None:
-        if data.get("outcome") == "failed":
-            return "failed", data
+        if data.get("outcome") in ("failed", "row_order"):
+            return str(data["outcome"]), data
         return ("changed" if data.get("changed") else "unchanged"), data
     if run.status in ("failed", "cancelled"):
         return "failed", {"error": run.error}
@@ -725,22 +728,31 @@ async def _settle(run_id: str) -> str | None:
         await _mark(run_id, {**base, "outcome": "failed", "changed": None, "error": CARD_UNREADABLE})
         return "failed"
     results = compare(meta, values)
-    changed = any(r["changed"] for r in results)
+    # 行序不定的（recall_card 记的 row_order）比不出变没变：不算「没变」。全部都行序不定的结论写核对不了、开待办请人核对；
+    # 部分行序不定的，已判定的照常判，行序不定的那几项在结论、待办、原答案的记账里照实列出
+    unstable = [r["name"] for r in results if r.get("row_order") == "unstable"]
+    decided = [r for r in results if r.get("row_order") != "unstable"]
+    changed = any(r["changed"] for r in decided)
+    outcome = "row_order" if not decided else "changed" if changed else "unchanged"
     answers = list(dict.fromkeys(str(u.get("run_id")) for u in origin.get("uses") or [] if u.get("run_id")))
     tasks: list[str] = []
-    if changed:
-        tasks = await open_tasks(run, reason="caliber")
+    if outcome in ("changed", "row_order"):
+        tasks = await open_tasks(run, reason="caliber", row_order=outcome == "row_order")
     else:
         for answer in answers:
-            await _mark_answer(answer, run)
-    await _mark(run_id, {**base, "outcome": "changed" if changed else "unchanged", "changed": changed,
+            await _mark_answer(answer, run, row_order=unstable)
+    await _mark(run_id, {**base, "outcome": outcome, "changed": changed if decided else None,
                          "answers": len(answers), "tasks": tasks,
-                         "metrics": [{"k": r["k"], "name": r["name"], "changed": r["changed"]} for r in results]})
-    return "changed" if changed else "unchanged"
+                         **({"row_order_metrics": unstable} if unstable else {}),
+                         "metrics": [{"k": r["k"], "name": r["name"], "changed": r["changed"],
+                                      **({"row_order": "unstable"} if r.get("row_order") else {})}
+                                     for r in results]})
+    return outcome
 
 
-async def _mark_answer(answer_id: str, recall_run: Run) -> None:
-    """数没变：在原答案的运行上记一条 ledger.recalled（已按 vN 核对，不影响）。这次召回记过的不再记。"""
+async def _mark_answer(answer_id: str, recall_run: Run, *, row_order: Sequence[str] = ()) -> None:
+    """数没变：在原答案的运行上记一条 ledger.recalled（已按 vN 核对，不影响）。这次召回记过的不再记。row_order 是行序不定、
+    没能核对的那几项（照实列出，不算进「不影响」）。"""
     origin = recall_run.origin if isinstance(recall_run.origin, dict) else {}
     async with SessionLocal() as session:
         rows = (await session.execute(select(RunEvent.data).where(
@@ -749,7 +761,8 @@ async def _mark_answer(answer_id: str, recall_run: Run) -> None:
             return
     await _mark(answer_id, {"role": "answer", "changed": False, "recall_run": recall_run.id,
                             "workflow_id": origin.get("workflow_id"), "from_version": origin.get("from_version"),
-                            "to_version": origin.get("to_version")})
+                            "to_version": origin.get("to_version"),
+                            **({"row_order_metrics": list(row_order)} if row_order else {})})
 
 
 # ==========================================================================
@@ -784,7 +797,7 @@ def correlation(run_id: str, owner: tuple[str, str], reason: str = "caliber", ex
 
 
 async def open_tasks(run: Run, *, reason: str, owners: Iterable[tuple[str, str]] | None = None,
-                     title: str | None = None, extra: str = "") -> list[str]:
+                     title: str | None = None, extra: str = "", row_order: bool = False) -> list[str]:
     """给答案的主人开 recall 待办（挂在 run 上）。owners 不给就按召回运行 origin 里各答案的主人。按 correlation_key 认同一条，
     开过的不再开。提交之后推通知。返回这次新开的待办 id。"""
     from app.process import tasks as process_tasks
@@ -796,7 +809,8 @@ async def open_tasks(run: Run, *, reason: str, owners: Iterable[tuple[str, str]]
         targets = await _owner_targets(session, owners)
         name = await ledger_uses.workflow_name(session, str(origin.get("workflow_id") or ""))
         if title is None:
-            title = CHANGED_TITLE.format(name=name or "口径", version=origin.get("to_version"))
+            title = (ROW_ORDER_TITLE if row_order else CHANGED_TITLE).format(name=name or "口径",
+                                                                            version=origin.get("to_version"))
         opened: list[str] = []
         for owner in targets:
             key = correlation(run.id, owner, reason, extra)
@@ -1016,7 +1030,10 @@ async def answer_marks(session: Any, run_id: str, principal: Any) -> dict[str, A
         data = data if isinstance(data, dict) else {}
         if etype == EventType.LEDGER_RECALLED and data.get("role") == "answer":
             checked.append({k: data.get(k) for k in ("workflow_id", "from_version", "to_version", "recall_run")}
-                           | {"changed": bool(data.get("changed")), "at": at(ts)})
+                           | {"changed": bool(data.get("changed")), "at": at(ts)}
+                           # 行序不定、没能核对的那几项（只列名字）：不算进「不影响」
+                           | ({"row_order_metrics": [str(x) for x in data["row_order_metrics"]]}
+                              if isinstance(data.get("row_order_metrics"), list) and data["row_order_metrics"] else {}))
             workflows.add(str(data.get("workflow_id") or ""))
         elif etype == EventType.LEDGER_BASIS_CHANGED:
             items = data.get("items")
@@ -1596,8 +1613,14 @@ async def _probe_skipped(change: Mapping[str, Any], caliber: Any, pinned: Any, g
     note = {"workflow_id": caliber.workflow_id, "version": caliber.version, "name": str(name), "outcome": "skipped",
             "missing": missing, "reason": reason, "inputs_from": inputs_from}
     async with SessionLocal() as session:
-        row = (await session.execute(select(SourceChange).where(SourceChange.id == change.get("id")))).scalar_one_or_none()
+        # 先占住那一行再读改写（SQLite 上拿到写锁、PostgreSQL 上锁住这一行）：几版口径同时记、探查同时刷新影响
+        # （source_changes.record）时一个接一个来，谁也不盖掉谁
+        await session.execute(update(SourceChange).where(SourceChange.id == change.get("id"))
+                              .values(seen_count=SourceChange.seen_count).execution_options(synchronize_session=False))
+        row = (await session.execute(select(SourceChange).where(SourceChange.id == change.get("id"))
+                                     .execution_options(populate_existing=True))).scalar_one_or_none()
         if row is None:
+            await session.rollback()
             return
         affects = dict(row.affects or {})
         kept = [p for p in affects.get(PROBES_KEY) or [] if isinstance(p, dict)

backend/app/notify/service.py

@@ -409,7 +409,9 @@ async def notify_recall(task_id: str, users: list[str]) -> int:
 
         def message(answers: str, tasks: int, link: tuple[str, str] | None) -> dict[str, Any]:
             return {"event": event, "status": "pending", "ok": False,
-                    "title": recall.CHANGED_TITLE.format(name=name[:80] or "口径", version=origin.get("to_version")),
+                    # 待办开出来时按结论起的标题（数变了 / 行序不定、没能核对）：通知和待办说同一句
+                    "title": (recall.ROW_ORDER_TITLE if "行序不定" in (task.title or "") else recall.CHANGED_TITLE)
+                    .format(name=name[:80] or "口径", version=origin.get("to_version")),
                     "lines": [*head, ("答案", answers), *([("待办", f"{tasks} 条")] if tasks > 1 else []), *tail],
                     "links": [link] if base and link else [],
                     "subject": {"task_id": task_id, "run_id": task.run_id, "workflow_id": origin.get("workflow_id")}}

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 {}

backend/app/scheduler/outbox.py

@@ -559,11 +559,17 @@ async def pending_for_many(session: Any, schedules: list[Schedule]) -> dict[str,
     counts = dict((await session.execute(select(TriggerDelivery.schedule_id, func.count()).where(
         TriggerDelivery.schedule_id.in_(ids), TriggerDelivery.status == "pending")
         .group_by(TriggerDelivery.schedule_id))).all())
+    # 每个任务各读最早的 DRAIN_BATCH 条（同 _drain 一拍读的）：合起来一个上限的话，积压多的任务会把别的挤掉，它们的 due_at 为空
+    ranked = (select(TriggerDelivery.id.label("did"), func.row_number().over(
+        partition_by=TriggerDelivery.schedule_id,
+        order_by=(TriggerDelivery.occurred_at, TriggerDelivery.created_at, TriggerDelivery.id)).label("rn"))
+        .where(TriggerDelivery.schedule_id.in_([i for i in ids if counts.get(i)]), TriggerDelivery.status == "pending")
+        .subquery())
     waiting = (await session.execute(
         select(TriggerDelivery.schedule_id, TriggerDelivery.id, TriggerDelivery.created_at, TriggerEvent)
         .join(TriggerEvent, TriggerEvent.id == TriggerDelivery.event_id)
-        .where(TriggerDelivery.schedule_id.in_([i for i in ids if counts.get(i)]), TriggerDelivery.status == "pending")
-        .order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at).limit(DRAIN_BATCH))).all() if counts else []
+        .join(ranked, ranked.c.did == TriggerDelivery.id).where(ranked.c.rn <= DRAIN_BATCH)
+        .order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at))).all() if counts else []
     by: dict[str, list[tuple[str, TriggerEvent, datetime]]] = {}
     stale: dict[str, int] = {}
     for sid, did, routed, ev in waiting:
@@ -585,8 +591,14 @@ async def pending_for_many(session: Any, schedules: list[Schedule]) -> dict[str,
 
         cond = or_(*[and_(TriggerEvent.event == kind, TriggerEvent.subject_id.in_(sorted(subs)))
                      for kind, subs in pairs.items()])
+        # 每个（事件种类, 对象）各读前 ROUTE_BATCH 条（同改前每个任务各读的那么多）：合起来一个上限会把别的对象挤掉、少算
+        ranked = (select(TriggerEvent.id.label("eid"), func.row_number().over(
+            partition_by=(TriggerEvent.event, TriggerEvent.subject_id),
+            order_by=(TriggerEvent.created_at, TriggerEvent.id)).label("rn"))
+            .where(TriggerEvent.status == "pending", cond).subquery())
         unrouted = (await session.execute(select(TriggerEvent.event, TriggerEvent.subject_id, TriggerEvent.payload)
-                                          .where(TriggerEvent.status == "pending", cond).limit(ROUTE_BATCH))).all()
+                                          .join(ranked, ranked.c.eid == TriggerEvent.id)
+                                          .where(ranked.c.rn <= ROUTE_BATCH))).all()
         for kind, subject, payload in unrouted:
             for sid in ids:
                 if triggers.matches(trig[sid], kind, subject, payload):

backend/tests/test_ledger_recall.py

@@ -954,3 +954,90 @@ async def test_a_probe_that_cannot_assemble_its_inputs_says_which_are_missing(en
         await session.commit()
         again = (await session.execute(select(SourceChange).where(SourceChange.id == out["id"]))).scalar_one()
     assert again.affects["probes"] == [note]
+
+
+async def test_probe_notes_and_change_refreshes_do_not_overwrite_each_other(engine_up):
+    """同一条变化上几处同时写 affects（几版口径各记一条没试算、探查又刷新了一次影响）：先占住那一行再读改写，谁也不盖掉谁。"""
+    import asyncio
+    from types import SimpleNamespace
+
+    from app.data import source_changes as sc
+    from app.db.models import DataSource
+    from app.db.models_sources import SourceChange
+
+    async with SessionLocal() as session:
+        source = (await session.execute(select(DataSource).where(DataSource.name == S.CURRENT["source"]))).scalar_one()
+        change, _ = await sc.record(session, source, sc.NewChange(kind="column_removed", table="orders", column="amount",
+                                                                  severity="action", key="v4",
+                                                                  affects={"catalog": [], "templates": [],
+                                                                           "rules_touched": False}),
+                                    sc.Detected(job_kind="introspect"))
+        await session.commit()
+        out = sc.change_out(change)
+        source_id = source.id
+
+    async def refresh() -> None:
+        async with SessionLocal() as session:
+            src = await session.get(DataSource, source_id)
+            await sc.record(session, src, sc.NewChange(kind="column_removed", table="orders", column="amount",
+                                                       severity="action", key="v4",
+                                                       affects={"catalog": [], "templates": [], "rules_touched": True}),
+                            sc.Detected(job_kind="introspect"))
+            await session.commit()
+
+    calibers = [SimpleNamespace(workflow_id=f"wf-{i}", version=1) for i in range(6)]
+    await asyncio.gather(*(recall._probe_skipped(out, c, None, {}, "defaults", "缺参数") for c in calibers), refresh())
+    async with SessionLocal() as session:
+        row = (await session.execute(select(SourceChange).where(SourceChange.id == out["id"]))).scalar_one()
+    assert sorted(p["workflow_id"] for p in row.affects["probes"]) == [f"wf-{i}" for i in range(6)], row.affects
+    assert row.affects["rules_touched"] is True, "探查刷新的那一份也在"
+
+
+# --------------------------------------------------------------------------
+# 行序不定的指标（召回比不出变没变）：不当「没变」
+# --------------------------------------------------------------------------
+
+
+async def test_all_metrics_with_unstable_row_order_cannot_be_checked_and_open_a_task(monkeypatch):
+    """全部指标都行序不定：结论写「核对不了（行序不定）」，开召回待办请人核对，不在原答案上记「已核对，不影响」。"""
+    wf = await S.publish()
+    a1 = await answer(monkeypatch)
+    graph = S.caliber_graph()
+    graph["nodes"][3]["data"]["config"]["instructions"] = "写周销售（新的写法）"
+    await publish_v2(wf, graph=graph)
+    monkeypatch.setattr(recall, "_rows_stable", lambda pinned, metric: False)
+    [done] = await run_recall(wf)
+    assert done.status == "succeeded", done.error
+    [settled] = [d for d in await events(done.id, EventType.LEDGER_RECALLED) if d.get("role") == "recall"]
+    assert settled["outcome"] == "row_order" and settled["changed"] is None
+    assert settled["row_order_metrics"] == ["销售额"] and settled["metrics"][0]["row_order"] == "unstable"
+    [task] = await recall_tasks(done.id)
+    assert task.title == "口径「周销售」升到 v2，答案里的数行序不定、没能核对，请人工核对"
+    assert await events(a1.id, EventType.LEDGER_RECALLED) == [], "核对不了的不在原答案上记「不影响」"
+    async with SessionLocal() as session:
+        view = await recall.batches(session, wf, 2)
+        state, _ = await recall.outcome_of(session, done)
+    assert state == "row_order" and view["groups"][0]["state"] == "row_order" and view["counts"]["row_order"] == 1
+
+
+async def test_some_metrics_with_unstable_row_order_are_listed_and_the_rest_decide(monkeypatch):
+    """部分行序不定：已判定的照常判（数没变就在原答案上记不影响），行序不定的那几项在结论和原答案的记账里照实列出。"""
+    two = S.caliber_graph()
+    card = two["nodes"][2]["data"]["config"]
+    card["metrics"] = [*card["metrics"], {**card["metrics"][0], "id": "gmv2", "name": "销售额（另算）"}]
+    wf = await S.publish(graph=two)
+    a1 = await answer(monkeypatch)
+    v2 = S.caliber_graph()
+    v2["nodes"][2]["data"]["config"]["metrics"] = card["metrics"]
+    v2["nodes"][3]["data"]["config"]["instructions"] = "写周销售（新的写法）"
+    await publish_v2(wf, graph=v2)
+    monkeypatch.setattr(recall, "_rows_stable", lambda pinned, metric: metric.get("id") != "gmv2")
+    [done] = await run_recall(wf)
+    assert done.status == "succeeded", done.error
+    [settled] = [d for d in await events(done.id, EventType.LEDGER_RECALLED) if d.get("role") == "recall"]
+    assert settled["outcome"] == "unchanged" and settled["changed"] is False
+    assert settled["row_order_metrics"] == ["销售额（另算）"]
+    assert [m.get("changed") for m in settled["metrics"]] == [False, None]
+    [booked] = await events(a1.id, EventType.LEDGER_RECALLED)
+    assert booked["changed"] is False and booked["row_order_metrics"] == ["销售额（另算）"]
+    assert await recall_tasks(done.id) == []

backend/tests/test_process_actions.py

@@ -96,7 +96,9 @@ async def _events(run_id: str, kind: str) -> list[RunEvent]:
 
 
 async def _confirm(run_id: str) -> Task:
-    """等确认卡出来，按卡上的内容指纹确认。"""
+    """等运行停到确认上、确认卡出来，按卡上的内容指纹确认。待办和审批同一个事务开出来，那一刻运行可能还没收成 interrupted、
+    还在跑——这时去确认会被拒（「运行正在运行，没有待处理的审批」）：先等运行停下来，再取待办。"""
+    await _wait(run_id, ("interrupted",))
     task = await _task(run_id, "action_confirm")
     card = (await tasks.detail(task.id, ANN))["payload"]
     out = await tasks.respond(task.id, ANN, decision="approve", payload_hash=card["payload_hash"])
@@ -166,6 +168,11 @@ async def test_unknown_result_with_lookup_is_reconciled_automatically_and_sent_o
     await _confirm(run_id)
     held = await _wait(run_id, ("interrupted", "failed", "succeeded"))
     assert held.status == "interrupted" and held.manifest_hash is None
+    # 运行先挂起、runner 再安排核实（actions.on_unknown）：等排上查询（记下 action.unknown）再看记录
+    for _ in range(100):
+        if await _events(run_id, "action.unknown"):
+            break
+        await asyncio.sleep(0.05)
     [rec] = await _records(run_id)
     assert rec.state == "unknown" and rec.next_lookup_at is not None
     assert len(await _events(run_id, "action.unknown")) == 1

backend/tests/test_schedule_events_scale.py

@@ -169,3 +169,46 @@ async def test_a_window_ending_where_another_fire_already_is_still_fires():
     assert len(await fires(sid)) == 2
     assert [d.status for d in await deliveries(sid)] == ["fired", "fired"]
     assert TriggerDelivery is not None
+
+
+async def test_trigger_option_search_ignores_case_and_takes_wildcards_literally(client):
+    """按名字查不分大小写（PostgreSQL 上 LIKE 分 ASCII 大小写、SQLite 不分：一律 ilike），% 和 _ 按字面匹配、不当通配符。"""
+    async with SessionLocal() as session:
+        session.add_all([Workflow(name=n, graph={"nodes": [], "edges": []})
+                         for n in ("Weekly_Report 样例", "WeeklyXReport 样例", "满 100% 样例", "满 1000 样例")])
+        await session.commit()
+
+    async def names(q: str) -> list[str]:
+        r = await client.get("/api/schedules/trigger-options", params={"event": "workflow.finished", "q": q})
+        assert r.status_code == 200, r.text
+        return [o["name"] for o in r.json()["events"][0]["objects"]]
+
+    assert await names("weekly_report") == ["Weekly_Report 样例"], "大小写不分；_ 不匹配任意一个字"
+    assert await names("100%") == ["满 100% 样例"], "% 不匹配任意几个字"
+
+
+async def test_a_backlog_on_one_schedule_does_not_crowd_out_the_others(monkeypatch):
+    """积压多的任务不把别的任务挤掉：等着的事件、还没派发的事件按任务（按对象）各读各的上限，不是合起来一个上限。"""
+    monkeypatch.setattr(outbox, "DRAIN_BATCH", 3)
+    monkeypatch.setattr(outbox, "ROUTE_BATCH", 3)
+    busy_up, quiet_up = await workflow("积压的上游"), await workflow("安静的上游")
+    downstream = await workflow("区域汇总")
+    busy = await event_schedule(downstream, {"event": "workflow.finished", "workflow_id": busy_up, "debounce_s": 600},
+                                name="积压的")
+    quiet = await event_schedule(downstream, {"event": "workflow.finished", "workflow_id": quiet_up, "debounce_s": 600},
+                                 name="安静的")
+    t0 = utcnow().replace(microsecond=0)
+    for i in range(5):
+        await put_event("workflow.finished", busy_up, {"workflow_id": busy_up, "run_id": f"b{i}"}, t0 + timedelta(seconds=i))
+    await put_event("workflow.finished", quiet_up, {"workflow_id": quiet_up, "run_id": "q0"}, t0 + timedelta(seconds=10))
+    for _ in range(4):                       # 一拍只派 ROUTE_BATCH 条：多派几拍，六条都派给任务
+        await outbox.route(t0 + timedelta(seconds=20))
+    for i in range(5):
+        await put_event("workflow.finished", busy_up, {"workflow_id": busy_up, "run_id": f"c{i}"}, t0 + timedelta(seconds=30 + i))
+    await put_event("workflow.finished", quiet_up, {"workflow_id": quiet_up, "run_id": "q1"}, t0 + timedelta(seconds=40))
+    async with SessionLocal() as session:
+        rows = list((await session.execute(select(Schedule).where(Schedule.id.in_([busy, quiet])))).scalars())
+        state = await outbox.pending_for_many(session, rows)
+    assert state[quiet]["due_at"] == (t0 + timedelta(seconds=610)).isoformat(), state
+    assert state[quiet]["events"] == 1 and state[busy]["events"] == 3, state
+    assert state[busy]["due_at"] == (t0 + timedelta(seconds=600)).isoformat() and state[busy]["waiting"] == 5

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"]

frontend/src/canvas/RecallPanel.tsx

@@ -79,6 +79,8 @@ export function RecallPreviewRow({ workflowId, version }: { workflowId: string;
 
 const STATE_COLOR: Record<string, string | undefined> = {
   changed: 'var(--st-waiting)', failed: 'var(--err)', not_recomputable: 'var(--text-faint)', unchanged: 'var(--ok)',
+  // 核对不了（行序不定）：要人去核对，和「数变了」同一种提醒色
+  row_order: 'var(--st-waiting)',
 }
 /** 列表先列几组，其余收起 */
 const SHOWN = 6

frontend/src/lib/termsLedger.ts

@@ -133,6 +133,8 @@ export const LEDGER_UI = {
   recall: {
     /** 口径升版后在同一份数据上重算过、数没变 */
     checked: (caliber: string) => `已按口径${caliber} 核对，不影响`,
+    /** 部分指标行序不定：那几项没能核对，照实列出 */
+    checkedRowOrder: (names: string[]) => `其中${names.map((x) => `「${x}」`).join('、')}行序不定，没能核对`,
     /** 依据有变：目录里影响数字的项被驳回或修改、出现新码值、处理变化时改了目录（目录类），表或列被删、类型变了（表结构） */
     basis: (what: 'catalog' | 'structure' | 'other', items: number) =>
       `${what === 'structure' ? '依据的表结构有变' : what === 'catalog' ? '依据的数据目录有变' : '答案的依据有变'}`

frontend/src/lib/termsRecall.ts

@@ -58,6 +58,7 @@ export const RECALL_TEXT = {
       const parts = [
         c.changed ? `数变了 ${n(c.changed)} 组` : '',
         c.unchanged ? `没变 ${n(c.unchanged)} 组` : '',
+        c.row_order ? `核对不了（行序不定）${n(c.row_order)} 组` : '',
         c.running ? `正在重算 ${n(c.running)} 组` : '',
         c.failed ? `重算失败 ${n(c.failed)} 组` : '',
         c.not_recomputable ? `无法重算 ${n(c.not_recomputable)} 组` : '',
@@ -65,7 +66,8 @@ export const RECALL_TEXT = {
       return parts.join('，')
     },
     state: {
-      changed: '数变了', unchanged: '没变', failed: '重算失败', running: '正在重算', pending: '未重算',
+      changed: '数变了', unchanged: '没变', row_order: '核对不了（行序不定）', failed: '重算失败', running: '正在重算',
+      pending: '未重算',
       not_recomputable: '无法重算',
     } as Record<string, string>,
     reason: (state: string, reason: string) => (state === 'not_recomputable' ? `无法重算：${reason}` : reason),

frontend/src/run/LedgerStep.tsx

@@ -127,7 +127,12 @@ export function RecallMarks({ recall }: { recall?: LedgerRecallMarks }) {
       {[...checked.values()].map((c) => (
         <li key={`c-${c.workflow_id ?? ''}`} className="flex items-start gap-1.5 text-dim" data-ledger-recall-checked={c.to_version ?? ''}>
           <CircleCheck size={11} className="mt-[2px] shrink-0 text-faint" aria-hidden />
-          <span>{R.checked(caliberName(c.workflow_name, c.to_version))}</span>
+          <span>
+            {R.checked(caliberName(c.workflow_name, c.to_version))}
+            {!!c.row_order_metrics?.length && (
+              <span className="ml-1" data-ledger-recall-row-order="">{R.checkedRowOrder(c.row_order_metrics)}</span>
+            )}
+          </span>
         </li>
       ))}
       {[...basis.entries()].map(([group, b]) => (

frontend/src/typesLedger.ts

@@ -220,6 +220,8 @@ export interface LedgerRecallMarks {
     recall_run?: string | null
     changed?: boolean
     at?: string | null
+    /** 行序不定、没能核对的那几项（服务端 ledger/recall.settle）：不算进「不影响」 */
+    row_order_metrics?: string[]
   }[]
   /** 依据有变：kind 是 catalog（目录里影响数字的项被驳回或修改）/ structure（表结构）/ values（新码值）/ catalog_edit */
   basis?: { kind?: string | null; items?: number; new_errors?: number; at?: string | null }[]

frontend/src/typesRecall.ts

@@ -42,7 +42,8 @@ export interface RecallPreview {
   reasons: { reason: string; groups: number }[]
 }
 
-export type RecallGroupState = 'changed' | 'unchanged' | 'failed' | 'running' | 'pending' | 'not_recomputable'
+/** row_order：全部指标行序不定、核对不了（服务端 ledger/recall.settle），已开待办请人核对 */
+export type RecallGroupState = 'changed' | 'unchanged' | 'row_order' | 'failed' | 'running' | 'pending' | 'not_recomputable'
 
 /** 召回批次里的一组 */
 export interface RecallGroup {