分析能力入口（七）：按 W1-a 算子库的实际接口对齐。工具参数名以算子为准：差异检验的均值是 mean / sd / n 或 sum / sumsq / n，逐行是 value，可选 pairs / reference；可比性检查另收可选的 measure、grain（不收 label）；分析卡的参数校验用同一份定义，运行当天、目录事实、下钻层数由算子自己记，分析项里写了不收。交给算子的输入是工件库里那份查询快照原样（含 SQL 检查结果 checks、「查询区间里这张表没有数据」no_data、查询当时记下的遮罩），加上出处和遮罩列（数据源设的、按 SQL 认出改了名字的、遮罩运行里查询时就遮掉的），合并查询写明 sqlite 方言；agent 的容器在分析前把每份换成完整快照（每份只取回一次）。探查器改用 analysis.make_prober(数据源名 → 按运行固定版本解析出的源，coverage.Cache 一个节点里共用)，剖析事实一律用 analysis.probe.profile_facts（查询工具的覆盖检查也改用它，原来那段同样的取法删掉）。分析次数用完交回 analysis.exhausted 的那句；口径卡的可比性写成 {工件, 判定码, 判定语, analysis.comparable_reasons 拆出的原因}，删掉自己拆原因的那段。加真算子的端到端测试：agent 变化拆解并下钻（父结论按内容哈希认）、审批恢复重放不再算、分析卡的波动基线和变化拆解（本期完整按现查的表范围判）、口径卡跑真可比性检查（可比、有条件可比带原因），每份结论都按封存的工件复算、逐字段相同

backend/app/engine/nodes/analysis.py

@@ -63,7 +63,7 @@ async def run_analysis(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
                 inputs[node_id] = await _load_input(state, ctx, node_id, queries)
     context = await context_of(list(inputs.values()), tool_ctx=tool_context(ctx.run, ctx.node.id, ctx.emit))
 
-    import app.analysis as analysis_ops   # 按模块取 run：测试、合并前的替身都只换模块上的这一个函数
+    import app.analysis as analysis_ops   # 按模块取 run：测试换替身时只换模块上的这一个函数
 
     exec_no = next_exec(state, ctx.node.id)
     done: dict[str, AnalysisResult] = {}
@@ -158,9 +158,8 @@ async def _load_input(state: GraphState, ctx: NodeContext, node_id: str,
         raise NodeError(ctx.node.id, f"{who}的查询快照与哈希不一致，疑似被修改，不能分析") from None
     if not isinstance(content, dict):
         raise NodeError(ctx.node.id, f"{who}的查询快照已不存在，无法分析。请重新运行它")
-    snapshot = input_snapshot(content, ref=node_id, artifact=artifact, node_id=node_id)
-    snapshot["masked"] = await masked_of(content, artifact)
-    return snapshot
+    return input_snapshot(content, ref=node_id, artifact=artifact, node_id=node_id,
+                          masked=await masked_of(content, artifact))
 
 
 __all__ = ["REFUSED_CODE", "card_view", "run_analysis"]

backend/app/engine/nodes/metrics.py

@@ -539,7 +539,7 @@ class _Skipped(Exception):
 
 async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]], ctx: NodeContext,
                             comparing: _Comparing, snapshots: dict[str, Any]) -> dict[str, Any]:
-    import app.analysis as analysis_ops   # 按模块取 run：测试、合并前的替身都只换模块上的这一个函数
+    import app.analysis as analysis_ops   # 按模块取 run：测试换替身时只换模块上的这一个函数
 
     from app.analysis.contract import VERDICTS
     from app.data.sql_semantics import normalize_sql
@@ -573,10 +573,9 @@ async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]],
         refs[artifact] = producer if producer and producer not in refs.values() else f"{producer or 'query'}·{role}"
     snaps = {}
     for artifact in order:
-        content = snapshots[artifact]
-        snap = input_snapshot(content, ref=refs[artifact], artifact=artifact, node_id=refs[artifact].split("·")[0])
-        snap["masked"] = await masked_of(content, artifact)
-        snaps[refs[artifact]] = snap
+        snaps[refs[artifact]] = input_snapshot(snapshots[artifact], ref=refs[artifact], artifact=artifact,
+                                               node_id=refs[artifact].split("·")[0],
+                                               masked=await masked_of(snapshots[artifact], artifact))
     settings = await settings_of(run_snapshot(ctx.run))
     context = await context_of(list(snaps.values()), tool_ctx=tool_context(ctx.run, ctx.node.id, ctx.emit))
     request = {"op": "comparable", "mode": None,
@@ -586,8 +585,9 @@ async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]],
     if result.get("status") != "ok" or verdict not in VERDICTS["comparable"]:
         raise _Skipped(str(result.get("reason") or "检查没有给出判定"))
     artifact = await store(result, run_id=ctx.run.run_id, node_id=ctx.node.id)
-    return {"artifact": artifact, "verdict": verdict, "phrase": str(result.get("phrase") or ""),
-            "reasons": comparability_reasons(result)}
+    # 原因的拆法只有算子那一处（analysis.comparable_reasons，按检查表的说明拆）
+    return {"artifact": artifact, "verdict": result["verdict"]["code"], "phrase": result["phrase"],
+            "reasons": analysis_ops.comparable_reasons(result)}
 
 
 async def _dialect_of(source: str, comparing: _Comparing) -> str | None:
@@ -625,24 +625,6 @@ def _period_order(a: tuple[str, ...], b: tuple[str, ...]) -> tuple[int, int]:
     return 1, 0
 
 
-def comparability_reasons(result: dict[str, Any]) -> list[str]:
-    """可比性结论里 fail、warn 的原因，一句一个（contract.MetricComparability.reasons）。结果表里有 status 一列和一列说明
-    （reason / text / detail / message）的照表取；没有的按判定语「两期不可比：原因；原因」拆。可比的没有原因。"""
-    table = result.get("table") or {}
-    columns = [str(c) for c in table.get("columns") or []]
-    text_col = next((c for c in ("reason", "text", "detail", "message") if c in columns), None)
-    if "status" in columns and text_col is not None:
-        si, ti = columns.index("status"), columns.index(text_col)
-        found = [str(row[ti]) for row in table.get("rows") or []
-                 if isinstance(row, list) and len(row) > max(si, ti) and row[si] in ("fail", "warn") and row[ti]]
-        if found:
-            return found
-    phrase = str(result.get("phrase") or "")
-    if (result.get("verdict") or {}).get("code") == "comparable" or "：" not in phrase:
-        return []
-    return [part.strip() for part in phrase.split("：", 1)[1].split("；") if part.strip()]
-
-
 #: 清单里输入取自「查询区间里这张表没有数据」的列的指标后面那句：直接读清单的模型节点不能把 0 当成真的是 0
 LISTED_NO_DATA = "（这段时间表里没有数据：不要当成 0 解读）"
 #: 指标的 no_data_reason 打头那半句（后面接查询记下的说明）

backend/app/tools/analysis.py

@@ -10,8 +10,9 @@
   那一步按 checkpoint 里的结果照样放一遍，编号还是同一套。
 - **阈值**：发起时的快照（run.options["analysis"]，engine/launch.py）← 分析卡上写的覆盖；没有快照的现读设置。agent 改不了：
   分析工具的参数里没有阈值（契约 D4）。
