召回复核修正（后端）：人工节点不再挡召回、读不出召回卡记失败、旧版不再接着算、团队渠道一次召回只推一条、目录核对按数据源合并排队；账本接口带上召回标记，对账结果可以反馈

- W4R-01：召回只查新版图里会不会写外部系统（apps/backfill 抽出 graph_external_write，graph_side_effect 复用它）。召回图只有两版
  的取数和口径卡，图里别处的人工节点、按流程使用的问题碰不到它；口径卡上游的人工节点 derive 本来就认作数据部分不能复用。
  「按新版重新出具」走完整的正式运行，仍按 graph_side_effect 全套拒绝。原来带审批的口径升版后一组都不召回，L6 对它们失效。
- W4R-09：收尾时召回卡读不出、或缺了某个比较的旧值 / 新值，照实记重算失败（读不出召回卡），原答案一个不动。原来两边都是空，
  按渲染比被当成「没变」，给每个原答案永久盖上「已核对，不影响」，收过尾之后再也不核对。
- W4R-12：这一版已经不是当前发布版（之后又发布了更新的版本）时不再发起召回、不给接着算，批次里写 continue_reason，接着算的
  接口 409 带同一句。同启动时接着做只认当前发布版：新版本的召回覆盖更早的所有版本，再按旧版开「口径升到 vN」的待办只是添乱。
- W4R-13：开登录时一组答案几个主人就几条待办，全局渠道却按这一批（recall.batch_of_key：同一次召回、同一个来由）只在最早开的
  那条上推一次，答案、待办按这一批汇总，几条的链到待办箱；别的几条连 task.opened 也不再推全局渠道。个人渠道照旧每人一条。
- W4R-11：目录审阅之后的核对按数据源排队、批量审阅的几张表合成一遍（catalog_pass，同一数据源一把锁）；查询快照按数据源在库里
  筛、按运行结束时间分页、每张表凑够 BASIS_MAX_RUNS 次运行就停；冻结的表结构快照这一遍里按工件 id 只读一次。原来批量审阅
  几十张表就同时起几十个后台任务，每个都全表扫查询快照、反复读同一份几 MB 的表结构快照。
- W4R-03（本包那一半）：GET /api/runs/{id}/ledger 加 recall：checked（已按 vN 核对，不影响）、basis（依据有变：哪一种、几项、
  新出现几条错误级问题）、tasks（看的这个人手上说的是这个答案、还没处理的 recall 待办）。不放数、不放表名。几种状态都带。
- W4R-10（本包那一半）：新事件 ledger.feedback（对账运行上的封存后事件），POST /api/ledger/answers/{run}/reconcile/{rid}/feedback
  {verdict: useful | unclear} 记反馈，同一个人改主意只算最后一次（app/ledger/feedback.py）；账本接口的 reconciled 每一项带
  feedback {mine, useful, unclear}；统计 E2 的 feedback 数到它，去掉「第 4 波接上」的占位说法。离线核对页、封存后事件摘要
  都认得这个事件。

测试：test_ledger_recall 新增七个用例、改了人工节点那一个（召回照算、重新出具照实拒绝）；新增 test_ledger_feedback；
test_ledger_api 的 E2 断言跟上。每个修复都先写失败的测试、去掉修复回跑确认抓得住。后端全量 14237 通过。

backend/app/api/ledger.py

@@ -10,6 +10,8 @@
 - POST /api/ledger/answers/{run_id}/issue（answers_router，登录即可，第 3 波）：按口径正式出具（app/ledger/issue.py，拍板
   L1、L5：参数只从比对工件取，使用者只能按他有权使用的应用固定着的版本出具）；
 - POST /api/ledger/answers/{run_id}/reconcile（同上）：口径对账，点开才跑一次确定性派生的只读图（app/ledger/reconcile.py）；
+- POST /api/ledger/answers/{run_id}/reconcile/{reconcile_id}/feedback（同上，第 4 波）：对账结果上的「说明有用 / 看不懂」
+  （app/ledger/feedback.py，统计 E2）；
 - GET /api/ledger/certify-queue、POST /api/ledger/certify-queue/dismiss（router，搭建者，第 3 波）：可沉淀的问题、「不是口径」
   （app/ledger/certify.py）。
 - GET /api/ledger/recall/{workflow_id}/{version}、POST …/continue（router，搭建者，第 4 波）：口径升到这一版之后的召回批次——
@@ -172,6 +174,28 @@ async def reconcile_answer(run_id: str, payload: ReconcileIn) -> dict[str, Any]:
     return {"run_id": run.id, "reconciled": ledger_reconcile.reconciled_view(run)}
 
 
+class FeedbackIn(BaseModel):
+    """对账结果上的反馈：说明有用（useful）还是看不懂（unclear）。"""
+
+    verdict: str = Field(min_length=1, max_length=20)
+
+
+@answers_router.post("/{run_id}/reconcile/{reconcile_id}/feedback")
+async def reconcile_feedback(run_id: str, reconcile_id: str, payload: FeedbackIn) -> dict[str, Any]:
+    """对账结果上的「说明有用 / 看不懂」（app/ledger/feedback.py，统计 E2）：记成对账运行上的封存后事件，同一个人改主意只算
+    最后一次。交回这次对账现在的反馈 {mine, useful, unclear}。不认的反馈 422；不是这个答案的对账 404；读不了这个答案 403；
+    对账还没跑完 409。"""
+    from app.api.coded import CodedHTTPException
+    from app.api.deps import current_principal
+    from app.ledger import feedback
+
+    try:
+        out = await feedback.give(run_id, reconcile_id, payload.verdict, principal=current_principal())
+    except feedback.FeedbackRefused as e:
+        raise CodedHTTPException(e.status_code, e.message, e.code) from e
+    return {"run_id": reconcile_id, "feedback": out}
+
+
 @router.get("/recall/{workflow_id}/{version}")
 async def recall_batches(workflow_id: str, version: int, session: AsyncSession = Depends(get_session)
                          ) -> dict[str, Any]:
@@ -188,7 +212,7 @@ async def recall_batches(workflow_id: str, version: int, session: AsyncSession =
 async def recall_continue(workflow_id: str, version: int, session: AsyncSession = Depends(get_session)
                           ) -> dict[str, Any]:
     """接着算：把还没重算的组再发起一批（每批 recall_limit 组，上限是 0 时按缺省的组数），交回发起了的召回运行和新的批次
-    情况。都算过了的回 409。"""
+    情况。都算过了的、这一版已经不是当前发布版的（之后又发布了更新的版本）回 409，detail 写为什么。"""
     from app.api.coded import CodedHTTPException
     from app.api.deps import current_principal
     from app.ledger import recall
@@ -197,7 +221,7 @@ async def recall_continue(workflow_id: str, version: int, session: AsyncSession
         raise HTTPException(404, "工作流不存在，可能已被删除")
     current = await recall.batches(session, workflow_id, version)
     if not current["can_continue"]:
-        raise CodedHTTPException(409, current.get("problem") or "没有还没重算的组", recall.REFUSED)
+        raise CodedHTTPException(409, current.get("continue_reason") or "没有还没重算的组", recall.REFUSED)
     launched = await recall.launch(workflow_id, version, batch="manual", actor=current_principal().username)
     async with SessionLocal() as fresh:
         view = await recall.batches(fresh, workflow_id, version)

backend/app/apps/backfill.py

@@ -120,7 +120,6 @@ async def graph_side_effect(session: AsyncSession, graph_json: Any, *, level: st
     判断沿用应用检查对流程钉住项的那一套（apps/lint._lint_process）：人工节点、写外部系统的能力
     （capabilities.registry.writes_externally，和运行时、发布检查同一个判断）、process.* 问题（子工作流先读好，
     process.human_in_subgraph 才查得到）。level 是那一版发布时的级别（门禁按它查），没有按 published。"""
-    from app.capabilities.registry import effective_effect, get_capability, writes_externally
     from app.engine.governance import publish_issues
     from app.engine.lint_process import load_subgraphs, with_subgraphs
     from app.engine.schema import GraphSpec, NodeType
@@ -131,6 +130,34 @@ async def graph_side_effect(session: AsyncSession, graph_json: Any, *, level: st
         return "的图无法解析"
     if humans := [n.title for n in graph.nodes if n.type == NodeType.HUMAN]:
         return f"里有人工节点「{'、'.join(humans[:3])}」"
+    if reason := await _external_write(session, graph):
+        return reason
+    with with_subgraphs(await load_subgraphs(session, graph)):
+        issues = publish_issues(graph, level=level or "published")
+    if process := [i for i in issues if (i.code or "").startswith("process.")]:
+        return f"按流程使用不安全：{process[0].message.rstrip('。')}"
+    return None
+
+
+async def graph_external_write(session: AsyncSession, graph_json: Any) -> str | None:
+    """一张图里有没有调用工具节点会写外部系统（graph_side_effect 里的那一条）：有的交回一句原因，没有返回 None。
+
+    口径召回（ledger/recall）只用这一条：召回图是从口径卡往上派生的取数和口径卡，图里别处的人工节点、按流程使用的问题
+    碰不到它（细案做法 7 不重算的只有缺入口字段、数据部分不能复用、会写外部系统三种）；替人整张图重跑的补算、按口径出具
+    仍然走 graph_side_effect 全套。"""
+    from app.engine.schema import GraphSpec
+
+    try:
+        graph = GraphSpec.model_validate(graph_json)
+    except ValueError:
+        return "的图无法解析"
+    return await _external_write(session, graph)
+
+
+async def _external_write(session: AsyncSession, graph: Any) -> str | None:
+    from app.capabilities.registry import effective_effect, get_capability, writes_externally
+    from app.engine.schema import NodeType
+
     for node in graph.nodes:
         name = node.config.get("tool") if node.type == NodeType.TOOL else None
         if not isinstance(name, str) or not name:
@@ -138,10 +165,6 @@ async def graph_side_effect(session: AsyncSession, graph_json: Any, *, level: st
         cap = await get_capability(session, name)
         if cap is not None and writes_externally(cap.name, effective_effect(cap), cap.endpoint_id):
             return f"的「{node.title}」会写外部系统（{name}）"
-    with with_subgraphs(await load_subgraphs(session, graph)):
-        issues = publish_issues(graph, level=level or "published")
-    if process := [i for i in issues if (i.code or "").startswith("process.")]:
-        return f"按流程使用不安全：{process[0].message.rstrip('。')}"
     return None
 
 

backend/app/bundle/build.py

@@ -1197,6 +1197,10 @@ def _summary(etype: str, data: Any) -> str:
         items = data.get("items")
         count = len(items) if isinstance(items, list) else 0
         return f"依据的{kind or '东西'}有变" + (f"（{count} 项）" if count else "")
+    if etype == EventType.LEDGER_FEEDBACK:
+        # 对账结果上的反馈（next-round 02 做法 10）：只说是哪一种，不带谁给的
+        verdict = {"useful": "说明有用", "unclear": "看不懂"}.get(str(data.get("verdict") or ""))
+        return f"反馈：{verdict}" if verdict else "对账说明的反馈"
     if etype == EventType.LOG and isinstance(data.get("message"), str):
         return _clip(data["message"])
     return ""

backend/app/bundle/verifier.js

@@ -66,6 +66,7 @@
     'issuance': '出具判定', 'caliber.upgrade': '口径卡升版处置', 'catalog.drift': '数据目录变化',
     'schema.drift': '表结构变化', 'report.checked': '报告自查', 'evidence.judged': '按需裁判', 'file.extracted': '读入附件',
     'ledger.matched': '比对已发布口径', 'ledger.recalled': '口径召回核对', 'ledger.basis_changed': '答案的依据有变',
+    'ledger.feedback': '对账说明的反馈',
     'log': '日志', 'usage': '用量',
   }
   const T = {

backend/app/core/events.py

@@ -102,6 +102,10 @@ class EventType(StrEnum):
     # structure 用到的表或列被删、列的类型变了；values 触及有效记录条件的新码值；catalog_edit 处理变化时改了目录。带数据源、
     # 表、哪几项（路径和叫法），按新目录跑 SQL 检查新出现的错误级问题条数。同上，封存之后追加，不进封存范围
     LEDGER_BASIS_CHANGED = "ledger.basis_changed"
+    # 对账结果上的「说明有用 / 看不懂」（02 做法 10 的 E2，ledger/feedback）：追加在对账运行（origin_kind=reconcile）上，
+    # {verdict: useful | unclear, by, verified, source_run}。同一个人改主意再记一条，统计只算他最后一次。同上，封存之后
+    # 追加，是批注不是证据
+    LEDGER_FEEDBACK = "ledger.feedback"
 
     # 应用平台（docs/design/app-platform-and-look.md §3.11）
     # 运行来源：发起时写进 launch_run 的 notes，在封存范围内（engine/origin.py）

backend/app/ledger/feedback.py

@@ -0,0 +1,108 @@
+"""对账结果上的「说明有用 / 看不懂」（docs/design/next-round/02-口径账本.md 做法 10 的 E2）。
+
+对账（ledger/reconcile）把这次的答案和一份相关口径在同一份数据上各算一次，结果卡片上写差在哪、差多少。看的人觉得这段
+说明有没有帮他看懂，点一下「说明有用」或「看不懂」——这是 E2 里唯一要人给的数，衡量差异说明写得好不好。
+
+- 记在哪：对账运行上的封存后事件 ledger.feedback（{verdict, by, verified, source_run}，engine/run_append，post_seal）。
+  运行本身就是记录，不另建反馈表（同统计「全部现算、不另存」）；批注不是证据，封存核对照旧一致。
+- 谁能给：能读这个答案的人（同发起对账，ledger/reconcile：使用者对自己的答案也能对账）；对账要是这个答案的（origin.source_run），
+  而且已经跑完封存。
+- 怎么数：同一个人（开登录时按用户名；关登录时署名是自报的，按署名、没署名的算同一个）改主意再记一条，只算他最后一次。
+  账本接口 reconciled 每一项带上 {mine, useful, unclear}（feedback_of），统计 E2 数整个窗口（counts）。
+"""
+from __future__ import annotations
+
+from collections.abc import Collection, Iterable
+from typing import Any
+
+from sqlalchemy import select
+
+from app.core.events import EventType
+from app.db.base import SessionLocal
+from app.db.models import Run, RunEvent
+
+#: 两种反馈
+VERDICTS = ("useful", "unclear")
+#: 被拒的机器码
+REFUSED = "ledger_feedback_refused"
+NOT_FOUND = "这次对账不存在，或者不是这个答案的对账"
+NO_ACCESS = "你没有权限查看这个答案"
+NOT_SEALED = "对账还没运行完，等结果出来再评说明有没有用"
+BAD_VERDICT = "反馈只有两种：说明有用、看不懂"
+
+
+class FeedbackRefused(Exception):
+    def __init__(self, status_code: int, message: str, code: str = REFUSED) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.message = message
+        self.code = code
+
+
+def _who(principal: Any) -> str:
+    return str(getattr(principal, "username", None) or "")
+
+
+async def give(answer_id: str, reconcile_id: str, verdict: str, *, principal: Any) -> dict[str, Any]:
+    """记一条反馈（见模块说明），交回这次对账现在的反馈 {mine, useful, unclear}。不能记的抛 FeedbackRefused。"""
+    from app.api.deps import can_read_run
+    from app.engine.run_append import append_events
+    from app.ledger.reconcile import ORIGIN as RECONCILE
+
+    if verdict not in VERDICTS:
+        raise FeedbackRefused(422, BAD_VERDICT)
+    async with SessionLocal() as session:
+        answer = await session.get(Run, answer_id)
+        rec = await session.get(Run, reconcile_id)
+    origin = rec.origin if rec is not None and isinstance(rec.origin, dict) else {}
+    if answer is None or rec is None or rec.origin_kind != RECONCILE or origin.get("source_run") != answer_id:
+        raise FeedbackRefused(404, NOT_FOUND)
+    if getattr(principal, "mode", "off") != "off" and not await can_read_run(principal, answer):
+        raise FeedbackRefused(403, NO_ACCESS)
+    data = {"verdict": verdict, "by": _who(principal) or None, "verified": bool(getattr(principal, "verified", False)),
+            "source_run": answer_id}
+    if not await append_events(reconcile_id, [(EventType.LEDGER_FEEDBACK, None, data)], require="sealed"):
+        raise FeedbackRefused(409, NOT_SEALED)
+    async with SessionLocal() as session:
+        return (await feedback_of(session, [reconcile_id], principal))[reconcile_id]
+
+
+async def _latest(session: Any, run_ids: Collection[str]) -> dict[str, dict[str, str]]:
+    """{对账运行: {谁: 他最后一次给的}}。"""
+    out: dict[str, dict[str, str]] = {}
+    if not run_ids:
+        return out
+    rows = (await session.execute(select(RunEvent.run_id, RunEvent.data).where(
+        RunEvent.run_id.in_(list(run_ids)), RunEvent.type == EventType.LEDGER_FEEDBACK).order_by(RunEvent.seq))).all()
+    for run_id, data in rows:
+        if isinstance(data, dict) and data.get("verdict") in VERDICTS:
+            out.setdefault(run_id, {})[str(data.get("by") or "")] = str(data["verdict"])
+    return out
+
+
+async def feedback_of(session: Any, run_ids: Iterable[str], principal: Any) -> dict[str, dict[str, Any]]:
+    """每次对账的反馈：{对账运行: {mine: 看的这个人最后一次给的（没给过为 None）, useful: 几个人, unclear: 几个人}}。"""
+    ids = list(dict.fromkeys(run_ids))
+    latest = await _latest(session, ids)
+    me = _who(principal)
+    out: dict[str, dict[str, Any]] = {}
+    for run_id in ids:
+        votes = latest.get(run_id, {})
+        out[run_id] = {"mine": votes.get(me), "useful": sum(v == "useful" for v in votes.values()),
+                       "unclear": sum(v == "unclear" for v in votes.values())}
+    return out
+
+
+async def counts(session: Any, run_ids: Collection[str]) -> dict[str, Any]:
+    """统计 E2 的反馈（ledger/stats._e2）：这些对账运行上，说明有用、看不懂各几个（每人只算最后一次），给过反馈的对账几次，
+    有用的比率（分母为 0 时为 None）。"""
+    latest = await _latest(session, run_ids)
+    useful = sum(v == "useful" for votes in latest.values() for v in votes.values())
+    unclear = sum(v == "unclear" for votes in latest.values() for v in votes.values())
+    total = useful + unclear
+    return {"available": True, "useful": useful, "unclear": unclear, "runs": sum(1 for v in latest.values() if v),
+            "useful_rate": round(useful / total, 4) if total else None}
+
+
+__all__ = ["BAD_VERDICT", "NOT_FOUND", "NOT_SEALED", "REFUSED", "VERDICTS", "FeedbackRefused", "counts", "feedback_of",
+           "give"]

backend/app/ledger/recall.py

@@ -13,8 +13,10 @@
   metrics_from 指向召回卡；不加报告、不调模型）。两版在同一次运行里查同一份数据，差值只反映口径；数据随时间的变化另外拿
   原答案当时的值对照着说（task_payload 的 original）。
 - 不能重算的照实记「无法重算：原因」，不发起：口径的数据部分不能复用（derive.NotDerivable）、新版里没有这张口径卡、两版
-  没有同名的指标、新版入口要的参数这一组里没有或不合规（compose.check_inputs 同一套校验）、新版的图里有人工节点或会写外部
-  系统（apps/backfill.graph_side_effect，和按新口径补算、按口径出具同一个判断：替人重跑不代人做这些事）。
+  没有同名的指标、新版入口要的参数这一组里没有或不合规（compose.check_inputs 同一套校验）、新版的图里会写外部系统
+  （apps/backfill.graph_external_write，和按新口径补算、按口径出具同一个判断）。新版图里别处有人工节点、按流程使用不安全的
+  照样重算：召回图只有两版的取数和口径卡，碰不到它们（口径卡上游的人工节点 derive 已经认作数据部分不能复用）；「按新版
+  重新出具」走完整的正式运行，仍按 graph_side_effect 全套拒绝（ledger/issue.check_target）。
 - 上限：每次发布最多自动重算 ledger.recall_limit 组（设置，缺省 50；0 表示不自动重算，只列出受影响的答案）。超出的写
   「另有 M 组未重算」，在版本历史里点「接着算」手动接着发起下一批（每批同样的组数，上限是 0 时按缺省值）。
 - **计划从已有的召回运行重算**（照 apps/backfill 的写法，不建任务表）：一组算没算过，看有没有 origin 指向（W、V、N、
@@ -25,13 +27,14 @@
   不同就是数变了。数变了：给这一组答案的主人开 recall 待办（挂在召回运行上；开登录时每个主人一条，只列他的答案；关登录时
   一条派给任何人），推 ledger.recall 通知。数没变：只记账——在每个原答案的运行上追加 ledger.recalled（「已按 vN 核对，
   不影响」）。召回运行自己也追加一条 ledger.recalled 记结论（只记变没变、几个答案、几条待办，不记数），它同时是「收过尾」的
-  记号。召回运行没成功的照实记重算失败。
+  记号。召回运行没成功的、成功了却读不出召回卡（或缺了某个比较的旧值、新值）的照实记重算失败，原答案一个不动。
 
 **二、目录事实被驳回或修改**（只管影响数字的那一组：有效记录条件、去重、业务日期、关联、业务主键、度量类型，同
 data/catalog_review.FIELD_GROUP 的 facts）：审阅写入之后（data/catalog 的审阅、人工编辑、保存提案，catalog_review 的批量审阅）
 在后台找出查过这张表、而且查询快照冻结的目录里看得到这一项的探索运行，追加 ledger.basis_changed（kind=catalog），界面标
 「依据有变」；对这些运行执行过的 SQL 按新目录跑一遍 SQL 检查（data/sqlcheck，纯计算），比冻结目录下多出错误级问题的才开
-一条 recall 待办（reason=catalog，只有「知道了」）通知答案的主人（拍板 L7：不自动重问）。
+一条 recall 待办（reason=catalog，只有「知道了」）通知答案的主人（拍板 L7：不自动重问）。同一个数据源排着做、批量审阅的几张表
+合成一遍（catalog_pass）：查询快照按数据源在库里筛、凑够就停，冻结的表结构快照（大库的能有几 MB）同一份只读一次。
 
 **三、表结构变动**：消费 data/source_changes（list_changes(since_seq) 按 seq 游标拉，游标记在设置表里；探查、剖析任务结束、
 处理变化改了目录时触发）。只看 triggers_recall 认的：表或列被删、列的类型变了标「依据的表结构已变」（kind=structure）；
@@ -41,7 +44,11 @@ data/catalog_review.FIELD_GROUP 的 facts）：审阅写入之后（data/catalog
 recall 待办（reason=structure）通知发布人。
 
 **通知**：ledger.recall 只推口径升版、数真变了的那一种待办（消息照 notify/service 的规矩：标题、状态、条数、链接，不放
-新旧值和差值）；目录和表结构的两种待办照别的待办推 task.opened。
+新旧值和差值）；目录和表结构的两种待办照别的待办推 task.opened。全局渠道按一批待办（一次召回开给几个主人的那几条，
+batch_of_key）只推一次，个人渠道每个主人各一条（notify/service.notify_recall）。
+
+**答案上的标记**（answer_marks）：账本接口 GET /api/runs/{id}/ledger 的 recall——已按 vN 核对、依据有变、看的这个人手上说的
+是这个答案的 recall 待办，给问数据的轮次和证据面板「口径」一步。
 
 **自动的那几样**（发布后自动那一批、探查剖析之后拉表结构变化、启动时接着做）和调度循环同一个开关（automatic，
 AGENTLAB_SCHEDULER）：它们是在后台替人查业务库，测试、拷来的数据目录（检查脚本）不做；关着时版本历史里照样列出受影响的
@@ -95,6 +102,11 @@ NEW_INPUTS = "新版 v{version} 的入口要的参数这一组里没有或不合
 NEW_GONE = "新版本不存在，可能已被删除"
 LAUNCH_REFUSED = "没能发起召回运行：{reason}"
 RUN_FAILED = "重算失败：{reason}"
+#: 召回运行成功了，召回卡（或其中某个比较的旧值、新值）却读不出来：照实算重算失败，不当成数没变
+CARD_UNREADABLE = "读不出召回卡"
+#: 这一版已经不是当前发布版：不再接着算（新版本的召回本来就覆盖更早的所有版本，同 resume 只接着做当前发布版）
+SUPERSEDED = "已有更新的发布版 v{latest}，召回在 v{latest} 那一版接着看"
+NOT_CURRENT = "v{version} 已不是当前发布版，召回不再接着算"
 #: 收尾
 CHANGED_TITLE = "口径「{name}」升到 v{version}，答案的数变了"
 CATALOG_TITLE = "答案依据的数据目录有变：按新目录检查出了新的问题"
@@ -152,6 +164,8 @@ class Plan:
     workflow_id: str
     workflow_name: str
     new_version: int
+    #: 工作流现在的发布版（不是 new_version 的不再接着算，见 stale）
+    published_version: int | None = None
     groups: list[Group] = field(default_factory=list)
     #: 每个旧版本的口径卡对、旧版本自己的问题（不是口径、数据部分不能复用……）
     pairs: dict[int, list[Pair]] = field(default_factory=dict)
@@ -165,6 +179,15 @@ class Plan:
         """还没算、能算的组（按最近一次用到的先后）。"""
         return [g for g in self.groups if g.run is None and g.problem is None]
 
+    def stale(self) -> str | None:
+        """这一版已经不是当前发布版时不再发起（手动接着算也不）的原因：之后又发布了更新的版本，新版本的召回覆盖更早的所有
+        版本（同 resume 只接着做当前发布版），再按这一版开「口径升到 vN」的待办只是添乱。是当前发布版为 None。"""
+        if self.published_version == self.new_version:
+            return None
+        if self.published_version is not None and self.published_version > self.new_version:
+            return SUPERSEDED.format(latest=self.published_version)
+        return NOT_CURRENT.format(version=self.new_version)
+
 
 def _aware(moment: datetime | None) -> datetime | None:
     if moment is None:
@@ -212,19 +235,20 @@ async def recall_runs(session: Any, workflow_id: str, new_version: int | None =
 async def build_plan(session: Any, workflow_id: str, new_version: int) -> Plan:
     """W 升到 N 的召回计划（见模块说明）。只读，不发起。"""
     from app.apps import versions as app_versions
-    from app.apps.backfill import graph_side_effect
+    from app.apps.backfill import graph_external_write
     from app.ledger import compose, derive
 
     workflow = await session.get(Workflow, workflow_id)
     plan = Plan(workflow_id=workflow_id, workflow_name=workflow.name if workflow is not None else "",
-                new_version=int(new_version))
+                new_version=int(new_version),
+                published_version=workflow.published_version if workflow is not None else None)
     new_row = await app_versions.version_row(session, workflow_id, int(new_version)) if workflow else None
     if new_row is None:
         plan.problem = NEW_GONE
         return plan
     found = await ledger_uses.find_uses(session, workflow_id, below=int(new_version))
     plan.found = found
-    side = await graph_side_effect(session, new_row.graph, level=new_row.level)
+    side = await graph_external_write(session, new_row.graph)
     for version in found.versions():
         old_row = await app_versions.version_row(session, workflow_id, version)
         cards = source_cards(old_row.graph if old_row is not None else None)
@@ -405,7 +429,7 @@ async def launch(workflow_id: str, new_version: int, *, batch: str = "auto", act
         async with SessionLocal() as session:
             plan = await build_plan(session, workflow_id, int(new_version))
             limit = int((await ledger_settings(session))["recall_limit"])
-        if plan.problem is not None:
+        if plan.problem is not None or plan.stale() is not None:
             return []
         if batch == "auto":
             size = max(0, limit - plan.auto_launched)
@@ -496,7 +520,8 @@ def _group_state(group: Group) -> tuple[str, str | None]:
 
 async def batches(session: Any, workflow_id: str, version: int) -> dict[str, Any]:
     """版本历史里「这一版的召回」：每组的结论（数变了 / 没变 / 无法重算：原因 / 重算失败 / 还在算 / 未重算），「另有 M 组
-    未重算」和能不能接着算。接口只给搭建者（api/ledger 的 router）：参数原样给，同他在召回运行的输入里读到的一样；使用者
+    未重算」和能不能接着算（还有没算的组却不能接着算时 continue_reason 写为什么：之后又发布了更新的版本……）。
+    接口只给搭建者（api/ledger 的 router）：参数原样给，同他在召回运行的输入里读到的一样；使用者
     看的是 recall 待办（task_payload，按遮罩换过）。"""
     from app.api.settings import LEDGER_DEFAULTS, ledger_settings
 
@@ -526,10 +551,11 @@ async def batches(session: Any, workflow_id: str, version: int) -> dict[str, Any
                       "latest": group.latest.isoformat() if group.latest else None,
                       "state": state, **({"reason": reason} if reason else {}), **extra})
     pending = counts["pending"]
+    why_not = (plan.problem or plan.stale()) if pending > 0 else None
     return {"workflow_id": workflow_id, "workflow_name": plan.workflow_name, "version": int(version), "limit": limit,
             "batch_size": limit or int(LEDGER_DEFAULTS["recall_limit"]), "problem": plan.problem,
             "caliber": bool(plan.pairs) or bool(plan.groups), "counts": counts, "groups": items,
-            "remaining": pending, "can_continue": pending > 0 and plan.problem is None,
+            "remaining": pending, "can_continue": pending > 0 and why_not is None, "continue_reason": why_not,
             "truncated": bool(plan.found and plan.found.truncated)}
 
 
@@ -644,7 +670,14 @@ async def settle(run_id: str) -> str | None:
             await _mark(run_id, {**base, "outcome": "failed", "changed": None,
                                  "error": str(run.error or "运行没有成功结束")[:300]})
             return "failed"
-        results = compare(origin.get("metrics") or [], await card_values(run_id))
+        meta = origin.get("metrics") or []
+        values = await card_values(run_id)
+        # 读不出召回卡、少了哪个比较的旧值或新值：两边都是空，按渲染比会被当成「没变」，给每个原答案盖上「已核对，不影响」
+        # ——照实记重算失败，原答案一个不动（收过尾就不再核对，盖错了收不回来）
+        if not meta or any(f"{side}_{m.get('k')}" not in values for m in meta for side in ("old", "new")):
+            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)
         answers = list(dict.fromkeys(str(u.get("run_id")) for u in origin.get("uses") or [] if u.get("run_id")))
         tasks: list[str] = []
@@ -738,6 +771,13 @@ def _owner_of_key(key: str) -> tuple[str, str]:
     return (parts[-2], parts[-1]) if len(parts) >= 4 else ("anyone", "")
 
 
+def batch_of_key(key: str) -> str:
+    """correlation_key 去掉派给谁的那两段：同一次召回（同一个来由、同一次运行、同一条依据）给各个主人开的那一批待办是同一个
+    值。全局渠道按它只推一次（notify/service.notify_recall）。"""
+    parts = (key or "").split(":")
+    return ":".join(parts[:-2]) if len(parts) >= 4 else key or ""
+
+
 async def task_payload(session: Any, task: Any, principal: Any) -> dict[str, Any]:
     """recall 待办给处理人看的内容（待办箱、待办详情的 payload，process/tasks._describe 调）：旧口径的值、新口径的值、两版之差
     和原答案当时的值（每个都带出处：哪次运行、哪张卡、哪个指标，点得开），这一组的参数，能做的动作。使用者看到的值照样
@@ -896,6 +936,72 @@ def _cell(item: Mapping[str, Any], value: Any, text: Any, hide: Any, *, missing:
     return out
 
 
+# ==========================================================================
+# 答案上的召回标记（账本接口 GET /api/runs/{id}/ledger 的 recall，ledger/view.run_view）
+# ==========================================================================
+
+
+async def answer_marks(session: Any, run_id: str, principal: Any) -> dict[str, Any]:
+    """一个答案（探索运行、正式运行）身上的召回标记，给问数据的轮次和证据面板「口径」一步（细案做法 7：界面标「依据有变」、
+    「已按 vN 核对，不影响」）：
+
+    - checked：数没变、追加在这次运行上的 ledger.recalled（role=answer）——按哪份口径、从哪一版核对到哪一版、哪次召回、什么时候；
+    - basis：ledger.basis_changed——哪一种依据（目录事实、表结构、取值、处理时改了目录）、几项、按新目录新出现几条错误级问题、
+      什么时候；
+    - tasks：看的这个人手上还没处理的、说的是这个答案的 recall 待办（数变了的那种挂在召回运行上，列着这个答案；目录有变的
+      挂在这次运行上），界面链到待办箱。
+
+    不放数（新旧值、差值只在待办里给，按遮罩换过），不放表名、取值。"""
+    from app.process.tasks import Task, _assignments, handles
+
+    checked: list[dict[str, Any]] = []
+    basis: list[dict[str, Any]] = []
+    rows = (await session.execute(select(RunEvent.type, RunEvent.data, RunEvent.ts).where(
+        RunEvent.run_id == run_id, RunEvent.type.in_((EventType.LEDGER_RECALLED, EventType.LEDGER_BASIS_CHANGED)))
+        .order_by(RunEvent.seq))).all()
+    def at(ts: Any) -> str | None:
+        return datetime.fromtimestamp(ts, timezone.utc).isoformat(timespec="seconds") \
+            if isinstance(ts, (int, float)) else None
+
+    workflows: set[str] = set()
+    for etype, data, ts in rows:
+        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)})
+            workflows.add(str(data.get("workflow_id") or ""))
+        elif etype == EventType.LEDGER_BASIS_CHANGED:
+            items = data.get("items")
+            basis.append({"kind": data.get("kind"), "items": len(items) if isinstance(items, list) else 0,
+                          "new_errors": int(data.get("new_errors") or 0), "at": at(ts)})
+    names = dict((await session.execute(select(Workflow.id, Workflow.name).where(
+        Workflow.id.in_(sorted(w for w in workflows if w))))).all()) if workflows - {""} else {}
+    for item in checked:
+        item["workflow_name"] = names.get(str(item.get("workflow_id") or ""))
+    # 说的是这个答案、还没处理的 recall 待办：挂在这次运行上的（目录有变），和召回运行 origin.uses 里列着它的（数变了）。
+    # 运行 id 是 32 位十六进制，在 uses 的 JSON 文字里找子串不会误中，找到的再逐条核对
+    candidates = (await session.execute(
+        select(Task, Run.origin).join(Run, Run.id == Task.run_id)
+        .where(Task.kind == TASK_KIND, Task.status == "pending",
+               (Task.run_id == run_id) | ((Run.origin_kind == ORIGIN)
+                                          & (func.instr(func.json_extract(Run.origin, "$.uses"), run_id) > 0)))
+        .order_by(Task.created_at, Task.id))).all()
+    mine = await _assignments(session, [t.id for t, _ in candidates])
+    tasks: list[dict[str, Any]] = []
+    for task, origin in candidates:
+        key = task.correlation_key or ""
+        reason = key.split(":", 1)[0] if key.split(":", 1)[0] in REASONS else "caliber"
+        if task.run_id != run_id:
+            origin = origin if isinstance(origin, dict) else {}
+            if not any(u.get("run_id") == run_id for u in _uses_of(origin, _owner_of_key(key))):
+                continue
+        if not handles(mine.get(task.id, []), principal):
+            continue
+        tasks.append({"id": task.id, "reason": reason,
+                      **({"to_version": (origin or {}).get("to_version")} if reason == "caliber" else {})})
+    return {"checked": checked, "basis": basis, "tasks": tasks}
+
+
 # ==========================================================================
 # 「按新版重新出具」「知道了」
 # ==========================================================================
@@ -999,51 +1105,85 @@ def fact_changes(before: Mapping[str, Any] | None, after: Mapping[str, Any] | No
     return out
 
 
-async def _explore_snapshots(session: Any, source: Any, table: str, *, before: datetime | None = None
-                             ) -> list[tuple[str, str]]:
-    """查过 source 这张表的探索运行（计入统计的、成功封存的）和那条查询快照：[(运行, 快照)]，新的在前，最多 BASIS_MAX_RUNS
-    次运行。表名按 name_key 比，表已经不在表结构里也认（表被删了正要找用过它的）。"""
+#: 查询快照一页取几行（按运行结束时间从新到旧翻，每张表凑够 BASIS_MAX_RUNS 次运行就不再往下翻）
+SNAPSHOT_PAGE = 500
+
+
+async def _explore_snapshots(session: Any, source: Any, tables: Iterable[str], *, before: datetime | None = None
+                             ) -> dict[str, list[tuple[str, str]]]:
+    """查过 source 这几张表的探索运行（计入统计的、成功封存的）和那几条查询快照：{表: [(运行, 快照)]}，新的在前，每张表最多
+    BASIS_MAX_RUNS 次运行。表名按 name_key 比，表已经不在表结构里也认（表被删了正要找用过它的）。
+
+    数据源在库里就筛掉（引用的 meta 里记着源 id 和源名，query_snapshot_meta）：部署久了查询快照有几万条，别的源的不该读回来；
+    按运行结束时间、运行 id 排序分页翻（同一次运行的几条挨在一起），每张表都凑够了就停。表名不在库里筛：meta 里的表名是
+    SQL 里写的样子（带 schema 前缀、大小写不一），按 name_key 比只能在这里做。"""
+    from sqlalchemy import or_
+
     from app.data.catalog import QUERY_SNAPSHOT_KIND
     from app.data.names import name_key
     from app.db.models import Artifact
     from app.engine.origin import counted_runs
 
-    want = name_key(table)
+    want: dict[str, list[str]] = {}
+    for t in dict.fromkeys(tables):
+        want.setdefault(name_key(t), []).append(t)
+    out: dict[str, list[tuple[str, str]]] = {t: [] for names in want.values() for t in names}
+    runs: dict[str, set[str]] = {t: set() for t in out}
     stmt = (select(Artifact.id, Artifact.meta, Run.id, Run.finished_at).join(Run, Run.id == Artifact.run_id)
             .where(Artifact.kind == QUERY_SNAPSHOT_KIND, Run.run_class == "exploratory", Run.status == "succeeded",
-                   Run.manifest_hash.is_not(None), counted_runs()).order_by(Run.finished_at.desc()))
-    out: list[tuple[str, str]] = []
-    runs: set[str] = set()
+                   Run.manifest_hash.is_not(None), counted_runs(),
+                   or_(func.json_extract(Artifact.meta, "$.source_id") == source.id,
+                       func.json_extract(Artifact.meta, "$.source") == source.name))
+            .order_by(Run.finished_at.desc(), Run.id, Artifact.id))
     cut = _aware(before)
-    for artifact, meta, run_id, finished in (await session.execute(stmt)).all():
-        if cut is not None and (_aware(finished) is None or _aware(finished) > cut):
-            continue
-        meta = meta if isinstance(meta, dict) else {}
-        if meta.get("source_id") != source.id and meta.get("source") != source.name:
-            continue
-        if not any(name_key(str(t).rsplit(".", 1)[-1]) == want for t in meta.get("tables") or []):
-            continue
-        if run_id not in runs and len(runs) >= BASIS_MAX_RUNS:
-            continue
-        runs.add(run_id)
-        out.append((run_id, artifact))
+    offset = 0
+    last_run: str | None = None
+    while want:
+        page = (await session.execute(stmt.limit(SNAPSHOT_PAGE).offset(offset))).all()
+        offset += len(page)
+        for artifact, meta, run_id, finished in page:
+            if run_id != last_run and all(len(seen) >= BASIS_MAX_RUNS for seen in runs.values()):
+                return out
+            last_run = run_id
+            if cut is not None and (_aware(finished) is None or _aware(finished) > cut):
+                continue
+            meta = meta if isinstance(meta, dict) else {}
+            if meta.get("source_id") != source.id and meta.get("source") != source.name:
+                continue
+            hit = {name_key(str(t).rsplit(".", 1)[-1]) for t in meta.get("tables") or []}
+            for table in (t for key in hit & want.keys() for t in want[key]):
+                if run_id not in runs[table] and len(runs[table]) >= BASIS_MAX_RUNS:
+                    continue
+                runs[table].add(run_id)
+                # 翻页之间进来了新的运行，偏移会错开一两行：同一条不收两次
+                if (run_id, artifact) not in out[table]:
+                    out[table].append((run_id, artifact))
+        if len(page) < SNAPSHOT_PAGE:
+            break
     return out
 
 
-def _frozen_of(snapshot_id: str) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
-    """查询快照 → (快照内容, 它当时冻结的表结构快照)。读不出为 None。"""
+def _frozen_of(snapshot_id: str, cache: dict[str, Any] | None = None
+               ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
+    """查询快照 → (快照内容, 它当时冻结的表结构快照)。读不出为 None。cache 是这一遍里读过的工件（{工件 id: 内容}）：工件按内容
+    寻址、不会变，几十次运行冻结的常常是同一份表结构快照（大库的能有几 MB），同一遍里只读一次。"""
     from app.core.artifact_store import load
 
-    try:
-        snap = load(snapshot_id)
-    except ValueError:
-        return None, None
+    def read(artifact_id: str) -> Any:
+        if cache is not None and artifact_id in cache:
+            return cache[artifact_id]
+        try:
+            content = load(artifact_id)
+        except ValueError:
+            content = None
+        if cache is not None:
+            cache[artifact_id] = content
+        return content
+
+    snap = read(snapshot_id)
     if not isinstance(snap, dict):
         return None, None
-    try:
-        schema = load(str(snap.get("schema_artifact") or "")) if snap.get("schema_artifact") else None
-    except ValueError:
-        schema = None
+    schema = read(str(snap["schema_artifact"])) if snap.get("schema_artifact") else None
     return snap, schema if isinstance(schema, dict) else None
 
 
@@ -1096,28 +1236,77 @@ async def _answer_owner(session: Any, run_id: str) -> list[tuple[str, str]]:
     return use.owners
 
 
+@dataclass
+class CatalogEdit:
+    """一张表这一次影响数字的改动（目录审阅写入、新码值、处理变化时改了目录）：catalog_changed 一遍里可以有好几张表。"""
+
+    table: str
+    changes: Sequence[Mapping[str, Any]]
+    catalog_version: int | None = None
+    kind: str = "catalog"
+    change_id: str | None = None
+
+
 async def catalog_changed(source_id: str, table: str, changes: Sequence[Mapping[str, Any]], *,
                           catalog_version: int | None, kind: str = "catalog", change_id: str | None = None) -> int:
     """目录里这张表影响数字的几项被驳回或改了（见模块说明二）：冻结目录里看得到其中某项的探索运行追加 ledger.basis_changed；
-    按新目录多出错误级问题的开 recall 待办（reason=catalog）。返回标了几次运行。"""
+    按新目录多出错误级问题的开 recall 待办（reason=catalog）。返回标了几次运行。同一个数据源的排着做（_catalog_lock）。"""
+    return await catalog_pass(source_id, [CatalogEdit(table=table, changes=changes, catalog_version=catalog_version,
+                                                      kind=kind, change_id=change_id)])
+
+
+#: 同一个数据源的目录核对排着做：{事件循环: {数据源: 锁}}。批量审阅几十张表、几个人同时审阅，每一遍都要翻查询快照、读冻结
+#: 的表结构快照，同时跑几十遍会把单进程的服务拖住
+_catalog_locks: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]]" = \
+    weakref.WeakKeyDictionary()
+
+
+def _catalog_lock(source_id: str) -> asyncio.Lock:
+    return _catalog_locks.setdefault(asyncio.get_running_loop(), {}).setdefault(source_id, asyncio.Lock())
+
+
+async def catalog_pass(source_id: str, edits: Sequence[CatalogEdit]) -> int:
+    """一遍核对同一个数据源的几张表（catalog_changed、审阅之后排着的那一批）：查询快照只翻一次，冻结的表结构快照同一份只读
+    一次。返回标了几次运行。"""
+    async with _catalog_lock(source_id):
+        return await _catalog_pass(source_id, edits)
+
+
+async def _catalog_pass(source_id: str, edits: Sequence[CatalogEdit]) -> int:
     from app.data import catalog
     from app.db.models import DataSource
 
