数字推演接口 POST /api/runs/{run_id}/evidence/whatif

证据面板点开口径卡上的一个数，改输入的值、排除它背后查询结果的几行，看指标和报告里引用它的地方会变成多少。
推演不是出具：不追加事件（last_seq 不变）、不落工件、不改报告。

- 只认这次运行封存范围内的工件：沿用证据接口的 _sealed / _sealed_loader 白名单，另加封存的节点产出和封存的指标集
  里写着的 value_artifact；排除行只认封存范围内的查询快照
- 口径卡按台账（一期以前按节点产出）找，循环里算过几次的取报告引用的那一份；nodes.* 按口径卡那次执行之前的产出还原
- 遮罩和证据面板查询步骤同一套：查询当时记下的加数据源现在设的，合并查询按它合并的数据源算
- 未封存（停在审批、还在跑）的运行也能推演，sealed 为假；封存核对没过的同样标出
- 请求形状不对回 422 whatif_bad_request，改动不合法回 422 whatif_bad_input（中文说明），卡、指标、报告不存在回 404
- 只在确有指标变化时列出引用了这张卡的下游口径卡

backend/app/api/whatif.py

@@ -1,9 +1,226 @@
 """数字推演：在一次运行的基础上改动输入，推演指标会怎样变化，不改原运行。
 
-目前只登记了路由前缀，接口由数字推演的工作包补上。路径挂在 /api/runs/{run_id}/… 下面。
+POST /api/runs/{run_id}/evidence/whatif：证据面板里点开口径卡上的一个数，改某个输入的值、或者排除它背后查询结果里的
+几行，看这个指标和报告里引用它的地方会变成多少。推演永远不是出具：不追加任何事件（last_seq 不变）、不落工件、
+不改报告，结果只交回给发请求的人。
+
+只认这次运行封存范围内的工件，和证据接口同一套白名单（api/evidence.py 的 _sealed / _sealed_loader）：口径卡的指标集
+从封存的 node.finished 台账找，排除行只认封存范围内交回过的查询快照，nodes.* 的完整值从封存的节点产出工件取，
+vars.* 的完整值（value_artifact）只认封存的指标集里写着的。还在跑、停在审批的运行也能推演，结果里 sealed 为假：
+数据可能还会变。遮罩列（查询当时记下的加数据源现在设的）不给值、也不能拿来推演，见 engine/recompute.py。
+
+重算本身是纯函数（engine/recompute.py）。这里负责：认请求、找卡、准备工件和遮罩、找报告里引用的地方、标出可能受影响
+的下游口径卡。
 """
 from __future__ import annotations
 
-from fastapi import APIRouter
+import asyncio
+from dataclasses import replace
+from typing import Any
+
+from fastapi import APIRouter, Body
+
+from app.api.coded import CodedHTTPException
+from app.api.evidence import (
+    REPORT_NOT_FOUND,
+    _Masks,
+    _merge_masks,
+    _recorded_masks,
+    _reports,
+    _Sealed,
+    _sealed,
+    _sealed_cards,
+    _sealed_loader,
+)
+from app.core.artifact_store import load
+from app.engine import recompute
+from app.engine.merge_query import MERGE_SOURCE
 
 router = APIRouter(prefix="/api/runs", tags=["whatif"])