-- **运行期依赖**（contract.Context）：运行当天（服务端时区）、现查表的日期范围（W1-a 的 make_prober，没有这个模块时不给）、
-  数据目录（按运行的目录视图读，和 db_schema 给模型看的同一份）、剖析事实。
+- **运行期依赖**（contract.Context）：运行当天（服务端时区）、现查表的日期范围（analysis.make_prober，按这次运行固定的数据
+  版本解析数据源）、数据目录（按运行的目录视图读，和 db_schema 给模型看的同一份）、剖析事实（analysis.probe.profile_facts，
+  和查询工具的覆盖检查同一个取法）。
 - **遮罩列**：数据源设的遮罩列、查询当时记下的、按 SQL 认出改了名字的（和证据面板、运行对比同一个函数，api/evidence 的
   _mask_sets 的 hidden，宁可多遮），加上遮罩运行里查询时就整列遮掉的。算子不拿它们当维度、度量。
 - **结论**：整份落成内容寻址的工件（contract.ARTIFACT_KIND），台账记 contract.ledger_entry；交回 agent 的是一段短的
@@ -64,9 +65,10 @@ def call_error_text(error: CallError) -> str:
 class AnalysisScope:
     """一次节点执行里分析工具读的容器（agent 节点、协作成员的一次执行各一个）。
 
-    queries：编号 Qn → 输入快照（contract.InputSnapshot，masked 留空，用到时由 prepare 补）。results：编号 An → 整份
-    结论；artifacts：An → 结论的工件 id。snapshot：发起时的分析设置快照（没有为 None，现读设置）。mask_cache：快照工件 →
-    遮罩列；source_cache：数据源和表 → 目录、剖析事实（一次节点执行里各读一次，只在工具执行时用，不影响编号）。"""
+    queries：编号 Qn → 输入快照（contract.InputSnapshot；记台账时先放交回模型的那份 payload，分析前 prepare 换成工件库里
+    那份完整快照，补上遮罩列）。results：编号 An → 整份结论；artifacts：An → 结论的工件 id。snapshot：发起时的分析设置快照
+    （没有为 None，现读设置）。mask_cache：快照工件 → 遮罩列；source_cache：数据源和表 → 目录、查询用的源；probe_cache：
+    数据源名 → coverage.Cache（一次节点执行里各读、各查一次，只在工具执行时用，不影响编号）。"""
 
     node_id: str
     snapshot: dict[str, Any] | None = None
@@ -75,6 +77,9 @@ class AnalysisScope:
     artifacts: dict[str, str] = field(default_factory=dict)
     mask_cache: dict[str, list[str]] = field(default_factory=dict)
     source_cache: dict[str, Any] = field(default_factory=dict)
+    probe_cache: dict[str, Any] = field(default_factory=dict)
+    #: 已经换成工件库里那份完整快照的编号（prepare 每份只取回一次）
+    prepared: set[str] = field(default_factory=set)
 
     def add_query(self, alias: str, snapshot: InputSnapshot) -> None:
         self.queries[alias] = snapshot
@@ -130,25 +135,18 @@ async def settings_of(snapshot: Mapping[str, Any] | None,
 # --------------------------------------------------------------------------
 
 
-def input_snapshot(content: Mapping[str, Any], *, ref: str, artifact: str, node_id: str) -> InputSnapshot:
-    """一份查询快照（数据源查询工具交回的、工件库里的）→ 算子的一份输入（masked 留空，prepare 时补）。列的类型对齐成和
-    columns 同序的列表（快照里记的是 {列: 类型}，没记的为 None）；行一律写成和 columns 同序的列表。"""
-    columns = [str(c) for c in content.get("columns") or []]
-    types = content.get("column_types")
-    rows: list[list[Any]] = []
-    for row in content.get("rows") or []:
-        if isinstance(row, (list, tuple)):
-            rows.append(list(row))
-        elif isinstance(row, Mapping):
-            rows.append([row.get(c) for c in columns])
-        else:
-            rows.append([row])
-    return {
-        "ref": ref, "artifact": artifact, "node_id": node_id, "source": str(content.get("source") or ""),
-        "sql": str(content.get("sql") or ""), "columns": columns,
-        "column_types": [types.get(c) if isinstance(types, Mapping) else None for c in columns],
-        "rows": rows, "truncated": bool(content.get("truncated")), "masked": [],
-    }
+def input_snapshot(content: Mapping[str, Any], *, ref: str, artifact: str, node_id: str,
+                   masked: Iterable[str] = ()) -> InputSnapshot:
+    """一份查询快照 → 算子的一份输入：查询工具的 payload（工件库里那份）原样——列、行、类型、SQL、数据源，连同 SQL 检查结果
+    （checks）、「查询区间里这张表没有数据」（no_data）、查询当时记下的遮罩（mask_columns、masked_at_query），可比性检查按
+    它们判数据问题——再加上出处（ref、artifact、node_id）和不能当维度、度量的列（masked，见 masked_of）。合并查询在 SQLite
+    里执行，方言写明；数据源的方言算子按探查器认（prober.dialect）。"""
+    from app.engine.merge_query import MERGE_SOURCE
+
+    out: dict[str, Any] = {**content, "ref": ref, "artifact": artifact, "node_id": node_id, "masked": list(masked)}
+    if content.get("source") == MERGE_SOURCE and not content.get("dialect"):
+        out["dialect"] = "sqlite"
+    return out  # type: ignore[return-value]
 
 
 async def masked_of(content: Mapping[str, Any] | None, artifact: str | None) -> list[str]:
@@ -171,23 +169,28 @@ async def masked_of(content: Mapping[str, Any] | None, artifact: str | None) ->
 
 
 async def prepare(scope: AnalysisScope) -> None:
-    """补上容器里每份输入快照的遮罩列（每份工件只认一次）。快照取不回来、和哈希对不上的照实报错：查询明明交回过，
-    分析读不到它的快照，不是 agent 写错了参数。"""
+    """分析前把容器里的每份输入换成工件库里那份完整的查询快照（取回时复验哈希），补上遮罩列（每份工件只认一次）。快照
+    取不回来、和哈希对不上的照实报错：查询明明交回过，分析读不到它的快照，不是 agent 写错了参数。"""
     import asyncio
 
     from app.core import artifact_store
 
-    for alias, snapshot in scope.queries.items():
-        artifact = snapshot.get("artifact") or ""
+    for alias, snapshot in list(scope.queries.items()):
+        if alias in scope.prepared:
+            continue
+        artifact = str(snapshot.get("artifact") or "")
+        try:
+            content = await asyncio.to_thread(artifact_store.load, artifact)
+        except ValueError as e:
+            raise RuntimeError(f"{alias} 的查询快照与哈希不一致，疑似被修改，不能分析") from e
+        if not isinstance(content, Mapping):
+            raise RuntimeError(f"{alias} 的查询快照已不存在，不能分析")
         if artifact not in scope.mask_cache:
-            try:
-                content = await asyncio.to_thread(artifact_store.load, artifact)
-            except ValueError as e:
-                raise RuntimeError(f"{alias} 的查询快照与哈希不一致，疑似被修改，不能分析") from e
-            if content is None:
-                raise RuntimeError(f"{alias} 的查询快照已不存在，不能分析")
             scope.mask_cache[artifact] = await masked_of(content, artifact)
-        snapshot["masked"] = list(scope.mask_cache[artifact])
+        scope.queries[alias] = input_snapshot(content, ref=alias, artifact=artifact,
+                                              node_id=str(snapshot.get("node_id") or scope.node_id),
+                                              masked=scope.mask_cache[artifact])
+        scope.prepared.add(alias)
 
 
 # --------------------------------------------------------------------------
@@ -214,21 +217,26 @@ def _tables_of(snapshots: Iterable[Mapping[str, Any]]) -> dict[str, set[str]]:
 
 
 async def context_of(snapshots: Iterable[Mapping[str, Any]], *, tool_ctx: Any,
-                     cache: dict[str, Any] | None = None) -> contract.Context:
-    """算子要的运行期依赖（contract.Context）。
+                     cache: dict[str, Any] | None = None,
+                     probe_cache: dict[str, Any] | None = None) -> contract.Context:
+    """算子要的运行期依赖（contract.Context）。每次分析调用一份（探查器有总时限，一次调用建一个）。
 
     - today：运行当天，服务端时区（算子据此判本期完不完整）
     - catalogs：数据源名 → {表名: 这张表的目录（app/data/catalog 的 notes）}，按运行的目录视图读（tool_ctx.catalog_view，
       和 db_schema 给模型看的、冻结进表结构快照的同一份），只给快照的 SQL 里用到的表
-    - facts：数据源名 → {表名: 剖析事实（catalog_profile_jobs.last_facts）}；上传表格的剖析不是这次查的那一版的不给
-    - probe：现查一列的实际范围（W1-a 的 make_prober，按这次运行固定的数据版本解析数据源；没有这个模块时为 None）
+    - facts：数据源名 → 取剖析事实的函数（analysis.probe.profile_facts：上传表格的剖析不是这次查的那一版的不给）
+    - probe：analysis.make_prober(数据源名 → 查询用的源)：按这次运行固定的数据版本解析数据源（上传表格查的是固定的那一版），
+      遮罩列按数据源设的不探，探不成由算子退回剖析事实；一个源都解析不出来时为 None
+
+    tool_ctx 是查询用的那个 ToolContext（固定的数据版本、目录视图都从它取）。cache 是调用方的缓存（同一个节点里分析好几次，
+    目录、数据源读一次）；probe_cache 是数据源名 → coverage.Cache（同一张表的范围一个节点里只查一次）。读不到的源当没有：
+    目录、剖析、探查都是附加的，不能挡住分析。"""
+    from app.analysis.probe import profile_facts
 