-    paths = [str(c.get("path")) for c in changes if c.get("path")]
-    if not paths and kind == "catalog":
+    # 目录审阅那一种得有改了的项（路径）才核对；新码值、处理时改了目录的没有逐项路径，照样核对
+    edits = [e for e in edits if e.kind != "catalog" or any(c.get("path") for c in e.changes)]
+    if not edits:
         return 0
     async with SessionLocal() as session:
         source = await session.get(DataSource, source_id)
         if source is None:
             return 0
-        found = await _explore_snapshots(session, source, table)
+        found = await _explore_snapshots(session, source, list(dict.fromkeys(e.table for e in edits)))
         current = {t: e.notes for t, e in (await catalog.read_catalog(session, source_id)).items()}
-    by_run: dict[str, list[tuple[dict[str, Any], dict[str, Any]]]] = {}
-    for run_id, snapshot in found:
-        snap, frozen = await asyncio.to_thread(_frozen_of, snapshot)
-        if snap is None or frozen is None:
-            continue
-        by_run.setdefault(run_id, []).append((snap, frozen))
+    cache: dict[str, Any] = {}
+    marked = 0
+    for edit in edits:
+        by_run: dict[str, list[tuple[dict[str, Any], dict[str, Any]]]] = {}
+        for run_id, snapshot in found.get(edit.table) or []:
+            snap, frozen = await asyncio.to_thread(_frozen_of, snapshot, cache)
+            if snap is None or frozen is None:
+                continue
+            by_run.setdefault(run_id, []).append((snap, frozen))
+        marked += await _mark_catalog(source, edit, by_run, current)
+    return marked
+
+
+async def _mark_catalog(source: Any, edit: CatalogEdit,
+                        by_run: Mapping[str, list[tuple[dict[str, Any], dict[str, Any]]]],
+                        current: Mapping[str, Mapping[str, Any]]) -> int:
+    """一张表的这次改动：冻结目录里看得到改动项的运行追加 ledger.basis_changed，按新目录多出错误级问题的开待办。"""
+    from app.engine.run_append import append_events
+
+    table, changes = edit.table, edit.changes
+    paths = [str(c.get("path")) for c in changes if c.get("path")]
     marked = 0
     labels = {str(c.get("path")): c for c in changes}
     for run_id, items in by_run.items():