+
+BAD_REQUEST = "whatif_bad_request"
+BAD_INPUT = "whatif_bad_input"
+CARD_NOT_FOUND = "whatif_card_not_found"
+CARD_TAMPERED = "whatif_card_tampered"
+METRIC_NOT_FOUND = "whatif_metric_not_found"
+
+BAD_REQUEST_TEXT = ("请求格式有误：card 应为口径卡节点的 ID；report、metric 可省略，填写时为节点 ID 和指标 ID；"
+                    "overrides.inputs 应为「输入路径 → 新值」，overrides.exclude_rows 应为「查询快照的工件 ID → 行号列表」")
+#: 一次最多改几个输入、排除几份快照里的行、每份最多几行：再多就是请求写错了
+MAX_INPUTS = 200
+MAX_SNAPSHOTS = 50
+MAX_ROWS = 10000
+
+
+@router.post("/{run_id}/evidence/whatif")
+async def evidence_whatif(run_id: str, payload: Any = Body(None)) -> dict[str, Any]:
+    """在一次运行的口径卡上推演：请求体 {report?, card, metric?, overrides: {inputs, exclude_rows}}。
+
+    返回 {sealed, seal, card, metrics, segments, unsupported, affected_cards}：每个指标的旧值、新值、变化量和输入，
+    报告里引用这些指标的地方会写成什么，不能用于推演的输入和原因，以及引用了这张卡、可能受影响（未重算）的下游口径卡。
+    """
+    report, card_node, metric, changes = _request(payload)
+    sealed = await _sealed(run_id)
+    reports = _reports(sealed)
+    if report is not None and not any(r.node_id == report for r in reports):
+        raise CodedHTTPException(404, "本次运行中没有所选报告撰写节点的文档", REPORT_NOT_FOUND)
+    cards = _sealed_cards(sealed)
+    artifact, card, started = _card_of(sealed, cards, card_node, reports, report)
+    if metric is not None and not any(isinstance(m, dict) and str(m.get("id")) == metric
+                                      for m in card.get("metrics") or []):
+        raise CodedHTTPException(404, f"口径卡中没有指标「{metric}」", METRIC_NOT_FOUND)
+    loader = _loader(sealed, cards)
+    sources = recompute.Sources(load=loader, queries=tuple(sealed.queries),
+                                outputs=_outputs_before(sealed, started), masked=await _hidden(sealed, loader))
+    docs = [(r.node_id, r.doc) for r in reports
+            if r.doc is not None and r.hash_ok is True and (report is None or r.node_id == report)]
+    try:
+        # 纯计算，连同读工件放进线程：快照大的时候不挡住同一进程里别的请求
+        result, segments = await asyncio.to_thread(_compute, card, changes, sources, metric, docs, card_node, artifact)
+    except recompute.WhatIfError as e:
+        raise CodedHTTPException(422, str(e), BAD_INPUT) from None
+    changed = any(m.get("changed") for m in result["metrics"])
+    return {
+        "run_id": run_id,
+        "card": {"node_id": card_node, "artifact": artifact, "caliber": card.get("caliber") or "",
+                 "version": card.get("caliber_version") or ""},
+        "report": report,
+        # 封存了、而且封存核对通过：推演依据的数据不会再变。没封存（还在跑、停在审批）或封存核对没过的照样能推演
+        "sealed": sealed.trusted,
+        "seal": {"sealed": sealed.sealed, "ok": sealed.seal["ok"]},
+        "metrics": result["metrics"],
+        "segments": segments,
+        "unsupported": result["unsupported"],
+        # 下游口径卡只在这次推演确实改变了指标时才标：什么都没变，下游也不会变
+        "affected_cards": recompute.downstream_cards(cards, card_node) if changed else [],
+    }
+
+
+def _compute(card: dict[str, Any], changes: recompute.Changes, sources: recompute.Sources, metric: str | None,
+             docs: list[tuple[str, dict[str, Any]]], card_node: str, artifact: str
+             ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    result = recompute.whatif(card, changes, sources, metric=metric)
+    segments = recompute.report_segments(docs, card_node=card_node, card_artifact=artifact, results=result["metrics"])
+    return result, segments
+
+
+def _request(payload: Any) -> tuple[str | None, str, str | None, recompute.Changes]:
+    """(report, card, metric, 改动)。形状不对一律 422 whatif_bad_request；值对不对（路径在不在、行号越没越界）由重算时核对。"""
+    def bad() -> CodedHTTPException:
+        return CodedHTTPException(422, BAD_REQUEST_TEXT, BAD_REQUEST)
+
+    if not isinstance(payload, dict):
+        raise bad()
+    card, report, metric = payload.get("card"), payload.get("report"), payload.get("metric")
+    if not isinstance(card, str) or not card.strip():
+        raise bad()
+    if any(v is not None and (not isinstance(v, str) or not v.strip()) for v in (report, metric)):
+        raise bad()
+    overrides = payload.get("overrides")
+    overrides = {} if overrides is None else overrides
+    if not isinstance(overrides, dict):
+        raise bad()
+    inputs = {} if overrides.get("inputs") is None else overrides["inputs"]
+    exclude = {} if overrides.get("exclude_rows") is None else overrides["exclude_rows"]
+    if not isinstance(inputs, dict) or not isinstance(exclude, dict) \
+            or len(inputs) > MAX_INPUTS or len(exclude) > MAX_SNAPSHOTS:
+        raise bad()
+    rows: dict[str, frozenset[int]] = {}
+    for artifact, picked in exclude.items():
+        if not isinstance(picked, list) or len(picked) > MAX_ROWS \
+                or not all(isinstance(r, int) and not isinstance(r, bool) for r in picked):
+            raise bad()
+        if picked:
+            rows[str(artifact)] = frozenset(picked)
+    return (report.strip() if report else None, card.strip(), metric.strip() if metric else None,
+            recompute.Changes(inputs={str(k): v for k, v in inputs.items()}, exclude=rows))
+
+
+def _card_of(sealed: _Sealed, cards: list[tuple[str, str, dict[str, Any]]], node_id: str, reports: list[Any],
+             report: str | None) -> tuple[str, dict[str, Any], int]:
+    """要推演的那份指标集：(工件 id, 内容, 那次执行的 node.started 序号)。
+
+    哪些 node.finished 算这张卡的执行，和 _sealed_cards 一样认：台账里的 metric_set 条目；一期以前没有台账的，画布上确实
+    是口径卡的节点，产出本身就是指标集。同一张卡在循环里算过几次时，选报告引用的那一份（指定了 report 看那份报告，
+    没指定看任意一份），都没引用就取最后一次。内容只认 _sealed_cards 取回、复验过哈希的：执行记录里有、内容却对不上
+    哈希的说清是被改过，不说「没有这张卡」。
+    """
+    calibers = set(sealed.nodes_of("metrics"))
+    runs: list[tuple[int, str]] = []
+    for seq, etype, nid, data in sealed.events:
+        if etype != "node.finished" or nid != node_id or not isinstance(data, dict):
+            continue
+        arts = [str(e["artifact"]) for e in data.get("evidence") or []
+                if isinstance(e, dict) and e.get("kind") == "metric_set" and e.get("artifact")]
+        if not arts and nid in calibers and data.get("artifact"):
+            arts = [str(data["artifact"])]
+        runs.extend((seq, art) for art in arts)
+    if not runs:
+        raise CodedHTTPException(404, "这次运行中没有这张口径卡的结果：节点不存在、不是口径卡、尚未执行或执行失败",
+                                 CARD_NOT_FOUND)
+    cited = {e.get("artifact") for r in reports if r.doc is not None and (report is None or r.node_id == report)
+             for e in (r.doc.get("catalog") or {}).values()
+             if isinstance(e, dict) and e.get("kind") == "metric" and e.get("node_id") == node_id}
+    seq, artifact = next(((s, a) for s, a in reversed(runs) if a in cited), runs[-1])
+    good = {art: card for nid, art, card in cards if nid == node_id}
+    if artifact not in good:
+        raise CodedHTTPException(409, "口径卡的指标集与哈希不一致或无法读取，疑似被修改，不能推演", CARD_TAMPERED)
+    started = max((s for s, etype, nid, _ in sealed.events if etype == "node.started" and nid == node_id and s < seq),
+                  default=seq)
+    return artifact, good[artifact], started
+
+
+def _outputs_before(sealed: _Sealed, started: int) -> dict[str, str]:
+    """口径卡那次执行开始之前，每个节点最后一次的产出工件：口径卡当时读到的 nodes.* 就是它们。"""
+    out: dict[str, str] = {}
+    for seq, etype, node_id, data in sealed.events:
+        if seq >= started:
+            break
+        if etype == "node.finished" and node_id and isinstance(data, dict) and data.get("artifact"):
+            out[node_id] = str(data["artifact"])
+    return out
+
+
+def _loader(sealed: _Sealed, cards: list[tuple[str, str, dict[str, Any]]]) -> Any:
+    """只认这次运行封存范围内的工件：证据接口的白名单（查询、表结构、检索快照和台账记过的），加上封存的节点产出工件，
+    以及封存的指标集里写着的完整值（value_artifact——写在内容寻址、又在封存范围内的指标集里，属于封存链的延伸）。
+
+    未封存、封存核对没过的运行也能推演（结果里 sealed 为假）：白名单按同一套规则算，只是不要求封存完好。
+    取回时照旧复验哈希，对不上抛 ValueError（artifact_store.load 的约定），由重算照实说。
+    """
+    base = _sealed_loader(replace(sealed, seal={**sealed.seal, "sealed": True, "ok": True}))
+    extra = {str(data["artifact"]) for _, etype, _, data in sealed.events
+             if etype == "node.finished" and isinstance(data, dict) and data.get("artifact")}
+    extra |= {str(item["value_artifact"]) for _, _, card in cards
+              for metric in card.get("metrics") or [] if isinstance(metric, dict)
+              for item in metric.get("inputs") or [] if isinstance(item, dict) and item.get("value_artifact")}
+    memo: dict[str, Any] = {}
+
+    def loader(artifact: str) -> Any:
+        key = str(artifact)
+        if key not in memo:
+            memo[key] = load(key) if key in extra else base(key)
+        return memo[key]
+    return loader
+
+
+async def _hidden(sealed: _Sealed, loader: Any) -> dict[str, frozenset[str]]:
+    """封存范围内每份查询快照要遮的列（小写）：查询当时记下的加数据源现在设的，合并查询按它合并的那几个数据源算。
+    和证据面板查询步骤的遮罩同一套（api/evidence.py 的 _query_step）。取不回来、被改过的快照不算：重算也不会用它。
+    快照经推演用的同一个 loader 取，读过的重算时不再读盘。"""
+    masks = _Masks()
+    out: dict[str, frozenset[str]] = {}
+    for artifact, info in sealed.queries.items():
+        try:
+            snap = loader(artifact)
+        except ValueError:
+            continue
+        if not isinstance(snap, dict):
+            continue
+        if snap.get("source") == MERGE_SOURCE:
+            hidden, _ = await _merge_masks(snap, masks)
+        else:
+            hidden = (await masks.of(info.get("source") or snap.get("source")))[0] | _recorded_masks(snap)
+        if hidden:
+            out[artifact] = frozenset(hidden)
+    return out

backend/tests/test_whatif_api.py

@@ -0,0 +1,446 @@
+"""数字推演接口：POST /api/runs/{run_id}/evidence/whatif。
+
+真跑一遍图（虚构的 SQLite 库、MockChatModel 写报告），再在这次运行上推演：
+- 什么都不改：每个指标原样交回，报告里引用它们的地方一处不变
+- 改输入、排除查询快照里的行：新值和手算一致，报告里会变的那几处按报告的写法渲染
+- 只读：不追加事件（last_seq 不变）、不落工件、封存核对照旧一致
+- 只认这次运行封存范围内的工件；遮罩列不给值、不能拿来推演；输入不合法回 422 并说清哪里不对
+- 输入的还原：nodes.* 从节点产出、agent 数组字段按 locator.rows、新运行的 vars.* 从 value_artifact；
+  老运行（没有 value_artifact）的 vars 容器照实说推演不了
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sqlite3
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from langchain_core.messages import AIMessage, ToolMessage
+from sqlalchemy import func, select
+
+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
+from app.engine.runner import run_manager, verify_manifest
+from app.main import app
+
+SQL = "SELECT id, region, amount FROM orders ORDER BY id"
+REGIONS = ["华东", "华南", "华北", "西南", "东北"]
+ROWS = [(i, REGIONS[i % 5], float(i * 100) + 0.5) for i in range(1, 11)]
+AMOUNT5 = "cell(nodes.fetch, 5, 'amount')"
+METRICS = [
+    {"id": "amount5", "name": "第六单金额", "unit": "元", "decimals": 1, "format": "thousands", "expression": AMOUNT5},
+    {"id": "cnt", "name": "订单数", "unit": "单", "format": "thousands", "expression": "len(vars.res.rows)"},
+    {"id": "per", "name": "平均每单", "unit": "元", "decimals": 2, "format": "thousands",
+     "expression": f"round({AMOUNT5} / len(vars.res.rows), 2)"},
+]
+TEXT = "第六单金额 [[m:amount5]]，共 [[m:cnt]]，平均每单 [[m:per]]。"
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    yield
+    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 set_source(path, options=None):
+    async with SessionLocal() as session:
+        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
+        if row is None:
+            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
+            session.add(row)
+        row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
+        row.options = dict(options or {})
+        await session.commit()
+        return row.id
+
+
+@pytest.fixture(autouse=True)
+async def shop(tmp_path):
+    path = tmp_path / "shop.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, amount REAL, week TEXT);")
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)",
+                   [(i, r, a, "2026-W37" if i > 3 else "2026-W36") for i, r, a in ROWS])
+    db.commit()
+    db.close()
+    source_id = await set_source(path)
+    await engines.invalidate(source_id)
+    yield path
+    await engines.invalidate(source_id)
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "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 query_graph(*, between=(), metrics=METRICS, sql=SQL):
+    return chain(
+        node("start", "input"),
+        node("fetch", "tool", tool="db_query__shop", args={"sql": sql}),
+        node("parse", "transform", mode="json", template="{{ nodes.fetch }}", assign_to="res"),
+        node("card", "metrics", caliber="推演口径", caliber_version="v3", metrics=metrics),
+        *between,
+        node("write", "report", instructions="写周报", on_violation="flag"),
+        node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
+    )
+
+
+def script(monkeypatch, text=TEXT):
+    from app.providers import mock_model
+
+    monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
+
+
+async def wait(run_id: str, statuses=("succeeded", "failed")) -> Run:
+    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 statuses and (row.status == "interrupted" or row.manifest_seq is not None):
+            return row
+    raise AssertionError(f"run {run_id} 没跑完：{row.status}")
+
+
+async def run_graph(monkeypatch, graph, *, text=TEXT, statuses=("succeeded", "failed")) -> Run:
+    script(monkeypatch, text)
+    row = await wait((await run_manager.start(graph=graph, input_payload={})).id, statuses)
+    assert row.status in statuses and row.status != "failed", row.error
+    return row
+
+
+async def snapshot_of(run_id: str, node_id: str = "fetch") -> str:
+    async with SessionLocal() as session:
+        [end] = (await session.execute(select(RunEvent).where(
+            RunEvent.run_id == run_id, RunEvent.type == "tool.end", RunEvent.node_id == node_id))).scalars()
+    return end.data["query_artifact"]
+
+
+async def whatif(client, run_id: str, body: dict, status: int = 200) -> dict:
+    resp = await client.post(f"/api/runs/{run_id}/evidence/whatif", json=body)
+    assert resp.status_code == status, resp.text
+    return resp.json()
+
+
+def by_id(body: dict) -> dict:
+    return {m["id"]: m for m in body["metrics"]}
+
+
+async def footprint(run_id: str) -> tuple:
+    """这次运行留在库里、盘上的痕迹：last_seq、事件数、工件引用行数、工件文件数。"""
+    async with SessionLocal() as session:
+        run = await session.get(Run, run_id)
+        events = await session.scalar(select(func.count()).select_from(RunEvent).where(RunEvent.run_id == run_id))
+        artifacts = await session.scalar(select(func.count()).select_from(Artifact))
+    files = sum(1 for _ in (settings.data_dir / "artifacts").rglob("*.json"))
+    return run.last_seq, run.manifest_hash, events, artifacts, files
+
+
+# --------------------------------------------------------------------------
+# 不改、改输入、排除行
+# --------------------------------------------------------------------------
+
+
+async def test_a_baseline_changes_nothing(client, monkeypatch):
+    row = await run_graph(monkeypatch, query_graph())
+    snapshot = await snapshot_of(row.id)
+    body = await whatif(client, row.id, {"card": "card"})
+
+    assert body["run_id"] == row.id and body["sealed"] is True and body["seal"] == {"sealed": True, "ok": True}
+    assert body["card"]["node_id"] == "card" and body["card"]["caliber"] == "推演口径" and body["card"]["version"] == "v3"
+    metrics = by_id(body)
+    assert {k: (m["old"], m["new"], m["status"], m["changed"]) for k, m in metrics.items()} == {
+        "amount5": (600.5, 600.5, "unchanged", False), "cnt": (10, 10, "unchanged", False),
+        "per": (60.05, 60.05, "unchanged", False)}
+    assert [m["rendered_old"] for m in body["metrics"]] == ["600.5元", "10单", "60.05元"]
+    [cell] = metrics["amount5"]["inputs"]
+    assert cell["path"] == AMOUNT5 and cell["editable"] is True and cell["artifact"] == snapshot
+    assert cell["locator"] == {"row": 5, "column": "amount"} and cell["old"] == 600.5
+    [rows] = metrics["cnt"]["inputs"]
+    assert rows["kind"] == "container" and rows["editable"] is False and rows["old"] == "[10 项]"
+    assert rows["artifact"] == snapshot, "整份查询结果赋给变量：按内容对上了那份快照"
+    assert body["unsupported"] == [] and body["affected_cards"] == []
+    # 报告里引用这三个指标的地方，都不变
+    assert [(s["report"], s["metric"], s["rendered_old"], s["rendered_new"], s["changed"]) for s in body["segments"]] == [
+        ("write", "amount5", "600.5元", "600.5元", False), ("write", "cnt", "10单", "10单", False),
+        ("write", "per", "60.05元", "60.05元", False)]
+
+
+async def test_excluding_rows_recomputes_like_by_hand(client, monkeypatch):
+    row = await run_graph(monkeypatch, query_graph())
+    snapshot = await snapshot_of(row.id)
+    body = await whatif(client, row.id, {"card": "card", "overrides": {"exclude_rows": {snapshot: [2, 7]}}})
+    metrics = by_id(body)
+    assert metrics["amount5"]["status"] == "unchanged"
+    assert metrics["cnt"]["new"] == 8 and metrics["cnt"]["rendered_new"] == "8单" and metrics["cnt"]["delta"] == -2
+    assert metrics["per"]["new"] == round(600.5 / 8, 2) == 75.06 and metrics["per"]["rendered_delta"] == "+15.01元"
+    [rows] = metrics["cnt"]["inputs"]
+    assert rows["excluded_rows"] == [2, 7] and (rows["old"], rows["new"]) == ("[10 项]", "[8 项]")
+    changed = [(s["metric"], s["rendered_old"], s["rendered_new"]) for s in body["segments"] if s["changed"]]
+    assert changed == [("cnt", "10单", "8单"), ("per", "60.05元", "75.06元")]
+
+    # 排除了 cell() 那一格所在的行：这个指标排除后无法计算
+    body = await whatif(client, row.id, {"card": "card", "overrides": {"exclude_rows": {snapshot: [5]}}})
+    metrics = by_id(body)
+    assert metrics["amount5"]["status"] == "excluded" and metrics["amount5"]["new"] is None
+    assert "排除后无法计算" in metrics["amount5"]["reason"] and "第 6 行" in metrics["amount5"]["reason"]
+    assert metrics["amount5"]["rendered_new"] == "—" and metrics["per"]["status"] == "excluded"
+    assert metrics["cnt"]["new"] == 9
+    seg = next(s for s in body["segments"] if s["metric"] == "amount5")
+    assert seg["rendered_new"] == "—" and seg["changed"] is True
+
+
+async def test_changing_an_input_of_one_metric(client, monkeypatch):
+    row = await run_graph(monkeypatch, query_graph())
+    body = await whatif(client, row.id, {"card": "card", "metric": "per", "report": "write",
+                                         "overrides": {"inputs": {AMOUNT5: 1000}}})
+    [per] = body["metrics"]
+    assert per["id"] == "per" and per["new"] == 100.0 and per["rendered_new"] == "100.00元"
+    assert per["substituted"] == "round(1000 / len(vars.res.rows), 2)"
+    # 只推演一个指标：报告里只列引用它的那一处
+    assert [(s["metric"], s["rendered_new"]) for s in body["segments"]] == [("per", "100.00元")]
+
+
+async def test_whatif_is_read_only(client, monkeypatch):
+    row = await run_graph(monkeypatch, query_graph())
+    snapshot = await snapshot_of(row.id)
+    before = await footprint(row.id)
+    await whatif(client, row.id, {"card": "card"})
+    await whatif(client, row.id, {"card": "card", "overrides": {"inputs": {AMOUNT5: 1}, "exclude_rows": {snapshot: [0]}}})
+    await whatif(client, row.id, {"card": "card", "overrides": {"inputs": {"vars.nope": 1}}}, status=422)
+    assert await footprint(row.id) == before, "推演不能追加事件、不能落工件"
+    verdict = await verify_manifest(row.id)
+    assert verdict["ok"] is True
+
+
+# --------------------------------------------------------------------------
+# 不合法的请求
+# --------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("body", [
+    [], {"card": 3}, {"card": " "}, {"card": "card", "report": 1}, {"card": "card", "metric": ["per"]},
+    {"card": "card", "overrides": []}, {"card": "card", "overrides": {"inputs": []}},
+    {"card": "card", "overrides": {"exclude_rows": {"abc": 3}}},
+    {"card": "card", "overrides": {"exclude_rows": {"abc": ["3"]}}},
+    {"card": "card", "overrides": {"exclude_rows": {"abc": [True]}}},
+])
+async def test_malformed_requests_are_rejected(client, monkeypatch, body):
+    row = await run_graph(monkeypatch, query_graph())
+    resp = await client.post(f"/api/runs/{row.id}/evidence/whatif", json=body)
+    assert resp.status_code == 422 and resp.json()["code"] == "whatif_bad_request", resp.text
+    assert "请求格式有误" in resp.json()["detail"]
+
+
+async def test_invalid_overrides_say_what_is_wrong(client, monkeypatch):
+    row = await run_graph(monkeypatch, query_graph())
+    snapshot = await snapshot_of(row.id)
+    # 别的运行的快照（内容不同，工件 id 也就不同）：工件库里确有其物，只是不在这次运行的封存范围内
+    other = await run_graph(monkeypatch, query_graph(sql="SELECT id, region, amount FROM orders ORDER BY id DESC"))
+    foreign = await snapshot_of(other.id)
+    assert foreign != snapshot
+    card_artifact = (await whatif(client, row.id, {"card": "card"}))["card"]["artifact"]
+    cases = [
+        ({"inputs": {"vars.nope": 1}}, ["vars.nope"]),
+        ({"inputs": {AMOUNT5: "很多"}}, ["数值", "很多"]),
+        ({"inputs": {AMOUNT5: True}}, ["数值"]),
+        ({"inputs": {"vars.res.rows": 3}}, ["列表", "排除"]),
+        ({"exclude_rows": {snapshot: [10]}}, ["行号 10", "10 行"]),
+        ({"exclude_rows": {snapshot: [-1]}}, ["行号 -1"]),
+        ({"exclude_rows": {foreign: [0]}}, ["不属于这次运行", "封存范围"]),
+        ({"exclude_rows": {card_artifact: [0]}}, ["不属于这次运行"]),
+    ]
+    for overrides, words in cases:
+        resp = await client.post(f"/api/runs/{row.id}/evidence/whatif", json={"card": "card", "overrides": overrides})
+        assert resp.status_code == 422, (overrides, resp.text)
+        body = resp.json()
+        assert body["code"] == "whatif_bad_input" and all(w in body["detail"] for w in words), (overrides, body)
+
+
+async def test_only_artifacts_sealed_in_this_run_are_loaded(monkeypatch):
+    """推演取工件只走封存范围内的白名单：别的运行的快照、只在工件表里的东西一律当取不到。"""
+    from app.api.evidence import _sealed, _sealed_cards
+    from app.api.whatif import _loader
+    from app.core.artifact_store import put_json
+
+    row = await run_graph(monkeypatch, query_graph())
+    other = await run_graph(monkeypatch, query_graph(sql="SELECT id, region, amount FROM orders ORDER BY id LIMIT 10"))
+    stray = await put_json({"path": "vars.res.rows", "value": [1, 2, 3]}, kind="metric_input", run_id=row.id)
+    sealed = await _sealed(row.id)
+    loader = _loader(sealed, _sealed_cards(sealed))
+    assert loader(await snapshot_of(row.id)) is not None
+    assert loader(await snapshot_of(other.id)) is None and loader(stray) is None
+    card = loader(sealed.outputs["card"])
+    value_artifacts = {i["value_artifact"] for m in card["metrics"] for i in m["inputs"] if i.get("value_artifact")}
+    assert value_artifacts and all(loader(a) is not None for a in value_artifacts)
+
+
+async def test_missing_things_are_coded_404s(client, monkeypatch):
+    row = await run_graph(monkeypatch, query_graph())
+    cases = [
+        ("/api/runs/nope/evidence/whatif", {"card": "card"}, "run_not_found"),
+        (f"/api/runs/{row.id}/evidence/whatif", {"card": "write"}, "whatif_card_not_found"),
+        (f"/api/runs/{row.id}/evidence/whatif", {"card": "ghost"}, "whatif_card_not_found"),
+        (f"/api/runs/{row.id}/evidence/whatif", {"card": "card", "metric": "ghost"}, "whatif_metric_not_found"),
+        (f"/api/runs/{row.id}/evidence/whatif", {"card": "card", "report": "ghost"}, "evidence_report_not_found"),
+    ]
+    for url, body, code in cases:
+        resp = await client.post(url, json=body)
+        assert resp.status_code == 404 and resp.json()["code"] == code, (body, resp.text)
+
+
+# --------------------------------------------------------------------------
+# 遮罩、封存
+# --------------------------------------------------------------------------
+
+
+async def test_masked_columns_are_neither_shown_nor_used(client, monkeypatch, shop):
+    await set_source(shop, {"mask_columns": ["amount"]})
+    row = await run_graph(monkeypatch, query_graph())
+    snapshot = await snapshot_of(row.id)
+    body = await whatif(client, row.id, {"card": "card"})
+    metrics = by_id(body)
+    [cell] = metrics["amount5"]["inputs"]
+    assert cell["masked"] is True and cell["old"] is None and cell["new"] is None and cell["editable"] is False
+    reasons = {(u["metric"], u["input"]): u["reason"] for u in body["unsupported"]}
+    assert "遮罩" in reasons[("amount5", AMOUNT5)] and "遮罩" in reasons[("cnt", "vars.res.rows")]
+
+    resp = await client.post(f"/api/runs/{row.id}/evidence/whatif",
+                             json={"card": "card", "overrides": {"inputs": {AMOUNT5: 1}}})
+    assert resp.status_code == 422 and "遮罩" in resp.json()["detail"]
+    metrics = by_id(await whatif(client, row.id, {"card": "card", "overrides": {"exclude_rows": {snapshot: [1]}}}))
+    assert metrics["cnt"]["status"] == "masked" and metrics["cnt"]["new"] is None
+    assert metrics["per"]["status"] == "masked" and metrics["amount5"]["status"] == "unchanged"
+
+
+async def test_a_run_waiting_for_approval_can_be_explored_but_is_not_sealed(client, monkeypatch):
+    row = await run_graph(monkeypatch, query_graph(between=[node("gate", "human", mode="approve", title="确认")]),
+                          statuses=("interrupted",))
+    before = await footprint(row.id)
+    body = await whatif(client, row.id, {"card": "card", "overrides": {"inputs": {AMOUNT5: 1000}}})
+    assert body["sealed"] is False and body["seal"]["sealed"] is False
+    assert by_id(body)["amount5"]["new"] == 1000 and body["segments"] == []
+    assert await footprint(row.id) == before
+
+
+# --------------------------------------------------------------------------
+# 容器输入的还原
+# --------------------------------------------------------------------------
+
+
+async def test_an_old_run_without_full_values_says_so(client, monkeypatch):
+    """加 value_artifact 之前的运行：vars 的容器没存完整值，用到它的指标推演不了，原因写清楚。"""
+    from app.engine.nodes import metrics as metrics_node
+
+    async def nothing(*args, **kwargs):
+        return None
+
+    monkeypatch.setattr(metrics_node, "_store_value", nothing)
+    row = await run_graph(monkeypatch, query_graph())
+    body = await whatif(client, row.id, {"card": "card", "overrides": {"inputs": {AMOUNT5: 1000}}})
+    reasons = {(u["metric"], u["input"]): u["reason"] for u in body["unsupported"]}
+    assert reasons[("cnt", "vars.res.rows")] == "无法推演：当时没有保存这个输入的完整值"
+    per = by_id(body)["per"]
+    assert per["status"] == "unsupported" and per["new"] is None and per["rendered_new"] is None
+    assert "当时没有保存这个输入的完整值" in per["reason"]
+    assert by_id(body)["amount5"]["new"] == 1000
+    assert next(s for s in body["segments"] if s["metric"] == "per")["rendered_new"] is None
+
+
+async def test_node_containers_come_from_the_sealed_node_output(client, monkeypatch):
+    row = await run_graph(monkeypatch, chain(
+        node("start", "input"),
+        node("fetch", "transform", mode="expression", expression="{'items': [3, 5, 7], 'k': 2}", assign_to="kpi"),
+        node("card", "metrics", caliber="推演口径", metrics=[
+            {"id": "n", "name": "加权条数", "expression": "len(nodes.fetch.items) * vars.kpi.k"}]),
+        node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}])))
+    body = await whatif(client, row.id, {"card": "card", "overrides": {"inputs": {"vars.kpi.k": 5}}})
+    [n] = body["metrics"]
+    assert n["status"] == "ok" and (n["old"], n["new"]) == (6, 15)
+    items = next(i for i in n["inputs"] if i["path"] == "nodes.fetch.items")
+    assert items["kind"] == "container" and items["old"] == "[3 项]"
+    assert body["segments"] == []
+
+
+class AgentScript:
+    """agent 查一次按周汇总，抽取把两周的金额交成一个数组字段（出处是 Q1 的第 0–1 行）。"""
+
+    WEEKLY = "SELECT week, SUM(amount) AS amount FROM orders GROUP BY week ORDER BY week"
+
+    def __init__(self, monkeypatch):
+        from app.providers import mock_model
+
+        extracted = {"amounts": {"value": [601.5, 4903.5], "from": {"call": "Q1", "rows": "0-1", "column": "amount"}}}
+
+        def decide(model, messages):
+            if model.response_format:
+                return AIMessage(content=json.dumps(extracted, ensure_ascii=False))
+            if not model.tools:
+                return AIMessage(content="收尾")
+            if not any(isinstance(m, ToolMessage) for m in messages):
+                return AIMessage(content="查。", tool_calls=[
+                    {"name": "db_query__shop", "args": {"sql": self.WEEKLY}, "id": "call_0"}])
+            return AIMessage(content="两周金额已汇总。")
+
+        monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
+
+
+async def test_agent_array_fields_follow_the_excluded_rows(client, monkeypatch):
+    AgentScript(monkeypatch)
+    schema = {"type": "object", "properties": {"amounts": {"type": "array", "items": {"type": "number"}}}}
+    graph = chain(
+        node("start", "input"),
+        node("bot", "agent", prompt="按周汇总金额", tools=["db_query__shop"], approval="never",
+             output_schema=schema, cite_fields=True, assign_to="kpi"),
+        node("card", "metrics", caliber="推演口径", metrics=[
+            {"id": "via_node", "name": "合计", "decimals": 1, "expression": "sum(nodes.bot.data.amounts)"},
+            {"id": "via_vars", "name": "合计", "decimals": 1, "expression": "sum(vars.kpi.amounts)"}]),
+        node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}]))
+    row = await wait((await run_manager.start(graph=graph, input_payload={})).id)
+    assert row.status == "succeeded", row.error
+    snapshot = await snapshot_of(row.id, "bot")
+    weeks = {"2026-W36": sum(a for i, _, a in ROWS if i <= 3), "2026-W37": sum(a for i, _, a in ROWS if i > 3)}
+    total = round(weeks["2026-W36"] + weeks["2026-W37"], 1)
+
+    baseline = by_id(await whatif(client, row.id, {"card": "card"}))
+    assert baseline["via_node"]["old"] == baseline["via_vars"]["old"] == total
+    body = await whatif(client, row.id, {"card": "card", "overrides": {"exclude_rows": {snapshot: [0]}}})
+    for metric in by_id(body).values():
+        assert metric["status"] == "ok" and metric["new"] == round(weeks["2026-W37"], 1), metric
+        [amounts] = metric["inputs"]
+        assert amounts["excluded_rows"] == [0] and (amounts["old"], amounts["new"]) == ("[2 项]", "[1 项]")
+
+
+# --------------------------------------------------------------------------
+# 下游口径卡
+# --------------------------------------------------------------------------
+
+
+async def test_downstream_cards_are_flagged_only_when_something_changes(client, monkeypatch):
+    row = await run_graph(monkeypatch, chain(
+        node("start", "input"),
+        node("fetch", "transform", mode="expression", expression="{'gmv': 100.0, 'k': 2}", assign_to="kpi"),
+        node("card", "metrics", caliber="周口径", caliber_version="v1", assign_to="m", metrics=[
+            {"id": "gmv", "name": "销售额", "decimals": 1, "expression": "vars.kpi.gmv"}]),
+        node("yearly", "metrics", caliber="年口径", caliber_version="v2", metrics=[
+            {"id": "year", "name": "年化", "decimals": 1, "expression": "vars.m.metrics[0].value * 52"}]),
+        node("out", "output", fields=[{"name": "r", "value": "{{ nodes.yearly.text }}"}])))
+    assert (await whatif(client, row.id, {"card": "card"}))["affected_cards"] == []
+    body = await whatif(client, row.id, {"card": "card", "overrides": {"inputs": {"vars.kpi.gmv": 120}}})
+    assert by_id(body)["gmv"]["new"] == 120
+    [flagged] = body["affected_cards"]
+    assert (flagged["node_id"], flagged["caliber"], flagged["version"], flagged["metrics"], flagged["direct"]) == (
+        "yearly", "年口径", "v2", ["year"], True)