-    tool_ctx 是查询用的那个 ToolContext（固定的数据版本、目录视图都从它取）。cache 是调用方的缓存（同一个节点里分析
-    好几次，目录、剖析事实读一次）。读不到的源当没有：目录、剖析是附加说明，不能挡住分析。"""
     cache = cache if cache is not None else {}
     catalogs: dict[str, Any] = {}
     facts: dict[str, Any] = {}
-    found: dict[str, Any] = {}
+    sources: dict[str, Any] = {}
     for name, tables in _tables_of(snapshots).items():
         key = f"{name}\n" + "\n".join(sorted(tables))
         if key not in cache:
@@ -236,17 +244,16 @@ async def context_of(snapshots: Iterable[Mapping[str, Any]], *, tool_ctx: Any,
         entry = cache[key]
         if entry is None:
             continue
-        found[name] = entry["source"]
-        catalogs[name], facts[name] = entry["catalog"], entry["facts"]
-    return {"today": date.today(), "probe": _prober(found, facts), "catalogs": catalogs, "facts": facts}
+        sources[name] = entry["source"]
+        catalogs[name] = entry["catalog"]
+        facts[name] = profile_facts(entry["source"])
+    return {"today": date.today(), "probe": _prober(sources, probe_cache), "catalogs": catalogs, "facts": facts}
 
 
 async def _source_context(name: str, tables: set[str], tool_ctx: Any) -> dict[str, Any] | None:
-    """一个数据源这几张表的目录、剖析事实，和查询用的那个源（探查用）。读不到为 None。"""
+    """一个数据源这几张表的目录，和查询用的那个源（探查、剖析事实都按它）。读不到为 None。"""
     from app.data import catalog as data_catalog
     from app.data import catalog_view
-    from app.data.catalog_profile_jobs import last_facts
-    from app.db.base import SessionLocal
     from app.tools.datasource import resolved_source
 
     try:
@@ -255,61 +262,24 @@ async def _source_context(name: str, tables: set[str], tool_ctx: Any) -> dict[st
             return None
         row, source = resolved
         entries = await catalog_view.read(catalog_view.CatalogView.parse(getattr(tool_ctx, "catalog_view", None)), row)
-        names = {data_catalog.resolve_table_name(getattr(source, "schema_cache", None), t) or t for t in tables}
-        async with SessionLocal() as session:
-            profiled = await last_facts(session, str(row.id), sorted(names)) if names else {}
-    except Exception:  # noqa: BLE001 - 目录、剖析读不到当没有（目录视图指定的快照不在了另由查询工具照实报）
-        logger.warning("分析读取数据源 %s 的目录、剖析事实失败，这次不附", name, exc_info=True)
+    except Exception:  # noqa: BLE001 - 目录读不到当没有（目录视图指定的快照不在了另由查询工具照实报）
+        logger.warning("分析读取数据源 %s 的目录失败，这次不附", name, exc_info=True)
         return None
-    snapshot = getattr(source, "snapshot_id", None)
-    return {
-        "source": source,
-        "catalog": {t: dict(entries[t].notes or {}) for t in sorted(names) if t in entries},
-        # 上传表格：剖析时是另一版数据，范围对不上这次读的这一版（同 tools/datasource 的 _coverage）
-        "facts": {t: f for t, f in profiled.items() if not (snapshot and f.get("snapshot_id") != snapshot)},
-    }
+    names = {data_catalog.resolve_table_name(getattr(source, "schema_cache", None), t) or t for t in tables}
+    return {"source": source, "catalog": {t: dict(entries[t].notes or {}) for t in sorted(names) if t in entries}}
 
 
-def _probe_factory() -> Any:
-    """W1-a 的 app.analysis.probe.make_prober；还没有这个模块时为 None（算子只用剖析事实，核对不了的写「核对不了」）。"""
-    try:
-        from app.analysis.probe import make_prober  # type: ignore[import-not-found]
-    except ModuleNotFoundError as e:
-        if e.name not in ("app.analysis.probe", "app.analysis"):
-            raise
-        return None
-    return make_prober
-
-
-def _prober(sources: Mapping[str, Any], facts: Mapping[str, Any]) -> contract.Prober | None:
-    """按数据源名分派的探查（contract.Prober）。每个源第一次被问到时才建它自己的探查：
-
-        make_prober(查询用的源, masked=遮罩列, cache=coverage.Cache(), facts=按表取剖析事实的协程函数)
-
-    签名以 W1-a 的交付为准（合并时由编排者对齐这一处）。遮罩列上的不探（data/coverage 的规矩），探查失败由它自己退回
-    剖析事实。"""
-    make = _probe_factory()
-    if make is None or not sources:
+def _prober(sources: Mapping[str, Any], probe_cache: dict[str, Any] | None) -> contract.Prober | None:
+    """这次分析调用的探查器（analysis.make_prober）。遮罩列、剖析事实按缺省（数据源设的遮罩列；profile_facts 现读）；
+    coverage.Cache 按数据源名放在调用方给的 probe_cache 里，同一个节点里的几次分析共用。"""
+    if not sources:
         return None
+    from app import analysis
     from app.data import coverage
-    from app.data.engine import masked_columns
 
-    built: dict[str, Any] = {}
-
-    async def probe(source: str, table: str, column: str) -> contract.Probe | None:
-        if source not in built:
-            obj = sources.get(source)
-            known = (facts.get(source) or {}) if isinstance(facts, Mapping) else {}
-
-            async def by_table(name: str, known: Mapping[str, Any] = known) -> Mapping[str, Any] | None:
-                return known.get(name)
-
-            built[source] = None if obj is None else make(
-                obj, masked=masked_columns(getattr(obj, "options", None)), cache=coverage.Cache(), facts=by_table)
-        found = built[source]
-        return await found(source, table, column) if found is not None else None
-
-    return probe
+    shared = probe_cache if probe_cache is not None else {}
+    return analysis.make_prober(dict(sources), cache={name: shared.setdefault(name, coverage.Cache())
+                                                      for name in sources})
 
 
 def tool_context(run_ctx: Any, node_id: str, emit: Any = None) -> Any:

backend/app/tools/builtin/analysis.py

@@ -42,9 +42,6 @@ NO_SCOPE = ("分析工具只能由 Agent 在本节点查过数之后调用：它
             "模板里要固定做的分析，请改用「分析卡」节点")
 #: 交回 agent 的说法：本节点还没有成功的查询
 NO_QUERIES = "本节点还没有成功的查询可以分析：先用数据库查询工具查出要分析的数，再按交回的编号（Q1、Q2……）调用分析工具"
-#: 这个节点的分析次数用完
-MAX_CALLS = ("这个节点已经做了 {done} 次分析，达到「{label}」的上限（{limit} 次），不能再分析了。"
-             "请基于已有的分析结论给出结论，没查清的照实说")
 
 _QUERY = "要分析的查询结果：本节点交回过的编号，如 Q3（和「【证据 Q3】」同一个）"
 _LABEL = "度量的叫法，系统写判定语时用（如「销售额」）；不要写数字、百分号和单位，留空就用列名"
@@ -118,20 +115,20 @@ class DifferArgs(BaseModel):
         description="rate：比率（转化率、退款率），给分子、分母两列；mean：均值，给每组的均值、标准差、样本数三列，"
                     "或合计、平方和、样本数三列；rows：逐行的值，算子自己算均值和方差（结果不能截断）")
     query: str = Field(description=_QUERY)
-    by: list[str] | None = Field(default=None, description="分组列的列名")
-    reference: str | None = Field(default=None, description="对照组：分组列里的一个值；留空和整体比")
+    by: list[str] | None = Field(default=None, description="按组比时：分组列的列名（一列）")
     pairs: list[list[str]] | None = Field(default=None, description="只比这几对：每对写两个组在分组列里的值")
-    period: str | None = Field(default=None, description="按期次比同一组的两期时：期次列")
+    reference: str | None = Field(default=None, description="对照组：分组列里的一个值，其余各组各和它比")
+    period: str | None = Field(default=None, description="按期次比同一组的两期时：期次列（SQL 里先筛好那一组）")
     base: str | None = Field(default=None, description="按期次比时：基期")
     target: str | None = Field(default=None, description="按期次比时：本期")
     numerator: str | None = Field(default=None, description="分子列：成功数（rate 用，如退款单数）")
     denominator: str | None = Field(default=None, description="分母列：试验数（rate 用，如订单数）")
-    mean: str | None = Field(default=None, description="每组的均值列（mean 用）")
-    std: str | None = Field(default=None, description="每组的标准差列（mean 用）")
-    sum: str | None = Field(default=None, description="每组的合计列（mean 用，没有标准差时和平方和一起给）")
-    sum_sq: str | None = Field(default=None, description="每组的平方和列（mean 用）")
+    mean: str | None = Field(default=None, description="每组的均值列（mean 用，和标准差、样本数一起给）")
+    sd: str | None = Field(default=None, description="每组的标准差列（mean 用）")
+    sum: str | None = Field(default=None, description="每组的合计列（mean 用，没有标准差时和平方和、样本数一起给）")
+    sumsq: str | None = Field(default=None, description="每组的平方和列（mean 用）")
     n: str | None = Field(default=None, description="每组的样本数列（mean 用）")
-    measure: str | None = Field(default=None, description="逐行的值列（rows 用）")
+    value: str | None = Field(default=None, description="逐行的值列（rows 用）")
     label: str | None = Field(default=None, description=_LABEL)
     hypothesis: str | None = Field(default=None, description=_HYPOTHESIS)
 
@@ -143,7 +140,8 @@ class ComparableArgs(BaseModel):
     period: str | None = Field(default=None, description="同一份查询时：期次列")
     base: str | None = Field(default=None, description="同一份查询时：基期")
     target: str | None = Field(default=None, description="同一份查询时：本期")
-    label: str | None = Field(default=None, description=_LABEL)
+    measure: str | None = Field(default=None, description="度量列：按它找时间条件、认它是不是可累加的量；只有一列数值列时可以不写")
+    grain: Literal["day", "week", "month", "quarter"] | None = Field(default=None, description=_GRAIN)
     hypothesis: str | None = Field(default=None, description=_HYPOTHESIS)
 
 
@@ -179,10 +177,10 @@ async def run_tool(op: str, ctx: ToolContext, args: BaseModel) -> dict[str, Any]
     """一次 agent 分析：读本节点的查询容器、按设置快照取阈值、调算子、结论落成工件。交回 {analysis_artifact, result}：
     记台账、编号 An、给模型看的那一段由节点在记台账时做（engine/nodes/llm.py 的 _record_call），重放时照样做一遍。
 