@@ -1128,17 +1317,15 @@ async def catalog_changed(source_id: str, table: str, changes: Sequence[Mapping[
         for snap, frozen in items:
             if isinstance(snap.get("sql"), str):
                 errors += new_errors(snap["sql"], frozen, current, source.kind)
-        match = {"kind": kind, "source_id": source_id, "table": table,
-                 **({"catalog_version": catalog_version} if catalog_version is not None else {}),
-                 **({"change": change_id} if change_id else {})}
+        match = {"kind": edit.kind, "source_id": source.id, "table": table,
+                 **({"catalog_version": edit.catalog_version} if edit.catalog_version is not None else {}),
+                 **({"change": edit.change_id} if edit.change_id else {})}
         async with SessionLocal() as session:
             if await _basis_marked(session, run_id, match):
                 continue
         data = {**match, "source": source.name,
                 "items": [{"path": p, "label": labels.get(p, {}).get("label"), "change": labels.get(p, {}).get("change")}
                           for p in seen], "new_errors": len(errors), "checks": errors[:10]}
-        from app.engine.run_append import append_events
-
         written = await append_events(run_id, [(EventType.LEDGER_BASIS_CHANGED, None, data)], require="sealed")
         if not written:
             continue
@@ -1159,20 +1346,48 @@ async def _open_basis_task(run_id: str, seq: int) -> list[str]:
     return await open_tasks(run, reason="catalog", owners=owners, title=CATALOG_TITLE, extra=str(seq))
 
 
+#: 审阅之后排着的目录改动：{事件循环: {数据源: [CatalogEdit]}}。一个数据源一个后台任务（_drain_catalog），批量审阅逐表写完
+#: 逐表调 after_review，这期间（不让出事件循环）进来的几张表合成一遍做；做的时候又进来的，做完这一遍接着做
+_catalog_queues: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, list[CatalogEdit]]]" = \
+    weakref.WeakKeyDictionary()
+
+
 def after_review(source_id: str, table: str, before: Mapping[str, Any] | None, after: Mapping[str, Any] | None,
                  version: int | None) -> int:
-    """审阅写入之后（data/catalog、catalog_review 调）：有影响数字的项被驳回或改了，在后台标依据有变。返回改了几项。
-    出错只记日志：审阅已经写进去了。"""
+    """审阅写入之后（data/catalog、catalog_review 调）：有影响数字的项被驳回或改了，排进这个数据源的后台任务标依据有变（同一个
+    数据源的几张表合成一遍）。返回改了几项。出错只记日志：审阅已经写进去了。"""
     try:
         changes = fact_changes(before, after)
         if changes:
-            spawn(catalog_changed(source_id, table, changes, catalog_version=version))
+            queues = _catalog_queues.setdefault(asyncio.get_running_loop(), {})
+            edit = CatalogEdit(table=table, changes=changes, catalog_version=version)
+            if source_id in queues:
+                queues[source_id].append(edit)
+            else:
+                queues[source_id] = [edit]
+                spawn(_drain_catalog(source_id))
         return len(changes)
     except Exception:  # noqa: BLE001 - 辅助步骤
         logger.exception("审阅写入后排队标依据有变时出错（%s · %s）", source_id, table)
         return 0
 
 
+async def _drain_catalog(source_id: str) -> None:
+    """一个数据源排着的目录改动一遍一遍做完。一遍出错只记日志，接着做后面排进来的。"""
+    queues = _catalog_queues.setdefault(asyncio.get_running_loop(), {})
+    try:
+        while edits := queues.get(source_id):
+            queues[source_id] = []
+            try:
+                await catalog_pass(source_id, edits)
+            except asyncio.CancelledError:
+                raise
+            except Exception:  # noqa: BLE001 - 辅助步骤
+                logger.exception("审阅之后标依据有变时出错（%s）", source_id)
+    finally:
+        queues.pop(source_id, None)
+
+
 # ==========================================================================
 # 三、表结构变动
 # ==========================================================================
@@ -1222,7 +1437,8 @@ async def _mark_structure(change: Mapping[str, Any]) -> int:
             return 0
         detected = change.get("detected_at")
         at = datetime.fromisoformat(detected) if isinstance(detected, str) else None
-        found = await _explore_snapshots(session, source, str(change["table"]), before=at)
+        table = str(change["table"])
+        found = (await _explore_snapshots(session, source, [table], before=at)).get(table) or []
     marked = 0
     for run_id in dict.fromkeys(r for r, _ in found):
         match = {"kind": "structure", "change": change.get("id")}
@@ -1459,9 +1675,10 @@ async def shutdown() -> None:
     await asyncio.gather(*tasks, return_exceptions=True)
 
 
-__all__ = ["BASIS_KINDS", "BATCHES", "CURSOR_KEY", "OUTCOMES", "ORIGIN", "REASONS", "TASK_KIND", "Group", "Pair", "Plan",
-           "RecallRefused", "after_publish", "after_review", "automatic", "basis_kind", "batches", "build_graph", "build_plan",
-           "card_values", "catalog_changed", "compare", "consume_source_changes", "drain", "fact_changes", "launch",
-           "new_errors", "normalize", "open_tasks", "original_values", "outcome_of", "preview", "process_change",
-           "recall_card", "recall_runs", "reissue", "reissue_action", "resume", "schedule_settle", "settle", "shutdown",
+__all__ = ["BASIS_KINDS", "BATCHES", "CURSOR_KEY", "OUTCOMES", "ORIGIN", "REASONS", "TASK_KIND", "CatalogEdit",
+           "Group", "Pair", "Plan", "RecallRefused", "after_publish", "after_review", "answer_marks", "automatic",
+           "basis_kind", "batch_of_key", "batches", "build_graph", "build_plan", "card_values", "catalog_changed",
+           "catalog_pass", "compare", "consume_source_changes", "drain", "fact_changes", "launch", "new_errors",
+           "normalize", "open_tasks", "original_values", "outcome_of", "preview", "process_change", "recall_card",
+           "recall_runs", "reissue", "reissue_action", "resume", "schedule_settle", "settle", "shutdown",
            "source_cards", "spawn", "start_resume", "task_payload", "version_recall_counts"]

backend/app/ledger/stats.py

@@ -11,8 +11,8 @@
   的查询里各自的比率；证明等价的绑定是在哪一级配上的；没证明的原因（不可比、参数绑不上、配上了但不等价）；不等价的差异类型
   分布（time / filter / aggregate / join / other / parse，每份口径的差异每类只计一次）；逐个数的标签各几个、整题等价几次。
 - E2 对账：对账运行（origin=reconcile，第 3 波起由 ledger/reconcile 发起）发起、完成、失败各几次，对过账的答案几个，其中参数
-  经人核对过的、对照的指标由人选的各几次；相关口径的差异说明里兜底说明的占比。「说明有用 / 看不懂」的反馈要等对账面板
-  （第 4 波）记成对账运行上的封存后事件，这之前照实写没有数据。
+  经人核对过的、对照的指标由人选的各几次；相关口径的差异说明里兜底说明的占比；对账结果上「说明有用 / 看不懂」的反馈
+  （第 4 波，ledger/feedback：对账运行上的封存后事件 ledger.feedback，每人只算最后一次）各几个、有用的比率。
 - E3 重复度：问数据运行里执行过的查询按（数据源、normalize_sql 文本）分组，出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS
   段对话的算重复的形状，给出这样的形状几个、有重复形状的运行占几成；问题聚类（app/chat/questions：全站唯一的聚类，只取问题
   原文、不含 history、排除系统派生的运行）里 ≥ REPEAT_RUNS 个问题的类占几成。认证队列的转化（第 3 波，ledger/certify）：固化成
@@ -42,8 +42,6 @@ ASPECT_GROUP = {"time": "time", "time_range": "time", "filter": "filter", "aggre
                 "group_by": "aggregate", "join": "join", "tables": "join", "parse": "parse"}
 DIFF_GROUPS = ("time", "filter", "aggregate", "join", "other", "parse")
 STATUSES = ("proven", "composed", "related", "none", "skipped")
-#: 没有数据时的说法（这一块的功能还没做）
-NOT_YET_RECONCILE = "对账面板的反馈在第 4 波接上，目前没有数据"
 #: 聚类最多取几个问题（同治理接口的上限）
 QUESTION_LIMIT = 500
 
@@ -154,8 +152,11 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic
             stmt = stmt.where(Run.created_at >= scope["since"])
         if scope.get("until") is not None:
             stmt = stmt.where(Run.created_at < scope["until"])
+    from app.ledger import feedback as ledger_feedback
+
     by_status = {s: int(n) for s, n in (await session.execute(stmt.group_by(Run.status))).all()}
     origins = [o for (o,) in (await session.execute(stmt.with_only_columns(Run.origin))).all() if isinstance(o, dict)]
+    reconcile_ids = [r for (r,) in (await session.execute(stmt.with_only_columns(Run.id))).all()]
     answers = {o.get("source_run") for o in origins if o.get("source_run")}
     explained = fallback = 0
     for content in contents:
@@ -170,7 +171,7 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic
             "pairs_chosen": sum(1 for o in origins if o.get("pairs_chosen")),
             "explained": explained, "fallback": fallback,
             "fallback_rate": _ratio(fallback, explained),
-            "feedback": {"available": False, "note": NOT_YET_RECONCILE}}
+            "feedback": await ledger_feedback.counts(session, reconcile_ids)}
 
 
 #: 查询快照 → (数据源, 形状)。工件按内容寻址、不会变，记下来不会过期

