评测的两种分析期望（贡献、判定）：期望 SQL 的探针快照上用平台的算子（app.analysis.run，随数据变化的调用错误记成算不了）算出期望的判定，评分沿用五种结论

- 题目（cases.py）：contribution 贡献恒为变化拆解（可加、比率、乘积）、另写 top_n（期望点名主要来源的前几段，缺省 1）；verdict 判定是波动基线、趋势判断、差异检验、可比性检查。参数同分析工具（不写输入、总数、下钻，不写阈值），文字参数可以带 {{ schedule.* }}；来源、方向、判定不许人填（HUMAN_VALUE_KEYS 加 segment、subjects、direction、verdict 等）；口径题、期望反问的题配不了
- 算期望（gold.py、judgements.py）：遮罩列、目录事实、现查表的日期范围和分析工具同一套取法；阈值用这次评测的分析设置快照（Context.analysis），试算时现读设置；期望 SQL 只返回两期时可以不写基期、本期；算不了的这一项照实记（评分时不能评分），参数和结果对不上的整题没有可用的值；截断的快照照旧拒收。作答前后两次期望、和上一次比的「数据变了」都不比运行当天、现查时刻、输入快照的工件（stable_sets）
- 评分（score.py、judgements.py，SCORER_VERSION 3）：判定码归到同一套判定类再比；答案引用了（判定语、具名的数、结果表的格、句末依据）判定对得上的分析结论是对且可追溯，引用它的那句却写反了不算；没引用但正文按说法表（engine/analysis_claims 同一个 scan、同一套绑定规则）写对了是对但没引用；判定、来源、方向对不上或写成相反、根本没做也没说是错。答案的分析结论从作答运行的台账读、引用从报告文档读，有判断词的句子一并读出；逐题 E1 引用分析结论时按它的输入查询比（下钻顺着上一层找）；判定语不再当成答案里的数；老结果的明细里没有这两项按空读
- 执行（runner.py、answer.py）：有分析期望的题集开评时拍一份分析设置，算期望、各臂作答的运行（Run.options 的 analysis）都按它，记进指纹（compare 的「分析设置」一项），重试沿用；别的题集的指纹、作答运行的选项不变。助手提示词的指纹加上分析方法（guide 的 id 和全文）、写作者的分析规则和说法表版本
- 总表（stats.py）：各期望的结论另按期望的种类各一份（outcomes_by_kind）

backend/app/eval/answer.py

@@ -14,7 +14,8 @@ app/chat/job.py），生成过程是 api/copilot.answer_ops，几个「决定」
   同一个规矩先带着原因重来一次，还不行才记建图失败。每次重来的原因照实记在 retries。
 - 运行：fill_input 填入口字段（单轮，history 为空），来源 eval，名字「评测：题集名·第 n 题」，选项带目录视图（Run.options
   的 catalog_view：运行时的 db_schema、冻结快照、自查都按它）、只读和遮罩（同使用者的问数据运行：评测页会显示答案里的数，
-  遮罩值不能从这里露出来；评测不写业务库）。写作模型按变体写进图的 defaults：节点自己点了模型的不覆盖，照实记在
+  遮罩值不能从这里露出来；评测不写业务库），有分析期望的题集另带这次评测的分析设置快照（判定用的阈值和算期望同一份）。
+  写作模型按变体写进图的 defaults：节点自己点了模型的不覆盖，照实记在
   own_models。
 - 终态：停在审批上（危险工具、人工节点）的直接取消、记「评测不代批」；被驳回的（cancelled + end_reason=rejected，总方案
   X9）记路径「被驳回」，不算运行失败；失败、被停止记运行失败。
@@ -68,6 +69,8 @@ class TurnContext:
     writer: Mapping[str, Any] = field(default_factory=dict)
     #: 目录视图（Run.options['catalog_view'] 的形状）
     view: Mapping[str, Any] | None = None
+    #: 这次评测的分析设置快照（有分析期望的题集才有）：作答的运行按它判（Run.options['analysis']），和算期望同一份
+    analysis: Mapping[str, Any] | None = None
 
 
 async def writer_usage(run_ids: list[str]) -> dict[str, Any]:
@@ -116,14 +119,18 @@ def with_writer(graph: dict[str, Any], writer: Mapping[str, Any]) -> tuple[dict[
     return out, own
 
 
-def _options(view: Mapping[str, Any] | None) -> dict[str, Any]:
-    """作答运行带的选项：目录视图、只读、遮罩（见模块说明）。"""
+def _options(view: Mapping[str, Any] | None, analysis: Mapping[str, Any] | None = None) -> dict[str, Any]:
+    """作答运行带的选项：目录视图、只读、遮罩（见模块说明）；有分析期望的题集另带这次评测的分析设置快照（发起时
+    engine/launch 见调用方给了就照它的，不再现读设置）：判「主要来自」「超出常规波动」的阈值和算期望的是同一份。"""
+    from app.analysis.contract import SETTINGS_KEY
     from app.data.catalog_view import OPTION_KEY
     from app.tools.datasource import MASK_RESULTS_OPTION, READ_ONLY_OPTION
 
     out: dict[str, Any] = {MASK_RESULTS_OPTION: True, READ_ONLY_OPTION: True}
     if view:
         out[OPTION_KEY] = dict(view)
+    if analysis is not None:
+        out[SETTINGS_KEY] = dict(analysis)
     return out
 
 
@@ -168,8 +175,8 @@ class Driver(TurnDriver):
             "suite_id": self.ctx.suite_id, "eval_run_id": self.ctx.eval_run_id, "case_id": self.ctx.case_id,
             "role": "answer", "arm": self.ctx.arm, "repeat": self.ctx.repeat, "attempt": self.turn.attempts}.items()
             if v is not None}
-        return {"actor": self.ctx.actor, "options": _options(self.ctx.view), "origin_kind": "eval", "origin": origin,
-                "workflow_name": run_name(self.ctx.suite_name, self.ctx.case_seq)}
+        return {"actor": self.ctx.actor, "options": _options(self.ctx.view, self.ctx.analysis), "origin_kind": "eval",
+                "origin": origin, "workflow_name": run_name(self.ctx.suite_name, self.ctx.case_seq)}
 
     async def on_interrupted(self, run_id: str, row: Any) -> Any | None:
         t = self.turn

backend/app/eval/cases.py

@@ -14,8 +14,14 @@ history 是多轮追问的位置：评测-D12 本轮只做单轮，非空一律
 - none 期望答「没有数据」：期望 SQL 返回 0 行。
 - ask 期望反问（问法含糊）：不能和要答出数的期望写在一起，不要期望答案的来源。
 - text 文字要点（points）：由裁判判，单列「模型判断」，不计入准确率（评测-D4），所以 required 恒为 false。
+- contribution 贡献：在期望 SQL 的结果上做一次变化拆解（mode：additive / ratio / factor；params 同分析工具的参数，
+  见 app/eval/judgements.py），top_n 是期望点名主要来源的前几段（缺省 1）。期望是算出的主要来源和方向。
+- verdict 判定：op 是 baseline / trend / differ / comparable（differ 另写 mode），params 同对应的分析工具。期望是算出的
+  判定类（常规 / 异常、上升 / 下降 / 无趋势、站得住 / 不足以下结论、可比 / 不可比）。
+  这两种的参数不写输入（就是期望 SQL 的结果）、不写阈值（按这次评测的分析设置）；文字参数可以带 {{ schedule.* }}。
 **数值期望不许人填**（评测-D4）：期望里写了 value、values、expected、answer、rows 这类键的一律拒收——期望值只能由 SQL、
-口径或从运行认定的 SQL 算出来（app/eval/gold.py），人填的数会过期、会填错，也说不清是哪个口径。
+口径或从运行认定的 SQL 算出来（app/eval/gold.py），人填的数会过期、会填错，也说不清是哪个口径。分析期望同样：写了
+segment、subjects、direction、verdict 这类键（人填的来源、方向、判定）的拒收，判定由平台的算子在期望 SQL 上算。
 
 **期望答案的来源**（gold；scalar、table、set、none 要它，ask、text 不要）：
 - sql：{datasource_id, sql}。SQL 可以带 {{ schedule.* }}，按评测的截至时间渲染；保存前过只读守卫。
@@ -48,14 +54,15 @@ GOLD_GUARD = "eval_gold_guard"
 #: 口径题指的卡不能当口径题（422）：不存在、不是口径卡、不能复用、取数节点引用了入口字段以外的值
 CALIBER_NOT_REUSABLE = "eval_caliber_not_reusable"
 
-EXPECTATION_KINDS = ("scalar", "table", "set", "none", "ask", "text")
-#: 要期望答案来源的期望：数都由 SQL / 口径算
-NEEDS_GOLD = ("scalar", "table", "set", "none")
+EXPECTATION_KINDS = ("scalar", "table", "set", "none", "ask", "text", "contribution", "verdict")
+#: 要期望答案来源的期望：数都由 SQL / 口径算，分析的判定由算子在期望 SQL 的结果上算
+NEEDS_GOLD = ("scalar", "table", "set", "none", "contribution", "verdict")
 #: 不计入准确率的（评测-D4：文字要点由裁判判，单列）
 UNCOUNTED = ("text",)
 GOLD_KINDS = ("sql", "caliber", "from_run")