-    调用写错（含这个节点的分析次数用完）抛 contract.CallError，节点作为工具错误交回模型、不记台账。"""
-    import app.analysis as analysis_ops   # 按模块取 run：测试、合并前的替身都只换模块上的这一个函数
+    调用写错（含这个节点的分析次数用完，analysis.exhausted）抛 contract.CallError，节点作为工具错误交回模型、不记台账。
+    结论原样存成工件、一个字段都不加：下钻时算子按上一层结论的内容哈希认 parent 的工件。"""
+    import app.analysis as analysis_ops   # 按模块取 run：测试换替身时只换模块上的这一个函数
 
-    from app.analysis.contract import LABELS
     from app.tools.analysis import context_of, prepare, scope_of, settings_of, store
 
     scope = scope_of(ctx.extra)
@@ -192,11 +190,11 @@ async def run_tool(op: str, ctx: ToolContext, args: BaseModel) -> dict[str, Any]
         raise CallError("no_queries", NO_QUERIES)
     settings = await settings_of(scope.snapshot)
     if len(scope.results) >= settings["max_calls"]:
-        raise CallError("max_calls", MAX_CALLS.format(done=len(scope.results), label=LABELS["max_calls"],
-                                                      limit=settings["max_calls"]))
+        raise analysis_ops.exhausted(settings)          # 次数用完的说法只有算子那一处
     request = request_of(op, args)
     await prepare(scope)
-    context = await context_of(list(scope.queries.values()), tool_ctx=ctx, cache=scope.source_cache)
+    context = await context_of(list(scope.queries.values()), tool_ctx=ctx, cache=scope.source_cache,
+                               probe_cache=scope.probe_cache)
     result = await analysis_ops.run(request, scope.resolve, settings=settings, context=context, parents=scope.parent)
     artifact = await store(result, run_id=ctx.run_id, node_id=scope.node_id)
     return {"analysis_artifact": artifact, "result": result}
@@ -242,7 +240,7 @@ async def analysis_trend(ctx: ToolContext, **kwargs: Any) -> dict[str, Any]:
     name=contract.TOOL_NAMES["differ"],
     category=CATEGORY,
     description="差异检验：判断两组（或同一组的两期）的差异站得住，还是可能只是样本波动。比率给分子、分母两列（rate），"