backend/app/ledger/view.py

@@ -6,7 +6,11 @@ ledger 键、报告导出的片段面板，三处读同一份比对工件、按
 
 界面层不放逻辑：标签、整题判断、能做的动作都由这里给。actions（第 3 波）：整题等价时按那份口径正式出具（能不能、能不能
 自动、不能的原因，ledger/issue.plan_issue，按看的这个人的权限判）；有相关口径的查询看差多少（口径对账，ledger/reconcile）。
-issued 是这次运行按口径出具过的正式运行，reconciled 是对过的账（各自的运行、状态）。
+issued 是这次运行按口径出具过的正式运行，reconciled 是对过的账（各自的运行、状态，第 4 波起带上结果上的「说明有用 /
+看不懂」：看的这个人给过的、各几个，ledger/feedback）。
+
+recall（第 4 波，ledger/recall.answer_marks）：这个答案身上的召回标记——「已按 vN 核对，不影响」（checked）、依据有变
+（basis）、看的这个人手上说的是它的 recall 待办（tasks）。几种状态都带：没比对过、不比对的答案照样可能被标依据有变。
 """
 from __future__ import annotations
 
@@ -108,11 +112,16 @@ async def run_view(run_id: str) -> dict[str, Any]:
     from app.api.coded import CodedHTTPException
     from app.api.evidence import RUN_NOT_FOUND, _sealed
 
+    from app.api.deps import current_principal
+    from app.ledger.recall import answer_marks
+
     async with SessionLocal() as session:
         run = await session.get(Run, run_id)
-    if run is None:
-        raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND)
-    base: dict[str, Any] = {"run_id": run_id, "note": ledger_labels.POST_SEAL_NOTE}
+        if run is None:
+            raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND)
+        principal = current_principal()
+        marks = await answer_marks(session, run_id, principal)
+    base: dict[str, Any] = {"run_id": run_id, "note": ledger_labels.POST_SEAL_NOTE, "recall": marks}
     recorded = await store.latest(run_id)
     if recorded is None:
         why = ledger_match.ineligible(run)
@@ -125,7 +134,7 @@ async def run_view(run_id: str) -> dict[str, Any]:
         return {**base, "state": "pending"}
     if recorded.content is None:
         return {**base, "state": "unreadable", "reason": recorded.problem, **_stamp(recorded)}
-    from app.api.deps import current_principal
+    from app.ledger import feedback as ledger_feedback
     from app.ledger import issue as ledger_issue
     from app.ledger import reconcile as ledger_reconcile
 
@@ -133,12 +142,14 @@ async def run_view(run_id: str) -> dict[str, Any]:
     sealed = await _sealed(run_id)
     hider = await _hider(sealed)
     names = await _names(_workflow_ids(content))
-    principal = current_principal()
     plans = await ledger_issue.plans_for(run_id, content, principal=principal, hide=hider.hide,
                                          artifact=recorded.artifact, retro=recorded.retro)
     async with SessionLocal() as session:
         issued = [ledger_issue.issued_view(r) for r in await ledger_issue.issued_runs(session, run_id)]
         reconciled = [ledger_reconcile.reconciled_view(r) for r in await ledger_reconcile.reconciled_runs(session, run_id)]
+        # 对账结果上的「说明有用 / 看不懂」（ledger/feedback）：看的这个人给过哪一个、各几个
+        votes = await ledger_feedback.feedback_of(session, [r["run_id"] for r in reconciled], principal)
+        reconciled = [{**r, "feedback": votes[r["run_id"]]} for r in reconciled]
     return {
         **base, "state": "matched", **_stamp(recorded),
         "calibers": [{**{k: c.get(k) for k in ("caliber", "workflow_id", "version", "level", "graph_hash")},

backend/app/notify/service.py

@@ -347,7 +347,12 @@ async def notify_recall(task_id: str, users: list[str]) -> int:
     自己的个人渠道；然后照别的待办推 task.opened（已经推过 ledger.recall 的渠道不再推，同一个渠道只推一次）。目录、表结构变了
     的那两种只推 task.opened。
 
-    消息照模块说明的规矩（X6）：口径名（搭建时起的工作流名）、新旧版本号、这条待办列了几个答案、召回编号、去处理的链接；不放
+    全局渠道按这一批待办只推一次（同一次召回运行、同一个来由开的那几条，recall.batch_of_key）：开登录时一组答案有几个主人就
+    开几条待办，团队群却只该收到一条——只在这一批里最早开的那条（开出时间、id 排序，几条并发推也认同一条）上推全局渠道，
+    答案、待办按这一批汇总，链到待办箱（各人在那里看到派给自己的那条）；别的几条连 task.opened 也不再推全局渠道。个人渠道
+    照旧每个主人各一条，只数他的答案、链到他的那条待办。
+
+    消息照模块说明的规矩（X6）：口径名（搭建时起的工作流名）、新旧版本号、几个答案（几条待办）、召回编号、去处理的链接；不放
     旧口径的值、新口径的值、差值和原答案的值。返回推了几个渠道。"""
     from app.ledger import recall
 