-#: 期望里出现这些键就是在人填数（评测-D4）
-HUMAN_VALUE_KEYS = ("value", "values", "expected", "answer", "rows", "number")
+#: 期望里出现这些键就是在人填数（评测-D4）；后几个是人填分析的结论（来源的段、方向、判定）
+HUMAN_VALUE_KEYS = ("value", "values", "expected", "answer", "rows", "number", "segment", "segments", "subjects",
+                    "direction", "verdict", "judgement", "driver")
 
 #: 每种期望认的键（另有 id、kind、required、label）
 _EXPECTATION_KEYS: dict[str, tuple[str, ...]] = {
@@ -65,6 +72,8 @@ _EXPECTATION_KEYS: dict[str, tuple[str, ...]] = {
     "none": (),
     "ask": (),
     "text": ("points",),
+    "contribution": ("op", "mode", "params", "top_n"),
+    "verdict": ("op", "mode", "params"),
 }
 _COMMON_KEYS = ("id", "kind", "required", "label")
 #: 期望答案来源的键（客户端能写的）；from_run 只能由服务端建
@@ -198,6 +207,9 @@ def _expectation(raw: Any, i: int, *, gold_kind: str | None, taken: set[str]) ->
     if kind not in EXPECTATION_KINDS:
         raise CaseInvalid(f"{where}的种类只能是 {'、'.join(EXPECTATION_KINDS)}", f"{path}.kind")
     human = [k for k in HUMAN_VALUE_KEYS if k in raw]
+    if human and kind in ("contribution", "verdict"):
+        raise CaseInvalid(f"{where}写了「{human[0]}」：分析期望的来源、方向、判定不许人填，由平台的分析方法在期望 SQL 的结果上"
+                          "算出来。请写清做哪种分析、按哪几列", f"{path}.{human[0]}")
     if human:
         raise CaseInvalid(f"{where}写了「{human[0]}」：数值期望不许人填，期望值只能由期望 SQL、口径或从运行认定的 SQL "
                           "算出来。请写清从期望答案的结果里取哪一格、哪一列或整张表", f"{path}.{human[0]}")
@@ -274,6 +286,42 @@ def _expectation(raw: Any, i: int, *, gold_kind: str | None, taken: set[str]) ->
         if len(points) > POINTS_MAX or any(len(p.strip()) > POINT_MAX for p in points):
             raise CaseInvalid(f"{where}最多 {POINTS_MAX} 条要点，每条不超过 {POINT_MAX} 个字符", f"{path}.points")
         out["points"] = [p.strip() for p in points]
+    elif kind in ("contribution", "verdict"):
+        out.update(_analysis_expectation(raw, kind, where, path))
+    return out
+
+
+def _analysis_expectation(raw: Mapping[str, Any], kind: str, where: str, path: str) -> dict[str, Any]:
+    """贡献、判定两种期望的分析方法、模式、参数（app/eval/judgements.py）。贡献恒为变化拆解，判定是另外四种分析。"""
+    from app.analysis.contract import MODES
+    from app.eval import judgements
+
+    ops = ("decompose",) if kind == "contribution" else judgements.VERDICT_OPS
+    op = raw.get("op") if raw.get("op") not in (None, "") else ("decompose" if kind == "contribution" else None)
+    if op not in ops:
+        said = "、".join(judgements.op_text(o) for o in ops)
+        raise CaseInvalid(f"{where}（{'贡献' if kind == 'contribution' else '判定'}）的分析方法只能是{said}"
+                          + ("" if op is None else f"（当前为「{op}」）"), f"{path}.op")
+    out: dict[str, Any] = {"op": op}
+    modes = MODES.get(op)
+    mode = raw.get("mode")
+    if modes:
+        if mode not in modes:
+            said = "、".join(judgements.op_text(op, m) for m in modes)
+            raise CaseInvalid(f"{where}要选一种模式：{said}", f"{path}.mode")
+        out["mode"] = mode
+    elif mode not in (None, ""):
+        raise CaseInvalid(f"{where}是{judgements.op_text(op)}，没有模式可选，请去掉模式", f"{path}.mode")
+    try:
+        params = judgements.check_params(op, out.get("mode"), raw.get("params"))
+    except judgements.ParamsInvalid as e:
+        raise CaseInvalid(f"{where}的{e.message}", f"{path}.params" + (f".{e.key}" if e.key else "")) from None
+    for key, text in judgements.template_texts(params):
+        check_variables(f"{where}的参数「{key}」", text, f"{path}.params.{key}")
+    out["params"] = params
+    if kind == "contribution":
+        out["top_n"] = _int(raw.get("top_n", 1), f"{where}点名主要来源的前几段", f"{path}.top_n", low=1,
+                            high=judgements.TOP_N_MAX)
     return out
 
 
@@ -300,8 +348,8 @@ def normalize_expectations(raw: Any, *, gold_kind: str | None, case_kind: str) -
     if "ask" in kinds and case_kind == "workflow":
         raise CaseInvalid("工作流题不会反问，期望反问只用于问数据题", "expectations")
     if gold_kind is not None and not kinds & set(NEEDS_GOLD):
-        raise CaseInvalid("有期望答案的来源，却没有要它算的期望：请加一个「一个数」「多行」「名单」或「没有数据」的期望，"
-                          "或者去掉期望答案的来源", "gold")
+        raise CaseInvalid("有期望答案的来源，却没有要它算的期望：请加一个「一个数」「多行」「名单」「没有数据」「贡献」"
+                          "或「判定」的期望，或者去掉期望答案的来源", "gold")
     return out
 
 

backend/app/eval/compare.py

@@ -11,7 +11,8 @@
 「只参数变」（多半是时间范围理解变了）；作答路径变了是「路径变」；其余「说不清」。SQL 的 same / params_only / changed 和运行
 对比（engine/compare）是同一条规则：data/sql_semantics.normalize_sql 的规范文本和字面量。
 
-**指纹差异**：助手模型、写作模型、目录快照、提示词指纹、表结构、题目、评分规则，哪几项变了。
+**指纹差异**：助手模型、写作模型、目录快照、提示词指纹、表结构、题目、评分规则、分析设置（有分析期望的题集开评时拍的
+那一份判定阈值），哪几项变了。
 
 **A/B**：同一次发起的几臂（共用 pair_id、共用期望）逐题配对，给两臂各自的准确率、配对表（都对 / 都错 / 只 A 对 / 只 B 对）和
 精确 McNemar 双侧 p 值；p ≥ 0.05 时照实写「这批题上看不出差别」。不能判的题不进配对表（单列个数）。
@@ -30,7 +31,8 @@ CAUSES = ("data", "method", "params", "path", "unknown")
 CAUSE_LABEL = {"data": "数据变", "method": "做法变", "params": "只参数变", "path": "路径变", "unknown": "说不清"}
 #: 指纹里比的几项和叫法
 FINGERPRINT_LABEL = {"assistant_model": "助手模型", "writer_model": "写作模型", "catalog": "目录快照",
-                     "prompt": "提示词", "schemas": "表结构", "cases": "题目", "scorer_version": "评分规则"}
+                     "prompt": "提示词", "schemas": "表结构", "cases": "题目", "scorer_version": "评分规则",
+                     "analysis": "分析设置"}
 NO_DIFFERENCE = "这批题上看不出差别"
 SIGNIFICANT = 0.05
 #: 基准能是哪几种状态

backend/app/eval/gold.py

@@ -3,8 +3,13 @@
 **期望不靠人填数，也不冻结数值。** 每次要期望（保存题目时试算一次、评测时作答前后各一次）都在业务库上重算一遍：
 
 - SQL 题、从运行认定：建一张「入口 → 调用工具（db_query__数据源，args.sql = 按截至时间渲染好的 SQL）→ 出口」的裸图，
-  以来源 eval 跑探索运行。期望值取这次运行封存的查询快照里指定的格、列或整张表（cases.py 的 expectations）。从运行认定
+  以来源 eval 发起探索运行。期望值取这次运行封存的查询快照里指定的格、列或整张表（cases.py 的 expectations）。从运行认定
   的题和 SQL 题算法一样，只是 SQL 出自某次运行里的一条查询、记着谁在什么时候认定的。
+- 分析期望（贡献、判定）：在同一份查询快照上用平台的算子算一次分析（app.analysis.run，和作答里的分析工具、分析卡同一个
+  函数；随数据变化的调用错误记成算不了，refuse_data_errors），判定就是期望（app/eval/judgements.py）。阈值用这次评测的
+  分析设置快照（Context.analysis，和作答的运行同一份；试算时现读设置），遮罩列、目录事实、现查表的日期范围和分析工具
+  同一套取法（app/tools/analysis.py）。算不了的这一项照实记（评分时不能评分），参数和期望 SQL 的结果对不上（列不在、期次
+  不在）的整题没有可用的值。
 - 口径题：app/ledger/derive.caliber_probe 从钉住版本拷出口径卡读的取数节点、接一张 caliber_from 固定到那一版的口径卡，
   跑探索运行，期望值就是口径卡的指标（取整、单位、比率都由口径卡自己定）。不可复用的卡拒收并给原因。
 
@@ -31,7 +36,8 @@ unresolved_caliber_upgrades（和正式运行用的是同一个判断），再
 - incomparable：旧题归档，按新版本另建一道新题、重开基线。
 
 **作答前后各算一次**（bracket / settle，第 3 波的执行调它）：两次一样才判；不一样就记「作答期间数据变了，不判」。A/B 两臂