-                "均值给均值、标准差、样本数（mean），也可以给逐行的值（rows）；样本小时照实标出。",
+                "均值给均值、标准差、样本数或合计、平方和、样本数（mean），也可以给逐行的值（rows）；样本小时照实标出。",
     args_schema=DifferArgs,
     guide=METHOD_GUIDE,
 )

backend/app/tools/datasource.py

@@ -804,29 +804,15 @@ def _unchecked_tables(sql: str, source: Any, outside: frozenset[str]) -> list[st
 
 async def _coverage(source: Any, sql: str, payload: Mapping[str, Any], mask: list[str]) -> list[dict[str, Any]]:
     """查询区间里这张表没有数据的那几列（data/coverage.result_gaps）。只在结果里有整列是 0 或空的数值列时才查库；遮罩列
-    不探查；现查表的范围不成时退回剖析、取值观察记下的范围（上传表格的快照对不上就不用）。从不抛异常。"""
+    不探查；现查表的范围不成时退回剖析、取值观察记下的范围（上传表格的快照对不上就不用）。从不抛异常。
+
+    剖析事实的取法只有一处（analysis.probe.profile_facts）：分析的可比性检查现查不成时退回的也是它。"""
+    from app.analysis.probe import profile_facts
     from app.data import coverage
 
     if not coverage.zero_columns(payload):
         return []
-
-    async def facts(table: str) -> Mapping[str, Any] | None:
-        from app.data.catalog_profile_jobs import last_facts
-        from app.db.base import SessionLocal
-
-        source_id = getattr(source, "id", None)
-        if not source_id:
-            return None
-        name = data_catalog.resolve_table_name(getattr(source, "schema_cache", None), table) or table
-        async with SessionLocal() as session:
-            found = (await last_facts(session, str(source_id), [name])).get(name)
-        snapshot = getattr(source, "snapshot_id", None)
-        if found and snapshot and found.get("snapshot_id") != snapshot:
-            # 上传表格：剖析时是另一版数据，范围对不上这次读的这一版
-            return None
-        return found
-
-    return await coverage.result_gaps(source, sql, payload, facts=facts, masked=mask)
+    return await coverage.result_gaps(source, sql, payload, facts=profile_facts(source), masked=mask)
 
 
 async def _sql_checks(checker: SqlChecker, sql: str) -> list[dict[str, Any]]:

backend/tests/test_analysis_card.py

@@ -158,6 +158,9 @@ def test_a_well_formed_card_validates():
      "输入里"),
     ([{**WOW, "params": {**WOW["params"], "bogus": 1}}], "items[0].params.bogus", "analysis.params_invalid",
      "没有「bogus」"),
+    # 运行当天、目录事实、下钻层数由系统填（算子自己记），分析项里写了不收
+    ([{**WOW, "params": {**WOW["params"], "today": "2026-10-21"}}], "items[0].params.today",
+     "analysis.params_invalid", "没有「today」"),
     ([{**WHERE, "params": {**WHERE["params"], "headline": {"base": "weekly.gmv", "target": "weekly.r1.gmv"}}}],
      "items[0].params.headline.base", "analysis.params_invalid", "节点 id.r行号.列名"),
     ([{**WHERE, "params": {**WHERE["params"], "headline": "weekly.r0.gmv"}}], "items[0].params.headline",

backend/tests/test_analysis_e2e.py

@@ -0,0 +1,314 @@
+"""分析入口接真算子跑一遍（不用替身）：agent 分析工具（含下钻）、分析卡（波动基线 + 变化拆解）、口径卡的可比性挂钩、
+审批恢复时的重放。断言工件、台账、交回模型的那一段、节点产出，并用 analysis.recompute 按封存的工件复算、逐字段相同。
+
+虚构零售库 shop：orders(day, region, channel, amount)，2026-03-02 起 12 个整周，每周每个地区、渠道两笔（周三、周日），
+最后一周华东掉三成。模型换成脚本（MockChatModel._decide），不调用任何模型。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sqlite3
+from datetime import date, timedelta
+
+import pytest
+from langchain_core.messages import AIMessage, ToolMessage
+from sqlalchemy import select
+
+import app.analysis as analysis_ops
+from app.analysis import contract
+from app.core import artifact_store
+from app.core.artifact_store import canonical_json, content_hash
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource, Run, RunEvent
+from app.engine.runner import run_manager
+
+TOOL = "db_query__shop"
+DECOMPOSE = contract.TOOL_NAMES["decompose"]
+WEEK = "date(day, '-6 days', 'weekday 1')"
+BY_REGION = (f"SELECT region, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' "
+             "GROUP BY region, week ORDER BY region, week")
+TOTALS = f"SELECT {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' GROUP BY week ORDER BY week"
+EAST = (f"SELECT channel, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE region = '华东' AND day >= '2026-05-11' "
+        "GROUP BY channel, week ORDER BY channel, week")
+WEEKLY = f"SELECT {WEEK} AS week, SUM(amount) AS gmv FROM orders GROUP BY week ORDER BY week"
+TWO_WEEKS = {"period": "week", "base": "2026-05-11", "target": "2026-05-18", "measure": "gmv"}
+A1_ARGS = {"mode": "additive", "query": "Q1", "by": ["region"], **TWO_WEEKS, "label": "销售额",
+           "headline": {"base": "Q2.r0.gmv", "target": "Q2.r1.gmv"}, "hypothesis": "下降是不是集中在某个地区"}
+A2_ARGS = {"mode": "additive", "query": "Q3", "by": ["channel"], **TWO_WEEKS, "parent": {"of": "A1", "key": ["华东"]},
+           "hypothesis": "华东是哪个渠道掉的"}
+
+
+def _orders():
+    out, i = [], 0
+    base = {"华东": 400, "华南": 300, "华北": 250, "西南": 50}
+    for w in range(12):
+        monday = date(2026, 3, 2) + timedelta(weeks=w)
+        for region, amount in base.items():
+            for channel, share in (("线上", 3), ("线下", 2)):
+                value = amount * share // 5 + (w % 3) * 2
+                if w == 11 and region == "华东":
+                    value = value * 7 // 10
+                for offset in (2, 6):
+                    i += 1
+                    out.append((i, (monday + timedelta(days=offset)).isoformat(), region, channel, value))
+    return out
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@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, day TEXT, region TEXT, channel TEXT, amount INTEGER);")
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", _orders())
+    db.commit()
+    db.close()
+    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, row.options = str(path), "sqlite", True, True, {}
+        await session.commit()
+        source_id = row.id
+    await engines.invalidate(source_id)
+    yield
+    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:])]}
+
+
+class Script:
+    """Agent 依次发 calls 里的工具调用，再给结论；seen 记下每次带工具的调用看到的消息。"""
+
+    def __init__(self, monkeypatch, calls):
+        from app.providers import mock_model
+
+        self.seen: list[list] = []
+
+        def decide(model, messages):
+            if not model.tools:
+                return AIMessage(content="收尾")
+            self.seen.append(list(messages))
+            done = sum(isinstance(m, ToolMessage) for m in messages)
+            if done < len(calls):
+                name, args = calls[done]
+                return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": f"call_{done}"}])
+            return AIMessage(content="销售额下降主要来自华东。")
+
+        monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
+
+    def replies(self):
+        return [m.content for m in self.seen[-1] if isinstance(m, ToolMessage)]
+
+
+class Counting:
+    """真算子外面包一层只数次数（重放时不该再算）。"""
+
+    def __init__(self, monkeypatch):
+        self.calls = 0
+        real = analysis_ops.run
+
+        async def run(*args, **kwargs):
+            self.calls += 1
+            return await real(*args, **kwargs)
+
+        monkeypatch.setattr(analysis_ops, "run", run)
+
+
+async def wait(run_id, statuses=("succeeded", "failed")):
+    for _ in range(600):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, run_id)
+        if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None):
+            return row
+    raise AssertionError(f"run {run_id} 没有结束：{row.status}")
+
+
+async def events(run_id, etype=None, node_id=None):
+    async with SessionLocal() as session:
+        q = select(RunEvent).where(RunEvent.run_id == run_id)
+        if etype:
+            q = q.where(RunEvent.type == etype)
+        if node_id:
+            q = q.where(RunEvent.node_id == node_id)
+        return [e.data for e in (await session.execute(q.order_by(RunEvent.seq))).scalars()]
+
+
+def agent_graph(approval="never"):
+    return chain(node("start", "input"),
+                 node("bot", "agent", prompt="销售额为什么下降", tools=[TOOL, DECOMPOSE], approval=approval),
+                 node("out", "output", fields=[{"name": "answer", "value": "{{ nodes.bot.text }}"}]))
+
+
+def recomputes(stored):
+    """按封存的工件复算（查询快照、上一层结论都按 sha256 取回），和封存的逐字段相同。"""
+    again = analysis_ops.recompute(stored, artifact_store.load, load_result=artifact_store.load)
+    assert contract.same(stored, again), {k: (stored.get(k), again.get(k)) for k in stored if stored.get(k) != again.get(k)}
+
+
+AGENT_CALLS = [(TOOL, {"sql": BY_REGION}), (TOOL, {"sql": TOTALS}), (TOOL, {"sql": EAST}),
+               (DECOMPOSE, A1_ARGS), (DECOMPOSE, A2_ARGS)]
+
+
+# --------------------------------------------------------------------------
+# agent 分析工具：拆解、下钻
+# --------------------------------------------------------------------------
+
+
+async def test_an_agent_decomposes_and_drills_down_with_the_real_operator(monkeypatch):
+    script = Script(monkeypatch, AGENT_CALLS)
+    run = await run_manager.start(graph=agent_graph(), input_payload={})
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+
+    ends = await events(row.id, "tool.end", "bot")
+    q1, q2, q3 = (ends[i]["query_artifact"] for i in range(3))
+    a1_artifact, a2_artifact = ends[3]["analysis_artifact"], ends[4]["analysis_artifact"]
+    a1, a2 = artifact_store.load(a1_artifact), artifact_store.load(a2_artifact)
+
+    # 结论：华东是主要来源，下钻到渠道是线上；总数闭合（headline 是 Q2 的两格）
+    assert a1["status"] == "ok" and a1["verdict"] == {"code": "driver", "subjects": [["华东"]], "direction": "down"}
+    assert "主要来自华东" in a1["phrase"] and a1["hypothesis"] == "下降是不是集中在某个地区"
+    assert {i["ref"] for i in a1["inputs"]} >= {"Q1", "Q2.r0.gmv", "Q2.r1.gmv"}
+    assert a2["verdict"]["code"] == "driver" and a2["verdict"]["subjects"] == [["线上"]]
+    # 下钻：算子按上一层结论的内容哈希认 parent，就是存下的那份工件
+    [parent] = [i for i in a2["inputs"] if i.get("role") == "parent"]
+    assert parent == {"ref": "A1", "artifact": a1_artifact, "role": "parent"}
+    assert content_hash(canonical_json(a1)) == a1_artifact
+
+    # 台账：两条 kind analysis，带 call_id，输入是那几份查询快照（下钻的另有上一层结论）
+    [finished] = await events(row.id, "node.finished", "bot")
+    analyses = [e for e in finished["evidence"] if e["kind"] == "analysis"]
+    assert [(e["call_id"], e["artifact"]) for e in analyses] == [("call_3", a1_artifact), ("call_4", a2_artifact)]
+    assert set(analyses[0]["inputs"]) == {q1, q2} and set(analyses[1]["inputs"]) == {q3, a1_artifact}
+    assert analyses[0] == contract.ledger_entry(a1, node_id="bot", exec_no=1, artifact=a1_artifact, call_id="call_3")
+
+    # 交回模型的：编号、判定语、关键数
+    replies = script.replies()
+    assert [r.split("\n", 1)[0] for r in replies] == ["【证据 Q1】", "【证据 Q2】", "【证据 Q3】", "【分析 A1】", "【分析 A2】"]
+    body = json.loads(replies[3].split("\n", 1)[1])
+    assert body["phrase"] == a1["phrase"] and body["verdict"] == "driver" and body["values"] == a1["values"]
+    assert body["table"]["rows"][0][0] == "华东"
+
+    # 按封存的工件复算，逐字段相同
+    recomputes(a1)
+    recomputes(a2)
+
+
+async def test_replaying_after_approvals_does_not_recompute(monkeypatch):
+    counted = Counting(monkeypatch)
+    script = Script(monkeypatch, AGENT_CALLS)
+    run = await run_manager.start(graph=agent_graph(approval="always"), input_payload={})
+    for _ in AGENT_CALLS:
+        row = await wait(run.id, ("interrupted", "failed", "succeeded"))
+        assert row.status == "interrupted", row.error
+        await run_manager.resume(run.id, {"approved": True})
+    row = await wait(run.id, ("succeeded", "failed", "interrupted"))
+    assert row.status == "succeeded", row.error
+
+    assert counted.calls == 2, "重放时分析又算了一次"
+    ends = await events(row.id, "tool.end", "bot")
+    assert len(ends) == len(AGENT_CALLS)
+    a1_artifact, a2_artifact = ends[3]["analysis_artifact"], ends[4]["analysis_artifact"]
+    assert [r.split("\n", 1)[0] for r in script.replies()][-2:] == ["【分析 A1】", "【分析 A2】"]
+    [parent] = [i for i in artifact_store.load(a2_artifact)["inputs"] if i.get("role") == "parent"]
+    assert parent["artifact"] == a1_artifact
+    [finished] = await events(row.id, "node.finished", "bot")
+    assert [e["artifact"] for e in finished["evidence"] if e["kind"] == "analysis"] == [a1_artifact, a2_artifact]
+
+
+# --------------------------------------------------------------------------
+# 分析卡：波动基线 + 变化拆解
+# --------------------------------------------------------------------------
+
+
+async def test_an_analysis_card_runs_the_real_operators():
+    items = [
+        {"id": "wow", "op": "baseline", "from": "weekly", "params": {"period": "week", "measure": "gmv", "label": "销售额"}},
+        {"id": "where", "op": "decompose", "mode": "additive", "from": "by_region",
+         "params": {"by": ["region"], **TWO_WEEKS, "label": "销售额",
+                    "headline": {"base": "weekly.r10.gmv", "target": "weekly.r11.gmv"}}},
+    ]
+    graph = chain(node("start", "input"), node("weekly", "tool", tool=TOOL, args={"sql": WEEKLY}),
+                  node("by_region", "tool", tool=TOOL, args={"sql": BY_REGION}),
+                  node("card", "analysis", items=items),
+                  node("out", "output", fields=[{"name": "判定", "value": "{{ nodes.card.results.wow.phrase }}"}]))
+    run = await run_manager.start(graph=graph, input_payload={})
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+
+    [finished] = await events(row.id, "node.finished", "card")
+    results = artifact_store.load(finished["artifact"])["results"]
+    wow, where = results["wow"], results["where"]
+    assert wow["status"] == "ok" and wow["verdict"] in ("outside_band", "far_outside_band") and wow["direction"] == "down"
+    assert where["status"] == "ok" and where["verdict"] == "driver" and "主要来自华东" in where["phrase"]
+    assert row.output["判定"] == wow["phrase"]
+
+    baseline = artifact_store.load(wow["artifact"])
+    # 本期完不完整按现查的表范围判（analysis.make_prober 按这次运行的数据源探 orders.day）：探到了，不再说核对不了
+    [probe] = baseline["probes"]
+    assert (probe["source"], probe["table"], probe["column"], probe["basis"]) == ("shop", "orders", "day", "live")
+    assert probe["max"] == "2026-05-24"
+    assert not [c for c in baseline["caveats"] if c["code"] == "comparability"]
+    decomposed = artifact_store.load(where["artifact"])
+    assert decomposed["verdict"]["subjects"] == [["华东"]]
+    entries = finished["evidence"]
+    assert [(e["kind"], e["item"], e["artifact"]) for e in entries] == [
+        ("analysis", "wow", wow["artifact"]), ("analysis", "where", where["artifact"])]
+    recomputes(baseline)
+    recomputes(decomposed)
+
+
+# --------------------------------------------------------------------------
+# 口径卡的可比性挂钩：真 comparable
+# --------------------------------------------------------------------------
+
+
+def _week_sql(start: str, end: str) -> str:
+    return f"SELECT SUM(amount) AS gmv FROM orders WHERE day BETWEEN '{start}' AND '{end}'"
+
+
+@pytest.mark.parametrize("this_week, verdict, has_reasons", [
+    (("2026-05-18", "2026-05-24"), "comparable", False),
+    (("2026-05-18", "2026-05-23"), "conditional", True),       # 本期 6 天对上期 7 天：窗口长度、周末天数都不一样
+])
+async def test_a_caliber_card_checks_two_periods_with_the_real_operator(this_week, verdict, has_reasons):
+    metrics = [{"id": "wow", "name": "销售额环比", "decimals": 4,
+                "expression": "(cell(nodes.this_week, 0, 'gmv') - cell(nodes.last_week, 0, 'gmv'))"
+                              " / cell(nodes.last_week, 0, 'gmv')"}]
+    graph = chain(node("start", "input"), node("this_week", "tool", tool=TOOL, args={"sql": _week_sql(*this_week)}),
+                  node("last_week", "tool", tool=TOOL, args={"sql": _week_sql("2026-05-11", "2026-05-17")}),
+                  node("card", "metrics", caliber="周报口径", metrics=metrics),
+                  node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}]))
+    run = await run_manager.start(graph=graph, input_payload={})
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+    [finished] = await events(row.id, "node.finished", "card")
+    [metric] = artifact_store.load(finished["artifact"])["metrics"]
+    found = metric["comparability"]
+    stored = artifact_store.load(found["artifact"])
+    assert stored["op"] == "comparable" and stored["params"]["base_query"] == "last_week"
+    assert stored["params"]["target_query"] == "this_week"
+    assert found == {"artifact": found["artifact"], "verdict": verdict, "phrase": stored["phrase"],
+                     "reasons": analysis_ops.comparable_reasons(stored)}
+    assert bool(found["reasons"]) is has_reasons
+    # 整期按现查的表范围判
+    assert any(p.get("table") == "orders" and p.get("basis") == "live" for p in stored["probes"])
+    recomputes(stored)
+    assert not [e for e in await events(row.id, "log", "card") if e.get("code") == "metric_comparability_skipped"]