@@ -361,26 +366,51 @@ async def notify_recall(task_id: str, users: list[str]) -> int:
         if task is None:
             return 0
         run = await session.get(Run, task.run_id)
-        reason = (task.correlation_key or "").split(":", 1)[0]
+        key = task.correlation_key or ""
+        reason = key.split(":", 1)[0]
+        batch = recall.batch_of_key(key)
+        siblings = [t for t, k in (await session.execute(
+            select(Task.id, Task.correlation_key).where(Task.run_id == task.run_id, Task.kind == recall.TASK_KIND)
+            .order_by(Task.created_at, Task.id))).all() if recall.batch_of_key(k or "") == batch]
+        first = not siblings or siblings[0] == task_id
         channels = await _channels(session, event, run, users) if reason == "caliber" else []
+        # 不是这一批最早的那条：全局渠道已经（或正由那一条）推过了，这条只推个人渠道，task.opened 也跳过全局渠道
+        shared = [] if first else list((await session.execute(select(NotifyChannel.id).where(
+            NotifyChannel.scope == "global"))).scalars())
+        channels = [c for c in channels if first or c.scope != "global"]
         origin = run.origin if run is not None and isinstance(run.origin, dict) else {}
         name = str(await session.scalar(select(Workflow.name).where(Workflow.id == origin.get("workflow_id")))
                    or "") if origin.get("workflow_id") else ""
         base = await _base_url(session) if channels else None
     pushed = 0
     if channels:
-        answers = len(recall._uses_of(origin, recall._owner_of_key(task.correlation_key or "")))
-        lines = [("状态", "待处理"), ("口径", f"「{name[:80]}」"),
-                 ("版本", f"v{origin.get('from_version')} → v{origin.get('to_version')}"), ("答案", f"{answers} 个"),
-                 ("召回", f"#{task.run_id[:6]}")]
-        message = {"event": event, "status": "pending", "ok": False,
-                   "title": recall.CHANGED_TITLE.format(name=name[:80] or "口径", version=origin.get("to_version")),
-                   "lines": lines, "links": [("去处理", f"{base}/apps/inbox/{task_id}")] if base else [],
-                   "subject": {"task_id": task_id, "run_id": task.run_id, "workflow_id": origin.get("workflow_id")}}
-        pushed = await _push(channels, message, subject_kind="task", subject_id=task_id, event=event,
-                             run_id=task.run_id, task_id=task_id)
+        head = [("状态", "待处理"), ("口径", f"「{name[:80]}」"),
+                ("版本", f"v{origin.get('from_version')} → v{origin.get('to_version')}")]
+        tail = [("召回", f"#{task.run_id[:6]}")]
+        mine = len(recall._uses_of(origin, recall._owner_of_key(key)))
+        everyone = len(recall._uses_of(origin, ("anyone", "")))
+
+        def message(answers: int, 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")),
+                    "lines": [*head, ("答案", f"{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")}}
+
+        own = ("去处理", f"{base}/apps/inbox/{task_id}")
+        group = [c for c in channels if c.scope == "global"]
+        if group:
+            # 这一批只有这一条（关登录时派给任何人）就链到它；几条的链到待办箱，各人看派给自己的那条
+            many = len(siblings) > 1
+            pushed += await _push(group, message(everyone, len(siblings), ("去待办箱", f"{base}/apps?tab=inbox")
+                                                 if many else own),
+                                  subject_kind="task", subject_id=task_id, event=event, run_id=task.run_id,
+                                  task_id=task_id)
+        if personal := [c for c in channels if c.scope != "global"]:
+            pushed += await _push(personal, message(mine, 1, own), subject_kind="task", subject_id=task_id,
+                                  event=event, run_id=task.run_id, task_id=task_id)
     return pushed + await notify_task(task_id, "task.opened", users=users,
-                                      exclude=frozenset(c.id for c in channels))
+                                      exclude=frozenset(c.id for c in channels) | frozenset(shared))
 
 
 async def notify_run(run_id: str, event: str) -> int:

backend/tests/test_ledger_api.py

@@ -233,7 +233,8 @@ async def test_stats_count_what_the_fixtures_did(client, monkeypatch):
     assert e1["with_candidates"] == 3 and e1["rates"]["proven"] == round(1 / 3, 4)
     assert e1["diff_aspects"]["filter"] == 1 and e1["labels"]["caliber"] == 1 and e1["labels"]["unknown"] == 1
     assert e1["whole_equivalent"] == 1
-    assert body["e2"]["launched"] == 0 and body["e2"]["feedback"]["available"] is False
+    assert body["e2"]["launched"] == 0 and body["e2"]["feedback"] == {"available": True, "useful": 0, "unclear": 0,
+                                                                    "runs": 0, "useful_rate": None}
     assert body["e3"]["chat_runs"] == 0 and body["e3"]["threshold"] == {"runs": 3, "conversations": 2}
     usage = {u["caliber"]: u for u in body["e4"]["usage"]}
     assert usage[f"{wf}@1"]["proven_runs"] == 1

backend/tests/test_ledger_feedback.py

@@ -0,0 +1,119 @@
+"""对账结果的「说明有用 / 看不懂」（next-round 02 做法 10 的 E2；app/ledger/feedback.py）。
+
+真的跑：虚构店铺库里的问数据答案（算上了退款）和受管口径「周销售」（只算已支付）对一次账，再在结果上给反馈：
+- 反馈记成对账运行上的封存后事件 ledger.feedback（批注不是证据，不进封存范围）；同一个人改主意只算最后一次；
+- 账本接口 reconciled 每一项带上反馈（看的这个人给过的、各几个），界面照着画选中的那一个；
+- 统计 E2 数到它（不再写「第 4 波接上」）；
+- 不认的反馈、不是这个答案的对账、还没跑完的对账照实拒绝。
+数据都是编的。
+"""
+from __future__ import annotations
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import select
+
+from app.core.events import EventType
+from app.db.base import SessionLocal
+from app.db.models import RunEvent
+from app.engine.runner import run_manager
+from app.ledger import match as ledger_match
+from app.main import app
+from tests import ledger_samples as S
+
+CHECKED = {"start": "2026-09-01", "end": "2026-09-08"}
+
+
+@pytest.fixture(autouse=True)
+async def engine_up(tmp_path):
+    await run_manager.setup()
+    path = tmp_path / "shop.db"
+    await S.fresh_source(path)
+    yield path
+    await ledger_match.drain()
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+async def reconciled(client, monkeypatch) -> tuple[str, str]:
+    """问一次（算上了退款）、按核对过的参数和「周销售」对一次账，等它跑完。返回 (答案, 对账运行)。"""
+    wf = await S.publish()
+    row = await S.explore(monkeypatch, S.WITH_REFUNDS)
+    body = (await client.get(f"/api/runs/{row.id}/ledger")).json()
+    [action] = [a for a in body["actions"] if a["kind"] == "reconcile"]
+    res = await client.post(f"/api/ledger/answers/{row.id}/reconcile",
+                            json={"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED,
+                                  "confirm": True})
+    assert res.status_code == 201, res.text
+    done = await S.wait(res.json()["run_id"])
+    assert done.status == "succeeded", done.error
+    return row.id, done.id
+
+
+async def test_feedback_is_a_post_seal_note_counted_by_e2(client, monkeypatch):
+    answer, rec = await reconciled(client, monkeypatch)
+    before = (await client.get("/api/ledger/stats")).json()["e2"]["feedback"]
+    assert before["available"] is True and "第 4 波" not in str(before)
+
+    res = await client.post(f"/api/ledger/answers/{answer}/reconcile/{rec}/feedback", json={"verdict": "useful"})
+    assert res.status_code == 200, res.text
+    assert res.json() == {"run_id": rec, "feedback": {"mine": "useful", "useful": 1, "unclear": 0}}
+    async with SessionLocal() as session:
+        [data] = (await session.execute(select(RunEvent.data).where(
+            RunEvent.run_id == rec, RunEvent.type == EventType.LEDGER_FEEDBACK))).scalars()
+    assert data["verdict"] == "useful" and data["post_seal"] is True and data["source_run"] == answer
+
+    # 改主意：同一个人只算最后一次
+    res = await client.post(f"/api/ledger/answers/{answer}/reconcile/{rec}/feedback", json={"verdict": "unclear"})
+    assert res.json()["feedback"] == {"mine": "unclear", "useful": 0, "unclear": 1}
+    [item] = (await client.get(f"/api/runs/{answer}/ledger")).json()["reconciled"]
+    assert item["run_id"] == rec and item["feedback"] == {"mine": "unclear", "useful": 0, "unclear": 1}
+
+    after = (await client.get("/api/ledger/stats")).json()["e2"]["feedback"]
+    assert after["useful"] - before["useful"] == 0 and after["unclear"] - before["unclear"] == 1
+    assert after["runs"] - before["runs"] == 1 and after["available"] is True
+
+
+async def test_feedback_refusals(client, monkeypatch):
+    answer, rec = await reconciled(client, monkeypatch)
+    bad = await client.post(f"/api/ledger/answers/{answer}/reconcile/{rec}/feedback", json={"verdict": "great"})
+    assert bad.status_code == 422
+    other = await S.explore(monkeypatch, S.ELSEWHERE, text="门店 [[v:Q1.r0.n]] 家。")
+    res = await client.post(f"/api/ledger/answers/{other.id}/reconcile/{rec}/feedback", json={"verdict": "useful"})
+    assert res.status_code == 404 and res.json()["code"] == "ledger_feedback_refused", "不是这个答案的对账"
+    res = await client.post(f"/api/ledger/answers/{answer}/reconcile/{answer}/feedback", json={"verdict": "useful"})
+    assert res.status_code == 404, "答案自己不是对账运行"
+    # 还没封存的对账（正在跑、刚发起）：没有结果可评
+    from app.db.models import Run
+
+    async with SessionLocal() as session:
+        row = await session.get(Run, rec)
+        sealed = row.manifest_hash
+        row.manifest_hash = None
+        await session.commit()
+    try:
+        res = await client.post(f"/api/ledger/answers/{answer}/reconcile/{rec}/feedback", json={"verdict": "useful"})
+        assert res.status_code == 409 and "还没运行完" in res.json()["detail"], res.text
+    finally:
+        async with SessionLocal() as session:
+            (await session.get(Run, rec)).manifest_hash = sealed
+            await session.commit()
+
+
+def test_offline_page_and_summary_know_the_feedback_event():
+    import re
+    from pathlib import Path
+
+    from app.bundle.build import _summary
+
+    script = (Path(__file__).resolve().parents[1] / "app" / "bundle" / "verifier.js").read_text(encoding="utf-8")
+    block = re.search(r"const EVENT_LABEL = \{(.*?)\n  \}", script, re.S)
+    labels = dict(re.findall(r"'([a-z_.]+)': '([^']+)'", block.group(1)))
+    assert labels["ledger.feedback"] == "对账说明的反馈"
+    assert _summary("ledger.feedback", {"verdict": "useful"}) == "反馈：说明有用"
+    assert _summary("ledger.feedback", {"verdict": "unclear"}) == "反馈：看不懂"