-共用同一次算出的期望：在 answer 里同时跑两臂即可。
+共用同一次算出的期望：在 answer 里同时跑两臂即可。两次比的是期望值本身：分析结论里随运行当天、现查时刻、输入快照的
+工件变的几项不比（judgements.steady）。
 """
 from __future__ import annotations
 
@@ -157,6 +163,8 @@ class Context:
     #: 记进运行 origin 的其余字段（臂、第几次重复、试算……）
     extra: Mapping[str, Any] = field(default_factory=dict)
     timeout_s: float = TRIAL_TIMEOUT_S
+    #: 这次评测的分析设置快照（runner 开评时拍一份，作答的运行带同一份）；没有（试算）时现读设置
+    analysis: Mapping[str, Any] | None = None
 
     def variables(self) -> dict[str, str]:
         return variable_values(self.as_of, self.tz, self.cron)
@@ -387,6 +395,83 @@ def _values(expectations: Sequence[Mapping[str, Any]], pick: Callable[[Mapping[s
     return {e["id"]: pick(e) for e in expectations if e.get("kind") in NEEDS_GOLD}
 
 
+# ==========================================================================
+# 分析期望：在期望 SQL 的快照上算一次分析（app/eval/judgements.py）
+# ==========================================================================
+
+
+@dataclass
+class _Analyses:
+    """一道题的几项分析期望共用的那几样：输入快照（含遮罩列）、阈值、算子的运行期依赖。用到时才备（_prepared）。"""
+
+    snapshot: dict[str, Any]
+    artifact: str
+    run_id: str
+    ctx: Context
+    ready: tuple[Any, dict[str, Any], Any] | None = None
+
+
+async def _prepared(found: _Analyses) -> tuple[Any, dict[str, Any], Any]:
+    """(算子的输入快照, 阈值, 运行期依赖)：遮罩列、目录事实、现查表的日期范围同分析工具（app/tools/analysis.py），按这次
+    探针运行固定的数据版本解析数据源；阈值是这次评测的分析设置快照，没有（试算）现读设置。"""
+    if found.ready is not None:
+        return found.ready
+    from app.db.base import SessionLocal
+    from app.db.models import Run
+    from app.eval.judgements import GOLD_REF
+    from app.tools.analysis import context_of, input_snapshot, masked_of, settings_of
+    from app.tools.registry import ToolContext
+
+    inp = input_snapshot(found.snapshot, ref=GOLD_REF, artifact=found.artifact, node_id=QUERY_NODE,
+                         masked=await masked_of(found.snapshot, found.artifact))
+    async with SessionLocal() as session:
+        row = await session.get(Run, found.run_id)
+        versions = dict(row.data_versions) if row is not None and row.data_versions is not None else None
+    tool_ctx = ToolContext(run_id=found.run_id, node_id=QUERY_NODE, data_versions=versions, mask_results=True,
+                           force_readonly=True, fixed_scope=True)
+    context = await context_of([inp], tool_ctx=tool_ctx)
+    found.ready = (inp, await settings_of(found.ctx.analysis), context)
+    return found.ready
+
+
+async def _analysis_value(exp: Mapping[str, Any], at: int, found: _Analyses) -> dict[str, Any]:
+    """一项分析期望的期望值（judgements.expected）。算不了（历史不够、本期不完整……）照实记成这一项的结论；参数和期望 SQL
+    的结果对不上（列不在、期次不在、不是乘积关系……）抛 _NoValue（整题没有可用的值，试算时照实交回去改）。"""
+    import app.analysis as analysis_ops   # 按模块取 run：测试换替身时只换模块上的这一个函数
+    from app.analysis.contract import CallError
+    from app.eval import judgements
+
+    where = f"第 {at + 1} 个期望（{'贡献' if exp.get('kind') == 'contribution' else '判定'}）"
+    if found.snapshot.get("truncated"):
+        raise _NoValue("期望 SQL 的结果被截断（只取回了前一部分行），分析期望算不准：请缩小范围或在 SQL 里汇总")
+    inp, settings, context = await _prepared(found)
+    try:
+        request = judgements.request_of(exp, found.ctx.variables(), found.snapshot)
+    except judgements.ParamsInvalid as e:
+        raise _NoValue(f"{where}的{e.message}") from None
+    try:
+        result = await analysis_ops.run(request, lambda name: inp if name == judgements.GOLD_REF else None,
+                                        settings=settings, context=context, refuse_data_errors=True)
+    except CallError as e:
+        raise _NoValue(f"{where}的分析参数和期望 SQL 的结果对不上：{e.message}") from None
+    return judgements.expected(exp, result)
+
+
+async def _sql_values(case: Mapping[str, Any], snapshot: dict[str, Any], artifact: str, run_id: str,
+                      ctx: Context) -> dict[str, dict[str, Any]]:
+    """SQL 题每个期望的值（按期望的先后）：取格、列、整张表的照快照取，分析期望在同一份快照上算。取不到抛 _NoValue。"""
+    from app.eval.judgements import ANALYSIS_KINDS
+
+    found = _Analyses(snapshot=snapshot, artifact=artifact, run_id=run_id, ctx=ctx)
+    out: dict[str, dict[str, Any]] = {}
+    for at, exp in enumerate(case["expectations"]):
+        if exp.get("kind") not in NEEDS_GOLD:
+            continue
+        out[exp["id"]] = await _analysis_value(exp, at, found) if exp.get("kind") in ANALYSIS_KINDS \
+            else _sql_value(exp, snapshot)
+    return out
+
+
 # ==========================================================================
 # SQL 探针（SQL 题、从运行认定）
 # ==========================================================================
@@ -437,7 +522,7 @@ async def _sql_gold(case: Mapping[str, Any], ctx: Context) -> GoldOutcome:
     if problem := masked_problem(snapshot):
         return _outcome("refused", ctx.as_of, problem, GOLD_MASKED, run_ids=runs, rendered=rendered)
     try:
-        values = _values(case["expectations"], lambda e: _sql_value(e, snapshot))
+        values = await _sql_values(case, snapshot, artifact_id, ran.run_id, ctx)
     except _NoValue as e:
         return _outcome("no_value", ctx.as_of, str(e), GOLD_NO_VALUE, run_ids=runs, rendered=rendered)
     return _outcome("ok", ctx.as_of, sets=[GoldSet(run_id=ran.run_id, snapshot=artifact_id, values=values)],
@@ -622,10 +707,23 @@ class Settled:
                 "after": self.after.to_dict() if self.after is not None else None}
 
 
+def stable_sets(sets: Sequence[Any]) -> list[list[Any]]:
+    """比较用的几份期望值：[[版本, {期望 id: 值}]]，分析结论里随运行当天、现查时刻、输入快照的工件变的几项去掉
+    （judgements.steady）。sets 是 GoldSet 或它的 to_dict。作答前后两次（_fingerprint）、和上一次比的「数据变了」
+    （runner.gold_key）都按它。"""
+    from app.eval.judgements import steady
+
+    out = []
+    for s in sets:
+        version, values = (s.version, s.values) if isinstance(s, GoldSet) else (s.get("version"), s.get("values"))
+        out.append([version, {k: steady(v) for k, v in values.items()} if isinstance(values, Mapping) else values])
+    return out
+
+
 def _fingerprint(outcome: GoldOutcome) -> str:
     from app.core.artifact_store import canonical_json
 
-    return canonical_json([[s.version, s.values] for s in outcome.sets])
+    return canonical_json(stable_sets(outcome.sets))
 
 
 def settle(before: GoldOutcome, after: GoldOutcome | None) -> Settled:
@@ -717,4 +815,4 @@ __all__ = ["CANCEL_GRACE_S", "Context", "DISPOSE_INVALID", "DISPOSE_NOTHING", "D
            "GOLD_DRIFTED", "GOLD_FAILED", "GOLD_MASKED", "GOLD_NO_VALUE", "GOLD_TIMEOUT", "GoldOutcome", "GoldSet",
            "QUERY_NODE", "ROLES", "STATUSES", "Settled", "TRIAL_TIMEOUT_S", "UPGRADE_LABEL", "UPGRADE_PENDING",
            "UpgradeState", "bracket", "compute_gold", "masked_problem", "plan_dispose", "settle", "sql_probe_graph",
-           "upgrade_state"]
+           "stable_sets", "upgrade_state"]

backend/app/eval/judgements.py

@@ -0,0 +1,520 @@
+"""分析期望：贡献（contribution）、判定（verdict）的算法和评分（docs/design/analysis-round/00-总方案.md §3.8）。
+
+**期望怎么算。** 期望 SQL 在业务库上查一次（gold.py 的 SQL 探针，和别的期望同一份快照），在这份快照上用平台的算子
+（app.analysis.run，和作答里的分析工具、分析卡同一个函数）算一次分析，判定就是期望：
+
+- 贡献：变化拆解（可加、比率、乘积），参数同分析工具（按维度 by、期次 period、基期 base、本期 target、度量 measure、分子
+  numerator、分母 denominator、乘积 product、因子 factors、叫法 label），另写 top_n（缺省 1）：期望点名主要来源的前几段。
+- 判定：波动基线、趋势判断、差异检验、可比性检查，参数同对应的分析工具。
+
+算子要的输入只有这一份快照（编号 GOLD_REF），不写 query、两份快照、总数、下钻。参数里的文字可以带 {{ schedule.* }}，按评测
+的截至时间渲染（同期望 SQL）；期望 SQL 只返回两期时可以不写基期、本期：先的一期是基期、后的一期是本期。阈值用这次评测的
+设置快照（和作答的运行同一份，runner.py），不写在期望里；算不了的（历史不够、本期不完整……）这一项不能评分，原因照实写。
+
+**判定类**（JUDGEMENTS）：期望和答案里的分析结论都按判定码归到同一套类再比——贡献：source 主要来源（可加的 driver、
+concentrated，乘积的 driver）、dispersed 分散、flat 基本持平、mix 主要来自结构变化、rate 主要来自各段自身的变化、mixed
+两方面都有影响（比率、乘积的 both）；判定：normal 常规波动（within_band）、anomaly 超出常规波动（outside_band、
+far_outside_band，看方向）、rising / falling / no_trend、differs 差异站得住（看哪两组、方向）、no_evidence 不足以下结论、
+comparable 可比（comparable、conditional）/ incomparable 不可比。
+
+**评分**（score；沿用五种结论）：
+- 对且可追溯：答案引用了（[[a:]]、[[v:A…]]、[[table:A…]]、[[see:A…]]）判定对得上的分析结论（同一个算子，变化拆解还要同一种
+  模式；主要来源要点名期望的那几段，方向一致；差异检验比的是同两组）；引用它的那句话里的判断词却和期望对不上的不算。
+- 对但没引用：没引用对得上的结论，但正文按说法表（engine/analysis_claims，同一套词表、同一套绑定规则）写对了：说法表认出的
+  这一类判断词，拿期望的那份分析结论核对得上（点名的段、方向、连续几期都核）。可比性没有对应的判断词，只能引用。
+- 错：引用的分析判定对不上、正文写的和期望相反（把常规波动说成异常、把上升说成下降、主要来源说成别的段），或者根本没做也
+  没说。
+- 不能评分：期望算不了。
+"""
+from __future__ import annotations
+
+import copy
+import re
+import unicodedata
+from collections.abc import Iterable, Mapping, Sequence
+from dataclasses import replace
+from typing import Any
+
+from app.analysis.contract import OP_LABELS, method_label
+
+#: 两种分析期望
+ANALYSIS_KINDS = ("contribution", "verdict")
+#: 判定期望能用的分析（贡献恒为变化拆解）
+VERDICT_OPS = ("baseline", "trend", "differ", "comparable")
+#: 判定类，见模块说明。前六个是贡献的，后九个是判定的
+JUDGEMENTS = ("source", "dispersed", "flat", "mix", "rate", "mixed",
+              "normal", "anomaly", "rising", "falling", "no_trend", "differs", "no_evidence", "comparable",
+              "incomparable")
+#: 分析的输入在算子里的名字（报错里照它说：「期望 SQL 里没有『地区』这一列」）
+GOLD_REF = "期望 SQL"
+#: 贡献期望最多点名前几段
+TOP_N_MAX = 10
+#: 期望里不写的参数：输入就是期望 SQL 这一份快照，没有两份快照、总数、下钻
+_NOT_IN_EXPECTATION = frozenset({"query", "base_query", "target_query", "headline", "parent"})
+#: 期望 SQL 只返回两期时，这几种分析可以不写基期、本期
+_TWO_PERIODS = frozenset({"decompose", "differ", "comparable"})
+
+#: 判定码 → 判定类（按算子、模式）
+_JUDGEMENT_OF: dict[tuple[str, str | None], dict[str, str]] = {
+    ("decompose", "additive"): {"driver": "source", "concentrated": "source", "dispersed": "dispersed", "flat": "flat"},
+    ("decompose", "ratio"): {"mix_driven": "mix", "rate_driven": "rate", "both": "mixed", "flat": "flat"},
+    ("decompose", "factor"): {"driver": "source", "both": "mixed", "flat": "flat"},
+    ("baseline", None): {"within_band": "normal", "outside_band": "anomaly", "far_outside_band": "anomaly"},
+    ("trend", None): {"rising": "rising", "falling": "falling", "no_trend": "no_trend"},
+    ("differ", None): {"differs": "differs", "no_evidence": "no_evidence"},
+    ("comparable", None): {"comparable": "comparable", "conditional": "comparable", "incomparable": "incomparable"},
+}
+#: 判定类要看方向的：主要来源、结构或自身为主、分散（总变化的涨跌）、超出常规波动（本期变化的涨跌）、差异站得住（A 相对 B）
+_DIRECTED = frozenset({"source", "mix", "rate", "mixed", "dispersed", "anomaly", "differs"})
+#: 说法表的类（engine/analysis_claims 的 Claim.key）→ 它说出的判定类；「连续…」「趋势」看方向，见 _said_judgement
+_CLAIM_JUDGEMENT = {"source": "source", "dispersed": "dispersed", "mix": "mix", "anomaly": "anomaly", "normal": "normal",
+                    "no_trend": "no_trend", "differ": "differs", "no_diff": "no_evidence"}
+#: 期望是这一类时，正文里哪几类判断词算说出了它（别的类只看对不对得上：对不上算说反了）
+_AFFIRMS: dict[str, frozenset[str]] = {
+    "source": frozenset({"source"}), "dispersed": frozenset({"dispersed"}), "mix": frozenset({"source", "mix"}),
+    "rate": frozenset({"source"}), "mixed": frozenset({"mix"}), "flat": frozenset(), "normal": frozenset({"normal"}),
+    "anomaly": frozenset({"anomaly"}), "rising": frozenset({"trend"}), "falling": frozenset({"trend"}),
+    "no_trend": frozenset({"no_trend"}), "differs": frozenset({"differ"}), "no_evidence": frozenset({"no_diff"}),
+    "comparable": frozenset(), "incomparable": frozenset(),
+}
+#: 判断词所在的句子最多留多少字给人看
+_SNIPPET_MAX = 40
+
+
+class ParamsInvalid(ValueError):
+    """期望的分析参数写得不对（保存题目时、算期望时都可能）。message 给人看，key 是出错的那个参数。"""
+
+    def __init__(self, message: str, key: str | None = None) -> None:
+        super().__init__(message)
+        self.message = message
+        self.key = key
+
+
+# ==========================================================================
+# 参数：校验（cases.py 保存题目时）、组成分析请求（gold.py 算期望时）
+# ==========================================================================
+
+
+def param_names(op: str) -> list[str]:
+    """期望里这种分析可以写的参数（分析工具的参数去掉输入、总数、下钻），按工具里的先后。"""
+    from app.tools.builtin.analysis import param_names as tool_params
+
+    return [k for k in tool_params(op) if k not in _NOT_IN_EXPECTATION]
+
+
+def op_text(op: str, mode: str | None = None) -> str:
+    """「变化拆解（可加）」「波动基线」：叫法只取契约那一份。"""
+    return method_label(op, mode) if mode else OP_LABELS.get(op, op)
+
+
+def template_texts(params: Mapping[str, Any]) -> Iterable[tuple[str, str]]:
+    """参数里的文字（含列表、成对列表里的）：(参数名, 文字)。{{ schedule.* }} 的校验、渲染按它走。"""
+    for key, value in params.items():
+        if isinstance(value, str):
+            yield key, value
+        elif isinstance(value, list):
+            for item in value:
+                if isinstance(item, str):
+                    yield key, item
+                elif isinstance(item, list):
+                    yield from ((key, x) for x in item if isinstance(x, str))
+
+
+def check_params(op: str, mode: str | None, params: Any) -> dict[str, Any]:
+    """参数的形状：只能写这种分析的参数，不能写阈值，类型照分析工具的参数定义。交回规范化的参数（去掉空值）。不对抛
+    ParamsInvalid。列在不在期望 SQL 的结果里要等试算才知道（算子按快照核）。"""
+    from pydantic import ValidationError
+
+    from app.analysis.contract import DEFAULTS, LABELS
+    from app.tools.builtin.analysis import ARGS
+
+    if params is None:
+        params = {}
+    if not isinstance(params, Mapping):
+        raise ParamsInvalid("分析参数需要是一个对象")
+    allowed = param_names(op)
+    out: dict[str, Any] = {}
+    for key, value in params.items():
+        if key in DEFAULTS:
+            label = LABELS[key] if LABELS[key].startswith("「") else f"「{LABELS[key]}」"
+            raise ParamsInvalid(f"参数里写了阈值{label}：阈值不写在期望里，评测按这次评测的分析设置算（和作答用同一份）",
+                                key)
+        if key not in allowed:
+            raise ParamsInvalid(f"参数里没有「{key}」，{op_text(op)}可以写：{'、'.join(allowed)}", key)
+        if value is None or value == "" or value == []:
+            continue
+        out[key] = copy.deepcopy(value)
+    probe = {**out, "query": GOLD_REF, **({"mode": mode} if mode else {})}
+    try:
+        ARGS[op].model_validate(probe)
+    except ValidationError as e:
+        error = e.errors()[0]
+        key = str((error.get("loc") or ("",))[0])
+        field = ARGS[op].model_fields.get(key)
+        said = (field.description or "") if field is not None else ""
+        verb = "没写" if error.get("type") == "missing" else "写得不对"
+        raise ParamsInvalid(f"参数「{key}」{verb}" + (f"：{said}" if said else ""), key) from None
+    return out
+
+
+def _render(value: Any, variables: Mapping[str, str]) -> Any:
+    from app.eval.cases import render
+
+    if isinstance(value, str):
+        return render(value, variables)
+    if isinstance(value, list):
+        return [_render(v, variables) for v in value]
+    return value
+
+
+def _two_periods(snapshot: Mapping[str, Any], period: str) -> tuple[Any, Any]:
+    """期望 SQL 只返回两期时的基期、本期：按期次先后（periods.parse_compared 认得出的按它，否则按写法）。"""
+    from app.analysis import periods
+
+    columns = [str(c) for c in snapshot.get("columns") or []]
+    hits = [i for i, c in enumerate(columns) if c == period] or \
+        [i for i, c in enumerate(columns) if c.lower() == period.lower()]
+    if not hits:
+        raise ParamsInvalid(f"期望 SQL 的结果里没有期次列「{period}」（结果的列：{'、'.join(columns) or '（没有）'}）",
+                            "period")
+    i = hits[0]
+    values = list(dict.fromkeys(row[i] for row in snapshot.get("rows") or []
+                                if isinstance(row, (list, tuple)) and i < len(row) and row[i] is not None))
+    if len(values) != 2:
+        raise ParamsInvalid(f"期望 SQL 的期次列「{period}」里有 {len(values)} 期：不写基期、本期时，期望 SQL 只能返回两期"
+                            "（先的一期是基期、后的一期是本期）", "base")
+    parsed = periods.parse_compared(values, str(snapshot.get("sql") or ""), columns[i], snapshot.get("dialect"))
+    if parsed is not None:
+        items, _ = parsed
+        order = sorted(range(2), key=lambda k: items[k].ordinal)
+    else:
+        order = sorted(range(2), key=lambda k: str(values[k]))
+    return values[order[0]], values[order[1]]
+
+
+def request_of(exp: Mapping[str, Any], variables: Mapping[str, str], snapshot: Mapping[str, Any]) -> dict[str, Any]:
+    """一项分析期望 → 一次分析的请求（contract.AnalysisRequest）：参数按截至时间渲染，输入是期望 SQL 这一份快照；只返回两期
+    又没写基期、本期的按两期补上。不对抛 ParamsInvalid。"""
+    from pydantic import ValidationError
+
+    from app.tools.builtin.analysis import ARGS, request_of as tool_request
+
+    op, mode = str(exp.get("op")), exp.get("mode")
+    params = {k: _render(v, variables) for k, v in (exp.get("params") or {}).items()}
+    if op in _TWO_PERIODS and isinstance(params.get("period"), str) and params.get("base") in (None, "") \
+            and params.get("target") in (None, ""):
+        params["base"], params["target"] = _two_periods(snapshot, params["period"])
+    params["query"] = GOLD_REF
+    if mode:
+        params["mode"] = mode
+    try:
+        args = ARGS[op].model_validate(params)
+    except ValidationError as e:
+        key = str((e.errors()[0].get("loc") or ("",))[0])
+        raise ParamsInvalid(f"参数「{key}」写得不对", key) from None
+    return tool_request(op, args)  # type: ignore[return-value]
+
+
+# ==========================================================================
+# 判定类
+# ==========================================================================
+
+
+def judgement_code(op: Any, mode: Any, code: Any) -> str | None:
+    """一份分析结论的判定码 → 判定类；没有对应的（拆解对不上总数）为 None。"""
+    key = (str(op), str(mode) if op == "decompose" else None)
+    return _JUDGEMENT_OF.get(key, {}).get(str(code)) if code else None
+
+
+def _segment(subject: Any) -> str:
+    """判定点名的一段（维度键是一个列表，两列用「·」连，同判定语的写法）。"""
+    from app.analysis.phrases import key_text
+
+    return key_text(list(subject)) if isinstance(subject, (list, tuple)) else str(subject)
+
+
+def norm(name: Any) -> str:
+    """比段名、因子名用的写法：全半角、大小写归一，去掉首尾空白。"""
+    return unicodedata.normalize("NFKC", str(name or "")).casefold().strip()
+
+
+def expected(exp: Mapping[str, Any], result: Mapping[str, Any]) -> dict[str, Any]:
+    """期望 SQL 上算出的那份分析结论 → 这一项的期望值（GoldSet.values 里的一项）：判定类、点名的段、方向、判定语，连同整份
+    结论（评分时说法表按它核对正文；比较作答前后两次期望时去掉随运行当天、现查时刻变的几项，见 steady）。算不了的照实记。"""
+    op, mode = result.get("op"), result.get("mode")
+    out: dict[str, Any] = {"kind": exp.get("kind"), "status": result.get("status"), "op": op, "mode": mode,
+                           "method": op_text(str(op), mode), "phrase": result.get("phrase") or ""}
+    if result.get("status") != "ok":
+        out.update(reason_code=result.get("reason_code"), reason=result.get("reason") or "原因未知")
+        out["result"] = copy.deepcopy(dict(result))
+        return out
+    verdict = result.get("verdict") or {}
+    judgement = judgement_code(op, mode, verdict.get("code"))
+    subjects = [_segment(s) for s in verdict.get("subjects") or []]
+    if judgement == "source":
+        subjects = subjects[:int(exp.get("top_n") or 1)]
+    elif judgement not in ("differs", "no_evidence"):
+        subjects = []
+    out.update(verdict=verdict.get("code"), judgement=judgement, subjects=subjects,
+               direction=verdict.get("direction") if judgement in _DIRECTED else None,
+               criterion=result.get("criterion") or "")
+    out["result"] = copy.deepcopy(dict(result))
+    return out
+
+
+def steady(value: Any) -> Any:
+    """比较用的期望值：分析结论里随运行当天（params.today）、现查时刻（probes[*].at）、输入快照的工件（inputs[*].artifact，
+    快照里记着查询用时，同一份数据每次查出来工件也不同）变的几项去掉。别的期望原样。作答前后两次期望（gold._fingerprint）、
+    和上一次比的「数据变了」（runner.gold_key）都按它比。"""
+    if not (isinstance(value, Mapping) and value.get("kind") in ANALYSIS_KINDS and isinstance(value.get("result"), Mapping)):
+        return value
+    result = copy.deepcopy(dict(value["result"]))
+    if isinstance(result.get("params"), dict):
+        result["params"].pop("today", None)
+    for probe in result.get("probes") or []:
+        if isinstance(probe, dict):
+            probe.pop("at", None)
+    for item in result.get("inputs") or []:
+        if isinstance(item, dict):
+            item.pop("artifact", None)
+    return {**value, "result": result}
+
+
+def view(value: Mapping[str, Any]) -> dict[str, Any]:
+    """评分结果里给人看的那一份（期望、答案各一份）：判定类、点名的段、方向、判定语、哪一份结论（A 编号）。界面照原样写，
+    叫法在前端 termsEval。"""
+    keys = ("judgement", "subjects", "direction", "phrase", "method", "alias", "status", "text")
+    out = {k: copy.deepcopy(value[k]) for k in keys if value.get(k) not in (None, "", [])}
+    if value.get("status") == "refused" and value.get("reason"):
+        out["reason"] = value["reason"]
+    return out
+
+
+# ==========================================================================
+# 答案里的分析结论
+# ==========================================================================
+
+
+def record_of(result: Mapping[str, Any], *, artifact: str, info: Mapping[str, Any] | None = None) -> dict[str, Any]:
+    """答案运行台账里的一份分析结论 → 评分用的那一份：算子、模式、判定类、点名的段、方向、判定语；输入（查询快照、上一层
+    结论的工件）给逐题 E1 找产出答案的查询。alias、cited 由读答案的地方填。"""
+    verdict = result.get("verdict") or {}
+    op, mode = result.get("op"), result.get("mode")
+    out: dict[str, Any] = {
+        "artifact": artifact, "op": op, "mode": mode, "method": op_text(str(op), mode), "status": result.get("status"),
+        "verdict": verdict.get("code"), "judgement": judgement_code(op, mode, verdict.get("code")),
+        "subjects": [_segment(s) for s in verdict.get("subjects") or []],
+        "phrase": result.get("phrase") or "", "cited": [],
+        "inputs": list((info or {}).get("inputs") or [i.get("artifact") for i in result.get("inputs") or []
+                                                      if isinstance(i, Mapping) and i.get("artifact")]),
+    }
+    judgement = out["judgement"]
+    out["direction"] = verdict.get("direction") if judgement in _DIRECTED else None
+    if result.get("status") != "ok":
+        out["reason"] = result.get("reason") or "原因未知"
+    if (info or {}).get("node_id"):
+        out["node_id"] = info["node_id"]
+    return out
+
+
+def _same_family(want: Mapping[str, Any], record: Mapping[str, Any]) -> bool:
+    """答案里的这份结论是不是这一项期望的同类分析：同一个算子；变化拆解还要同一种模式（可加、比率、乘积说的不是一回事）。"""
+    if record.get("op") != want.get("op"):
+        return False
+    return want.get("op") != "decompose" or record.get("mode") == want.get("mode")
+
+
+def _names(items: Iterable[Any]) -> set[str]:
+    return {norm(x) for x in items if norm(x)}
+
+
+def mismatch(want: Mapping[str, Any], got: Mapping[str, Any]) -> str | None:
+    """答案里一份同类的分析结论和期望对不对得上：对得上为 None，否则是对不上的那一句（给人看）。"""
+    if got.get("status") != "ok":
+        return f"这份分析没有做成（{got.get('reason') or '原因未知'}）"
+    judgement = want.get("judgement")
+    if got.get("judgement") != judgement:
+        return "判定和期望不是一类"
+    if judgement in ("differs", "no_evidence"):
+        mine, theirs = [norm(s) for s in got.get("subjects") or []], [norm(s) for s in want.get("subjects") or []]
+        if len(theirs) == 2 and (len(mine) != 2 or set(mine) != set(theirs)):
+            return f"比的是{'和'.join(got.get('subjects') or []) or '别的组'}，期望比的是{'和'.join(want.get('subjects') or [])}"
+        if judgement == "differs" and len(theirs) == 2 and want.get("direction") and got.get("direction"):
+            same_order = mine == theirs
+            if (got["direction"] == want["direction"]) != same_order:
+                return "高低写反了"
+        return None
+    if judgement == "source":
+        missing = [s for s in want.get("subjects") or [] if norm(s) not in _names(got.get("subjects") or [])]
+        if missing:
+            return f"主要来源是{'、'.join(got.get('subjects') or []) or '别的段'}，不是{'、'.join(missing)}"
+    if judgement in _DIRECTED and want.get("direction") and got.get("direction") \
+            and want["direction"] != got["direction"]:
+        return "方向相反"
+    return None
+
+
+# ==========================================================================
+# 正文里的判断词：按说法表认，拿期望的那份分析结论核对（engine/analysis_claims 同一套词表和绑定规则）
+# ==========================================================================
+
+_SENTENCE_END = re.compile(r"(?<=[。！？!?\n])")
+
+
+def sentences_of_doc(doc: Mapping[str, Any], aliases: Iterable[str]) -> list[dict[str, Any]]:
+    """报告文档里有判断词的句子：{text: 写作者自己写的字（引用渲染出来的值、判定语、引文涂成 \\x00）, names: 点名用的样子
+    （引用渲染出来的值留着）, analyses: 这句挂的分析结论编号}。标题、代码块、表格不算（同说法核对）。aliases 是文档目录里
+    分析结论的编号。"""
+    from app.engine import analysis_claims as said
+    from app.engine.evidence import iter_units
+
+    known = {a: {} for a in aliases}
+    out: list[dict[str, Any]] = []
+    for block, unit in iter_units(dict(doc)):
+        if block.get("type") in ("heading", "code", "table") or unit.get("kind") in ("heading", "code") \
+                or unit.get("loc") is not None:
+            continue
+        sentence = said._sentence(unit, unit.get("cites") or [], known)
+        if sentence is not None and said.scan(sentence.text):
+            out.append({"text": sentence.text, "names": sentence.names, "analyses": list(sentence.analyses)})
+    return out
+
+
+def sentences_of_text(text: str) -> list[dict[str, Any]]:
+    """没有报告文档时（成果里的文字）：按句末标点切开，有判断词的句子。"""
+    from app.engine import analysis_claims as said
+
+    out = []
+    for piece in _SENTENCE_END.split(text or ""):
+        piece = piece.strip()
+        if piece and said.scan(piece):
+            out.append({"text": piece, "names": piece, "analyses": []})
+    return out
+
+
+def _snippet(names: str, start: int, end: int) -> str:
+    """判断词连同它所在的分句（给人看）。"""
+    from app.engine import analysis_claims as said
+
+    left, right = said._clause(names, start, end)
+    text = names[left:right].replace("\x00", "").strip()
+    return text[:_SNIPPET_MAX] + ("…" if len(text) > _SNIPPET_MAX else "")
+
+
+def _said_judgement(claim: str, direction: str | None) -> str | None:
+    if claim == "trend":
+        return {"up": "rising", "down": "falling"}.get(direction or "")
+    return _CLAIM_JUDGEMENT.get(claim)
+
+
+def said_in(want: Mapping[str, Any], sentences: Sequence[Mapping[str, Any]], *,
+            only: Iterable[str] | None = None) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    """正文里和这项期望同一类的判断词，拿期望的那份分析结论（want.result）按说法表的绑定规则核对：(说出了期望的那几处,
+    和期望对不上的那几处)。每处 {text: 所在分句, judgement: 它说出的判定类, subjects: 点名的段, reason: 对不上的原因}。
+    only 给了时只看挂了这几份结论的句子。
+
+    规则同 engine/analysis_claims.check_sentence（同一个 _check）：因果词点名的是期望拆解里的段的按来源核；点名的段要在期望
+    点名的段里，拖累、拉动按那一段的涨跌核；「连续 N 期」按末尾连续同向的期数核。另外按期望的方向核主要来源那句话的涨跌
+    （「下降主要来自华东」）。"""
+    from app.engine import analysis_claims as said
+
+    gold = want.get("result") or {}
+    op = want.get("op")
+    affirms = _AFFIRMS.get(str(want.get("judgement")), frozenset())
+    lex = said.lexicon()
+    wanted = set(only) if only is not None else None
+    good: list[dict[str, Any]] = []
+    bad: list[dict[str, Any]] = []
+    for raw in sentences:
+        if wanted is not None and not wanted & set(raw.get("analyses") or []):
+            continue
+        sentence = said.Sentence(text=str(raw.get("text") or ""), names=str(raw.get("names") or ""))
+        for hit in said.scan(sentence.text):
+            claim = lex.claim(hit.claim)
+            left, right = said._clause(sentence.text, hit.start, hit.end)
+            names = sentence.names[left:right]
+            spans = said._object(sentence, hit, left, right)
+            if claim.key == "causal":
+                if op != "decompose":
+                    continue
+                cause = said._cause(sentence, hit, left, right)
+                if not said._named(gold, cause):
+                    continue
+                hit, claim, spans = replace(hit, claim="source", side="cause"), lex.claim("source"), (cause,)
+            if claim.op != op:
+                continue
+            miss = said._check(lex, hit, gold, names, spans=spans)
+            reason = miss.human if miss is not None else None
+            named = next((found for span in spans if (found := said._named(gold, span))), [])
+            if reason is None and hit.claim == "source" and hit.side != "pull" and want.get("direction") \
+                    and hit.direction and hit.direction != want["direction"]:
+                reason = "涨跌写反了"
+            item = {"text": _snippet(sentence.names, hit.start, hit.end),
+                    "judgement": _said_judgement(hit.claim, hit.direction), "subjects": named}
+            if reason is not None:
+                bad.append({**item, "reason": reason})
+            elif hit.claim in affirms:
+                good.append(item)
+    return good, bad
+
+
+def _covered(want: Mapping[str, Any], good: Sequence[Mapping[str, Any]]) -> list[str]:
+    """主要来源期望点名的段里，正文说对的那几处一处都没点到的（top_n 要点名的前几段都要说到）。"""
+    if want.get("judgement") != "source":
+        return []
+    named = _names(s for item in good for s in item.get("subjects") or [])
+    return [s for s in want.get("subjects") or [] if norm(s) not in named]
+
+
+# ==========================================================================
+# 评分
+# ==========================================================================
+
+
+def score(exp: Mapping[str, Any], value: Mapping[str, Any], answer: Any) -> dict[str, Any]:
+    """一项分析期望的结论（score.OUTCOMES 之一）。value 是期望值（expected 的形状），answer 是 score.Answer（analyses：
+    答案运行台账里的分析结论，cited 记着报告怎么引用了它；sentences：有判断词的句子）。纯计算。"""
+    want = view(value)
+    if value.get("status") != "ok":
+        return {"outcome": "ungradable", "want": want, "reason": f"期望算不了：{value.get('reason') or '原因未知'}"}
+    if value.get("judgement") is None:
+        return {"outcome": "ungradable", "want": want, "reason": "期望的分析没有给出判定"}
+    records = [r for r in getattr(answer, "analyses", None) or [] if _same_family(value, r)]
+    cited = [r for r in records if r.get("cited") and r.get("alias")]
+    sentences = list(getattr(answer, "sentences", None) or [])
+    matched = [r for r in cited if mismatch(value, r) is None]
+    if matched:
+        found = matched[0]
+        _, bad = said_in(value, sentences, only=[r["alias"] for r in matched])
+        if bad:
+            return {"outcome": "wrong", "want": want, "found": view({**found, "text": bad[0]["text"]}),
+                    "reason": f"引用了 {found['alias']}，同一句却写成「{bad[0]['text']}」：{bad[0]['reason']}"}
+        return {"outcome": "traced", "want": want, "found": view(found)}
+    good, bad = said_in(value, sentences)
+    if good:
+        missing = _covered(value, good)
+        if not missing:
+            return {"outcome": "untraced", "want": want, "found": view(good[0])}
+        return {"outcome": "wrong", "want": want, "found": view(good[0]),
+                "reason": f"正文只点名了{'、'.join(s for g in good for s in g.get('subjects') or []) or '别的段'}，"
+                          f"期望的主要来源还有{'、'.join(missing)}"}
+    if cited:
+        first = cited[0]
+        return {"outcome": "wrong", "want": want, "found": view(first),
+                "reason": f"答案引用的 {first['alias']}（{first.get('phrase') or first.get('method')}）"
+                          f"和期望对不上：{mismatch(value, first)}"}
+    if bad:
+        return {"outcome": "wrong", "want": want, "found": view(bad[0]),
+                "reason": f"正文写的「{bad[0]['text']}」和期望对不上：{bad[0]['reason']}"}
+    method = op_text(str(value.get("op")), value.get("mode"))
+    if records:
+        first = records[0]
+        shown = f"（{first['alias']}）" if first.get("alias") else ""
+        return {"outcome": "wrong", "want": want, "found": view(first),
+                "reason": f"答案做了{method}{shown}，但没有引用，正文里也没有写出这个判断"}
+    return {"outcome": "wrong", "want": want, "reason": f"答案里没有做{method}，正文里也没有写出这个判断"}
+
+
+__all__ = ["ANALYSIS_KINDS", "GOLD_REF", "JUDGEMENTS", "ParamsInvalid", "TOP_N_MAX", "VERDICT_OPS", "check_params",
+           "expected", "judgement_code", "mismatch", "norm", "op_text", "param_names", "record_of", "request_of",
+           "said_in", "score", "sentences_of_doc", "sentences_of_text", "steady", "template_texts", "view"]

backend/app/eval/runner.py

@@ -95,17 +95,22 @@ def leader_of(run: Any) -> str:
 
 
 def prompt_fingerprint() -> str:
-    """助手提示词的指纹：节点手册、输出协议、问数据的请求包法、重来的两种措辞、复核的规则。改了任何一段，和上一次比时
-    就标「提示词变了」。"""
+    """助手提示词的指纹：节点手册、输出协议、问数据的请求包法、重来的两种措辞、复核的规则；平台的分析方法（分析工具的
+    guide：id 带版本，连同全文）、写作者的分析规则（写作目录里分析结论那一段、判断词的写法、重写时补的那句）和说法表的
+    版本。改了任何一段，和上一次比时就标「提示词变了」。"""
     from app.api import copilot
     from app.chat import turn
     from app.core.artifact_store import canonical_json, content_hash
-    from app.engine import review
+    from app.engine import analysis_claims, evidence, review
+    from app.tools.builtin.analysis import METHOD_GUIDE
 
     parts = [copilot.NODE_REFERENCE, copilot._STREAM_PROTOCOL,
              copilot._user_message(copilot.GenerateIn(instruction="{question}", intent="answer"), patch=False),
              turn.RETRY_WITH_BASE, turn.RETRY_FROM_SCRATCH,
-             str(getattr(review, "REVIEW_SYSTEM", "") or getattr(review, "SYSTEM_PROMPT", ""))]
+             str(getattr(review, "REVIEW_SYSTEM", "") or getattr(review, "SYSTEM_PROMPT", "")),
+             METHOD_GUIDE.id, METHOD_GUIDE.text, evidence.ANALYSIS_PROMPT, evidence.ANALYSIS_SEE_ONLY_PROMPT,
+             analysis_claims.ANALYSIS_RULES_PROMPT, analysis_claims.ANALYSIS_NONE_PROMPT,
+             analysis_claims.ANALYSIS_REPAIR_PROMPT, analysis_claims.LEXICON_VERSION]
     return content_hash(canonical_json(parts))[:16]
 
 
@@ -300,6 +305,8 @@ class EvalExecution(runtime.JobRun):
         self.copilot_reason = ""
         self.timeouts: list[bool] = []
         self.tasks: set[asyncio.Task[Any]] = set()
+        #: 这次评测的分析设置快照（有分析期望的题集才拍，_prepare）：算期望、作答的运行都按它
+        self.analysis: dict[str, Any] | None = None
 
     def lane_of(self, job: Any) -> str | None:
         return None
@@ -354,6 +361,13 @@ class EvalExecution(runtime.JobRun):
                         raise runtime.JobFailed(f"拍目录快照没有成功：{e.message}") from e
                     snapshot = snap.artifact_id
             versions = await source_versions(session, sources)
+            # 有分析期望的题集拍一份分析设置（判定用的阈值）：算期望、各臂作答的运行都按这一份（重试沿用拍过的那一份），
+            # 评测期间有人改了设置也不会一边按旧阈值判、一边按新阈值判；记进指纹，和上一次比时阈值变了照实标出
+            self.analysis = (leader.fingerprint or {}).get("analysis")
+            if self.analysis is None and self._analyzes():
+                from app.api.settings import analysis_settings
+
+                self.analysis = await analysis_settings(session)
             for run in runs:
                 fp = dict(run.fingerprint or {})
                 if "sources_before" not in fp:
@@ -366,6 +380,8 @@ class EvalExecution(runtime.JobRun):
                               sources_before=versions,
                               schemas={sid: v.get("schema") for sid, v in versions.items()},
                               cases=_cases_key(results, run.id))
+                if self.analysis is not None:
+                    fp["analysis"] = dict(self.analysis)
                 run.fingerprint = fp
                 run.catalog_snapshot = snapshot if (run.variant.get("catalog") or {}).get("mode") in (
                     "live", "reviewed") else fp.get("snapshot_artifact")
@@ -384,6 +400,13 @@ class EvalExecution(runtime.JobRun):
                 units.setdefault(key, _Unit(key[0], key[1], {})).results[r.eval_run_id] = r.id
         return list(units.values())
 
+    def _analyzes(self) -> bool:
+        """这次评测的题里有没有分析期望（贡献、判定）。"""
+        from app.eval.judgements import ANALYSIS_KINDS
+
+        return any(e.get("kind") in ANALYSIS_KINDS for case in self.cases.values()
+                   for e in case.get("expectations") or [])
+
     async def _sources(self, session: AsyncSession) -> list[str]:
         """这次评测用得到的数据源：题目限定的、期望查的；有不限数据源的问数据题时，全部启用的源（助手会在它们里面挑）。"""
         from app.db.models import DataSource
@@ -490,7 +513,8 @@ class EvalExecution(runtime.JobRun):
         timeout = float(self.budget.get("per_case_timeout_s") or 900)
         ctx = gold_rules.Context(as_of=self.as_of, tz=self.suite["timezone"], cron=self.cron, actor=self.actor,
                                  suite_id=self.suite["id"], suite_name=self.suite["name"], eval_run_id=self.job_id,
-                                 role="gold", extra={"repeat": unit.repeat}, timeout_s=timeout)
+                                 role="gold", extra={"repeat": unit.repeat}, timeout_s=timeout,
+                                 analysis=self.analysis)
 
         async def answer(_before: Any) -> list[Any]:
             return await asyncio.gather(*(self._answer(arm, unit, case, question, values, timeout) for arm in arms))
@@ -515,7 +539,7 @@ class EvalExecution(runtime.JobRun):
                           case_id=unit.case_id, case_seq=case.get("seq"), actor=self.actor,
                           arm=arm.name, repeat=unit.repeat,
                           assistant=arm.variant.get("assistant") or {}, writer=arm.variant.get("writer") or {},
-                          view=arm.view)
+                          view=arm.view, analysis=self.analysis)
         driver = Driver(question, datasource_ids=list(case.get("datasource_ids") or []), ctx=ctx, on_phase=on_phase)
         try:
             if case.get("kind") == "workflow":
@@ -806,13 +830,15 @@ def _cases_key(results: Sequence[EvalResult], run_id: str) -> str:
 
 
 def gold_key(gold: Mapping[str, Any] | None) -> str | None:
-    """期望值的指纹（作答前那一份）：和上一次比时认「数据变了」。"""
+    """期望值的指纹（作答前那一份）：和上一次比时认「数据变了」。分析结论里随运行当天、现查时刻、输入快照的工件变的几项
+    不算（gold.stable_sets：不然分析题每次评测都成了「数据变」）。"""
     from app.core.artifact_store import canonical_json, content_hash
+    from app.eval.gold import stable_sets
 
     before = (gold or {}).get("before") or {}
     if (gold or {}).get("status") not in ("ok", "drifted"):
         return None
-    return content_hash(canonical_json([[s.get("version"), s.get("values")] for s in before.get("sets") or []]))[:16]
+    return content_hash(canonical_json(stable_sets(before.get("sets") or [])))[:16]
 
 
 def _sql_shapes(answer: Any) -> list[dict[str, Any]]:

backend/app/eval/score.py

@@ -26,6 +26,10 @@ wrong 错、scale 量纲不同（差 10 的整数次幂，比如元和万元、
   没有数据」引用了地区那一格，说的仍是没有数据。
 - ask：期望反问：作答路径是反问才对。
 - text：文字要点，由裁判判（covered / partial / missing / unjudged），单列「模型判断」，不进准确率（评测-D4）。
+- contribution、verdict（分析期望，规则在 app/eval/judgements.py）：答案引用了判定对得上的分析结论（[[a:]]、[[v:A…]]、
+  [[table:A…]]、[[see:A…]]）是可追溯；没引用、但正文按说法表写对了是对但没引用；判定、来源、方向对不上，或者答成相反、
+  根本没做也没说是错；期望算不了的这一项不能评分。答案的分析结论从作答运行的台账（kind analysis）读，报告怎么引用了它们
+  从报告文档读；正文里有判断词的句子一并读出来（说法表按句子认）。
 
 **题目结论**（verdict）：路径不对（要答出数的题直接回答了、反问了、只建不跑、建图或运行失败）直接不通过、写明原因；
 必需的期望全是 traced / untraced 就通过，另报「可追溯通过」（全是 traced）；有 wrong / scale 的不通过；剩下的（有不能评分的、
@@ -34,8 +38,8 @@ wrong 错、scale 量纲不同（差 10 的整数次幂，比如元和万元、
 
 **逐题的 E1**（sql_match）：产出答案的那条查询和期望的取数（SQL 题是期望 SQL，口径题是口径卡的取数节点）用口径账本同一个
 比对函数（ledger/match.compare_queries：先绑参数、再由 explain_sql_diff 判等价）比一次，给出等价与否和差在哪个方面（总方案
-X2：E1 只有一份实现）。哪条是产出答案的：答案引用了就是被引用的；没引用、只查了一条就是那一条（写明依据）；没引用、查了
-几条的不硬比（照实写没法对上是哪条，不计入 E1）。
+X2：E1 只有一份实现）。哪条是产出答案的：答案引用了就是被引用的（引用的是分析结论的，按它的输入找到查询，下钻的顺着上一层
+找下去）；没引用、只查了一条就是那一条（写明依据）；没引用、查了几条的不硬比（照实写没法对上是哪条，不计入 E1）。
 """
 from __future__ import annotations
 