backend/tests/test_analysis_tools.py

@@ -23,7 +23,7 @@ from app.db.base import SessionLocal
 from app.db.models import Artifact, DataSource, Run, RunEvent
 from app.engine.runner import run_manager
 from app.tools.analysis import CALL_ERROR_HEAD, REFUSED_CODE
-from app.tools.builtin.analysis import ANALYSIS_METHOD_PROMPT, ARGS, MAX_CALLS, METHOD_GUIDE, NO_SCOPE
+from app.tools.builtin.analysis import ANALYSIS_METHOD_PROMPT, ARGS, METHOD_GUIDE, NO_SCOPE
 from tests.fixtures.analysis_ops import FakeOps
 
 TOOL = "db_query__shop"
@@ -173,8 +173,11 @@ def test_the_five_tools_take_no_thresholds_and_share_one_method():
     assert {"mode", "by", "period", "base", "target", "measure", "numerator", "denominator", "product", "factors",
             "headline", "parent", "label", "hypothesis"} <= set(ARGS["decompose"].model_fields)
     assert {"basis", "grain", "lag"} <= set(ARGS["baseline"].model_fields)
-    assert {"reference", "pairs", "mode"} <= set(ARGS["differ"].model_fields)
-    assert {"base_query", "target_query"} <= set(ARGS["comparable"].model_fields)
+    # 参数名以算子为准（app/analysis 各算子文件头）：差异检验的均值是 mean / sd / n 或 sum / sumsq / n，逐行是 value
+    assert {"reference", "pairs", "mode", "mean", "sd", "sum", "sumsq", "n", "value"} <= set(ARGS["differ"].model_fields)
+    assert {"base_query", "target_query", "measure", "grain"} <= set(ARGS["comparable"].model_fields)
+    # 运行当天、目录事实、下钻层数由系统填，哪个工具的参数里都没有
+    assert not any({"today", "catalog", "depth"} & set(model.model_fields) for model in ARGS.values())
 
 
 def test_unknown_parameters_and_thresholds_are_rejected_like_any_tool():