backend/tests/test_ledger_recall.py

@@ -7,7 +7,8 @@
 - 召回运行：派生图里新旧两版的取数和口径卡各一份（前缀不撞），召回卡里算两版之差；同一份数据上两版各算一次；
 - 收尾：数变了才开 recall 待办（挂在召回运行上，旧值、新值、差值、原答案当时的值都给），推 ledger.recall，消息里没有数；
   数没变的在原答案上记 ledger.recalled（已按 vN 核对，不影响），不开待办；
-- 不能重算的照实记：新版入口要的参数这一组里没有、图里有人工节点（有副作用）；
+- 不能重算的照实记：新版入口要的参数这一组里没有、图里会写外部系统；图里别处有人工节点的照样重算（召回图碰不到它），
+  只是「按新版重新出具」不代人审批；
 - 待办「按新版重新出具」走按口径出具同一套判法、「知道了」；
 - 目录里影响数字的一项被改了：冻结目录里看得到它的答案标「依据有变」，按新目录多出错误级问题的才开待办；
 - 表结构变了：用过这张表的答案标「依据的表结构已变」，口径试算算不出来才开待办；
@@ -146,6 +147,32 @@ async def test_limit_continue_and_idempotent_replanning(monkeypatch):
     assert await recall.launch(wf, 2, batch="manual") == []
 
 
+async def test_superseded_version_offers_no_continue(client, monkeypatch):
+    """v2 自动算了一批、还剩几组没算，接着又发布了 v3：v2 那一版不再给「接着算」（v3 的召回覆盖更早的所有版本，同启动时
+    接着做只认当前发布版），写明去 v3 那一版看；手动点也照样拒绝，不再开「口径升到 v2」的待办。"""
+    await ledger_settings(auto_issue=False, issue_daily_limit=100_000, recall_limit=1)
+    wf = await S.publish()
+    await answer(monkeypatch)
+    await answer(monkeypatch, "2026-09-08", "2026-09-15")
+    await publish_v2(wf)
+    assert len(await run_recall(wf)) == 1
+    async with SessionLocal() as session:
+        view = await recall.batches(session, wf, 2)
+    assert view["remaining"] == 1 and view["can_continue"] is True and view["continue_reason"] is None
+    await S.publish(graph=S.caliber_graph(WITH_REFUNDED), version=3, workflow_id=wf)
+    async with SessionLocal() as session:
+        view = await recall.batches(session, wf, 2)
+    assert view["remaining"] == 1 and view["can_continue"] is False
+    assert view["continue_reason"] == "已有更新的发布版 v3，召回在 v3 那一版接着看"
+    res = await client.post(f"/api/ledger/recall/{wf}/2/continue")
+    assert res.status_code == 409 and res.json()["detail"] == view["continue_reason"], res.text
+    assert await recall.launch(wf, 2, batch="manual") == [], "直接调也不再发起"
+    async with SessionLocal() as session:
+        assert len(await recall.recall_runs(session, wf, 2)) == 1
+        current = await recall.batches(session, wf, 3)
+    assert current["can_continue"] is True and current["continue_reason"] is None
+
+
 async def test_zero_limit_lists_but_does_not_recompute(monkeypatch):
     await ledger_settings(auto_issue=False, issue_daily_limit=100_000, recall_limit=0)
     wf = await S.publish()