@@ -52,7 +56,9 @@ logger = logging.getLogger(__name__)
 #: 评分规则的版本：规则改了就加一，结果里记着按哪一版评的，可以按新规则重评。
 #: 2：期望「没有数据」时引用文字格、空值格不算给了数；判错时「答案里的数」给配对上的那一格；答案没引用查询时逐题 E1
 #: 不硬比第一条查询
-SCORER_VERSION = 2
+#: 3：两种分析期望（贡献、判定）：引用了判定对得上的分析结论是可追溯，正文按说法表写对了是对但没引用；答案引用分析结论
+#: 时逐题 E1 按它的输入查询比（以前把结论的工件当查询，比不上）；分析结论的判定语不当成答案里的数
+SCORER_VERSION = 3
 
 OUTCOMES = ("traced", "untraced", "wrong", "scale", "ungradable")
 OUTCOME_LABEL = {"traced": "对且可追溯", "untraced": "对但没引用", "wrong": "错", "scale": "量纲不同",
@@ -115,15 +121,38 @@ def _masked_columns(snapshot: Mapping[str, Any]) -> set[str]:
     return {str(c) for c in snapshot.get("masked_at_query") or []}
 
 
+def _result_dims(result: Mapping[str, Any], locator: Mapping[str, Any]) -> dict[str, Any]:
+    """分析结论结果表的一格所在行的维度：记成文字的列（地区、期次、因子的叫法……）。结果表里的数一律是十进制字符串，
+    按列类型分，不按值的样子分；遮罩列不会出现在结果表里。"""
+    table = result.get("table") if isinstance(result.get("table"), Mapping) else {}
+    columns = [str(c) for c in table.get("columns") or []]
+    types = table.get("column_types") if isinstance(table.get("column_types"), Mapping) else {}
+    rows = table.get("rows") or []
+    row_index = locator.get("row")
+    dims: dict[str, Any] = {}
+    if isinstance(row_index, int) and 0 <= row_index < len(rows) and isinstance(rows[row_index], (list, tuple)):
+        row = rows[row_index]
+        dims = {c: row[i] for i, c in enumerate(columns) if i < len(row) and c != locator.get("column")
+                and row[i] is not None and types.get(c) != "number"}
+    return {"dims": dims, "masked": False}
+
+
 def _cell_facts(cite: Mapping[str, Any], catalog: Mapping[str, Any], cache: dict[str, Any]) -> dict[str, Any]:
-    """引用的一格：哪份快照、第几行哪一列、这一行的维度（非数值列）、是不是遮罩格。"""
+    """引用的一格：哪份快照、第几行哪一列、这一行的维度（非数值列）、是不是遮罩格。分析结论的结果表、具名的数也是这样的
+    一格（工件是那份结论）。"""
+    from app.analysis.contract import LEDGER_KIND, is_result
+
     entry = catalog.get(cite.get("alias") or "") or {}
     locator = cite.get("locator") or {}
     artifact = entry.get("artifact")
     out: dict[str, Any] = {"artifact": artifact, "row": locator.get("row"), "column": locator.get("column"), "dims": {}}
+    if locator.get("key"):
+        out["column"] = locator["key"]
     snapshot = _snapshot(artifact, cache)
     if snapshot is None:
         return out
+    if entry.get("kind") == LEDGER_KIND or is_result(snapshot):
+        return {**out, **_result_dims(snapshot, locator)}
     columns = [str(c) for c in snapshot.get("columns") or []]
     masked = _masked_columns(snapshot)
     out["masked"] = bool(snapshot.get("masked_whole")) or str(locator.get("column")) in masked
@@ -139,7 +168,11 @@ def _cell_facts(cite: Mapping[str, Any], catalog: Mapping[str, Any], cache: dict
 
 @dataclass
 class Answer:
-    """从封存的作答运行里读出来的答案（评分只看这些）。problem 不为空时整份答案读不了（封存核对不上、没有封存）。"""
+    """从封存的作答运行里读出来的答案（评分只看这些）。problem 不为空时整份答案读不了（封存核对不上、没有封存）。
+
+    analyses：作答运行台账里的分析结论（judgements.record_of 的形状：算子、判定类、点名的段、方向、判定语、输入；alias 是
+    报告文档目录里的编号，cited 记报告怎么引用了它：phrase 判定语、value 具名的数、cell 结果表的格、see 句末依据）。
+    sentences：正文里有判断词的句子（judgements.sentences_of_doc）。老结果的明细里没有这两项，按空读。"""
 
     run_id: str | None
     text: str = ""
@@ -151,18 +184,22 @@ class Answer:
     tier: str | None = None
     report: str | None = None
     problem: str | None = None
+    analyses: list[dict[str, Any]] = field(default_factory=list)
+    sentences: list[dict[str, Any]] = field(default_factory=list)
 
     def to_dict(self) -> dict[str, Any]:
         return {"run_id": self.run_id, "text": self.text, "values": self.values, "bare": self.bare,
                 "queries": self.queries, "cited": self.cited, "stats": self.stats, "tier": self.tier,
-                "report": self.report, "problem": self.problem}
+                "report": self.report, "problem": self.problem, "analyses": self.analyses,
+                "sentences": self.sentences}
 
     @classmethod
     def from_dict(cls, raw: Mapping[str, Any]) -> Answer:
         return cls(run_id=raw.get("run_id"), text=str(raw.get("text") or ""), values=list(raw.get("values") or []),
                    bare=list(raw.get("bare") or []), queries=list(raw.get("queries") or []),
                    cited=list(raw.get("cited") or []), stats=dict(raw.get("stats") or {}), tier=raw.get("tier"),
-                   report=raw.get("report"), problem=raw.get("problem"))
+                   report=raw.get("report"), problem=raw.get("problem"), analyses=list(raw.get("analyses") or []),
+                   sentences=list(raw.get("sentences") or []))
 
 
 def _output_text(output: Any) -> str:
@@ -180,11 +217,66 @@ def _output_text(output: Any) -> str:
     return "\n".join(parts)
 
 
+def _ledger_analyses(sealed: Any, cache: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """作答运行台账里的分析结论：{结论工件: judgements.record_of}。取不回、不是分析结论的跳过（证据面板照实说取不回）。"""
+    from app.analysis.contract import is_result
+    from app.eval.judgements import record_of
+
+    out: dict[str, dict[str, Any]] = {}
+    for artifact, info in (getattr(sealed, "analyses", None) or {}).items():
+        content = _snapshot(artifact, cache)
+        if is_result(content):
+            out[artifact] = record_of(content, artifact=artifact, info=info)
+    return out
+
+
+def _named_analyses(catalog: Mapping[str, Any], found: dict[str, dict[str, Any]],
+                    cache: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """报告文档目录里的分析结论编号 → 那一份（found 里的同一个对象，记上 alias）。目录里有、台账里没收的照工件补上。"""
+    from app.analysis.contract import LEDGER_KIND, is_result
+    from app.eval.judgements import record_of
+
+    out: dict[str, dict[str, Any]] = {}
+    for alias, entry in catalog.items():
+        if not isinstance(entry, Mapping) or entry.get("kind") != LEDGER_KIND or not entry.get("artifact"):
+            continue
+        artifact = str(entry["artifact"])
+        if artifact not in found:
+            content = _snapshot(artifact, cache)
+            if not is_result(content):
+                continue
+            found[artifact] = record_of(content, artifact=artifact, info={"inputs": entry.get("inputs")})
+        record = found[artifact]
+        record.setdefault("alias", str(alias))
+        out[str(alias)] = record
+    return out
+
+
+def _query_inputs(record: Mapping[str, Any], found: Mapping[str, Mapping[str, Any]], queries: Mapping[str, Any],
+                  seen: set[str] | None = None) -> list[str]:
+    """一份分析结论的输入里的查询快照（逐题 E1 拿它们当产出答案的查询）：下钻的顺着上一层的结论找下去。"""
+    seen = seen if seen is not None else set()
+    out: list[str] = []
+    for artifact in record.get("inputs") or []:
+        if artifact in queries:
+            out.append(artifact)
+        elif artifact in found and artifact not in seen:
+            seen.add(artifact)
+            out += _query_inputs(found[artifact], found, queries, seen)
+    return out
+
+
+def _cited_as(record: dict[str, Any], how: str) -> None:
+    if how not in record["cited"]:
+        record["cited"].append(how)
+
+
 async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer:
     """读一次作答运行的答案。没有运行（直接回答、反问）的只有原文：reply。"""
     from fastapi import HTTPException
 
     from app.api.evidence import _choose, _reports, _sealed
+    from app.eval import judgements
 
     if not run_id:
         text = reply or ""
@@ -209,6 +301,7 @@ async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer
                                "columns": [str(c) for c in snap.get("columns") or []],
                                "row_count": snap.get("row_count", len(rows)), "truncated": bool(snap.get("truncated")),
                                "role": snap.get("role")})