@@ -211,6 +214,22 @@ async def test_calling_it_outside_an_agent_says_where_it_belongs():
     assert found.value.message == NO_SCOPE
 
 
+def test_input_snapshots_carry_the_payload_as_is():
+    """交给算子的一份输入：查询工具的 payload 原样（SQL 检查结果、「查询区间里这张表没有数据」、查询当时记下的遮罩，可比性
+    检查按它们判数据问题），加上出处和遮罩列；合并查询写明方言（在 SQLite 里执行）。"""
+    from app.engine.merge_query import MERGE_SOURCE
+    from app.tools.analysis import input_snapshot
+
+    content = {"columns": ["region", "gmv"], "rows": [["华东", 1]], "sql": "SELECT …", "source": "shop",
+               "column_types": {"region": "text", "gmv": "number"}, "truncated": False,
+               "checks": [{"level": "error", "code": "fanout_sum"}], "no_data": [{"columns": ["gmv"]}],
+               "mask_columns": ["phone"], "masked_at_query": ["tel"]}
+    snap = input_snapshot(content, ref="Q1", artifact="a" * 64, node_id="bot", masked=["tel", "phone"])
+    assert snap == {**content, "ref": "Q1", "artifact": "a" * 64, "node_id": "bot", "masked": ["tel", "phone"]}
+    merged = input_snapshot({**content, "source": MERGE_SOURCE}, ref="m", artifact="b" * 64, node_id="m")
+    assert merged["dialect"] == "sqlite" and merged["masked"] == []
+
+
 # --------------------------------------------------------------------------
 # 方法说明：绑了分析工具的 Agent 注入一次，模型调用节点、没绑的 Agent 不注入
 # --------------------------------------------------------------------------
