合入期 2 变化三分法（后端）：同一工作流两次运行的指标变化拆成数据、口径、模板与 SQL，口径与数据都变时能严格拆分两种效应，数据变化按维度归因，每项追到两边的查询快照；基准优先取同一定时任务的上一次

backend/app/api/compare.py

@@ -1,10 +1,175 @@
-"""运行对比：两次运行按「变化三分法」对照。
+"""运行对比：同一个工作流的两次运行按「变化三分法」对照——数据变了、口径变了、还是模板 / SQL 变了（engine/compare.py）。
 
-目前只登记了路由前缀，接口由运行对比的工作包补上。路径挂在 /api/runs/{run_id}/… 下面；
-这个路由登记在 runs.router 之前，以后加 /api/runs/compare 这类固定路径也不会被 /{run_id} 吃掉。
+GET /api/runs/{run_id}/compare/base：默认拿哪一次来比，{base_run_id, reason, base}。这次运行是定时任务发起的，取同一
+定时任务、结束得比它早的最近一次终态运行（reason: same_schedule）；不是、或者同一任务还没有更早的，取同一工作流、
+同一运行类别、结束得比它早的最近一次终态运行（reason: same_workflow）。只挑封存过的：没封存的拿来也比不了。
+找不到 404。
+
+GET /api/runs/{run_id}/compare?base={base_id}：{target, base, summary, template, metrics, queries, notes}，每个字段
+的形状见 engine/compare.py。两次运行都必须已结束、封存且封存核对一致，封存判定照搬证据接口（api/evidence._sealed）：
+没封存的运行事件还会变，封存核对不一致的事件被改过，拿它们比出来的结论都不作数，回 409（还没结束的和已经结束却没有
+封存清单的分开说：后者再等也等不来封存）。不是同一个工作流、或者是同一次运行，回 422。
+
+只读：两次运行都不追加事件、不落工件、不改运行记录；结果只交回给发请求的人。只认各自封存范围内的工件（证据接口的
+白名单，加上封存的节点产出和指标集里写着的完整值，同推演），遮罩列（查询当时记下的加数据源现在设的）的值不出现在
+结果里。读工件、比 SQL、重算都在线程里做：快照大的时候不挡住同一进程里别的请求。
+
+路径挂在 /api/runs/{run_id}/… 下面；这个路由登记在 runs.router 之前，以后加 /api/runs/compare 这类固定路径也不会
+被 /{run_id} 吃掉。定时任务结束后自动对比（期 4）直接调 default_base、compare_runs。
 """
 from __future__ import annotations
 
+import asyncio
+from collections.abc import Callable
+from typing import Any
+
 from fastapi import APIRouter
+from sqlalchemy import select
+
+from app.api.coded import CodedHTTPException
+from app.api.evidence import _TERMINAL, RUN_NOT_FOUND, _Masks, _Sealed, _sealed, _sealed_cards
+from app.api.whatif import _finished_before, _loader, _mask_of, _outputs_before, _returned
+from app.db.base import SessionLocal
+from app.db.models import DataSource, Run
+from app.engine import compare as engine
+from app.engine import recompute
 
 router = APIRouter(prefix="/api/runs", tags=["compare"])
+
+BAD_REQUEST = "compare_bad_request"
+MISMATCH = "compare_mismatch"
+UNSEALED = "compare_unsealed"
+SEAL_BROKEN = "compare_seal_broken"
+NO_BASE = "compare_no_base"
+
+BAD_REQUEST_TEXT = "请求格式有误：base 应为作为对比基准的运行 ID"
+BASE_NOT_FOUND = "作为对比基准的运行记录不存在"
+SAME_RUN = "不能和同一次运行自己对比，请选择另一次运行作为基准"
+UNSAVED = "{who}不属于任何已保存的工作流，无法确认两次运行是同一个工作流，不能对比"
+OTHER_WORKFLOW = "两次运行不属于同一个工作流（「{target}」与「{base}」），不能对比"
+UNSEALED_TEXT = "{who}尚未结束并封存：对比只用封存的运行记录，运行结束后才能对比"
+#: 已经结束、只是没有封存清单（封存上线前结束的、结束时没封上的）：再等也等不来封存，不能说「运行结束后才能对比」
+NO_MANIFEST_TEXT = "{who}没有封存清单，不能用于对比"
+SEAL_BROKEN_TEXT = "{who}的封存核对不一致：封存后有事件被修改、删除或插入，不能用于对比。可在记录页点「核对」查看核对结果"
+NO_BASE_TEXT = ("没有找到可以对比的上一次运行：需要同一工作流、同一运行类别（定时任务发起的优先找同一定时任务）、"
+                "更早结束并已封存的运行")
+TARGET = "这次运行"
+BASE = "作为基准的运行"
+
+
+@router.get("/{run_id}/compare/base")
+async def compare_base(run_id: str) -> dict[str, Any]:
+    """默认拿来对比的那一次运行：{base_run_id, reason: same_schedule | same_workflow, base}。"""
+    found = await default_base(run_id)
+    if found is None:
+        raise CodedHTTPException(404, NO_BASE_TEXT, NO_BASE)
+    run, reason = found
+    return {"base_run_id": run.id, "reason": reason, "base": run_ref(run)}
+
+
+@router.get("/{run_id}/compare")
+async def compare_with(run_id: str, base: str | None = None) -> dict[str, Any]:
+    """这次运行和 base 那一次的对比，形状见 engine/compare.py。"""
+    if base is None or not base.strip():
+        raise CodedHTTPException(422, BAD_REQUEST_TEXT, BAD_REQUEST)
+    return await compare_runs(run_id, base.strip())
+
+
+async def default_base(run_id: str) -> tuple[Run, str] | None:
+    """(默认的基准运行, 怎么选的)；找不到返回 None，这次运行本身不存在 404。
+
+    「早」按结束时间比：这次运行还没结束（还在跑、停在审批）的按开始时间。只挑终态、封存过的运行——没封存的拿来
+    对比只会得到 409。定时任务发起的先找同一定时任务（同一个工作流）的；它是这个任务的第一次时，退回同一工作流、
+    同一运行类别。没挂在已保存的工作流上的运行（画布上没保存就运行的图）没有可比的。"""
+    async with SessionLocal() as session:
+        run = await session.get(Run, run_id)
+        if run is None:
+            raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND)
+        if not run.workflow_id:
+            return None
+        before = run.finished_at or run.started_at or run.created_at
+        query = (select(Run)
+                 .where(Run.id != run.id, Run.workflow_id == run.workflow_id, Run.status.in_(_TERMINAL),
+                        Run.manifest_hash.is_not(None), Run.finished_at.is_not(None), Run.finished_at < before)
+                 .order_by(Run.finished_at.desc(), Run.created_at.desc()).limit(1))
+        if run.schedule_id:
+            found = (await session.execute(query.where(Run.schedule_id == run.schedule_id))).scalars().first()
+            if found is not None:
+                return found, "same_schedule"
+        found = (await session.execute(query.where(Run.run_class == run.run_class))).scalars().first()
+    return (found, "same_workflow") if found is not None else None
+
+
+async def compare_runs(run_id: str, base_id: str) -> dict[str, Any]:
+    """对比两次运行（run_id 是这次，base_id 是基准）。拒绝时抛 CodedHTTPException：运行不存在 404，同一次运行、
+    不是同一个工作流 422，没结束封存、封存核对不一致 409。"""
+    async with SessionLocal() as session:
+        target = await session.get(Run, run_id)
+        base = await session.get(Run, base_id)
+        kinds = {str(name): kind for name, kind in (await session.execute(select(DataSource.name, DataSource.kind)))}
+    if target is None:
+        raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND)
+    if base is None:
+        raise CodedHTTPException(404, BASE_NOT_FOUND, RUN_NOT_FOUND)
+    if base.id == target.id:
+        raise CodedHTTPException(422, SAME_RUN, MISMATCH)
+    for run, who in ((target, TARGET), (base, BASE)):
+        if not run.workflow_id:
+            raise CodedHTTPException(422, UNSAVED.format(who=who), MISMATCH)
+    if target.workflow_id != base.workflow_id:
+        raise CodedHTTPException(422, OTHER_WORKFLOW.format(target=target.workflow_name, base=base.workflow_name),
+                                 MISMATCH)
+    sealed = []
+    for run, who in ((target, TARGET), (base, BASE)):
+        found = await _sealed(run.id)
+        if found.status not in _TERMINAL:
+            raise CodedHTTPException(409, UNSEALED_TEXT.format(who=who), UNSEALED)
+        if not found.sealed:
+            raise CodedHTTPException(409, NO_MANIFEST_TEXT.format(who=who), UNSEALED)
+        if not found.trusted:
+            raise CodedHTTPException(409, SEAL_BROKEN_TEXT.format(who=who), SEAL_BROKEN)
+        sealed.append(found)
+    masks = _Masks()
+    target_side = await _side(target, sealed[0], masks, kinds)
+    base_side = await _side(base, sealed[1], masks, kinds)
+    return await asyncio.to_thread(engine.compare, base_side, target_side)
+
+
+def run_ref(run: Run) -> dict[str, Any]:
+    """一次运行的概要（RunRef）：对比结果和默认基准里都用这一个形状。时间带时区（UTC）。"""
+    return {"id": run.id, "workflow_name": run.workflow_name, "version": run.version, "run_class": run.run_class,
+            "started_at": run.started_at.isoformat() if run.started_at else None,
+            "finished_at": run.finished_at.isoformat() if run.finished_at else None,
+            "schedule_id": run.schedule_id}
+
+
+async def _side(run: Run, sealed: _Sealed, masks: _Masks, kinds: dict[str, Any]) -> engine.Side:
+    """一次运行交给对比的东西（engine.compare.Side）：封存范围内的事件、复验过哈希的口径卡、只认封存范围内工件的
+    读法、每份查询快照要遮的列，以及按口径卡那次执行还原输入用的 Sources。"""
+    cards = await asyncio.to_thread(_sealed_cards, sealed)
+    loader = _loader(sealed, cards)
+    # 遮罩要查库（数据源现在设的），在这里每份快照算一遍交进去，对比本身在线程里不碰库。快照先在线程里读好，
+    # 同一个 loader 记着读过的，对比时不再读盘
+    snaps = await asyncio.to_thread(_read_all, loader, list(sealed.queries))
+    masked = {artifact: frozenset(await _mask_of(snaps.get(artifact), info, masks))
+              for artifact, info in sealed.queries.items()}
+    returned = _returned(sealed)
+
+    def sources(started: int) -> recompute.Sources:
+        return recompute.Sources(load=loader, queries=tuple(sealed.queries), outputs=_outputs_before(sealed, started),
+                                 masked=masked, returned=returned, finished=_finished_before(sealed, started))
+
+    return engine.Side(run=run_ref(run), graph=sealed.graph, events=sealed.events, cards=cards, load=loader,
+                       masked=masked, sources=sources, dialects=kinds, inputs=dict(run.input or {}))
+
+
+def _read_all(loader: Callable[[str], Any], artifacts: list[str]) -> dict[str, Any]:
+    """读出这几份查询快照：读不出来、被改过的记 None（遮罩照样按台账记的数据源算）。"""
+    out: dict[str, Any] = {}
+    for artifact in artifacts:
+        try:
+            content = loader(artifact)
+        except ValueError:
+            content = None
+        out[artifact] = content if isinstance(content, dict) else None
+    return out

backend/tests/fixtures/compare_runs.py

@@ -0,0 +1,250 @@
+"""运行对比的虚构运行（tests/test_compare_engine.py、tests/test_compare_api.py 共用）。
+
+两次周报运行照真实运行的事件形状手写（区域、金额、日期都是编的）：取数节点 fetch 查一条按区域汇总的 SQL，Agent 节点
+bot 自己再查一次、把各区域的金额作为数组字段 amounts 交给口径卡，口径卡算几个指标。工件按内容寻址放在字典里，
+接口测试把它们原样落进工件库、事件落进运行记录。
+"""
+from __future__ import annotations
+
+import json
+
+from app.core.artifact_store import canonical_json, content_hash
+from app.engine import compare, recompute
+from app.engine.evidence import render_number
+from app.engine.expressions import parse_expression, substitute
+from app.engine.merge_query import MERGE_SOURCE
+
+TOOL = "db_query__shop"
+WEEK36 = ("2026-08-31", "2026-09-06")
+WEEK37 = ("2026-09-07", "2026-09-13")
+COLUMNS = ["region", "gmv", "refund"]
+TYPES = {"region": "text", "gmv": "number", "refund": "number"}
+BASE_ROWS = [["东区", 1200.5, 100.0], ["南区", 800.0, 50.0], ["西区", 600.0, 0.0]]
+NEXT_ROWS = [["东区", 1500.5, 120.0], ["南区", 700.0, 40.0], ["西区", 600.0, 10.0]]
+EAST = "cell(nodes.fetch, 0, 'gmv')"
+REFUND = "cell(nodes.fetch, 0, 'refund')"
+
+
+def region_sql(days=WEEK36, extra=""):
+    return ("SELECT region, SUM(amount) AS gmv, SUM(refund) AS refund FROM orders "
+            f"WHERE day BETWEEN '{days[0]}' AND '{days[1]}'{extra} GROUP BY region ORDER BY region")
+
+
+def node(nid, ntype, label, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label, "config": config}}
+
+
+def chain(*nodes):
+    return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
+def weekly_graph(sql=None, *, between=(), after_fetch=(), card_label="周报口径卡"):
+    """一条链：开始 → 取数 → [after_fetch] → Agent → [between] → 口径卡 → 写周报 → 成果。after_fetch 里的节点不在
+    Agent 的下游（它的上游只有取数节点），between 里的在。"""
+    return chain(
+        node("start", "input", "开始"),
+        node("fetch", "tool", "区域取数", tool=TOOL, args={"sql": sql or region_sql()}),
+        *after_fetch,
+        node("bot", "agent", "区域分析", prompt="按区域列出金额", tools=[TOOL], cite_fields=True),
+        *between,
+        node("card", "metrics", card_label, caliber="周报口径", caliber_version="v1"),
+        node("write", "report", "写周报"),
+        node("out", "output", "成果"),
+    )
+
+
+class Fake:
+    """一次虚构的运行：工件按内容寻址放在字典里，事件照真实运行的形状写（tool.start / tool.end / node.finished
+    的证据台账、口径卡的 metric_set 条目、节点产出工件）。"""
+
+    def __init__(self, run_id, *, graph=None, version=3, inputs=None, masked=None, source="shop"):
+        self.id = run_id
+        self.source = source
+        self.graph = graph or weekly_graph()
+        self.version = version
+        self.inputs = inputs or {}
+        self.masked = masked or {}
+        self.store: dict[str, object] = {}
+        self.tampered: set[str] = set()
+        self.events: list[tuple[int, str, str | None, dict]] = []
+        self.cards: list[tuple[str, str, dict]] = []
+        self.queries: list[str] = []
+        self.elapsed = 0
+        self.add("run.started", None, {"nodes": len(self.graph["nodes"])})
+
+    def put(self, obj):
+        text = canonical_json(obj)
+        artifact = content_hash(text)
+        self.store[artifact] = json.loads(text)
+        return artifact
+
+    def load(self, artifact):
+        if artifact in self.tampered:
+            raise ValueError("工件内容与哈希不符，疑似被篡改")
+        return self.store.get(artifact)
+
+    def add(self, etype, node_id, data):
+        self.events.append((len(self.events) + 1, etype, node_id, data))
+
+    def snapshot(self, sql, columns, rows, *, truncated=False, types=None, masks=None):
+        self.elapsed += 1      # 真实快照带着耗时：同一条 SQL 查两次也是两件工件
+        snap = {"columns": columns, "rows": rows, "row_count": len(rows), "truncated": truncated,
+                "elapsed_ms": self.elapsed, "sql": sql, "source": self.source,
+                **({"column_types": types} if types else {}), **({"mask_columns": masks} if masks else {})}
+        artifact = self.put(snap)
+        self.queries.append(artifact)
+        return artifact
+
+    def entry(self, node_id, artifact, columns, rows, *, tool=TOOL, call_id=None, truncated=False):
+        return {"kind": "query", "node_id": node_id, "exec": 1, "artifact": artifact, "tool": tool,
+                "source": self.source,
+                "columns": columns, "rows": len(rows), "truncated": truncated,
+                **({"call_id": call_id} if call_id else {})}
+
+    def tool(self, node_id, sql, columns=COLUMNS, rows=BASE_ROWS, *, tool=TOOL, truncated=False, types=TYPES,
+             masks=None):
+        """调用工具节点查一次库。masks 是查询当时数据源设的遮罩（记进快照的 mask_columns）。"""
+        artifact = self.snapshot(sql, columns, rows, truncated=truncated, types=types, masks=masks)
+        output = self.put({"columns": columns, "rows": rows, "sql": sql, "source": self.source, "artifact": artifact})
+        self.add("node.started", node_id, {"node_type": "tool"})
+        self.add("tool.start", node_id, {"tool": tool, "args": {"sql": sql}})
+        self.add("tool.end", node_id, {"tool": tool, "duration_ms": 3, "query_artifact": artifact})
+        self.add("node.finished", node_id, {"duration_ms": 4, "attempt": 1, "artifact": output, "evidence": [
+            self.entry(node_id, artifact, columns, rows, tool=tool, truncated=truncated)]})
+        return artifact
+
+    def agent(self, node_id, sql, columns=COLUMNS, rows=BASE_ROWS, *, types=TYPES, truncated=False, masks=None,
+              verify=(), column="gmv", span=None):
+        """Agent 节点自己查一次库，按出处交回数组字段 amounts（默认是快照 gmv 列的全部行；column、span 改引用的列和
+        行段）。verify 是同一节点里核查类的查询（role: verify，不进台账）。"""
+        artifact = self.snapshot(sql, columns, rows, truncated=truncated, types=types, masks=masks)
+        first, last = span or (0, len(rows) - 1)
+        at = columns.index(column)
+        data = {"amounts": [r[at] for r in rows[first:last + 1]]}
+        evidence = {"amounts": {"ref": f"Q1.r{first}-{last}.{column}", "call": "Q1", "artifact": artifact,
+                                "locator": {"rows": [first, last], "column": column}, "status": "verified"}}
+        output = self.put({"text": "按区域列出了金额", "data": data, "data_evidence": evidence})
+        self.add("node.started", node_id, {"node_type": "agent"})
+        self.add("tool.start", node_id, {"tool": TOOL, "args": {"sql": sql}, "call_id": "c1"})
+        self.add("tool.end", node_id, {"tool": TOOL, "call_id": "c1", "duration_ms": 5, "query_artifact": artifact})
+        for i, extra in enumerate(verify):
+            check = self.snapshot(extra, ["n"], [[1]])
+            self.add("tool.end", node_id, {"tool": TOOL, "call_id": f"v{i}", "role": "verify", "query_artifact": check})
+        self.add("node.finished", node_id, {"duration_ms": 9, "attempt": 1, "artifact": output, "evidence": [
+            self.entry(node_id, artifact, columns, rows, call_id="c1", truncated=truncated)]})
+        return artifact
+
+    def plain_agent(self, node_id, sqls, data, columns=COLUMNS, rows=BASE_ROWS, *, types=TYPES):
+        """没开按出处核对的 Agent：查几次库，交回的 data 没有出处（data_evidence），口径卡只知道是它写的。"""
+        self.add("node.started", node_id, {"node_type": "agent"})
+        entries = []
+        for i, sql in enumerate(sqls):
+            artifact = self.snapshot(sql, columns, rows, types=types)
+            self.add("tool.start", node_id, {"tool": TOOL, "args": {"sql": sql}, "call_id": f"c{i}"})
+            self.add("tool.end", node_id, {"tool": TOOL, "call_id": f"c{i}", "duration_ms": 5,
+                                           "query_artifact": artifact})
+            entries.append(self.entry(node_id, artifact, columns, rows, call_id=f"c{i}"))
+        output = self.put({"text": "按区域分析了金额", "data": data})
+        self.add("node.finished", node_id, {"duration_ms": 9, "attempt": 1, "artifact": output, "evidence": entries})
+
+    def merge(self, node_id, sql, inputs, columns, rows, *, types=None):
+        """合并查询：几份上游快照按别名落进内存库，再执行一条合并 SQL。快照里记着各个输入（别名、节点、快照），
+        事件照合并节点的形状写（merge.end 加 node.finished 台账里的 query 条目，没有 tool.end）。
+        inputs 是 [(别名, 上游节点, 上游快照)]。"""
+        snap = {"columns": columns, "rows": rows, "row_count": len(rows), "truncated": False, "sql": sql,
+                "source": MERGE_SOURCE, "sources": [self.source],
+                "inputs": [{"alias": alias, "node_id": upstream, "artifact": artifact,
+                            "rows": len(self.store[artifact]["rows"]), "columns": self.store[artifact]["columns"],
+                            "source": self.store[artifact].get("source")} for alias, upstream, artifact in inputs],
+                "warnings": [], **({"column_types": types} if types else {})}
+        artifact = self.put(snap)
+        self.queries.append(artifact)
+        output = self.put({**snap, "artifact": artifact})
+        view = [{k: v for k, v in item.items() if k != "columns"} for item in snap["inputs"]]
+        self.add("node.started", node_id, {"node_type": "merge"})
+        self.add("merge.end", node_id, {"inputs": view, "sql": sql, "rows": len(rows), "query_artifact": artifact})
+        self.add("node.finished", node_id, {"duration_ms": 2, "attempt": 1, "artifact": output, "evidence": [
+            {"kind": "query", "node_id": node_id, "exec": 1, "artifact": artifact, "source": MERGE_SOURCE,
+             "columns": columns, "rows": len(rows), "truncated": False}]})
+        return artifact
+
+    def code(self, node_id, value, var):
+        """沙箱代码节点算出一个数写进变量：它不查库，对不上具体的查询。"""
+        output = self.put({var: value})
+        self.add("node.started", node_id, {"node_type": "code"})
+        self.add("node.finished", node_id, {"duration_ms": 2, "attempt": 1, "artifact": output})
+
+    def card(self, metrics, *, node_id="card", caliber="周报口径", version="v1", runs=1):
+        card = {"kind": "metric_set", "caliber": caliber, "caliber_version": version, "metrics": metrics}
+        artifact = self.put(card)
+        for _ in range(runs):
+            self.add("node.started", node_id, {"node_type": "metrics"})
+            self.add("node.finished", node_id, {"duration_ms": 1, "attempt": 1, "artifact": self.put({**card, "text": "…"}),
+                                                "evidence": [{"kind": "metric_set", "node_id": node_id, "exec": 1,
+                                                              "artifact": artifact, "caliber": caliber,
+                                                              "version": version}]})
+        self.cards.append((node_id, artifact, card))
+        return artifact
+
+    def side(self):
+        def sources(started):
+            outputs = {}
+            for seq, etype, nid, data in self.events:
+                if seq < started and etype == "node.finished" and data.get("artifact"):
+                    outputs[nid] = data["artifact"]
+            return recompute.Sources(load=self.load, queries=tuple(self.queries), outputs=outputs, masked=self.masked)
+
+        run = {"id": self.id, "workflow_name": "区域周报", "version": self.version, "run_class": "formal",
+               "started_at": "2026-09-07T01:00:00+00:00", "finished_at": "2026-09-07T01:02:00+00:00",
+               "schedule_id": None}
+        return compare.Side(run=run, graph=self.graph, events=self.events, cards=self.cards, load=self.load,
+                            masked=self.masked, sources=sources, dialects={self.source: "sqlite"}, inputs=self.inputs)
+
+
+def cell_in(row, column, value, artifact, node_id="fetch"):
+    return {"path": f"cell(nodes.{node_id}, {row}, '{column}')", "value": value, "node_id": node_id, "via": "tool_cell",
+            "locator": {"row": row, "column": column}, "artifact": artifact,
+            "eid": f"ev:cell:{artifact[:8]}:{row}:{column}", "status": "verified"}
+
+
+def amounts_in(artifact, rows, *, column="gmv", span=None):
+    first, last = span or (0, len(rows) - 1)
+    return {"path": "vars.kpi.amounts", "value": f"[{last - first + 1} 项]", "node_id": "bot", "via": "agent_field",
+            "field": "amounts", "ref": f"Q1.r{first}-{last}.{column}", "artifact": artifact,
+            "locator": {"rows": [first, last], "column": column}, "eid": f"ev:rows:{artifact[:8]}:{first}:{column}",
+            "status": "verified"}
+
+
+def scalar_in(path, value, node_id, via):
+    return {"path": path, "value": value, "node_id": node_id, "via": via, "field": path.split(".", 2)[-1],
+            "status": "ok" if value is not None else "missing"}
+
+
+def metric(mid, expression, value, inputs, *, name=None, unit="元", decimals=1, fmt="thousands"):
+    tree = parse_expression(expression)[0]
+    scalars = {i["path"]: i["value"] for i in inputs if not str(i["value"]).endswith(" 项]")}
+    return {"id": mid, "name": name or mid, "unit": unit, "value": value, "expression": expression,
+            "decimals": decimals, "format": fmt,
+            "rendered": render_number(value, unit=unit, decimals=decimals, fmt=fmt),
+            "inputs": inputs, "substituted": substitute(tree, scalars), "recompute_ok": True,
+            "status": "ok" if value is not None else "missing_input"}
+
+
+def week(run_id, rows=BASE_ROWS, *, sql=None, metrics=None, graph=None, version="v1", bot_rows=None, **kw):
+    """一次周报运行：fetch 和 bot 各查一次，口径卡算 east（东区金额）和 total（各区域合计）。
+    metrics(fetch, bot, rows) 返回口径卡的指标，不给就是这两个。"""
+    run = Fake(run_id, graph=graph or weekly_graph(sql), **kw)
+    sql = sql or region_sql()
+    fetch = run.tool("fetch", sql, rows=rows)
+    bot_rows = bot_rows if bot_rows is not None else rows
+    bot = run.agent("bot", sql, rows=bot_rows)
+    if metrics is None:
+        metrics = default_metrics
+    run.card(metrics(fetch, bot, rows, bot_rows), version=version)
+    return run
+
+
+def default_metrics(fetch, bot, rows, bot_rows):
+    total = sum(r[1] for r in bot_rows)
+    return [metric("east", EAST, rows[0][1], [cell_in(0, "gmv", rows[0][1], fetch)], name="东区金额"),
+            metric("total", "sum(vars.kpi.amounts)", total, [amounts_in(bot, bot_rows)], name="各区域合计")]

backend/tests/test_compare_api.py

@@ -0,0 +1,392 @@
+"""运行对比的接口：GET /api/runs/{run_id}/compare/base 和 GET /api/runs/{run_id}/compare?base=。
+
+两次周报运行用 tests/fixtures/compare_runs.py 的虚构运行：工件落进工件库，事件落进运行记录，按 runner 的算法封存
+（不连库、不调模型）。另有一组真跑：虚构的 SQLite 库在两次运行之间改了数，再一次代进去的日期参数不同。守的是：
+- 两次都结束、封存且核对一致才比：没封存、还在运行、封存后事件被改过都回 409，说清是哪一次；不是同一个工作流、
+  同一次运行回 422；运行不存在 404
+- 对比只读：两次运行的 last_seq、事件、工件引用、工件文件一样不多
+- 遮罩列（查询当时记下的、数据源现在设的）的值不出现在结果里
+- 默认基准：同一定时任务优先，否则同一工作流、同一运行类别；只挑更早结束、封存过的终态运行
+- 真跑：同一条 SQL 数据变了是 data_only；只是代进去的日期不同是 params_only，图一样
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sqlite3
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import func, select
+
+from app.core.artifact_store import manifest_hash, put_json
+from app.core.config import settings
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import Artifact, DataSource, Run, RunEvent, Workflow
+from app.engine.runner import run_manager, verify_manifest
+from app.main import app
+from tests.fixtures.compare_runs import (
+    EAST,
+    NEXT_ROWS,
+    Fake,
+    amounts_in,
+    cell_in,
+    chain,
+    metric,
+    node,
+    week,
+)
+from tests.fixtures.sources import drop_source
+
+SOURCE = "cmp_shop"
+T0 = datetime(2026, 9, 7, 1, 0, tzinfo=timezone.utc)
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+async def new_workflow(name="区域周报") -> str:
+    async with SessionLocal() as session:
+        workflow = Workflow(name=name)
+        session.add(workflow)
+        await session.commit()
+        return workflow.id
+
+
+async def persist(fake: Fake, workflow_id: str | None, *, finished=T0, status="succeeded", run_class="formal",
+                  schedule_id=None, seal=True) -> str:
+    """把一次虚构运行落库：工件进工件库（内容寻址，id 和虚构时一样），事件进运行记录，按 runner 的算法封存。"""
+    for content in fake.store.values():
+        await put_json(content, kind="compare_fixture")
+    async with SessionLocal() as session:
+        run = Run(workflow_id=workflow_id, workflow_name="区域周报", status=status, graph=fake.graph,
+                  input=dict(fake.inputs), run_class=run_class, version=fake.version, schedule_id=schedule_id,
+                  started_at=finished - timedelta(minutes=2), finished_at=finished if status != "running" else None,
+                  agent_limits={})
+        session.add(run)
+        await session.flush()
+        rows = []
+        for seq, etype, node_id, data in fake.events:
+            session.add(RunEvent(run_id=run.id, seq=seq, type=etype, node_id=node_id, ts=float(seq), data=data))
+            rows.append((seq, etype, node_id, data))
+        run.last_seq = len(rows)
+        if seal:
+            run.manifest_hash, run.manifest_seq = manifest_hash(rows), rows[-1][0]
+        await session.commit()
+        return run.id
+
+
+async def pair(base=None, target=None, **kw) -> tuple[str, str, str]:
+    """(工作流, 基准运行, 这次运行)：同一个工作流的两次周报，这次的数变了。"""
+    workflow = await new_workflow()
+    base_id = await persist(base or week("r1", source=SOURCE), workflow, finished=T0)
+    target_id = await persist(target or week("r2", NEXT_ROWS, source=SOURCE), workflow,
+                              finished=T0 + timedelta(days=7), **kw)
+    return workflow, base_id, target_id
+
+
+async def footprint(*run_ids: str) -> tuple:
+    """几次运行留在库里、盘上的痕迹：各自的 last_seq、封存、事件数，加上全部工件引用行数和工件文件数。"""
+    async with SessionLocal() as session:
+        runs = []
+        for run_id in run_ids:
+            run = await session.get(Run, run_id)
+            count = await session.scalar(select(func.count()).select_from(RunEvent).where(RunEvent.run_id == run_id))
+            runs.append((run.last_seq, run.manifest_hash, run.manifest_seq, run.status, count))
+        artifacts = await session.scalar(select(func.count()).select_from(Artifact))
+    files = sum(1 for _ in (settings.data_dir / "artifacts").rglob("*.json"))
+    return runs, artifacts, files
+
+
+async def get(client, path: str, status: int = 200) -> dict:
+    resp = await client.get(path)
+    assert resp.status_code == status, resp.text
+    return resp.json()
+
+
+# --------------------------------------------------------------------------
+# 对比
+# --------------------------------------------------------------------------
+
+
+async def test_compares_two_sealed_runs(client):
+    _, base_id, target_id = await pair()
+    body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}")
+    assert body["target"]["id"] == target_id and body["base"]["id"] == base_id
+    assert set(body["target"]) == {"id", "workflow_name", "version", "run_class", "started_at", "finished_at",
+                                   "schedule_id"}
+    assert body["target"]["finished_at"].endswith("+00:00") and body["base"]["run_class"] == "formal"
+    assert body["summary"]["data_only"] == 2 and sum(body["summary"].values()) == 2
+    east = next(m for m in body["metrics"] if m["metric_id"] == "east")
+    assert (east["classification"], east["base"], east["target"], east["delta_rendered"]) == (
+        "data_only", 1200.5, 1500.5, "+300.0元")
+    total = next(m for m in body["metrics"] if m["metric_id"] == "total")
+    assert [r["share"] for r in total["attribution"]["rows"]] == [1.5, -0.5, 0.0]
+    assert [(q["node_id"], q["structure"]) for q in body["queries"]] == [("fetch", "same"), ("bot", "same")]
+    assert body["template"] == {"base_version": 3, "target_version": 3, "same_graph": True, "nodes": []}
+    assert body["notes"] == []
+
+
+async def test_comparing_writes_nothing(client):
+    _, base_id, target_id = await pair()
+    before = await footprint(base_id, target_id)
+    await get(client, f"/api/runs/{target_id}/compare?base={base_id}")
+    await get(client, f"/api/runs/{target_id}/compare/base")
+    assert await footprint(base_id, target_id) == before
+    for run_id in (base_id, target_id):
+        assert (await verify_manifest(run_id))["ok"] is True
+
+
+async def test_unsealed_runs_are_refused(client):
+    workflow = await new_workflow()
+    base_id = await persist(week("r1", source=SOURCE), workflow)
+    target_id = await persist(week("r2", NEXT_ROWS, source=SOURCE), workflow, finished=T0 + timedelta(days=7),
+                              seal=False)
+    body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}", 409)
+    assert body["code"] == "compare_unsealed" and body["detail"].startswith("这次运行")
+    # 已经结束、只是没有封存清单的：说没有封存清单，不说「运行结束后才能对比」
+    assert "没有封存清单" in body["detail"] and "尚未结束" not in body["detail"]
+    body = await get(client, f"/api/runs/{base_id}/compare?base={target_id}", 409)
+    assert body["code"] == "compare_unsealed" and body["detail"].startswith("作为基准的运行")
+    # 封存过、又被接着运行（还在运行）的：封存范围之后还会变，同样不比
+    running = await persist(week("r3", NEXT_ROWS, source=SOURCE), workflow, status="running")
+    body = await get(client, f"/api/runs/{running}/compare?base={base_id}", 409)
+    assert body["code"] == "compare_unsealed" and "尚未结束" in body["detail"]
+
+
+async def test_a_broken_seal_is_refused(client):
+    _, base_id, target_id = await pair()
+    async with SessionLocal() as session:
+        event = (await session.execute(select(RunEvent).where(
+            RunEvent.run_id == base_id, RunEvent.type == "tool.end"))).scalars().first()
+        event.data = {**event.data, "duration_ms": 999}
+        await session.commit()
+    body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}", 409)
+    assert body["code"] == "compare_seal_broken" and body["detail"].startswith("作为基准的运行")
+    assert "封存核对不一致" in body["detail"]
+
+
+async def test_mismatched_runs_are_refused(client):
+    _, base_id, target_id = await pair()
+    other = await persist(week("r9", source=SOURCE), await new_workflow("另一个周报"))
+    body = await get(client, f"/api/runs/{target_id}/compare?base={other}", 422)
+    assert body["code"] == "compare_mismatch" and "同一个工作流" in body["detail"]
+    body = await get(client, f"/api/runs/{target_id}/compare?base={target_id}", 422)
+    assert body["code"] == "compare_mismatch" and "同一次运行" in body["detail"]
+    unsaved = await persist(week("r8", source=SOURCE), None)
+    body = await get(client, f"/api/runs/{unsaved}/compare?base={base_id}", 422)
+    assert body["code"] == "compare_mismatch"
+    body = await get(client, f"/api/runs/{target_id}/compare", 422)
+    assert body["code"] == "compare_bad_request"
+    await get(client, f"/api/runs/no-such-run/compare?base={base_id}", 404)
+    body = await get(client, f"/api/runs/{target_id}/compare?base=no-such-run", 404)
+    assert "基准" in body["detail"]
+
+
+async def test_masked_columns_never_leave_the_server(client):
+    """快照里记着查询当时遮了 refund，数据源现在又遮了 manager：两样的值都不出现在结果里。"""
+    columns = ["region", "manager", "gmv", "refund"]
+    types = {"region": "text", "manager": "text", "gmv": "number", "refund": "number"}
+    sql = "SELECT region, manager, SUM(amount) AS gmv, SUM(refund) AS refund FROM orders GROUP BY region, manager"
+
+    def make(run_id, rows, expression, version):
+        run = Fake(run_id, source=SOURCE)
+        fetch = run.tool("fetch", sql, columns, rows, types=types, masks=["refund"])
+        bot = run.agent("bot", sql, columns, rows, types=types, masks=["refund"])
+        inputs = [cell_in(0, "gmv", rows[0][2], fetch)]
+        value = rows[0][2]
+        if "refund" in expression:
+            inputs.append(cell_in(0, "refund", rows[0][3], fetch))
+            value = rows[0][2] - rows[0][3]
+        run.card([metric("net", expression, value, inputs),
+                  metric("total", "sum(vars.kpi.amounts)", sum(r[2] for r in rows), [amounts_in(bot, rows)])],
+                 version=version)
+        return run
+
+    async with SessionLocal() as session:
+        session.add(DataSource(name=SOURCE, kind="sqlite", database=":memory:", options={"mask_columns": ["manager"]}))
+        await session.commit()
+    try:
+        base = make("r1", [["东区", "经理甲", 1200.5, 101.25], ["南区", "经理乙", 800.0, 52.75]], EAST, "v1")
+        target = make("r2", [["东区", "经理甲", 1500.5, 133.5], ["南区", "经理乙", 700.0, 47.25]],
+                      f"{EAST} - cell(nodes.fetch, 0, 'refund')", "v2")
+        _, base_id, target_id = await pair(base, target)
+        resp = await client.get(f"/api/runs/{target_id}/compare?base={base_id}")
+        assert resp.status_code == 200, resp.text
+        for secret in ("经理甲", "经理乙", "101.25", "52.75", "133.5", "47.25"):
+            assert secret not in resp.text, secret
+        body = resp.json()
+        net = next(m for m in body["metrics"] if m["metric_id"] == "net")
+        refund = next(i for i in net["inputs"] if "refund" in i["path"])
+        assert refund["masked"] is True and net["decomposition"] is None and "遮罩" in net["decomposition_note"]
+        total = next(m for m in body["metrics"] if m["metric_id"] == "total")
+        assert total["attribution"]["columns"] == ["region"]
+    finally:
+        await drop_source(name=SOURCE)
+
+
+# --------------------------------------------------------------------------
+# 默认基准
+# --------------------------------------------------------------------------
+
+
+async def bare(workflow_id, finished, *, status="succeeded", run_class="formal", schedule_id=None, sealed=True,
+               resumed=False) -> str:
+    """只为挑基准用的运行记录：挑基准只看运行记录上的几列。resumed：结束、封存过又被接着运行（状态已回到运行中，
+    结束时间和封存还没清掉），只能靠状态把它挡在外面。"""
+    done = status in ("succeeded", "failed", "cancelled") or resumed
+    async with SessionLocal() as session:
+        run = Run(workflow_id=workflow_id, workflow_name="区域周报", status=status, run_class=run_class,
+                  schedule_id=schedule_id, started_at=finished - timedelta(minutes=2),
+                  finished_at=finished if done else None,
+                  manifest_hash="ab" * 32 if sealed else None, manifest_seq=1 if sealed else None)
+        session.add(run)
+        await session.commit()
+        return run.id
+
+
+async def test_the_default_base_prefers_the_same_schedule(client):
+    workflow = await new_workflow()
+    day = timedelta(days=1)
+    scheduled = await bare(workflow, T0, schedule_id="sched-weekly")
+    manual = await bare(workflow, T0 + day)                        # 更近，但不是同一个定时任务
+    current = await bare(workflow, T0 + 2 * day, schedule_id="sched-weekly")
+    body = await get(client, f"/api/runs/{current}/compare/base")
+    assert (body["base_run_id"], body["reason"]) == (scheduled, "same_schedule")
+    assert body["base"]["id"] == scheduled and body["base"]["schedule_id"] == "sched-weekly"
+    # 不是定时任务发起的：同一工作流、同一运行类别里最近的一次（定时的也算）
+    later = await bare(workflow, T0 + 3 * day)
+    body = await get(client, f"/api/runs/{later}/compare/base")
+    assert (body["base_run_id"], body["reason"]) == (current, "same_workflow")
+    # 定时任务的第一次：退回同一工作流、同一运行类别
+    first = await bare(workflow, T0 + 4 * day, schedule_id="sched-daily")
+    body = await get(client, f"/api/runs/{first}/compare/base")
+    assert (body["base_run_id"], body["reason"]) == (later, "same_workflow")
+    assert manual != body["base_run_id"]
+
+
+async def test_the_default_base_skips_what_cannot_be_compared(client):
+    workflow = await new_workflow()
+    day = timedelta(days=1)
+    good = await bare(workflow, T0)
+    await bare(workflow, T0 + day, sealed=False)                       # 没封存
+    await bare(workflow, T0 + 2 * day, status="running")               # 还在运行
+    await bare(workflow, T0 + 2 * day, status="running", resumed=True)  # 结束过、又被接着运行
+    await bare(workflow, T0 + 3 * day, run_class="exploratory")        # 别的运行类别
+    await bare(await new_workflow("另一个周报"), T0 + 4 * day)          # 别的工作流
+    current = await bare(workflow, T0 + 5 * day)
+    await bare(workflow, T0 + 6 * day)                                 # 比它晚结束
+    body = await get(client, f"/api/runs/{current}/compare/base")
+    assert (body["base_run_id"], body["reason"]) == (good, "same_workflow")
+    # 失败、取消的终态运行也算（封存过）
+    failed = await bare(workflow, T0 + 5 * day - timedelta(hours=1), status="failed")
+    assert (await get(client, f"/api/runs/{current}/compare/base"))["base_run_id"] == failed
+    # 第一次运行、没保存的工作流：没有可比的
+    body = await get(client, f"/api/runs/{good}/compare/base", 404)
+    assert body["code"] == "compare_no_base" and "上一次运行" in body["detail"]
+    await get(client, f"/api/runs/{await bare(None, T0)}/compare/base", 404)
+    await get(client, "/api/runs/no-such-run/compare/base", 404)
+
+
+# --------------------------------------------------------------------------
+# 真跑：虚构的 SQLite 库，两次运行之间改了数、换了日期参数
+# --------------------------------------------------------------------------
+
+E2E_SOURCE = "cmp_e2e"
+E2E_SQL = ("SELECT region, SUM(amount) AS gmv FROM orders WHERE day BETWEEN '{{ input.start }}' AND '{{ input.end }}' "
+           "GROUP BY region ORDER BY region")
+WEEK36_INPUT = {"start": "2026-08-31", "end": "2026-09-06"}
+WEEK37_INPUT = {"start": "2026-09-07", "end": "2026-09-13"}
+
+
+@pytest.fixture
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def orders(tmp_path):
+    path = tmp_path / "orders.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, amount REAL);")
+    db.executemany("INSERT INTO orders (day, region, amount) VALUES (?, ?, ?)", [
+        ("2026-09-01", "东区", 1200.5), ("2026-09-02", "南区", 800.0), ("2026-09-03", "西区", 600.0),
+        ("2026-09-08", "东区", 900.0), ("2026-09-09", "南区", 650.0), ("2026-09-10", "西区", 640.0)])
+    db.commit()
+    db.close()
+    async with SessionLocal() as session:
+        row = DataSource(name=E2E_SOURCE, kind="sqlite", database=str(path), readonly=True, enabled=True)
+        session.add(row)
+        await session.commit()
+        source_id = row.id
+    await engines.invalidate(source_id)
+    yield path, source_id
+    await drop_source(source_id)
+
+
+def e2e_graph():
+    return chain(
+        node("start", "input", "开始"),
+        node("fetch", "tool", "区域取数", tool=f"db_query__{E2E_SOURCE}", args={"sql": E2E_SQL}),
+        node("card", "metrics", "周报口径卡", caliber="周报口径", caliber_version="v1", metrics=[
+            {"id": "east", "name": "东区金额", "unit": "元", "decimals": 1, "format": "thousands",
+             "expression": "cell(nodes.fetch, 0, 'gmv')"}]),
+        node("out", "output", "成果", fields=[{"name": "指标", "value": "{{ nodes.card.text }}"}]),
+    )
+
+
+async def finished_run(workflow_id: str, inputs: dict) -> str:
+    run = await run_manager.start(graph=e2e_graph(), input_payload=inputs, workflow_id=workflow_id,
+                                  workflow_name="区域周报")
+    for _ in range(400):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, run.id)
+        if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
+            assert row.status == "succeeded", row.error
+            return run.id
+    raise AssertionError(f"运行 {run.id} 没有结束：{row.status}")
+
+
+async def test_real_runs_split_data_and_parameter_changes(client, engine_up, orders):
+    path, source_id = orders
+    workflow = await new_workflow()
+    first = await finished_run(workflow, WEEK36_INPUT)
+    db = sqlite3.connect(path)
+    db.execute("UPDATE orders SET amount = 1500.5 WHERE day = '2026-09-01'")
+    db.commit()
+    db.close()
+    await engines.invalidate(source_id)
+    second = await finished_run(workflow, WEEK36_INPUT)
+    third = await finished_run(workflow, WEEK37_INPUT)
+
+    body = await get(client, f"/api/runs/{second}/compare?base={first}")
+    [east] = body["metrics"]
+    assert (east["classification"], east["base"], east["target"], east["delta"]) == ("data_only", 1200.5, 1500.5, 300.0)
+    [fetch] = body["queries"]
+    assert (fetch["node_id"], fetch["structure"], fetch["base_rows"], fetch["target_rows"]) == ("fetch", "same", 3, 3)
+    assert east["inputs"][0]["base_artifact"] == fetch["base_artifact"]
+    # 东区金额只是三个区域里的一格：各区域的变化照列，占比为空
+    assert [(r["key"], r["delta"]) for r in east["attribution"]["rows"]] == [
+        (["东区"], 300.0), (["南区"], 0), (["西区"], 0)]
+
+    body = await get(client, f"/api/runs/{third}/compare?base={second}")
+    [east] = body["metrics"]
+    assert east["classification"] == "params_only" and east["target"] == 900.0
+    [fetch] = body["queries"]
+    assert fetch["structure"] == "params_only"
+    assert fetch["literals"] == [{"base": "'2026-08-31'", "target": "'2026-09-07'"},
+                                 {"base": "'2026-09-06'", "target": "'2026-09-13'"}]
+    assert body["template"]["same_graph"] is True
+    assert any("start" in n for n in body["notes"]), body["notes"]
+
+    body = await get(client, f"/api/runs/{third}/compare/base")
+    assert (body["base_run_id"], body["reason"]) == (second, "same_workflow")
+    json.dumps(body, allow_nan=False)