+    found = _ledger_analyses(sealed, cache)
     from app.db.base import SessionLocal
     from app.db.models import Run
 
@@ -222,6 +315,8 @@ async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer
     if chosen is None or chosen.doc is None or chosen.hash_ok is False:
         answer.text = _output_text(final)
         answer.bare = _numbers(answer.text)
+        answer.analyses = list(found.values())
+        answer.sentences = judgements.sentences_of_text(answer.text)
         if chosen is not None and chosen.hash_ok is False:
             answer.problem = "报告文档与哈希不一致，疑似被修改，答案不能当证据"
         return answer
@@ -230,11 +325,22 @@ async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer
     answer.report = chosen.node_id
     answer.text = str(doc.get("markdown") or "")
     answer.stats = dict(doc.get("stats") or {})
-    from app.engine.evidence import iter_segments
+    from app.engine.evidence import iter_segments, iter_units
 
+    named = _named_analyses(catalog, found, cache)
     cited: list[str] = []
     for seg in iter_segments(doc):
         cite = seg.get("cite") or {}
+        record = named.get(str(cite.get("alias") or "")) if cite.get("status") == "resolved" else None
+        if record is not None:
+            # 引用了分析结论：判定语（[[a:A1]]）、具名的数（[[a:A1.键]]）、结果表的格（[[v:A1.r0.列]]、[[table:A1 …]]）。
+            # 产出答案的查询是它的输入（逐题 E1）；判定语是一句话，不是答案里的数
+            how = "cell" if cite.get("kind") == "cell" else "value" if (cite.get("locator") or {}).get("key") \
+                else "phrase"
+            _cited_as(record, how)
+            cited += _query_inputs(record, found, sealed.queries)
+            if how == "phrase":
+                continue
         if cite.get("role") == "value" and cite.get("status") == "resolved":
             item: dict[str, Any] = {"value": cite.get("value"), "kind": cite.get("kind"), "text": seg.get("text"),
                                     "segment": seg.get("id"), "alias": cite.get("alias")}