@@ -265,11 +284,15 @@ async def test_an_analysis_reads_this_nodes_query_and_is_recorded(monkeypatch, s
     assert snap["ref"] == "Q1" and snap["artifact"] == query_artifact and snap["node_id"] == "bot"
     assert snap["source"] == "shop" and snap["sql"] == SQL and snap["truncated"] is False
     assert snap["columns"] == ["region", "week", "gmv"] and snap["rows"][0] == ["华东", "2026-W41", 820.0]
-    assert snap["column_types"] == ["text", "text", "number"] and snap["masked"] == []
+    # 查询工具的 payload 原样（工件库里那份），加上出处和遮罩列
+    assert snap["column_types"] == {"region": "text", "week": "text", "gmv": "number"} and snap["masked"] == []
+    assert snap["row_count"] == 6 and snap == {**artifact_store.load(query_artifact), "ref": "Q1",
+                                               "artifact": query_artifact, "node_id": "bot", "masked": []}
     # 阈值：没有发起时的快照，现读设置（没存过就是缺省）
     assert call["settings"] == contract.DEFAULTS
     assert set(call["context"]) == {"today", "probe", "catalogs", "facts"}
-    assert call["context"]["probe"] is None, "还没有 W1-a 的 make_prober 时不给探查"
+    # 探查器按这次运行固定的数据版本解析数据源（analysis.make_prober），剖析事实是按表取的函数（probe.profile_facts）
+    assert call["context"]["probe"].dialect("shop") == "sqlite" and callable(call["context"]["facts"]["shop"])
 
     # 结论落成工件，tool.end 带着它；台账记一条 kind analysis（带 call_id），在那次查询之后
     analysis_artifact = ends[1]["analysis_artifact"]
@@ -353,7 +376,9 @@ async def test_the_number_of_analyses_per_node_is_capped(monkeypatch, shop):
     await run_graph(agent_graph(), options={contract.SETTINGS_KEY: {"max_calls": 1}})
     assert len(ops.calls) == 1
     capped = tool_messages(script)[2]
-    assert capped == CALL_ERROR_HEAD + MAX_CALLS.format(done=1, label=contract.LABELS["max_calls"], limit=1)
+    import app.analysis as analysis_ops
+
+    assert capped == CALL_ERROR_HEAD + analysis_ops.exhausted({"max_calls": 1}).message
     # 阈值来自发起时的快照
     assert ops.calls[0]["settings"]["max_calls"] == 1
 

backend/tests/test_metric_comparability.py

@@ -1,9 +1,12 @@
 """口径卡的可比性挂钩（docs/design/analysis-round/00-总方案.md §3.4 第 3 条）：一个指标的输入落在两份不同的查询快照上、
 两份 SQL 同形只差字面量（本周对上周）时，口径卡执行时自动做一次可比性检查，结论落成工件、记在指标的 comparability 上。
 
-- 同形只差日期 → 有 comparability，早的那期是基期（和表达式里谁先写无关）；同一对快照只查一次
+- 同形只差日期 → 有 comparability，早的那期是基期（和表达式里谁先写无关）；同一对快照只查一次；原因按算子的
+  analysis.comparable_reasons 拆
 - SQL 结构不同（退款额 / 销售额）、两次是同一条 SQL → 不查、不记日志
-- 算子出错、算不了、还没有算子（第 1 波的 run 抛 NotImplementedError）→ 不挡口径卡，记一条 metric_comparability_skipped
+- 算子出错、算不了、算子没有实现（NotImplementedError）→ 不挡口径卡，记一条 metric_comparability_skipped
+
+这里的算子是替身（tests/fixtures/analysis_ops.FakeOps）；真算子跑一遍的见 tests/test_analysis_e2e.py。
 """
 from __future__ import annotations
 
@@ -18,7 +21,7 @@ from app.core import artifact_store
 from app.data.engine import engines
 from app.db.base import SessionLocal
 from app.db.models import DataSource, Run, RunEvent
-from app.engine.nodes.metrics import COMPARABILITY_SKIPPED, comparability_reasons
+from app.engine.nodes.metrics import COMPARABILITY_SKIPPED
 from app.engine.runner import run_manager
 from tests.fixtures.analysis_ops import FakeOps
 
@@ -118,12 +121,23 @@ async def test_two_periods_of_the_same_query_get_a_comparability_check(monkeypat
     assert not [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED]
 
 
+#: 可比性结论的检查表（照算子的 comparable.table_of：检查项、结果、基期、本期、说明）
+CHECKS = {"columns": ["check", "status", "base", "target", "note"], "labels": {},
+          "rows": [["同口径", "通过", None, None, None], ["窗口长度", "通过", "7 天", "7 天", None],
+                   ["整期", "不通过", None, None, "本期数据只到 2026-10-15，还差 3 天"],
+                   ["星期构成", "提醒", None, None, "两段的周末天数不同：2 天对 1 天"]]}
+
+
 async def test_an_incomparable_pair_lists_its_reasons(monkeypatch):
-    FakeOps(monkeypatch, verdicts=["incomparable"])
+    import app.analysis as analysis_ops
+
+    FakeOps(monkeypatch, verdicts=["incomparable"], table=CHECKS)
     _, card, _ = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW}]))
     found = by_id(card)["wow"]["comparability"]
-    assert found["verdict"] == "incomparable" and found["phrase"] == "两期不可比：本期只有 3 天数据"
-    assert found["reasons"] == ["本期只有 3 天数据"]
+    stored = artifact_store.load(found["artifact"])
+    assert found == {"artifact": found["artifact"], "verdict": "incomparable", "phrase": stored["phrase"],
+                     "reasons": analysis_ops.comparable_reasons(stored)}
+    assert found["reasons"] == ["本期数据只到 2026-10-15，还差 3 天", "两段的周末天数不同：2 天对 1 天"]
 
 
 async def test_different_structures_and_the_same_query_are_not_checked(monkeypatch):
@@ -136,8 +150,9 @@ async def test_different_structures_and_the_same_query_are_not_checked(monkeypat
     assert not [log for log in [*logs, *logs2] if log.get("code") == COMPARABILITY_SKIPPED]
 
 
-async def test_without_the_operator_the_card_still_runs():
-    """第 1 波的 app.analysis.run 还抛 NotImplementedError：口径卡照常算，照实记一条没做成。"""
+async def test_an_operator_that_is_not_there_does_not_block_the_card(monkeypatch):
+    """算子没有实现（合并前的骨架抛 NotImplementedError）：口径卡照常算，照实记一条没做成。"""
+    FakeOps(monkeypatch, errors=[NotImplementedError("analysis.run 还没有实现")])
     _, card, logs = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW, "decimals": 4}]))
     wow = by_id(card)["wow"]
     assert "comparability" not in wow and wow["value"] == pytest.approx((450 + 300 - 900) / 900, abs=1e-4)
@@ -157,12 +172,3 @@ async def test_failures_do_not_block_the_card(monkeypatch, ops_kw):
     assert "comparability" not in by_id(card)["wow"]
     [log] = [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED]
     assert log["reason"]
-
-
-def test_reasons_come_from_the_check_table_when_there_is_one():
-    result = {"verdict": {"code": "conditional"}, "phrase": "两期有条件可比：两段的周末天数不同",
-              "table": {"columns": ["check", "status", "reason"], "labels": {},
-                        "rows": [["同口径", "pass", None], ["窗口长度", "warn", "窗口长度不同：7 天对 6 天"],
-                                 ["星期构成", "warn", "两段的周末天数不同：2 天对 1 天"], ["整期", "unknown", "核对不了"]]}}
-    assert comparability_reasons(result) == ["窗口长度不同：7 天对 6 天", "两段的周末天数不同：2 天对 1 天"]
-    assert comparability_reasons({"verdict": {"code": "comparable"}, "phrase": "两期可比"}) == []