@@ -189,6 +216,9 @@ async def test_changed_numbers_open_a_recall_task_with_all_three_values(client,
     assert await events(a1.id, EventType.LEDGER_RECALLED) == [], "数变了的不在原答案上记「不影响」"
     [task] = await recall_tasks(done.id)
     assert task.title == "口径「周销售」升到 v2，答案的数变了"
+    # 问数据的轮次、证据面板读的账本接口：这个答案有一条没处理的召回待办（界面链到待办箱），没有「已核对」
+    marks = (await client.get(f"/api/runs/{a1.id}/ledger")).json()["recall"]
+    assert marks == {"checked": [], "basis": [], "tasks": [{"id": task.id, "reason": "caliber", "to_version": 2}]}
     res = await client.get(f"/api/tasks/{task.id}")
     assert res.status_code == 200, res.text
     payload = res.json()["payload"]
@@ -220,11 +250,43 @@ async def test_unchanged_numbers_are_only_booked_on_the_answer(monkeypatch):
     [booked] = await events(a1.id, EventType.LEDGER_RECALLED)
     assert booked == {"role": "answer", "changed": False, "recall_run": done.id, "workflow_id": wf,
                       "from_version": 1, "to_version": 2, "post_seal": True}
+    # 账本接口带上「已按 v2 核对，不影响」：问数据的轮次、证据面板不用再去翻运行时间线
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        marks = (await c.get(f"/api/runs/{a1.id}/ledger")).json()["recall"]
+    [checked] = marks["checked"]
+    assert {k: checked[k] for k in ("workflow_id", "workflow_name", "from_version", "to_version", "changed",
+                                     "recall_run")} == {"workflow_id": wf, "workflow_name": "周销售", "from_version": 1,
+                                                        "to_version": 2, "changed": False, "recall_run": done.id}
+    assert checked["at"] and marks["basis"] == [] and marks["tasks"] == []
     # 再收一次尾不会重复记
     assert await recall.settle(done.id) is None
     assert len(await events(a1.id, EventType.LEDGER_RECALLED)) == 1
 
 
+@pytest.mark.parametrize("values", [{}, {"old_0": {"value": 1, "rendered": "1.00元"}}], ids=["none", "half"])
+async def test_unreadable_recall_card_is_a_failure_not_unchanged(monkeypatch, values):
+    """召回运行成功了、召回卡却读不出来（封存里找不到 recall 节点、读不出封存，或少了某个比较的旧值 / 新值）：照实记重算
+    失败，不当成「数没变」去给原答案盖「已核对，不影响」，也不开待办。"""
+    wf = await S.publish()
+    a1 = await answer(monkeypatch)
+    await publish_v2(wf)
+
+    async def unreadable(run_id):
+        return dict(values)
+
+    monkeypatch.setattr(recall, "card_values", unreadable)
+    [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"] == "failed" and settled["changed"] is None and settled["error"] == "读不出召回卡"
+    assert await events(a1.id, EventType.LEDGER_RECALLED) == [], "读不出的不在原答案上记「不影响」"
+    assert await recall_tasks(done.id) == []
+    async with SessionLocal() as session:
+        view = await recall.batches(session, wf, 2)
+    [group] = view["groups"]
+    assert group["state"] == "failed" and group["reason"] == "重算失败：读不出召回卡"
+
+
 async def test_incompatible_entry_and_side_effects_are_not_recomputed(monkeypatch):
     wf = await S.publish()
     await answer(monkeypatch)
@@ -240,14 +302,59 @@ async def test_incompatible_entry_and_side_effects_are_not_recomputed(monkeypatc
     assert view["counts"]["not_recomputable"] == 1 and view["groups"][0]["reason"] == group.problem
     assert view["can_continue"] is False
 
-    wf2 = await S.publish(name="周销售（要审批）")
+
+
+async def test_human_node_elsewhere_still_recomputes_but_reissue_refuses(client, monkeypatch):
+    """新版图里在口径卡下游有审批（取数 → 口径卡 → 报告 → 人工确认 → 出口）：召回图只有两版的取数和口径卡，人工节点碰不到，
+    照样重算、数变了照样开待办（做法 7 不重算的只有三种：缺入口字段、数据部分不能复用、会写外部系统）；「按新版重新出具」
+    走完整的正式运行，仍然不代人审批，照实写不能出具。"""
+    wf = await S.publish(name="周销售（要审批）")
     await answer(monkeypatch)
     human = S.caliber_graph(WITH_REFUNDED)
-    human["nodes"].insert(3, S.node("ok", "human", mode="approve", title="确认"))
-    await S.publish(graph=human, version=2, workflow_id=wf2)
+    human["nodes"].insert(4, S.node("ok", "human", mode="approve", title="确认"))
+    human["edges"] = [{"source": a["id"], "target": b["id"]} for a, b in zip(human["nodes"], human["nodes"][1:])]
+    await S.publish(graph=human, version=2, workflow_id=wf)
     async with SessionLocal() as session:
-        plan = await recall.build_plan(session, wf2, 2)
-    assert plan.groups and all("人工节点" in (g.problem or "") for g in plan.groups)
+        plan = await recall.build_plan(session, wf, 2)
+    assert plan.groups and all(g.problem is None for g in plan.groups), [g.problem for g in plan.groups]
+    assert plan.version_problem == {}
+    [done] = await run_recall(wf)
+    assert done.status == "succeeded", done.error
+    assert not [n for n in done.graph["nodes"] if n["type"] == "human"], "召回图里没有人工节点"
+    [task] = await recall_tasks(done.id)
+    payload = (await client.get(f"/api/tasks/{task.id}")).json()["payload"]
+    reissue = next(a for a in payload["actions"] if a["decision"] == "reissue")
+    assert reissue["available"] is False and "人工节点" in reissue["reason"], reissue
+    res = await client.post(f"/api/tasks/{task.id}/respond", json={"decision": "reissue"})
+    assert res.status_code == 409 and "人工节点" in res.json()["detail"], res.text
+
+
+async def test_writing_to_an_external_system_is_not_recomputed(monkeypatch):
+    """新版图里调用工具节点绑的能力会写外部系统（capabilities.registry.writes_externally，同按新口径补算、按口径出具）：
+    照实记「无法重算」，不发起。"""
+    from types import SimpleNamespace
+
+    from app.capabilities import registry
+
+    wf = await S.publish()
+    await answer(monkeypatch)
+    graph = S.caliber_graph(WITH_REFUNDED)
+    graph["nodes"].insert(4, S.node("pay", "tool", tool="erp_pay", args={}))
+    graph["edges"] = [{"source": a["id"], "target": b["id"]} for a, b in zip(graph["nodes"], graph["nodes"][1:])]
+    await publish_v2(wf, graph=graph)
+    real = registry.get_capability
+
+    async def get_capability(session, name):
+        if name == "erp_pay":
+            return SimpleNamespace(name=name, effect="write", effect_system="write", endpoint_id=None)
+        return await real(session, name)
+
+    monkeypatch.setattr(registry, "get_capability", get_capability)
+    async with SessionLocal() as session:
+        plan = await recall.build_plan(session, wf, 2)
+    assert plan.groups and all("会写外部系统（erp_pay）" in (g.problem or "") for g in plan.groups), \
+        [g.problem for g in plan.groups]
+    assert await recall.launch(wf, 2) == []
 
 
 async def test_plan_from_formal_runs_and_exporters_as_owners(monkeypatch):
@@ -362,6 +469,74 @@ async def test_recall_notification_has_no_numbers(monkeypatch):
     assert "周销售" in mine[0]["title"] and "v2" in mine[0]["title"]
 
 
+async def test_recall_pushes_the_shared_channel_once_per_recall_run(monkeypatch):
+    """开登录时一组答案有几个主人就开几条待办，但全局渠道（团队群）按召回运行只收一条（答案、待办都按这次召回汇总，链到
+    待办箱）；订阅了 task.opened 的全局渠道也不再按每条待办各推一次。每个主人自己的个人渠道各收他那一条。"""
+    from app.core.crypto import encrypt
+    from app.db.models import NotifyChannel, User
+    from app.notify import service
+
+    sent: list[dict] = []
+
+    async def capture(channel, message):
+        sent.append({"channel": channel.name, **message})
+        from app.notify.core import Outcome
+
+        return Outcome(True, 200, 1, None)
+
+    monkeypatch.setattr("app.notify.send", capture)
+    wf = await S.publish()
+    first = await answer(monkeypatch)
+    second = await answer(monkeypatch)                       # 同一周：同一组，两个主人
+    made: list[str] = []
+    async with SessionLocal() as session:
+        for name, run_id in (("analyst_a", first.id), ("analyst_b", second.id)):
+            user = User(username=name, role="user")
+            session.add(user)
+            await session.flush()
+            (await session.get(Run, run_id)).started_by = name
+            channel = NotifyChannel(name=f"个人-{name}", scope="user", owner_id=user.id, format="generic",
+                                    url=encrypt(f"https://hooks.example.com/{name}"), events=["ledger.recall"])
+            session.add(channel)
+            await session.flush()
+            made.append(channel.id)
+        for name, events_ in (("团队-召回", ["ledger.recall", "task.opened"]), ("团队-待办", ["task.opened"])):
+            channel = NotifyChannel(name=name, scope="global", url=encrypt("https://hooks.example.com/team"),
+                                    events=events_)
+            session.add(channel)
+            await session.flush()
+            made.append(channel.id)
+        await session.commit()
+    monkeypatch.setenv("AGENTLAB_AUTH_MODE", "local")
+    try:
+        await publish_v2(wf)
+        [done] = await run_recall(wf)
+        await service.drain()
+    finally:
+        async with SessionLocal() as session:
+            for channel_id in made:
+                await session.delete(await session.get(NotifyChannel, channel_id))
+            await session.commit()
+    tasks = await recall_tasks(done.id)
+    assert len(tasks) == 2, "每个主人一条待办"
+    team = [m for m in sent if m["channel"] == "团队-召回"]
+    assert [m["event"] for m in team] == ["ledger.recall"], [m["event"] for m in team]
+    lines = dict(team[0]["lines"])
+    assert lines["答案"] == "2 个" and lines["待办"] == "2 条", lines
+    assert [label for label, _ in team[0]["links"]] == ["去待办箱"] or team[0]["links"] == []
+    assert [m["event"] for m in sent if m["channel"] == "团队-待办"] == ["task.opened"], "只订阅了新待办的团队渠道也只收一条"
+    for name in ("analyst_a", "analyst_b"):
+        mine = [m for m in sent if m["channel"] == f"个人-{name}"]
+        assert [m["event"] for m in mine] == ["ledger.recall"] and dict(mine[0]["lines"])["答案"] == "1 个", mine
+    # 账本接口里的召回待办只列看的这个人手上的：第一个答案的主人看得到自己那条，另一个主人看不到
+    from app.api.deps import Principal
+
+    async with SessionLocal() as session:
+        own = await recall.answer_marks(session, first.id, Principal(username="analyst_a", role="user", mode="local"))
+        other = await recall.answer_marks(session, first.id, Principal(username="analyst_b", role="user", mode="local"))
+    assert len(own["tasks"]) == 1 and own["tasks"][0]["reason"] == "caliber" and other["tasks"] == []
+
+
 # --------------------------------------------------------------------------
 # 目录事实被改、表结构变了
 # --------------------------------------------------------------------------
@@ -411,6 +586,13 @@ async def test_catalog_fact_change_marks_answers_and_notifies_only_on_new_errors
     assert await events(unrelated.id, EventType.LEDGER_BASIS_CHANGED) == [], "没查过这张表的不标"
     [task] = await recall_tasks(row.id)
     assert task.correlation_key.startswith("catalog:")
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        marks = (await c.get(f"/api/runs/{row.id}/ledger")).json()["recall"]
+        quiet = (await c.get(f"/api/runs/{unrelated.id}/ledger")).json()["recall"]
+    [basis] = marks["basis"]
+    assert {k: basis[k] for k in ("kind", "items", "new_errors")} == {"kind": "catalog", "items": 1, "new_errors": 1}
+    assert basis["at"] and marks["tasks"] == [{"id": task.id, "reason": "catalog"}] and marks["checked"] == []
+    assert quiet == {"checked": [], "basis": [], "tasks": []}
     async with SessionLocal() as session:
         from app.api.deps import Principal
 
@@ -427,6 +609,89 @@ async def test_catalog_fact_change_marks_answers_and_notifies_only_on_new_errors
     assert len(await events(row.id, EventType.LEDGER_BASIS_CHANGED)) == 1
 
 
+async def test_catalog_edits_on_one_source_share_one_pass(engine_up, monkeypatch):
+    """批量审阅一口气改了同一个数据源的几张表：只在后台做一遍（同一个数据源排着做，几张表合在一次里查），查询快照按数据源
+    在库里筛，冻结的表结构快照（大库的能有几 MB）同一份只读一次。"""
+    from collections import Counter
+
+    from app.core import artifact_store
+    from app.data import catalog
+
+    source_id = await S.set_source(engine_up)
+    await _catalog(source_id, {"business_date": _item({"column": "paid_at"}),
+                               "columns": {"amount": {"measure": _item("flow")}}})
+    async with SessionLocal() as session:
+        await catalog.write_entry(session, source_id, "stores", {"valid_filter": _item("id > 0")}, if_version=0,
+                                  actor="tester")
+    week1 = await answer(monkeypatch)
+    week2 = await answer(monkeypatch, "2026-09-08", "2026-09-15")
+    shops = await S.explore(monkeypatch, S.ELSEWHERE, text="门店 [[v:Q1.r0.n]] 家。", origin="chat")
+    scans: list[list[str]] = []
+    loads: list[str] = []
+    real_scan, real_load = recall._explore_snapshots, artifact_store.load
+
+    async def scan(session, source, tables, **kw):
+        scans.append(sorted(tables))
+        return await real_scan(session, source, tables, **kw)
+
+    def load(artifact_id):
+        loads.append(artifact_id)
+        return real_load(artifact_id)
+
+    monkeypatch.setattr(recall, "_explore_snapshots", scan)
+    monkeypatch.setattr(artifact_store, "load", load)
+    async with SessionLocal() as session:
+        orders = await catalog.read_entry(session, source_id, "orders")
+        stores = await catalog.read_entry(session, source_id, "stores")
+    rejected = {**orders.notes, "business_date": {**orders.notes["business_date"], "status": "rejected"}}
+    # 批量审阅逐表写完逐表调审阅之后的钩子（catalog.after_review → recall.after_review），中间不让出事件循环
+    assert recall.after_review(source_id, "orders", orders.notes, rejected, orders.version + 1) == 1
+    assert recall.after_review(source_id, "stores", stores.notes,
+                               {"valid_filter": {**stores.notes["valid_filter"], "status": "rejected"}},
+                               stores.version + 1) == 1
+    await recall.drain()
+    assert scans == [["orders", "stores"]], "同一个数据源的几张表合成一遍"
+    for run in (week1, week2):
+        [marked] = await events(run.id, EventType.LEDGER_BASIS_CHANGED)
+        assert marked["table"] == "orders" and marked["items"][0]["path"] == "business_date"
+    [marked] = await events(shops.id, EventType.LEDGER_BASIS_CHANGED)
+    assert marked["table"] == "stores" and marked["items"][0]["path"] == "valid_filter"
+    schemas = set()
+    for snapshot in {a for a in loads}:
+        content = real_load(snapshot)
+        if isinstance(content, dict) and content.get("schema_artifact"):
+            schemas.add(content["schema_artifact"])
+    assert schemas and all(Counter(loads)[sid] == 1 for sid in schemas), Counter(loads)
+
+
+async def test_snapshot_scan_pages_newest_first_and_stops_when_full(engine_up, monkeypatch):
+    """翻查询快照：只取这个数据源的、新的在前，每张表凑够 BASIS_MAX_RUNS 次运行就不再往下翻。"""
+    from app.db.models import DataSource
+
+    source_id = await S.set_source(engine_up)
+    runs = [await answer(monkeypatch, f"2026-09-0{d}", "2026-09-08") for d in (1, 2, 3)]
+    pages: list[int] = []
+    monkeypatch.setattr(recall, "BASIS_MAX_RUNS", 2)
+    monkeypatch.setattr(recall, "SNAPSHOT_PAGE", 1)
+    async with SessionLocal() as session:
+        source = await session.get(DataSource, source_id)
+        real_execute = session.execute
+
+        async def counted(stmt, *args, **kwargs):
+            pages.append(1)
+            return await real_execute(stmt, *args, **kwargs)
+
+        monkeypatch.setattr(session, "execute", counted)
+        found = await recall._explore_snapshots(session, source, ["ORDERS"])
+        read = len(pages)
+        other = DataSource(id="elsewhere", name="elsewhere")
+        nothing = await recall._explore_snapshots(session, other, ["orders"])
+    assert [r for r, _ in found["ORDERS"]] == [runs[2].id, runs[1].id], "新的在前、凑够两次运行就停"
+    # 一页一行：第三页读到第三次运行时两次已经凑够，停下，不再翻第四页
+    assert read == 3, f"凑够了不再往下翻（翻了 {read} 页）"
+    assert nothing == {"orders": []}, "别的数据源的快照不读回来"
+
+
 def test_fact_changes_only_counts_items_that_affect_numbers():
     before = {"valid_filter": _item("status <> 'open'"), "label": _item("订单"),
               "columns": {"amount": {"measure": _item("flow"), "label": _item("金额")}}}