@@ -244,11 +350,19 @@ async def read_answer(run_id: str | None, *, reply: str | None = None) -> Answer
             else:
                 item.update(_cell_facts(cite, catalog, cache))
             answer.values.append(item)
-            if item.get("artifact"):
+            if item.get("artifact") and record is None:
                 cited.append(item["artifact"])
         elif seg.get("issue") == "uncited_number" or (seg.get("kind") == "number" and not cite):
             answer.bare += _numbers(str(seg.get("text") or ""))
+    for _, unit in iter_units(doc):
+        for support in unit.get("see") or []:
+            record = named.get(str(support.get("alias") or "")) if support.get("status") == "resolved" else None
+            if record is not None:
+                _cited_as(record, "see")
+                cited += _query_inputs(record, found, sealed.queries)
     answer.cited = list(dict.fromkeys(cited))
+    answer.analyses = list(found.values())
+    answer.sentences = judgements.sentences_of_doc(doc, named)
     return answer
 
 
@@ -513,6 +627,10 @@ def _expectation_scores(expectations: Sequence[Mapping[str, Any]], values: Mappi
             out.append({**base, **_score_set(exp, gold, answer, cache)})
         elif kind == "none":
             out.append({**base, **_score_none(answer)})
+        elif kind in ("contribution", "verdict"):
+            from app.eval import judgements
+
+            out.append({**base, **judgements.score(exp, gold, answer)})
     return out
 
 

backend/app/eval/stats.py

@@ -3,6 +3,8 @@
 
 - 准确率和可追溯准确率：分母是能判的题（通过 + 不通过）；不能评分、跳过的单列，既不算对也不算错，没跑的（停止、中断、
   到了上限、止损没开评）另列 not_run，出错的另列 errors。都带 Wilson 95% 区间和样本数——题集小、模型有随机性，差两三题就可能被读成「变好了」「变差了」（评测-D10：只展示、不卡门）。
+- 各期望的结论（必需的、文字要点除外）合起来一份，另按期望的种类各一份（outcomes_by_kind：贡献、判定这些分析题和
+  一个数、多行分开看）。
 - 作答路径、出具档位（声明了出具的按 formal / degraded / withheld，没声明的单列「探索问答」）、复核结论（verdict × severity）
   和自动重来的次数、复核用时超过问数据页等待（25 秒）的题数。
 - 结论裁判（打开了才有）：supported / partial / contradicted / insufficient / 未判的句数；文字要点的判断单列。
@@ -77,6 +79,11 @@ def summarize(results: Sequence[Mapping[str, Any]], *, judge_on: bool = False) -
     traced = [r for r in passed if r.get("traced")]
     outcomes = Counter(s.get("outcome") for r in done for s in r.get("scores") or []
                        if s.get("required") and s.get("kind") != "text")
+    by_kind: dict[str, Counter] = {}
+    for r in done:
+        for s in r.get("scores") or []:
+            if s.get("required") and s.get("kind") != "text":
+                by_kind.setdefault(str(s.get("kind")), Counter())[s.get("outcome")] += 1
     text = Counter(s.get("outcome") for r in done for s in r.get("scores") or [] if s.get("kind") == "text")
     answered = [r for r in done if r.get("path")]
     tiers = Counter((r.get("tier") or "explore") for r in answered if r.get("path") == "ran")
@@ -113,6 +120,7 @@ def summarize(results: Sequence[Mapping[str, Any]], *, judge_on: bool = False) -
         "accuracy": rate(len(passed), len(graded)),
         "traced_accuracy": rate(len(traced), len(graded)),
         "outcomes": dict(outcomes),
+        "outcomes_by_kind": {kind: dict(found) for kind, found in by_kind.items()},
         "text": dict(text),
         "paths": dict(Counter(r.get("path") for r in answered)),
         "tiers": {k: tiers.get(k, 0) for k in TIERS},

backend/tests/test_eval_analysis.py

@@ -0,0 +1,317 @@
+"""分析期望（贡献、判定）的规矩和评分（app/eval/judgements.py、cases.py、score.py；docs/design/analysis-round/00-总方案.md
+§3.8）。纯计算：期望的分析结论用 app/analysis/testing 的假结论（形状同真算子），答案直接构造。走真运行的在
+tests/test_eval_analysis_runs.py。地区、门店、数字都是虚构的。
+
+- 校验：贡献恒为变化拆解、要选模式；判定是另外四种分析；参数只能写分析工具的那几个、不能写阈值、不能写输入；来源、方向、
+  判定不许人填；文字参数里的变量只认定时任务那份；口径题配不了分析期望。
+- 期望值：判定码归到判定类，主要来源按 top_n 取前几段；比较作答前后两次期望时不看运行当天、现查时刻、输入快照的工件。
+- 评分五种结论：引用了判定对得上的结论 → 对且可追溯；正文按说法表写对了、没引用 → 对但没引用；引用的判定对不上、正文
+  写反了（常规波动说成异常、方向反了、来源说成别的段）、根本没做也没说 → 错；期望算不了 → 不能评分。
+"""
+from __future__ import annotations
+
+import pytest
+
+import app.db.models  # noqa: F401 - conftest 按已导入的模型建表：只跑这一个文件时也要有表
+from app.analysis.testing import fake_result
+from app.eval import cases, judgements, stats
+from app.eval.cases import CaseInvalid, normalize_case
+from app.eval.score import Answer, score_case
+
+SQL = {"kind": "sql", "datasource_id": "ds1", "sql": "SELECT region AS 地区, week AS 周, SUM(amount) AS 销售额 FROM orders"}
+SOURCE = {"id": "e1", "kind": "contribution", "required": True, "op": "decompose", "mode": "additive",
+          "params": {"by": ["地区"], "period": "周", "measure": "销售额"}, "top_n": 1}
+BAND = {"id": "e1", "kind": "verdict", "required": True, "op": "baseline",
+        "params": {"period": "周", "measure": "订单数"}}
+
+
+def invalid(expectation: dict, **gold) -> CaseInvalid:
+    with pytest.raises(CaseInvalid) as caught:
+        normalize_case({"question": "上周销售额为什么下降？", "gold": gold or SQL, "expectations": [expectation]})
+    return caught.value
+
+
+# --------------------------------------------------------------------------
+# 校验
+# --------------------------------------------------------------------------
+
+
+def test_the_two_kinds_are_normalized():
+    fields = normalize_case({"question": "上周销售额为什么下降？", "gold": SQL, "expectations": [
+        {"kind": "contribution", "mode": "additive", "params": {"by": ["地区"], "period": "周", "measure": "销售额",
+                                                                  "label": "", "base": None}},
+        {"kind": "verdict", "op": "comparable", "params": {
+            "period": "周", "base": "{{ schedule.last_week_start }}", "target": "{{ schedule.this_week_start }}"}}]})
+    contribution, verdict = fields["expectations"]
+    assert contribution == {"id": "e1", "kind": "contribution", "required": True, "op": "decompose", "mode": "additive",
+                            "params": {"by": ["地区"], "period": "周", "measure": "销售额"}, "top_n": 1}
+    assert verdict["op"] == "comparable" and "mode" not in verdict and verdict["params"]["base"].startswith("{{")
+    assert cases.case_hash(fields) == cases.case_hash(dict(fields)), "指纹照常按内容算"
+
+
+@pytest.mark.parametrize("expectation, field, words", [
+    ({"kind": "contribution", "op": "baseline", "params": {}}, "expectations.0.op", "只能是变化拆解"),
+    ({"kind": "contribution", "params": {"by": ["地区"]}}, "expectations.0.mode", "要选一种模式"),
+    ({"kind": "verdict", "params": {"period": "周", "measure": "订单数"}}, "expectations.0.op", "波动基线、趋势判断"),
+    ({"kind": "verdict", "op": "decompose", "mode": "additive"}, "expectations.0.op", "当前为「decompose」"),
+    ({"kind": "verdict", "op": "trend", "mode": "additive", "params": {"period": "周", "measure": "m"}},
+     "expectations.0.mode", "没有模式可选"),
+    ({"kind": "verdict", "op": "differ", "params": {}}, "expectations.0.mode", "差异检验（比率）"),
+    ({"kind": "verdict", "op": "baseline", "params": {"period": "周"}}, "expectations.0.params.measure", "没写"),
+    ({**SOURCE, "params": {**SOURCE["params"], "main_share": 0.3}}, "expectations.0.params.main_share",
+     "阈值不写在期望里"),
+    ({**SOURCE, "params": {**SOURCE["params"], "query": "Q1"}}, "expectations.0.params.query", "没有「query」"),
+    ({**SOURCE, "params": {**SOURCE["params"], "headline": {"base": "Q1.r0.x", "target": "Q1.r1.x"}}},
+     "expectations.0.params.headline", "没有「headline」"),
+    ({**SOURCE, "params": {**SOURCE["params"], "base": "{{ schedule.nope }}"}}, "expectations.0.params.base",
+     "不是可用的变量"),
+    ({**SOURCE, "top_n": 0}, "expectations.0.top_n", "1 到 10"),
+    ({**SOURCE, "params": "地区"}, "expectations.0.params", "需要是一个对象"),
+])
+def test_wrong_analysis_expectations_are_refused(expectation, field, words):
+    e = invalid(expectation)
+    assert e.field == field and words in e.message, (e.field, e.message)
+
+
+@pytest.mark.parametrize("key", ["segment", "subjects", "direction", "verdict", "judgement", "driver", "value"])
+def test_conclusions_cannot_be_filled_in_by_hand(key):
+    e = invalid({**SOURCE, key: "华东"})
+    assert e.field == f"expectations.0.{key}" and "不许人填" in e.message and "算出来" in e.message
+
+
+def test_a_caliber_gold_cannot_carry_an_analysis_expectation():
+    caliber = {"kind": "caliber", "workflow_id": "w", "version": 1, "node_id": "c", "inputs": {}}
+    assert invalid(BAND, **caliber).field == "expectations.0.kind"
+    # 期望反问的题同样不能带分析期望：问法含糊就不该下判断
+    with pytest.raises(CaseInvalid):
+        normalize_case({"question": "最近怎么样？", "gold": SQL, "expectations": [{"kind": "ask"}, BAND]})
+
+
+def test_the_request_renders_variables_and_fills_two_periods():
+    snapshot = {"columns": ["地区", "周", "销售额"], "sql": SQL["sql"],
+                "rows": [["华东", "2026-10-12", 900], ["华东", "2026-10-05", 1000], ["华南", "2026-10-05", 800]]}
+    values = {"last_week_start": "2026-10-12", "this_week_start": "2026-10-19"}
+    request = judgements.request_of(SOURCE, values, snapshot)
+    assert request["op"] == "decompose" and request["mode"] == "additive"
+    assert request["params"] == {"query": judgements.GOLD_REF, "by": ["地区"], "period": "周",
+                                 "base": "2026-10-05", "target": "2026-10-12", "measure": "销售额"}
+    explicit = {**SOURCE, "params": {**SOURCE["params"], "target": "{{ schedule.last_week_start }}",
+                                     "base": "2026-10-05"}}
+    assert judgements.request_of(explicit, values, snapshot)["params"]["target"] == "2026-10-12"
+    three = {**snapshot, "rows": [*snapshot["rows"], ["华南", "2026-09-28", 700]]}
+    with pytest.raises(judgements.ParamsInvalid) as caught:
+        judgements.request_of(SOURCE, values, three)
+    assert "有 3 期" in caught.value.message and "只能返回两期" in caught.value.message
+
+
+# --------------------------------------------------------------------------
+# 期望值、比较用的那一份
+# --------------------------------------------------------------------------
+
+
+def _gold(*pairs: tuple[dict, dict]) -> dict:
+    values = {exp["id"]: judgements.expected(exp, result) for exp, result in pairs}
+    sets = [{"run_id": "g", "snapshot": "x", "version": None, "values": values}]
+    return {"status": "ok", "reason": None, "code": None, "before": {"status": "ok", "sets": sets},
+            "after": {"status": "ok", "sets": sets}}
+
+
+def _case(*expectations: dict) -> dict:
+    return {"kind": "ask", "question": "虚构问题", "expectations": list(expectations), "gold": SQL}
+
+
+def test_expected_values_follow_the_judgement_classes():
+    driver = judgements.expected(SOURCE, fake_result())
+    assert (driver["judgement"], driver["subjects"], driver["direction"]) == ("source", ["华东"], "down")
+    assert driver["phrase"].startswith("销售额下降") and driver["result"]["op"] == "decompose"
+    two = judgements.expected({**SOURCE, "top_n": 2}, fake_result(verdict="concentrated", subjects=[["华东"], ["华北"]]))
+    assert two["judgement"] == "source" and two["subjects"] == ["华东", "华北"]
+    assert judgements.expected(SOURCE, fake_result(verdict="concentrated", subjects=[["华东"], ["华北"]])
+                               )["subjects"] == ["华东"], "top_n 缺省 1：只要点名第一段"
+    for code, judgement in (("within_band", "normal"), ("outside_band", "anomaly"), ("far_outside_band", "anomaly")):
+        assert judgements.expected(BAND, fake_result("baseline", None, code))["judgement"] == judgement
+    for code, judgement in (("comparable", "comparable"), ("conditional", "comparable"),
+                            ("incomparable", "incomparable")):
+        exp = {**BAND, "op": "comparable"}
+        assert judgements.expected(exp, fake_result("comparable", None, code))["judgement"] == judgement
+    differ = judgements.expected({**BAND, "op": "differ", "mode": "rate"},
+                                 fake_result("differ", "rate", "no_evidence", subjects=[["西南"], ["华东"]],
+                                             direction=None))
+    assert differ["judgement"] == "no_evidence" and differ["subjects"] == ["西南", "华东"]
+    refused = judgements.expected(BAND, fake_result("baseline", None, None, status="refused",
+                                                    reason_code="history_short"))
+    assert refused["status"] == "refused" and refused["reason"] == "历史只有 5 期，至少要 8 期"
+    assert set(judgements.JUDGEMENTS) >= {v for m in judgements._JUDGEMENT_OF.values() for v in m.values()}
+
+
+def test_comparing_two_golds_ignores_the_day_the_probe_time_and_the_snapshot_artifact():
+    from app.eval import gold as gold_rules
+
+    def outcome(day: str, at: str, artifact: str) -> gold_rules.GoldOutcome:
+        result = fake_result("baseline", None, "within_band")
+        result["params"]["today"] = day
+        result["probes"] = [{"source": "shop", "table": "orders", "column": "order_date", "min": "2026-04-01",
+                             "max": "2026-10-20", "basis": "live", "at": at}]
+        result["inputs"] = [{"ref": judgements.GOLD_REF, "artifact": artifact, "role": "data", "columns": ["周"],
+                             "rows": "all"}]
+        values = {"e1": judgements.expected(BAND, result), "e2": {"kind": "scalar", "value": 330, "column": "gmv"}}
+        return gold_rules.GoldOutcome(status="ok", as_of="", sets=[gold_rules.GoldSet("r", "s", values)])
+
+    before = outcome("2026-10-21", "2026-10-21T09:00:01", "a" * 64)
+    after = outcome("2026-10-22", "2026-10-22T00:00:03", "b" * 64)
+    assert gold_rules.settle(before, after).status == "ok"
+    from app.eval.runner import gold_key
+
+    assert gold_key({"status": "ok", "before": before.to_dict()}) == gold_key({"status": "ok", "before": after.to_dict()})
+    changed = outcome("2026-10-21", "x", "a" * 64)
+    changed.sets[0].values["e1"]["result"]["probes"][0]["max"] = "2026-10-19"
+    assert gold_rules.settle(before, changed).status == "drifted", "表的范围变了才算数据变了"
+    # 别的期望的指纹和以前一样（老结果和新结果比「数据变了」不受影响）
+    from app.core.artifact_store import canonical_json
+
+    plain = gold_rules.GoldOutcome(status="ok", as_of="", sets=[gold_rules.GoldSet("r", "s", {"e2": {"value": 1}})])
+    assert gold_rules._fingerprint(plain) == canonical_json([[None, {"e2": {"value": 1}}]])
+
+
+# --------------------------------------------------------------------------
+# 评分
+# --------------------------------------------------------------------------
+
+
+def _record(result: dict, alias: str | None = "A1", cited: tuple[str, ...] = ("phrase",)) -> dict:
+    record = judgements.record_of(result, artifact="c" * 64)
+    record.update(alias=alias, cited=list(cited))
+    return record
+
+
+def _answer(*records: dict, text: str = "", cited_in: tuple[str, ...] = ()) -> Answer:
+    sentences = judgements.sentences_of_text(text)
+    for s in sentences:
+        s["analyses"] = list(cited_in)
+    return Answer(run_id="r", text=text, analyses=list(records), sentences=sentences)
+
+
+def _score(exp: dict, result: dict, answer: Answer) -> dict:
+    graded = score_case(_case(exp), _gold((exp, result)), answer, path="ran")
+    [item] = graded.scores
+    return item
+
+
+def test_a_cited_conclusion_that_agrees_is_traced():
+    item = _score(SOURCE, fake_result(), _answer(_record(fake_result())))
+    assert item["outcome"] == "traced" and item["found"]["alias"] == "A1", item
+    assert item["want"] == {"judgement": "source", "subjects": ["华东"], "direction": "down", "status": "ok",
+                            "phrase": fake_result()["phrase"], "method": "变化拆解（可加）"}
+    # 主要来源点名了期望的那一段就行：答案的拆解判成集中在华东、华北也对得上
+    wider = fake_result(verdict="concentrated", subjects=[["华东"], ["华北"]])
+    assert _score(SOURCE, fake_result(), _answer(_record(wider)))["outcome"] == "traced"
+
+
+def test_the_text_right_without_a_citation_is_untraced():
+    item = _score(SOURCE, fake_result(), _answer(text="上周销售额下降主要来自华东。"))
+    assert item["outcome"] == "untraced" and item["found"]["text"] == "上周销售额下降主要来自华东", item
+    normal = _score(BAND, fake_result("baseline", None, "within_band"),
+                    _answer(text="订单数的下降在常规波动范围内，不算异常。"))
+    assert normal["outcome"] == "untraced", normal
+    rising = _score({**BAND, "op": "trend"}, fake_result("trend", None, "rising", direction="up"),
+                    _answer(text="线上销售额持续上升。"))
+    assert rising["outcome"] == "untraced", rising
+
+
+@pytest.mark.parametrize("text, words", [
+    ("订单数的下降属于异常。", "判定是"),                      # 常规波动说成异常
+    ("上周销售额下降主要来自华南。", "不是华南"),                # 主要来源说成别的段
+    ("上周销售额上升主要来自华东。", "涨跌写反了"),              # 方向反了
+    ("线上销售额呈下降趋势。", "相反"),                          # 趋势方向反了
+])
+def test_saying_the_opposite_is_wrong(text, words):
+    exp, result = {
+        "订单数": (BAND, fake_result("baseline", None, "within_band")),
+        "销售额": (SOURCE, fake_result()),
+        "线上": ({**BAND, "op": "trend"}, fake_result("trend", None, "rising", direction="up")),
+    }[next(k for k in ("订单数", "线上", "销售额") if k in text)]
+    item = _score(exp, result, _answer(text=text))
+    assert item["outcome"] == "wrong" and words in item["reason"], item
+
+
+def test_a_cited_conclusion_that_disagrees_is_wrong():
+    anomaly = fake_result("baseline", None, "outside_band", direction="down")
+    item = _score(BAND, fake_result("baseline", None, "within_band"), _answer(_record(anomaly)))
+    assert item["outcome"] == "wrong" and "A1" in item["reason"] and "不是一类" in item["reason"], item
+    assert item["found"]["judgement"] == "anomaly"
+    # 差异检验比的是别的两组：对不上
+    other = fake_result("differ", "rate", "no_evidence", subjects=[["华北"], ["华东"]], direction=None)
+    want = fake_result("differ", "rate", "no_evidence", subjects=[["西南"], ["华东"]], direction=None)
+    item = _score({**BAND, "op": "differ", "mode": "rate"}, want, _answer(_record(other)))
+    assert item["outcome"] == "wrong" and "期望比的是西南和华东" in item["reason"], item
+
+
+def test_citing_the_right_conclusion_but_writing_the_opposite_is_wrong():
+    normal = fake_result("baseline", None, "within_band")
+    item = _score(BAND, normal, _answer(_record(normal, cited=("see",)), text="订单数下降属于异常。",
+                                        cited_in=("A1",)))
+    assert item["outcome"] == "wrong" and "引用了 A1" in item["reason"], item
+
+
+def test_doing_nothing_and_saying_nothing_is_wrong():
+    item = _score(BAND, fake_result("baseline", None, "within_band"), _answer(text="上周订单数是 1,203 单。"))
+    assert item["outcome"] == "wrong" and "没有做波动基线" in item["reason"], item
+    uncited = _score(BAND, fake_result("baseline", None, "within_band"),
+                     _answer(_record(fake_result("baseline", None, "within_band"), cited=()), text="上周订单数是 1,203 单。"))
+    assert uncited["outcome"] == "wrong" and "但没有引用" in uncited["reason"], uncited
+    # 可比性没有对应的判断词：只能引用
+    exp = {**BAND, "op": "comparable"}
+    words = _score(exp, fake_result("comparable", None, "incomparable"), _answer(text="本周还没过完，两周不能直接比。"))
+    assert words["outcome"] == "wrong"
+    cited = _score(exp, fake_result("comparable", None, "incomparable"),
+                   _answer(_record(fake_result("comparable", None, "incomparable"))))
+    assert cited["outcome"] == "traced"
+
+
+def test_an_analysis_that_cannot_be_computed_cannot_be_graded():
+    refused = fake_result("baseline", None, None, status="refused", reason_code="history_short")
+    graded = score_case(_case(BAND, {"id": "e2", "kind": "scalar", "column": "gmv", "required": True}),
+                        {**_gold((BAND, refused)), "before": {"status": "ok", "sets": [{"values": {
+                            "e1": judgements.expected(BAND, refused),
+                            "e2": {"kind": "scalar", "value": 330.0, "column": "gmv", "row": 0, "dims": {}}}}]}},
+                        Answer(run_id="r", values=[{"value": 330.0, "kind": "cell", "artifact": "a", "column": "gmv",
+                                                    "dims": {}}]), path="ran")
+    band, sales = graded.scores
+    assert band["outcome"] == "ungradable" and band["reason"] == "期望算不了：历史只有 5 期，至少要 8 期"
+    assert sales["outcome"] == "traced", "同一题别的期望照常评"
+    assert graded.verdict == "ungradable"
+    # 答案里同类的分析做成了算不了：不算答对
+    item = _score(BAND, fake_result("baseline", None, "within_band"), _answer(_record(refused)))
+    assert item["outcome"] == "wrong" and "没有做成" in item["reason"], item
+
+
+def test_top_n_wants_every_named_source():
+    want = fake_result(verdict="concentrated", subjects=[["华东"], ["华北"]])
+    exp = {**SOURCE, "top_n": 2}
+    only_one = _score(exp, want, _answer(text="上周销售额下降集中在华东。"))
+    assert only_one["outcome"] == "wrong" and "还有华北" in only_one["reason"], only_one
+    both = _score(exp, want, _answer(text="上周销售额下降集中在华东、华北。"))
+    assert both["outcome"] == "untraced", both
+
+
+def test_old_answers_without_analyses_read_as_empty():
+    old = Answer.from_dict({"run_id": "r", "text": "上周销售额 330。", "values": [], "bare": []})
+    assert old.analyses == [] and old.sentences == []
+    item = _score(SOURCE, fake_result(), old)
+    assert item["outcome"] == "wrong", item
+    again = Answer.from_dict(_answer(_record(fake_result())).to_dict())
+    assert again.analyses[0]["alias"] == "A1"
+
+
+def test_the_summary_counts_outcomes_by_kind():
+    rows = [{"status": "ok", "verdict": "pass", "case_id": "c1", "scores": [
+        {"kind": "contribution", "required": True, "outcome": "traced"},
+        {"kind": "scalar", "required": False, "outcome": "wrong"}]},
+        {"status": "ok", "verdict": "fail", "case_id": "c2", "scores": [
+            {"kind": "verdict", "required": True, "outcome": "wrong"},
+            {"kind": "scalar", "required": True, "outcome": "untraced"},
+            {"kind": "text", "required": False, "outcome": "covered"}]}]
+    summary = stats.summarize(rows)
+    assert summary["outcomes_by_kind"] == {"contribution": {"traced": 1}, "verdict": {"wrong": 1},
+                                           "scalar": {"untraced": 1}}
+    assert summary["outcomes"] == {"traced": 1, "wrong": 1, "untraced": 1}