diff --git a/backend/app/api/schedules.py b/backend/app/api/schedules.py index ff6688cb..393a4998 100644 --- a/backend/app/api/schedules.py +++ b/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] diff --git a/backend/app/data/source_changes.py b/backend/app/data/source_changes.py index 6c3dd76f..adba3f79 100644 --- a/backend/app/data/source_changes.py +++ b/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 diff --git a/backend/app/engine/context.py b/backend/app/engine/context.py index 24bbac71..6aff018f 100644 --- a/backend/app/engine/context.py +++ b/backend/app/engine/context.py @@ -43,12 +43,6 @@ class RunContext: """这张图里的节点 id → 这次运行里的 id（事件、工件上记的那个）：最外层原样，子工作流里加前缀。""" return f"{self.prefix}{node_id}" - @property - def root_ids(self) -> frozenset[str]: - """这次运行最外层图上的节点 id：最外层就是这张图的，子工作流里是 multi.run_subgraph 交下来的。""" - given = self.extra.get(ROOT_IDS_KEY) - return frozenset(given) if given is not None else frozenset(n.id for n in self.spec.nodes) - def global_label(self, title: str) -> str: """这张图里的节点标题 → 这次运行里给人看的名字：最外层原样，子工作流里接上子工作流的名字（「子工作流 › x」，嵌套的 一层层接上，同 multi.run_subgraph 记在异常上的 label）。父运行的图上没有子工作流里的节点，各处照这个名字说。""" @@ -59,8 +53,6 @@ class RunContext: PREFIX_KEY = "node_prefix" #: 子工作流里的节点名前缀（「子工作流 › 」，同上）：RunContext.global_label LABEL_PREFIX_KEY = "node_label_prefix" -#: 最外层图上的节点 id（同上，只在子工作流里有）：RunContext.root_ids -ROOT_IDS_KEY = "root_node_ids" #: 父运行的事件流（RunContext.extra 里的键，同上）。子工作流在父节点里用 ainvoke 跑，LangGraph 不把它的 custom 事件交给 #: 父图的流（ainvoke 不开 custom 模式，get_stream_writer 拿到的是空的那个）：子工作流里每一步以前在运行的事件里一条都 #: 没有，时间线看不到、封存也不含它查过的数。所以 run_subgraph 把父图那一刻的写入口交下来，子工作流的节点经它发 diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index c2d5a152..f3ff3002 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -15,7 +15,7 @@ from app.core.events import EventType from app.db.base import SessionLocal from app.db.models import Workflow from app.engine.context import ( - LABEL_PREFIX_KEY, PREFIX_KEY, ROOT_IDS_KEY, STREAM_KEY, NodeContext, NodeError, stream_writer, + LABEL_PREFIX_KEY, PREFIX_KEY, STREAM_KEY, NodeContext, NodeError, stream_writer, ) from app.engine.lineage import whole_ref from app.engine.state import GraphState, message_text, template_context @@ -694,9 +694,7 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 子图里的节点经父运行的写入口发事件、节点 id 带前缀；嵌套的子图拿到的是同一个写入口、更长的前缀 extra={**{k: v for k, v in ctx.run.extra.items() if k not in ("resumed", "actors")}, PREFIX_KEY: f"{ctx.run.global_id(ctx.node.id)}/", LABEL_PREFIX_KEY: f"{ctx.run.global_label(ctx.node.title)} › ", - STREAM_KEY: stream_writer(ctx.run), - # 最外层图上的节点 id：子图里的写入认升级前按子图里的 id 记的幂等键时，排除父图同名节点记的那一条（tools._run_write） - ROOT_IDS_KEY: ctx.run.extra.get(ROOT_IDS_KEY) or frozenset(n.id for n in ctx.run.spec.nodes)}, + STREAM_KEY: stream_writer(ctx.run)}, ) # 这个子工作流节点之前几次执行里，子图各节点跑过的记录（交回父图 trail 的，见下）：子图从它们接着数第几次执行 # （engine/evidence.next_exec）。以前每次都从空的 trail 起步，循环里第二轮的子图节点又记成第 1 次：台账分不出是哪一次， @@ -772,7 +770,7 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 从这里接着数（上面的 seed）；走了驳回出口的那几步（outcome=rejected）运行收尾据此在终态事件上照实写「途经驳回」 # （runner._rejected_via），实例页、通知、定时任务的 Webhook 说得出「子工作流 › 把关」被驳回、转到哪（名字、转到哪几步 # 在子图里已经按这次运行的说法记好：global_label、global_id）。父图自己的节点按自己的 id 数，不受影响 - updates["trail"] = [{**t, "node_id": f"{prefix}{t.get('node_id')}"} for t in ran] + updates["trail"] = _forwarded_trail(ran, prefix) var_name = ctx.cfg("assign_to", "") if var_name: updates["vars"] = {var_name: output} @@ -783,6 +781,11 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: _LEDGER_NODE_KEYS = ("node_id", "card") +def _forwarded_trail(ran: list[dict[str, Any]], prefix: str) -> list[dict[str, Any]]: + """子图这一次跑过的步骤交回父图 trail 的样子：节点 id 接上「子工作流节点 id/」前缀。""" + return [{**t, "node_id": f"{prefix}{t.get('node_id')}"} for t in ran] + + def _forwarded(entry: dict[str, Any], node_id: str) -> dict[str, Any]: """子图台账里的一条交回父图：节点 id 加上「子工作流节点 id/」前缀，别的照抄。""" out = dict(entry) diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index f2f61386..3f15785b 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -275,10 +275,10 @@ async def _run_write(state: GraphState, ctx: NodeContext, name: str, args: dict[ exec_no = next_exec(state, ctx.node.id) # 幂等键按这次运行里的节点 id 算（子工作流里的写「sub/pay」）：以前按子图里的 id（pay）算，父图里同名的节点、套同一个 - # 工作流的两个子工作流节点撞同一个键，后发的那一笔交回先发的回执、不发送。升级前停在子图里的运行按子图里的 id 记过， - # 恢复时整个子图重跑：照旧认那个键（legacy_node_id），已经生效的不重发；父图里有同名节点时不认（那一条可能是它的） - node_id = ctx.run.global_id(ctx.node.id) - legacy = ctx.node.id if node_id != ctx.node.id and ctx.node.id not in ctx.run.root_ids else None + # 工作流的两个子工作流节点撞同一个键，后发的那一笔交回先发的回执、不发送。升级前子工作流里的写入按子图里的 id 记过键： + # 写入节点本身重新执行时（结果未知核实之后、失败后接着跑；审批恢复时子图从自己的断点接着跑，不走到这里）照旧认那个键 + # （legacy_node_id），是不是这个节点的那一笔、是不是这一次的内容，由 actions.prepare 按那条记录的事件和内容指纹认 + node_id, legacy = _action_ids(ctx) try: prep = await actions.prepare(run_id=ctx.run.run_id, node_id=node_id, exec_no=exec_no, capability=capability, args=args, legacy_node_id=legacy) @@ -329,6 +329,12 @@ async def _run_write(state: GraphState, ctx: NodeContext, name: str, args: dict[ return _write_updates(ctx, result) +def _action_ids(ctx: NodeContext) -> tuple[str, str | None]: + """写入的幂等键按哪个节点 id 算：(这次运行里的 id, 升级前按子图里的 id 记过的那种；最外层的为 None)。""" + node_id = ctx.run.global_id(ctx.node.id) + return node_id, ctx.node.id if node_id != ctx.node.id else None + + async def _confirmed_by(run_id: str, node_id: str) -> str | None: """确认这次发送的人：节点重新执行（结果未知核实后接着跑、重启后恢复）时 ctx.actor() 已经不是确认的那次恢复， 按待办上记的处理人认（会签时是凑齐的那一票）。""" diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 43eff374..552af462 100644 --- a/backend/app/engine/runner.py +++ b/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 {} diff --git a/backend/app/ledger/recall.py b/backend/app/ledger/recall.py index 5cc96e12..364b138d 100644 --- a/backend/app/ledger/recall.py +++ b/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) diff --git a/backend/app/notify/service.py b/backend/app/notify/service.py index d412c33e..17366503 100644 --- a/backend/app/notify/service.py +++ b/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")}} diff --git a/backend/app/process/actions.py b/backend/app/process/actions.py index 0f1d6edd..d0a68d83 100644 --- a/backend/app/process/actions.py +++ b/backend/app/process/actions.py @@ -244,6 +244,51 @@ class Prepared: return f"{self.key}:{self.attempt}" +#: 按自己的键找到别人的那一笔时换用的键的前缀（prepare）：节点 id 里不会有这个字符 +OWN_KEY_MARK = "\x1e" +#: 带 action_id 的那几种事件：认一条记录是哪个节点写的（_owner） +_ACTION_EVENTS = (EventType.ACTION_PREPARED, EventType.ACTION_SENT, EventType.ACTION_CONFIRMED, EventType.ACTION_FAILED, + EventType.ACTION_UNKNOWN, EventType.ACTION_SIMULATED) + + +async def _owner(session: AsyncSession, rec: ActionRecord) -> str | None: + """这条记录是哪个节点写的（这次运行里的 id，子工作流里的带前缀）：按它的 action.* 事件上记的 node_id。记录上的 node_id + 升级前子工作流里记的是子图里的 id，和父图同名的节点分不开；事件一向按这次运行里的 id 记。认不出（事件没写上）为 None。""" + from app.db.models import RunEvent + + rows = (await session.execute(select(RunEvent.node_id, RunEvent.data).where( + RunEvent.run_id == rec.run_id, RunEvent.type.in_([str(t) for t in _ACTION_EVENTS])))).all() + return next((str(node) for node, data in rows if node and isinstance(data, dict) and data.get("action_id") == rec.id), + None) + + +def _same_payload(rec: ActionRecord, capability: Any, args: dict[str, Any]) -> bool | None: + """这条记录是不是这一次要发的内容：能力相同、内容指纹等于这一次参数收好类型之后的指纹。收不成类型（参数定义改了）的 + 比不了，为 None。""" + name = capability.name + if rec.capability != name: + return False + try: + typed = typed_args(capability, args) + except ActionFailed: + return None + return rec.payload_hash == payload_hash(name, typed) + + +async def _belongs(session: AsyncSession, rec: ActionRecord, node_id: str, capability: Any, args: dict[str, Any], *, + strict: bool = False) -> bool: + """这条记录是不是这个节点（node_id，这次运行里的 id）的。事件认得出是谁写的就按事件；认不出的按内容（能力、内容指纹）， + 比不了的当是（以前就是这样认的）。strict（认升级前的旧键时）：事件说是这个节点的，内容也要对得上——循环跨了升级时， + 同一个节点第二轮按老算法同样是第 1 次执行，找到的是第一轮那一笔；比不了内容（参数定义改了）的照事件认。""" + owner = await _owner(session, rec) + same = _same_payload(rec, capability, args) + if owner is not None and owner != node_id: + return False + if owner is not None and not strict: + return True + return same is not False + + async def latest(session: AsyncSession, key: str) -> ActionRecord | None: return await session.scalar(select(ActionRecord).where(ActionRecord.action_key == key) .order_by(ActionRecord.attempt.desc()).limit(1)) @@ -268,12 +313,20 @@ async def prepare(*, run_id: str, node_id: str, exec_no: int, capability: Any, a run = await session.get(Run, run_id) endpoint = await _endpoint(session, capability) prev = await latest(session, key) + if prev is not None and not await _belongs(session, prev, node_id, capability, args): + # 按自己的键找到的是别的节点的那一笔：升级前子工作流里同名的节点按子图里的 id 记的键，和父图这个节点的一样。 + # 那一笔不是这个节点的，换一个只属于这个节点的键（同样的输入永远算出同一个） + key = action_key(run_id, f"{OWN_KEY_MARK}{node_id}", exec_no) + prev = await latest(session, key) if prev is None and legacy_node_id: - # 升级前子工作流里的写入按子图里的节点 id 记幂等键（node_id 也记成它）：恢复时整个子图重跑，照那个键认出已经 - # 记下的那一笔，接着用它的键（不重发、第几次发送接着数）。新记的子图写入 node_id 带前缀，认不到这里来 + # 升级前子工作流里的写入按子图里的节点 id 记幂等键（node_id 也记成它）：写入节点本身重新执行时（结果未知核实之后、 + # 失败后接着跑）照那个键认出已经记下的那一笔，接着用它的键（不重发、第几次发送接着数）。只认这个节点的、这一次的： + # 那一笔的事件记着是这个节点写的（父图里同名的节点按同一个键写过），内容指纹也和这一次的一样（循环跨了升级时， + # 第二轮按老算法同样是第 1 次执行，找到的是第一轮那一笔） old_key = action_key(run_id, legacy_node_id, exec_no) old = await latest(session, old_key) - if old is not None and old.node_id == legacy_node_id: + if old is not None and old.node_id == legacy_node_id \ + and await _belongs(session, old, node_id, capability, args, strict=True): key, prev, node_id = old_key, old, legacy_node_id facts = {"app_id": run.app_id, "workflow_id": run.workflow_id, "workflow_version": run.version, "graph_hash": run.version_hash, "started_by": run.started_by} if run is not None else {} diff --git a/backend/app/scheduler/outbox.py b/backend/app/scheduler/outbox.py index 722147e0..4df850a5 100644 --- a/backend/app/scheduler/outbox.py +++ b/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): diff --git a/backend/tests/test_ledger_recall.py b/backend/tests/test_ledger_recall.py index 751a7ed7..5610451a 100644 --- a/backend/tests/test_ledger_recall.py +++ b/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) == [] diff --git a/backend/tests/test_process_actions.py b/backend/tests/test_process_actions.py index c093f98d..2110b39c 100644 --- a/backend/tests/test_process_actions.py +++ b/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 diff --git a/backend/tests/test_schedule_events_scale.py b/backend/tests/test_schedule_events_scale.py index 547701a0..40bd429d 100644 --- a/backend/tests/test_schedule_events_scale.py +++ b/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 diff --git a/backend/tests/test_subgraph_repeat.py b/backend/tests/test_subgraph_repeat.py index e7ebac88..31e30365 100644 --- a/backend/tests/test_subgraph_repeat.py +++ b/backend/tests/test_subgraph_repeat.py @@ -137,22 +137,144 @@ async def test_a_namesake_in_the_parent_does_not_share_the_key(monkeypatch, erp) async def test_a_write_recorded_before_the_upgrade_is_still_found(monkeypatch, erp): - """升级前子图里的写入按子图里的 id（pay）记幂等键。停在子图里的运行恢复时整个子图重跑：照旧按那个键找到已经生效的 - 记录、交回回执，不重发。父图里有同名节点时不这样认（那一条可能是父图那个节点的）。""" + """升级前子图里的写入按子图里的 id（pay）记幂等键、记录上的 node_id 也是它；事件一向按这次运行里的 id（sub/pay）记。 + 写入节点本身重新执行时（结果未知核实之后接着跑）：事件说是这个节点的、内容指纹也是这一次的，照旧按那个键交回回执、不重发； + 事件说是父图同名节点的、或者内容不一样（跨升级的循环，第二轮按老算法也是第 1 次执行）的不认。""" + from app.core.events import EventType + from app.db.models import RunEvent + caps = install(monkeypatch, erp, await endpoint_row(test=True), lookup=False) - async with SessionLocal() as session: - run = Run(graph={"nodes": [node("sub", "subgraph")], "edges": []}, status="running", run_class="exploratory") - session.add(run) - await session.flush() - legacy = actions.action_key(run.id, "pay", 1) - session.add(ActionRecord(action_key=legacy, attempt=1, idem_key=f"{legacy}:1", run_id=run.id, node_id="pay", - exec_no=1, capability=PAY, environment="test", payload_hash="0" * 64, state="confirmed", - receipt={"summary": {"id": "PAY-OLD"}})) - await session.commit() - run_id = run.id args = {"amount": "5", "card": "6222020200112233"} + digest = actions.payload_hash(PAY, actions.typed_args(caps[PAY], args)) + + async def old_record(writer: str, payload: str) -> tuple[str, str]: + async with SessionLocal() as session: + run = Run(graph={"nodes": [node("sub", "subgraph")], "edges": []}, status="running", run_class="exploratory") + session.add(run) + await session.flush() + legacy = actions.action_key(run.id, "pay", 1) + rec = ActionRecord(action_key=legacy, attempt=1, idem_key=f"{legacy}:1", run_id=run.id, node_id="pay", + exec_no=1, capability=PAY, environment="test", payload_hash=payload, state="confirmed", + receipt={"summary": {"id": "PAY-OLD"}}) + session.add(rec) + await session.flush() + session.add(RunEvent(run_id=run.id, seq=1, type=str(EventType.ACTION_PREPARED), node_id=writer, ts=0.0, + data={"action_id": rec.id, "capability": PAY, "payload_hash": payload})) + await session.commit() + return run.id, legacy + + run_id, legacy = await old_record("sub/pay", digest) found = await actions.prepare(run_id=run_id, node_id="sub/pay", exec_no=1, capability=caps[PAY], args=args, legacy_node_id="pay") assert found.key == legacy and found.receipt is not None fresh = await actions.prepare(run_id=run_id, node_id="sub/pay", exec_no=1, capability=caps[PAY], args=args) assert fresh.key != legacy and fresh.receipt is None + # 父图里同名的 pay 按自己的键（和旧键一样）找到这一笔：事件说是 sub/pay 的，不是自己的，换一个只属于自己的键 + mine = await actions.prepare(run_id=run_id, node_id="pay", exec_no=1, capability=caps[PAY], args=args) + assert mine.key != legacy and mine.receipt is None + # 内容不一样（这一次是 7）：不当成这一次已经生效 + other = await actions.prepare(run_id=run_id, node_id="sub/pay", exec_no=1, capability=caps[PAY], + args={**args, "amount": "7"}, legacy_node_id="pay") + assert other.key != legacy and other.receipt is None + # 那一笔是父图 pay 写的（事件说 pay）：子图里的 pay 不认 + run_id, legacy = await old_record("pay", digest) + theirs = await actions.prepare(run_id=run_id, node_id="sub/pay", exec_no=1, capability=caps[PAY], args=args, + legacy_node_id="pay") + assert theirs.key != legacy and theirs.receipt is None + own = await actions.prepare(run_id=run_id, node_id="pay", exec_no=1, capability=caps[PAY], args=args) + assert own.key == legacy and own.receipt is not None + + +# -------------------------------------------------------------------------- +# 跨升级：升级前的运行按老算法记的键，升级后接着跑 +# -------------------------------------------------------------------------- + + +def _old_code(monkeypatch): + """装成升级前的代码：子工作流里的写入按子图里的 id 算幂等键、不认旧键，子图跑过的步骤不交回父图（exec 每次从 1 数）。 + 交回一个函数，调了就换回现在的代码（同一个进程里接力跑，等于升级后重启接着跑；能力登记的替身要重新装）。""" + from app.engine.nodes import multi, tools + + monkeypatch.setattr(tools, "_action_ids", lambda ctx: (ctx.node.id, None)) + monkeypatch.setattr(multi, "_forwarded_trail", lambda ran, prefix: []) + return monkeypatch.undo + + +def _amounts(erp) -> list[str]: + return [str((w.get("body") or {}).get("amount")) for w in erp.writes("test")] + + +async def _until_status(run_id: str, statuses: tuple[str, ...]) -> Run: + for _ in range(600): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in statuses and not run_manager.is_active(run_id): + return row + raise AssertionError(f"运行 {run_id} 停在 {row.status}") + + +async def test_an_unknown_write_inside_a_subworkflow_with_a_namesake_in_the_parent_survives_the_upgrade(monkeypatch, erp): + """升级前：子工作流里的 pay（1）发出去、结果未知、运行挂起，键按子图里的 id（pay）记——和父图里同名的 pay 一样。升级后 + 核实为已生效、接着跑：子图里的 pay 认出自己那一笔（事件记着是「sub/pay」写的），不再发；父图的 pay（99）按自己的键找到 + 的是子图那一笔、不是自己的，换用自己的键照常发。1 和 99 各只发一次。""" + from app.db.models import Task + from app.process import tasks + from tests.test_process_actions import ANN + + await _unattended(monkeypatch, erp) + inner = S.chain(node("s", "input", fields=[{"name": "n"}]), pay(), + node("o", "output", fields=[{"name": "r", "value": "x"}])) + sub_id = await S.publish("付款子流程", graph=inner, level="draft") + graph = S.chain(node("start", "input", fields=[{"name": "n"}]), + node("sub", "subgraph", "子流程", workflow_id=sub_id, workflow_version=1, input={"n": "1"}), + pay(), node("out", "output", fields=[{"name": "r", "value": "完成"}])) + upgrade = _old_code(monkeypatch) + erp.mode = "timeout_after_write" + run = await run_manager.start(graph=graph, input_payload={"n": 99}) + parked = await _until_status(run.id, ("interrupted", "failed", "succeeded")) + assert parked.status == "interrupted", parked.error + assert _amounts(erp) == ["1"] + upgrade() + await _unattended(monkeypatch, erp) # undo 连能力登记的替身一起撤了：重新装上 + erp.mode = "ok" + async with SessionLocal() as session: + task = (await session.execute(select(Task).where(Task.run_id == run.id, Task.kind == "reconcile", + Task.status == "pending"))).scalars().first() + assert task is not None + await tasks.respond(task.id, ANN, decision="confirmed", values={"external_id": "PAY-1"}) + final = await _until_status(run.id, ("succeeded", "failed")) + assert final.status == "succeeded", final.error + assert sorted(_amounts(erp)) == ["1", "99"], _amounts(erp) + + +async def test_a_loop_across_the_upgrade_sends_the_second_round(monkeypatch, erp): + """升级前跑完第一轮（子图里 pay 发了 1）、停在父图的审批上；升级后批准，第二轮按老算法同样是第 1 次执行，找到第一轮那一笔： + 内容指纹不一样（这一轮是 2），不当成已经生效，照常发；也不记「已经生效」的提示。""" + await _unattended(monkeypatch, erp) + inner = S.chain(node("s", "input", fields=[{"name": "n"}]), pay(), + node("o", "output", fields=[{"name": "r", "value": "x"}])) + sub_id = await S.publish("付款子流程", graph=inner, level="draft") + nodes = [node("start", "input"), + node("each", "loop", "逐项", mode="foreach", items="[1, 2]", item_var="it", max_iterations=5), + node("sub", "subgraph", "子流程", workflow_id=sub_id, workflow_version=1, input={"n": "{{ vars.it }}"}), + node("gate", "human", "把关", mode="approve", title="下一轮？"), + node("out", "output", fields=[{"name": "r", "value": "完成"}])] + edges = [edge("start", "each"), edge("each", "sub", "body"), edge("sub", "gate"), edge("gate", "each", "approved"), + edge("each", "out", "done")] + upgrade = _old_code(monkeypatch) + run = await run_manager.start(graph={"nodes": nodes, "edges": edges}, input_payload={}) + parked = await _until_status(run.id, ("interrupted", "failed", "succeeded")) + assert parked.status == "interrupted", parked.error + assert _amounts(erp) == ["1"] + upgrade() + await _unattended(monkeypatch, erp) + await run_manager.resume(run.id, {"approved": True}) + parked = await _until_status(run.id, ("interrupted", "failed", "succeeded")) + assert parked.status == "interrupted", parked.error + assert _amounts(erp) == ["1", "2"], "第二轮照发，不交回第一轮的回执" + assert not [e for e in await _events(run.id, "log") if e.data.get("code") == "action_already_applied"] + await run_manager.resume(run.id, {"approved": True}) + final = await _until_status(run.id, ("succeeded", "failed")) + assert final.status == "succeeded", final.error + assert _amounts(erp) == ["1", "2"] diff --git a/frontend/src/canvas/RecallPanel.tsx b/frontend/src/canvas/RecallPanel.tsx index a09532fd..3e29fe6d 100644 --- a/frontend/src/canvas/RecallPanel.tsx +++ b/frontend/src/canvas/RecallPanel.tsx @@ -79,6 +79,8 @@ export function RecallPreviewRow({ workflowId, version }: { workflowId: string; const STATE_COLOR: Record = { changed: 'var(--st-waiting)', failed: 'var(--err)', not_recomputable: 'var(--text-faint)', unchanged: 'var(--ok)', + // 核对不了（行序不定）：要人去核对，和「数变了」同一种提醒色 + row_order: 'var(--st-waiting)', } /** 列表先列几组，其余收起 */ const SHOWN = 6 diff --git a/frontend/src/lib/termsLedger.ts b/frontend/src/lib/termsLedger.ts index a5a93452..29165b0b 100644 --- a/frontend/src/lib/termsLedger.ts +++ b/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' ? '依据的数据目录有变' : '答案的依据有变'}` diff --git a/frontend/src/lib/termsRecall.ts b/frontend/src/lib/termsRecall.ts index a1931a94..e15ef124 100644 --- a/frontend/src/lib/termsRecall.ts +++ b/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, reason: (state: string, reason: string) => (state === 'not_recomputable' ? `无法重算：${reason}` : reason), diff --git a/frontend/src/run/LedgerStep.tsx b/frontend/src/run/LedgerStep.tsx index cf511dbf..4d247e7d 100644 --- a/frontend/src/run/LedgerStep.tsx +++ b/frontend/src/run/LedgerStep.tsx @@ -127,7 +127,12 @@ export function RecallMarks({ recall }: { recall?: LedgerRecallMarks }) { {[...checked.values()].map((c) => (
  • - {R.checked(caliberName(c.workflow_name, c.to_version))} + + {R.checked(caliberName(c.workflow_name, c.to_version))} + {!!c.row_order_metrics?.length && ( + {R.checkedRowOrder(c.row_order_metrics)} + )} +
    • ))} {[...basis.entries()].map(([group, b]) => ( diff --git a/frontend/src/typesLedger.ts b/frontend/src/typesLedger.ts index 745da48f..ca0ab914 100644 --- a/frontend/src/typesLedger.ts +++ 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 }[] diff --git a/frontend/src/typesRecall.ts b/frontend/src/typesRecall.ts index 90aca8c8..542bf4fe 100644 --- a/frontend/src/typesRecall.ts +++ b/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 {