口径标签逐格判：引用了好几列的查询，每个数按它自己那一格标

真实模型下问「上周的销售额是多少」，助手写 SUM(amount) AS 销售额, COUNT(*) AS 订单数, MIN / MAX(paid_date) AS 区间首日 /
末日，报告四列都引用了，口径「周销售」只取 SUM(amount)。按「全部引用列」投影后还是 4 列，对不上口径的 1 列，整条比、
四个数都标相关口径，说明整段兜底。

- ledger/match._cells / _cell（algo 4）：整条（含按全部引用列投影）没证明、又引用了这条查询两列以上的，被引用的每一列单独
  投影到那一列，和每份口径的每个指标读的那一列（card_reads；口径版本没有口径卡的按取数的每一列）逐个比，仍只走
  match_template / explain_sql_diff，上次的投影规则照用：投影不了（分组列、DISTINCT……）、过了时限、不可比的格照用整条
  查询的那一个候选。结果记在 queries[].cells {列名: {status, candidates}}，另记 one_row（sql_semantics.one_row：不分组的
  聚合，一定只出一行）。整条这一层的结论、候选一字不改；老工件没有 cells，照旧按整条读。
- ledger/labels：单元格按它那一格的结论标（_view）；指标用了同一条查询的好几格时，只有那条查询一定只出一行才把逐格的
  结论合起来用（多行的结果里每一列各自对得上，不等于每一行的几列一起对得上）。整题照拍板 L1 不变：每个数都证明来自
  同一份口径才等价；一格证明了、别的格只是相关的整题不等价，原因写「有的数和相关口径不一致，不来自口径」（以前一律说
  「这次的推理」）。出具入口照 ledger/issue 现有判法只在整题等价时给，没有放宽。
- 逐格证明了的口径各处都算上（labels.candidates_of）：工件 meta 的 calibers（召回按旧版本反查）、ledger/uses 的使用记录、
  召回时找原答案对应的那一格、E4；E1 另数各种结论各几格（cells），E2 的兜底比例按标签上给人看的那一份说明数。
- GET /runs/{id}/ledger 的 queries 带上逐格的 cells（遮罩同候选）；证据片段的 ledger 键、轮次摘要的计数本来就按每个数的
  标签，数据变了、形状不变。前端只补类型；check-ledger 加一段：同一条查询一格来自口径（正文挂小标记）、一格相关（差在哪
  先说多查了几列），整题不等价不给出具。
- 评测逐题比 SQL（eval/score.sql_match）取和期望配上的那一格（score.matched_cells：评分时配上的格），没给时取逐格证明了
  的那一格；结果带 column。runner、rescore 传它。

测试：tests/test_ledger_cells.py（同一条查询一列等价、一列不等价各自标对；两份口径只证明了一份标那一份；口径多个指标按列
配对；指标跨两格只在一行的查询上用逐格结论；分组列、DISTINCT 逐格也不投影；照真实例子造的虚构库真跑一遍：销售额标等价
已证明，订单数、首末日标相关，整题不等价、不给出具、不自动出具，证据片段的 ledger 键一致；评测取配上的那一格）。去掉逐格
再跑，10 条失败；去掉「一行才合起来用」、去掉评测取格，各 1 条失败。

backend/app/data/sql_semantics.py

@@ -9147,6 +9147,19 @@ def project_template(template_sql: str, keep: Sequence[int], dialect: str | None
         return Projection(None, keep, refused=_PROJ_TEXT)
 
 
+def one_row(sql: str, dialect: str | None = None) -> bool:
+    """这条查询一定只出一行（或者一行都不出）：一段 SELECT、没有 GROUP BY、本身是聚合（不分组的聚合，HAVING 至多筛掉那
+    一行）。逐格比出来的结论要合起来用（一个指标用了同一条查询的好几格）时，只在这种查询上成立：多行的结果里，每一列
+    各自对得上不等于每一行的几列一起对得上。认不出的为假。"""
+    from sqlglot import exp
+
+    read = _read(dialect)
+    root, _ = _parse(_mysql_dashes(sql) if read == "mysql" and isinstance(sql, str) else sql, read)
+    if not isinstance(root, exp.Select) or root.args.get("group") is not None or root.args.get("qualify") is not None:
+        return False
+    return _aggregating(root)
+
+
 def output_names(sql: str, dialect: str | None = None, *, template: bool = False) -> list[str] | None:
     """最外层选择列表各项的输出名（按位置；同 _result_columns，不展开 *：有 * 的为 None）。template 为真时 sql 是口径
     模板，按哨兵渲染之后认。解析不了、不是一段 SELECT 的为 None。"""
@@ -9329,5 +9342,5 @@ def sql_shape(sql: str, dialect: str | None = None) -> SqlShape | None:
 __all__ = ["BINDING_LEVELS", "MAX_SQL_CHARS", "PLACEHOLDER", "Aggregate", "ColumnRef", "Computed", "FacetDiff", "Facets",
            "JoinFacet", "LimitFacet", "NormalizedSql", "Part", "Projection", "SqlDiff", "SqlShape", "TemplateBinding",
            "TemplateMatch", "TemplateSlot", "TimePredicate", "bind_template", "catalog_labels", "diff_facets",
-           "explain_sql_diff", "facets", "match_template", "normalize_sql", "output_names", "project_select",
+           "explain_sql_diff", "facets", "match_template", "normalize_sql", "one_row", "output_names",
-           "project_template", "render_bound", "sql_shape"]
+           "project_select", "project_template", "render_bound", "sql_shape"]

backend/app/eval/rescore.py

@@ -53,7 +53,7 @@ async def rescore(eval_run_id: str) -> int:
             answer = await score.read_answer(row.run_id if row.path == "ran" else None, reply=turn.get("reply"))
             graded = score.score_case(case, row.gold, answer, path=row.path, path_reason=turn.get("reason"),
                                       judgement=(row.judge or {}).get("text"))
-            match = await score.sql_match(case, row.gold, answer)
+            match = await score.sql_match(case, row.gold, answer, cells=score.matched_cells(graded))
         if graded.verdict != row.verdict:
             changed += 1
         updates[row.id] = {"verdict": graded.verdict, "scores": graded.scores, "sql_match": match,

backend/app/eval/runner.py

@@ -576,7 +576,7 @@ class EvalExecution(runtime.JobRun):
             judge_cost += float(judge["text"].get("cost_usd") or 0)
         graded = score.score_case(case, gold_dict, answer, path=turn.path, path_reason=turn.reason,
                                   judgement=judge.get("text"))
-        match = await score.sql_match(case, gold_dict, answer)
+        match = await score.sql_match(case, gold_dict, answer, cells=score.matched_cells(graded))
         writer = await writer_usage(turn.runs)
         assistant = add_usage(turn.usage.get("assistant"), turn.usage.get("review"))
         usage = {"assistant": assistant, "writer": writer, "review": turn.usage.get("review"),

backend/app/eval/score.py

@@ -613,10 +613,14 @@ async def gold_calibers(case: Mapping[str, Any], gold: Mapping[str, Any] | None)
 
 
 async def sql_match(case: Mapping[str, Any], gold: Mapping[str, Any] | None, answer: Answer, *,
-                    timeout: float = 8.0) -> dict[str, Any] | None:
+                    timeout: float = 8.0, cells: Sequence[Any] = ()) -> dict[str, Any] | None:
-    """逐题的 E1：{status: proven|related|none|skipped, query, aspects, diff, reason, cited, basis}。没有作答运行、没有期望
+    """逐题的 E1：{status: proven|related|none|skipped, query, column?, aspects, diff, reason, cited, basis}。没有作答运行、
-    取数的为 None。比的是哪条查询见 _answering：basis 是 cited（答案引用的）或 only_query（答案没引用、运行只查了这一条）。
+    没有期望取数的为 None。比的是哪条查询见 _answering：basis 是 cited（答案引用的）或 only_query（答案没引用、运行只查了
-    比对在线程里跑（纯计算，和账本一样限时）。"""
+    这一条）。比对在线程里跑（纯计算，和账本一样限时）。
+
+    和账本同一个口径逐格判（ledger/match._cells）：引用了好几列、整条比不上的查询，每一列各比一次。cells 是评分时和期望
+    配上的那几格（各项期望的 cell：{artifact, column}）：取那一格的结论（column 写是哪一列）；没给、配上的格没逐格比过时，
+    逐格证明了的取那一格，再不然照旧按整条查询。"""
     import asyncio
     import time
 
@@ -653,8 +657,9 @@ async def sql_match(case: Mapping[str, Any], gold: Mapping[str, Any] | None, ans
     pool, basis = _answering(rows, answer)
     if isinstance(pool, dict):
         return pool
+    column, cell = _cell_of(pool, cells)
     proven = next((r for r in pool if r.get("status") == "proven"), None)
-    chosen = proven or next((r for r in pool if r.get("status") == "related"), None) or \
+    chosen = cell or proven or next((r for r in pool if r.get("status") == "related"), None) or \
         next((r for r in pool if r.get("status") == "skipped"), None) or pool[0]
     candidate = next((c for c in chosen.get("candidates") or [] if c.get("status") == chosen.get("status")), None) or \
         next(iter(chosen.get("candidates") or []), None) or {}
@@ -662,9 +667,34 @@ async def sql_match(case: Mapping[str, Any], gold: Mapping[str, Any] | None, ans
     # projection 那一条只是说「多查了几列，不影响这个数」，不是差在哪个方面（账本的 E1 差异类型也不数它）
     aspects = sorted({i.get("aspect") for i in diff.get("items") or [] if i.get("aspect") not in (None, "", "projection")})
     sql = next((q.get("sql") for q in answer.queries if q.get("artifact") == chosen.get("artifact")), None)
-    return {"status": chosen.get("status"), "query": chosen.get("artifact"), "sql": sql, "aspects": aspects,
+    return {"status": chosen.get("status"), "query": chosen.get("artifact"), **({"column": column} if column else {}),
-            "diff": diff or None, "via": candidate.get("via"), "binding": candidate.get("binding"),
+            "sql": sql, "aspects": aspects, "diff": diff or None, "via": candidate.get("via"),
-            "reason": chosen.get("reason") or candidate.get("reason"), "cited": basis == "cited", "basis": basis}
+            "binding": candidate.get("binding"), "reason": chosen.get("reason") or candidate.get("reason"),
+            "cited": basis == "cited", "basis": basis}
+
+
+def matched_cells(graded: "CaseScore") -> list[dict[str, Any]]:
+    """评分时各项期望和答案里配上的那一格（答对的、量纲不同的、答错时配对上的）：逐题 E1 取它们的逐格结论。"""
+    return [dict(item["cell"]) for item in graded.scores
+            if isinstance(item, Mapping) and isinstance(item.get("cell"), dict)]
+
+
+def _cell_of(pool: list[dict[str, Any]], cells: Sequence[Any]) -> tuple[str | None, dict[str, Any] | None]:
+    """逐格比过的格里挑一格：评分时和期望配上的那几格先（其中证明了的先），再不然逐格证明了的那一格。交回 (列名, 那一格的
+    结论，带上查询快照 artifact)；没有可挑的为 (None, None)。"""
+    hinted = [(str(c.get("artifact")), str(c.get("column"))) for c in cells
+              if isinstance(c, Mapping) and c.get("artifact") and c.get("column") is not None]
+    found: list[tuple[str, dict[str, Any]]] = []
+    for row in pool:
+        for name, entry in (row.get("cells") or {}).items() if isinstance(row.get("cells"), dict) else []:
+            if isinstance(entry, dict):
+                found.append((name, {**entry, "artifact": row.get("artifact"), "_hinted": (row.get("artifact"), name)
+                                     in hinted}))
+    ranked = sorted(found, key=lambda f: (not f[1]["_hinted"], f[1].get("status") != "proven"))
+    for name, entry in ranked:
+        if entry["_hinted"] or entry.get("status") == "proven":
+            return name, {k: v for k, v in entry.items() if k != "_hinted"}
+    return None, None
 
 
 def _answering(rows: list[dict[str, Any]], answer: Answer) -> tuple[list[dict[str, Any]] | dict[str, Any], str]:
@@ -692,4 +722,5 @@ def _answering(rows: list[dict[str, Any]], answer: Answer) -> tuple[list[dict[st
 
 
 __all__ = ["Answer", "CaseScore", "DEFAULT_TOLERANCE", "OUTCOMES", "OUTCOME_LABEL", "PATH_LABEL", "SCORER_VERSION",
-           "TEXT_LABEL", "TEXT_OUTCOMES", "VERDICT_LABEL", "gold_calibers", "read_answer", "score_case", "sql_match"]
+           "TEXT_LABEL", "TEXT_OUTCOMES", "VERDICT_LABEL", "gold_calibers", "matched_cells", "read_answer",
+           "score_case", "sql_match"]

backend/app/ledger/labels.py

@@ -8,10 +8,14 @@
 - unknown：没比对、比对没做完（超时、解析失败）、快照读不出来。
 
 怎么认：
-- 单元格（[[v:]]、[[table:]] 里的格）：按证据目录找到查询快照，取那条查询的比对状态。等价时按列映射写出口径那边的列名。
+- 单元格（[[v:]]、[[table:]] 里的格）：按证据目录找到查询快照，取那条查询的比对状态；这条查询逐格比过（algo 4：整条按全部
+  引用列投影没证明时，被引用的每一列单独投影、和口径的每个指标列逐个比，ledger/match._cells）的，按这一格所在的那一列的
+  结论——同一条查询里一列等价、一列不等价各标各的。等价时按列映射写出口径那边的列名。
 - 指标（[[m:]]）：指标集工件里这个指标的每个输入都要来自查询快照，而且这些快照都被证明等价于同一份口径的同一版、绑定一致；
   指标式还要和那一版里某张口径卡上的某个指标相同——两边的式子把每个输入换成「口径的哪个取数节点、第几行、哪一列」之后
   一字不差（caliber_from 转抄来的式子本来就是那一张卡的，同样按这个比）。输入有不来自查询的（运行输入、整形、代码）不算来自口径。
+  逐格比过的输入按它那一格的结论；同一条查询的好几格合起来用时，只在这条查询一定只出一行（one_row）时才按逐格的结论，
+  别的按整条查询：多行的结果里每一列各自对得上，不等于每一行的几列一起对得上。
 
 整题（whole_of）：主报告里所有带标签的数都来自同一口径的同一版、绑定一致，就记 equivalent，并写出还缺的入口字段（口径入口
 节点声明了、这次的绑定里没有的；required 且没有缺省值的另列）。
@@ -92,11 +96,38 @@ def _caliber_column(cand: Mapping[str, Any], column: Any) -> str | None:
     return None
 
 
+def cells_of(query: Mapping[str, Any]) -> list[Mapping[str, Any]]:
+    """这条查询逐格比出来的几格（algo 4 的 cells，见 ledger/match._cells）；老工件、没逐格比的为空。"""
+    cells = query.get("cells")
+    return [c for c in cells.values() if isinstance(c, Mapping)] if isinstance(cells, Mapping) else []
+
+
+def candidates_of(query: Mapping[str, Any], *, shown: bool = False) -> list[Mapping[str, Any]]:
+    """一条查询的候选：整条比的，加上逐格比出来的。召回反查、使用次数、工件 meta 都要算上逐格证明了的那几份口径。shown 为真
+    时只要标签上给人看的那一份：逐格比过的只要逐格的（整条的那一份不在标签上），别的照旧是整条的（统计兜底说明的比例用）。"""
+    cells = [c for cell in cells_of(query) for c in cell.get("candidates") or [] if isinstance(c, Mapping)]
+    whole = [c for c in query.get("candidates") or [] if isinstance(c, Mapping)]
+    if shown:
+        return cells or whole
+    return whole + cells
+
+
+def _view(query: Mapping[str, Any], column: Any, *, alone: bool = True) -> Mapping[str, Any]:
+    """这一格按什么判：这条查询逐格比过这一列（cells 里有）的按那一格的结论，别的按整条查询的。alone 为假（一个指标用了
+    这条查询的好几格）时，只有这条查询一定只出一行（one_row）才按逐格的（见模块说明）。老工件没有 cells，照旧按整条。"""
+    cells = query.get("cells")
+    cell = cells.get(column) if isinstance(cells, Mapping) and isinstance(column, str) else None
+    if isinstance(cell, Mapping) and (alone or query.get("one_row") is True):
+        return cell
+    return query
+
+
 def cell_label(artifact: str, column: Any, queries: Mapping[str, Mapping[str, Any]]) -> dict[str, Any]:
-    """单元格的标签：它所在的那条查询的比对状态。"""
+    """单元格的标签：它所在的那条查询的比对状态（逐格比过的按这一格，_view）。"""
-    query = queries.get(artifact)
+    found = queries.get(artifact)
-    if query is None:
+    if found is None:
         return _unknown(NO_RECORD)
+    query = _view(found, column)
     status = query.get("status")
     if status in ("proven", "composed"):
         cand = _proven(query)[0] if _proven(query) else {}
@@ -247,9 +278,18 @@ def metric_label(card: Mapping[str, Any] | None, metric_id: Any, queries: Mappin
     from_queries = [i for i in inputs if i.get("artifact") and i.get("via") in _SNAPSHOT_VIAS]
     if not from_queries:
         return {"kind": "reasoning"}
-    found = [queries.get(str(i["artifact"])) for i in from_queries]
+    rows = [queries.get(str(i["artifact"])) for i in from_queries]
-    if any(q is None for q in found):
+    if any(q is None for q in rows):
         return _unknown(NO_RECORD)
+    # 每个输入按它那一格判（_view）：同一条查询用了好几格的，那条查询一定只出一行才按逐格的结论
+    used: dict[str, set[Any]] = {}
+    for item in from_queries:
+        where = item.get("locator") if isinstance(item.get("locator"), dict) else {}
+        used.setdefault(str(item["artifact"]), set()).add(where.get("column"))
+    found: list[Mapping[str, Any] | None] = [
+        _view(q, (i.get("locator") if isinstance(i.get("locator"), dict) else {}).get("column"),
+              alone=len(used[str(i["artifact"])]) == 1) if q is not None else None
+        for q, i in zip(rows, from_queries)]
     if any(q.get("status") == "skipped" for q in found):      # type: ignore[union-attr]
         return _unknown(next(str(q.get("reason") or UNFINISHED) for q in found
                              if q.get("status") == "skipped"))  # type: ignore[union-attr]
@@ -353,8 +393,9 @@ def whole_of(labels: Mapping[str, Mapping[str, Any]], calibers: Mapping[str, Map
         return {"equivalent": False, "reason": "报告里没有可标口径的数", "counts": counts}
     keys = {label.get("caliber") for label in labels.values() if label.get("kind") == "caliber"}
     if counts["caliber"] != len(labels):
-        reason = ("有的数没有比对" if counts["unknown"] else "有的数是这次的推理" if counts["caliber"]
+        # 逐格判之后常见「一部分数来自口径、另几个只是相关」：照实说是不一致，不说成这次的推理
-                  else "没有数来自口径")
+        reason = ("有的数没有比对" if counts["unknown"] else "没有数来自口径" if not counts["caliber"]
+                  else "有的数和相关口径不一致，不来自口径" if counts["related"] else "有的数是这次的推理")
         return {"equivalent": False, "reason": reason, "counts": counts}
     if len(keys) != 1:
         return {"equivalent": False, "reason": "数来自不同的口径", "counts": counts}
@@ -431,5 +472,6 @@ def related_view(item: Mapping[str, Any], *, hide: Callable[[Any], bool], shown:
 
 
 __all__ = ["CERTIFIED_NOTE", "HOWS", "KINDS", "LEVEL_TEXT", "POST_SEAL_NOTE", "RETRO_NOTE", "binding_view", "caliber_key",
-           "card_expressions", "card_reads", "cell_label", "diff_view", "label_doc", "metric_label",
+           "candidates_of", "card_expressions", "card_reads", "cell_label", "cells_of", "diff_view", "label_doc",
+           "metric_label",
            "referenced_columns", "related_view", "view_label", "whole_of"]

backend/app/ledger/match.py

@@ -25,6 +25,14 @@ MAX_CANDIDATES 份（按工作流、版本、节点排好的先后）。
 答案多选了列的在最前面写一句「这次多查了 N 列，不影响这个数」）。没有投影可试的（报告没引用这条查询、两边都不投影、
 投影不了）照旧是整条比的结果，逐对一字不差。
 
+**逐格判**（_cells，algo 4）。报告常把一条查询的好几列都引用了（销售额、订单数、区间首日、末日），口径只取其中一列：按全部
+引用列投影还是好几列，对不上口径，整条比只能标相关。整条（含按全部引用列投影）没证明、又引用了这条查询两列以上的，被引用
+的每一列单独投影到那一列（_cell），和每份口径的每个指标读的那一列逐个比，照样只走 match_template——上面的投影规则一条
+不少（多一个分组列、DISTINCT 的照旧不投影，那一格照用整条的结论）。结果记在 queries[].cells {列名: {status, candidates}}，
+每个数按它那一格标（labels._view）；同一个指标用了这条查询的好几格时，只有这条查询一定只出一行（one_row）才把逐格的结论
+合起来用。整题照拍板 L1：每个数都证明来自同一份口径才算等价，一格证明了、别的格只是相关的，整题不等价、不自动出具。召回
+反查、使用次数、工件 meta 把逐格证明了的口径也算上（labels.candidates_of）。
+
 **助手选用的（composed）。** 问数据的助手 use_caliber 选用口径时，取数节点是服务端照那一版拷进来的（ledger/compose.py），
 口径卡的 caliber_from 带着改名映射 {原 id: 新 id}。这几个节点交回的查询照样要证明：证明了的那一份记 how=composed，查询
 状态记 composed——数确实来自口径，但「这份口径答的就是这个问题」是助手凭问法判断的，不能说成「等价已证明」（拍板 L1：
@@ -447,6 +455,89 @@ def _projected(caliber: Caliber, fetch: Fetch, query: Query, wanted: tuple[int,
     return cand
 
 
+def _cell(caliber: Caliber, fetch: Fetch, query: Query, index: int, deadline: float | None) -> dict[str, Any] | None:
+    """逐格比（algo 4，见模块说明）：答案只留这一格所在的那一列（project_select），口径只留它的指标读到的某一列
+    （口径卡上读这个取数节点的每一列各试一次；口径版本里没有口径卡的，取数选出来的每一列各试一次），投影后照样交
+    match_template 判。证明了的候选（带 projection，列映射就是这一列对口径那一列）；都没证明时第一种比法的那次比较
+    （差在哪按投影之后说，答案多选了列的先写一句）；投影不了、没有可比的为 None（这一格照旧用这条查询的结论）。"""
+    from app.data.sql_semantics import FacetDiff, match_template, output_names, project_select, project_template
+
+    width = len(query.columns or [])
+    ours = project_select(query.sql, (index,), query.dialect, width=width)
+    names = output_names(fetch.sql, query.dialect, template=True)
+    if ours.sql is None or not names:
+        return None
+    if caliber.cards:
+        columns = sorted({i for group in _caliber_sets(caliber, fetch, names) for i in group})
+    else:
+        columns = list(range(len(names)))
+    theirs = [project_template(fetch.sql, (i,), query.dialect) for i in columns] if len(names) > 1 \
+        else [project_template(fetch.sql, (0,), query.dialect)] if columns == [0] else []
+    first: tuple[Any, Any] | None = None
+    for c in theirs[:MAX_PROJECTIONS]:
+        if c.sql is None or (deadline is not None and first is not None and time.monotonic() > deadline):
+            continue
+        m = match_template(c.sql, list(caliber.fields), ours.sql, query.dialect, schema=query.schema,
+                           names=query.names, columns=[str((query.columns or [])[index])], deadline=deadline)
+        if m.equivalent:
+            return {**_candidate(caliber, fetch, m), "projection": _projection_view(query, ours, c)}
+        if first is None:
+            first = (c, m)
+    if first is None:
+        return None
+    c, m = first
+    cand = {**_candidate(caliber, fetch, m), "projection": _projection_view(query, ours, c)}
+    if ours.dropped and isinstance(cand.get("diff"), dict):
+        note = FacetDiff("projection", None, None, EXTRA_COLUMNS.format(n=ours.dropped)).as_dict()
+        cand["diff"] = {**cand["diff"], "items": [note, *cand["diff"]["items"]]}
+    return cand
+
+
+def _verdict(candidates: list[dict[str, Any]]) -> tuple[str, str | None]:
+    """一组候选的结论（查询、逐格同一个判法）：(status, 没比完时的原因)。"""
+    statuses = {c["status"] for c in candidates}
+    status = "proven" if "proven" in statuses else "skipped" if "skipped" in statuses else "related"
+    if status == "proven" and any(c.get("how") == "composed" for c in candidates if c["status"] == "proven"):
+        # 证明了的是助手选用的那一份：数来自口径，但不是这次独立写出来的（见模块说明）。同时和别的口径等价的也按选用的算
+        status = "composed"
+    reason = next((c.get("reason") for c in candidates if c["status"] == "skipped"), None) if status == "skipped" \
+        else None
+    return status, reason
+
+
+def _cells(query: Query, wanted: tuple[int, ...], pool: list[tuple[Caliber, Fetch]], whole: list[dict[str, Any]],
+           chosen: tuple[str, str] | None, deadline: float | None) -> dict[str, dict[str, Any]]:
+    """这条查询被引用的每一列各比一次（_cell）：{列名: {status, candidates, reason?}}。每份口径取数一个候选：逐格比出来的，
+    比不了（投影不了、过了时限、不可比、整条比就没比完）的照用这条查询的那一个候选。同名的几列认不出是哪一列，不逐格比。"""
+    columns = [str(c) for c in query.columns or []]
+    out: dict[str, dict[str, Any]] = {}
+    for index in wanted:
+        name = columns[index]
+        if columns.count(name) != 1:
+            continue
+        candidates: list[dict[str, Any]] = []
+        for (caliber, fetch), base in zip(pool, whole):
+            cand = None
+            if base.get("status") == "related" and base.get("comparable", True) and \
+                    (deadline is None or time.monotonic() <= deadline):
+                try:
+                    cand = _cell(caliber, fetch, query, index, deadline)
+                except Exception:  # noqa: BLE001 - 逐格比是附加的一步：出错这一格照用这条查询的结论
+                    logger.exception("逐格比对出错（运行查询 %s，口径 %s）", query.artifact[:12], caliber.key)
+            if cand is None:
+                candidates.append(dict(base))
+                continue
+            if chosen is not None and (caliber.key, fetch.node_id) == chosen:
+                if cand["status"] == "proven":
+                    cand["how"] = "composed"
+                else:
+                    cand.setdefault("reason", COMPOSED_UNPROVEN)
+            candidates.append(cand)
+        status, reason = _verdict(candidates)
+        out[name] = {"status": status, "candidates": candidates, **({"reason": reason} if reason else {})}
+    return out
+
+
 def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline: float | None = None,
                     composed: dict[str, tuple[str, str]] | None = None,
                     referenced: Mapping[str, Collection[str]] | None = None) -> list[dict[str, Any]]:
@@ -455,7 +546,7 @@ def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline:
     等价）/ none（没有相关的口径）/ skipped（没比完：超时、解析失败、快照读不出）。composed 是 composed_nodes 的结果，
     不给就没有助手选用的。referenced 是报告引用了哪条查询的哪几列（labels.referenced_columns：{查询快照: {列名}}）：
     给了的，整条比不等价时按引用的列投影再比（_projected）；不给就只整条比。纯计算，从不抛异常。评测逐题比 SQL 也调它。"""
-    from app.data.sql_semantics import match_template, normalize_sql
+    from app.data.sql_semantics import match_template, normalize_sql, one_row
 
     out: list[dict[str, Any]] = []
     for query in queries:
@@ -500,16 +591,17 @@ def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline:
                 else:
                     cand.setdefault("reason", COMPOSED_UNPROVEN)
             candidates.append(cand)
-        statuses = {c["status"] for c in candidates}
+        status, reason = _verdict(candidates)
-        status = "proven" if "proven" in statuses else "skipped" if "skipped" in statuses else "related"
-        if status == "proven" and any(c.get("how") == "composed" for c in candidates):
-            # 证明了的是助手选用的那一份：数来自口径，但不是这次独立写出来的（见模块说明）。同时和别的口径等价的也按选用的算
-            status = "composed"
         if len(pool) > MAX_CANDIDATES:
             row["note"] = MORE_CANDIDATES.format(n=len(pool) - MAX_CANDIDATES, max=MAX_CANDIDATES)
-        if status == "skipped":
+        if reason:
-            row["reason"] = next(c.get("reason") for c in candidates if c["status"] == "skipped")
+            row["reason"] = reason
-        out.append({**row, "status": status, "candidates": candidates})
+        row.update(status=status, candidates=candidates)
+        if wanted is not None and len(wanted) >= 2 and status == "related":
+            # 整条（按全部引用列投影）没证明：被引用的几列逐格各比一次，每个数按它自己那一格的结论标（labels.cell_label）
+            row["cells"] = _cells(query, wanted, pool[:MAX_CANDIDATES], candidates, chosen, deadline)
+            row["one_row"] = one_row(query.sql, query.dialect)
+        out.append(row)
     return out
 
 

backend/app/ledger/recall.py

@@ -871,12 +871,14 @@ async def original_values(use: Mapping[str, Any], meta: Sequence[Mapping[str, An
     from app.engine.evidence import render_number
     from app.engine.expressions import cell_parts
     from app.ledger import store
+    from app.ledger.labels import candidates_of
 
     recorded = await store.latest(run_id)
     content = recorded.content if recorded is not None else None
     query = next((q for q in (content or {}).get("queries") or [] if q.get("artifact") == use.get("query")), None)
-    cand = next((c for c in (query or {}).get("candidates") or [] if c.get("status") == "proven"
+    # 整条证明了的、逐格证明了的都算：口径的哪一列对这次的哪一列，按各自的列映射找（逐格的只映射那一列）
-                 and c.get("node_id") == use.get("node_id")), None)
+    proven = [c for c in candidates_of(query or {})
+              if c.get("status") == "proven" and c.get("node_id") == use.get("node_id")]
     try:
         snap = load(str(use.get("query") or "")) if use.get("query") else None
     except ValueError:
@@ -893,8 +895,8 @@ async def original_values(use: Mapping[str, Any], meta: Sequence[Mapping[str, An
                 column = ast.literal_eval(parts[2])
             except (ValueError, SyntaxError):
                 column = None
-        ours = next((m.get("ours") for m in (cand or {}).get("columns") or [] if m.get("caliber") == column), None) \
+        ours = next((m.get("ours") for c in proven for m in c.get("columns") or [] if m.get("caliber") == column),
-            if isinstance(column, str) else None
+                    None) if isinstance(column, str) else None
         if ours in columns and isinstance(rows, list) and rows and isinstance(rows[0], (list, tuple)):
             value = rows[0][columns.index(ours)]
             try:

backend/app/ledger/stats.py

@@ -9,9 +9,10 @@
 
 - E1 等价命中：比对过的查询里 proven / composed / related / none / skipped 各几条（skipped 单独列，不混进 none），有候选口径
   的查询里各自的比率；证明等价的绑定是在哪一级配上的，其中几份是按引用的列投影之后才证明的（proven_projected，ledger/match
-  的 _projected）；没证明的原因（不可比、参数绑不上、配上了但不等价）；不等价的差异类型分布（time / filter / aggregate /
+  的 _projected）；逐格比过的格（algo 4，ledger/match._cells）各种结论各几格（cells）；没证明的原因（不可比、参数绑不上、
-  join / other / parse，每份口径的差异每类只计一次；投影之后那句「这次多查了 N 列」不是差异，不计）；逐个数的标签各几个、
+  配上了但不等价）；不等价的差异类型分布（time / filter / aggregate / join / other / parse，每份口径的差异每类只计一次；
-  整题等价几次。
+  投影之后那句「这次多查了 N 列」不是差异，不计）；逐个数的标签各几个、整题等价几次。
+  E2 的兜底比例按标签上给人看的那一份说明数：逐格比过的按逐格的（labels.candidates_of 的 shown）。
 - E2 对账：对账运行（origin=reconcile，第 3 波起由 ledger/reconcile 发起）发起、完成、失败各几次，对过账的答案几个，其中参数
   经人核对过的、对照的指标由人选的各几次；相关口径的差异说明里兜底说明的占比；对账结果上「说明有用 / 看不懂」的反馈
   （第 4 波，ledger/feedback：对账运行上的封存后事件 ledger.feedback，每人只算最后一次）各几个、有用的比率。
@@ -103,6 +104,7 @@ def e1_of(contents: list[dict[str, Any]]) -> dict[str, Any]:
     with_candidates = Counter({s: 0 for s in STATUSES})
     via: Counter[str] = Counter()
     projected = 0
+    cells = Counter({s: 0 for s in STATUSES})
     unproven: Counter[str] = Counter()
     diffs = Counter({g: 0 for g in DIFF_GROUPS})
     labels = Counter({k: 0 for k in ledger_labels.KINDS})
@@ -113,6 +115,8 @@ def e1_of(contents: list[dict[str, Any]]) -> dict[str, Any]:
             queries += 1
             s = str(query.get("status") or "")
             status[s] += 1
+            for cell in ledger_labels.cells_of(query):
+                cells[str(cell.get("status") or "")] += 1
             if query.get("candidates"):
                 with_candidates[s] += 1
             for cand in query.get("candidates") or []:
@@ -142,7 +146,8 @@ def e1_of(contents: list[dict[str, Any]]) -> dict[str, Any]:
         "runs": len(contents), "queries": queries, "status": dict(status),
         "with_candidates": candidates,
         "rates": {s: _ratio(with_candidates[s], candidates) for s in STATUSES if s != "none"},
-        "proven_via": dict(via), "proven_projected": projected, "unproven_reasons": dict(unproven.most_common(20)),
+        "cells": dict(cells), "proven_via": dict(via), "proven_projected": projected,
+        "unproven_reasons": dict(unproven.most_common(20)),
         "diff_aspects": dict(diffs),
         "labels": dict(labels), "whole_equivalent": whole,
     }
@@ -169,7 +174,8 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic
     explained = fallback = 0
     for content in contents:
         for query in content.get("queries") or []:
-            for cand in query.get("candidates") or []:
+            # 标签上给人看的那一份：逐格比过的按逐格的说明数（ledger_labels.candidates_of 的 shown）
+            for cand in ledger_labels.candidates_of(query, shown=True):
                 if cand.get("status") == "related" and isinstance(cand.get("diff"), dict):
                     explained += 1
                     fallback += bool(cand["diff"].get("fallback"))
@@ -285,7 +291,7 @@ async def _e4(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic
 
     proven: Counter[str] = Counter()
     for content in contents:
-        keys = {c.get("caliber") for q in content.get("queries") or [] for c in q.get("candidates") or []
+        keys = {c.get("caliber") for q in content.get("queries") or [] for c in ledger_labels.candidates_of(q)
                 if c.get("status") == "proven" and c.get("caliber")}
         for key in keys:
             proven[str(key)] += 1

backend/app/ledger/store.py

@@ -32,7 +32,9 @@ FORMAT = "agentlab.caliber_match/1"
 #: 2（第 3 波）：认出助手选用的口径（查询 composed、标签 how=composed），指标的标签记它取自哪几条查询（queries，认证队列
 #: 认「被报告引用的查询」用）
 #: 3：整条比不等价时按报告引用的列投影再比（ledger/match 的 _projected），候选记 projection；没投影的照旧
-ALGO = "3"
+#: 4：按全部引用列投影仍没证明的查询，被引用的每一列逐格各比一次（queries[].cells、one_row），每个数按它那一格标；老工件
+#: 没有 cells，照旧按整条查询读
+ALGO = "4"
 #: 触发方式（见模块说明）。certified（第 3 波）：从问数据固化出的口径发布之后，对来源运行补比的（ledger/certify）
 TRIGGERS = ("sealed", "backfill", "on_demand", "certified")
 
@@ -73,10 +75,13 @@ def counts_of(queries: list[dict[str, Any]]) -> dict[str, int]:
 
 def meta_of(content: dict[str, Any]) -> dict[str, Any]:
     """工件引用的 meta：{calibers: [等价已证明的「工作流@版本」], related: [相关的], counts}。"""
+    from app.ledger.labels import candidates_of
+
     proven: list[str] = []
     related: list[str] = []
     for query in content.get("queries") or []:
-        for cand in query.get("candidates") or []:
+        # 逐格证明了的口径也算（召回按旧版本反查、统计按种类数都只读这一份 meta）
+        for cand in candidates_of(query):
             key = str(cand.get("caliber") or "")
             if not key:
                 continue

backend/app/ledger/uses.py

@@ -110,7 +110,7 @@ async def _answers(session: AsyncSession, workflow_id: str, below: int | None, o
     """问数回答：比对工件 meta 里等价已证明的那几份口径里有 W@V 的探索运行（见模块说明）。"""
     from app.engine.origin import counted_runs
     from app.ledger import store
-    from app.ledger.labels import caliber_key
+    from app.ledger.labels import caliber_key, candidates_of
 
     prefix = f"{workflow_id}@"
     rows = (await session.execute(
@@ -134,7 +134,8 @@ async def _answers(session: AsyncSession, workflow_id: str, below: int | None, o
             continue
         seen: set[tuple[int, str]] = set()
         for query in content.get("queries") or []:
-            for cand in query.get("candidates") or []:
+            # 逐格证明了的也是用了这份口径（ledger/labels.candidates_of）
+            for cand in candidates_of(query):
                 if cand.get("status") != "proven" or cand.get("workflow_id") != workflow_id:
                     continue
                 version = cand.get("version")

backend/app/ledger/view.py

@@ -78,21 +78,27 @@ def _whole_view(whole: dict[str, Any], hider: _Hide, names: dict[str, str]) -> d
     return out
 
 
+def _candidate_view(cand: dict[str, Any], hider: _Hide, names: dict[str, str]) -> dict[str, Any]:
+    item = ledger_labels.related_view(cand, hide=hider.hide, shown=hider.shown, names=names)
+    item["status"] = cand.get("status")
+    if cand.get("how"):
+        item["how"] = cand["how"]
+    if cand.get("columns") is not None:
+        item["columns"] = cand["columns"]
+    if isinstance(cand.get("projection"), dict):
+        # 按引用的列投影之后比的（ledger/match 的 _projected、_cell）：两边各留下了哪几列、去掉了几列（列名，不带值）
+        item["projection"] = cand["projection"]
+    return item
+
+
 def _query_view(query: dict[str, Any], hider: _Hide, names: dict[str, str]) -> dict[str, Any]:
     out = {k: query[k] for k in ("artifact", "node_id", "source", "tables", "status", "reason", "note") if k in query}
-    cands = []
+    out["candidates"] = [_candidate_view(c, hider, names) for c in query.get("candidates") or []]
-    for cand in query.get("candidates") or []:
+    if isinstance(query.get("cells"), dict):
-        item = ledger_labels.related_view(cand, hide=hider.hide, shown=hider.shown, names=names)
+        # 逐格比过的（algo 4）：每一列的结论和候选，标签按它标；遮罩同上
-        item["status"] = cand.get("status")
+        out["cells"] = {column: {**{k: cell[k] for k in ("status", "reason") if k in cell},
-        if cand.get("how"):
+                                 "candidates": [_candidate_view(c, hider, names) for c in cell.get("candidates") or []]}
-            item["how"] = cand["how"]
+                        for column, cell in query["cells"].items() if isinstance(cell, dict)}
-        if cand.get("columns") is not None:
-            item["columns"] = cand["columns"]
-        if isinstance(cand.get("projection"), dict):
-            # 按引用的列投影之后比的（ledger/match 的 _projected）：两边各留下了哪几列、去掉了几列（列名，不带值）
-            item["projection"] = cand["projection"]
-        cands.append(item)
-    out["candidates"] = cands
     return out
 
 

backend/tests/test_ledger_cells.py

@@ -0,0 +1,265 @@
+"""口径标签逐格判（app/ledger/match._cells、labels._view，algo 4）。
+
+真实模型下问「上周的销售额是多少」，助手写 SUM(amount) AS 销售额, COUNT(*) AS 订单数, MIN / MAX(paid_date) AS 区间首日 / 末日，
+报告四列都引用了；口径「周销售」只取 SUM(amount)。以前按「全部引用列」投影后还是 4 列，对不上口径的 1 列，整条比、四个数都
+标相关口径，说明整段兜底。现在每个被引用的格单独投影到那一列，和口径每个指标读的那一列逐个比（仍只走 match_template /
+explain_sql_diff，上次的投影规则照用）：
+- 销售额标「等价已证明」，订单数、首末日照实标相关口径；整题不等价（订单数不来自口径），不自动出具、不给出具入口；
+- 同一个数在两份口径里只证明了一份时标那一份；口径有多个指标时按列配对；一个指标用了同一条一行查询的两格，两格各自
+  证明了就算来自口径，多行的查询不拿逐格的结论合起来用；
+- 不错给信任：多一个分组列、DISTINCT 的查询逐格也不投影；
+- 评测逐题比 SQL 取和期望配上的那一格。
+数据是虚构的。
+"""
+from __future__ import annotations
+
+import sqlite3
+import uuid
+from types import SimpleNamespace
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.engine.runner import run_manager
+from app.ledger import labels as ledger_labels
+from app.ledger import match as ledger_match
+from app.ledger import store
+from tests import ledger_samples as S
+
+#: 口径「周销售」的取数（照真实例子：按支付日期、两头都含）
+WEEKLY = ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' "
+          "AND paid_date >= '{{ input.start }}' AND paid_date <= '{{ input.end }}'")
+WINDOW = "status = 'paid' AND paid_date >= '2026-10-05' AND paid_date <= '2026-10-11'"
+#: 助手写的（照真实例子）
+ANSWER = ("SELECT SUM(amount) AS 销售额, COUNT(*) AS 订单数, MIN(paid_date) AS 区间首日, MAX(paid_date) AS 区间末日 "
+          f"FROM orders WHERE {WINDOW}")
+COLUMNS = ["销售额", "订单数", "区间首日", "区间末日"]
+ROWS = [(1, "paid", 120.0, "2026-10-04"), (2, "paid", 80.0, "2026-10-05"), (3, "paid", 45.5, "2026-10-08"),
+        (4, "refunded", 30.0, "2026-10-09"), (5, "paid", 64.5, "2026-10-11"), (6, "open", 10.0, "2026-10-10"),
+        (7, "paid", 99.0, "2026-10-12")]
+
+
+async def week_source(tmp_path) -> None:
+    path = tmp_path / "week.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, status TEXT, amount REAL, paid_date TEXT);")
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", ROWS)
+    db.commit()
+    db.close()
+    S.CURRENT["source"] = f"week_{uuid.uuid4().hex[:8]}"
+    await S.set_source(path)
+
+
+def caliber(sql: str = WEEKLY, metrics: list[dict] | None = None, workflow_id: str = "wf") -> ledger_match.Caliber:
+    graph = S.caliber_graph(sql)
+    if metrics is not None:
+        graph["nodes"][2]["data"]["config"]["metrics"] = metrics
+    found = ledger_match.caliber_of(workflow_id, SimpleNamespace(graph=graph, version=1, level="governed",
+                                                                 graph_hash="h"))
+    assert found is not None
+    return found
+
+
+def compare(sql: str, columns: list[str], cited: set[str], calibers: list[ledger_match.Caliber] | None = None) -> dict:
+    query = ledger_match.Query("q1", "q", S.CURRENT["source"], sql, columns, frozenset({"orders"}), "sqlite")
+    [row] = ledger_match.compare_queries([query], calibers or [caliber()], referenced={"q1": cited})
+    return row
+
+
+def label(row: dict, column: str) -> dict:
+    return ledger_labels.cell_label("q1", column, {"q1": row})
+
+
+def texts(cand: dict) -> list[str]:
+    return [i["text"] for i in (cand.get("diff") or {}).get("items") or []]
+
+
+# --------------------------------------------------------------------------
+# 逐格判：同一条查询里一列等价、一列不等价各自标对
+# --------------------------------------------------------------------------
+
+
+def test_each_cited_column_is_judged_on_its_own():
+    row = compare(ANSWER, COLUMNS, set(COLUMNS))
+    assert row["status"] == "related", "按全部引用列投影还是 4 列，整条照旧不等价"
+    assert row["one_row"] is True and set(row["cells"]) == set(COLUMNS)
+    sales = label(row, "销售额")
+    assert sales["kind"] == "caliber" and sales["how"] == "independent" and sales["column"] == "gmv"
+    assert sales["binding"] == {"start": "2026-10-05", "end": "2026-10-11"}
+    assert [label(row, c)["kind"] for c in COLUMNS[1:]] == ["related", "related", "related"]
+    [orders] = row["cells"]["订单数"]["candidates"]
+    assert texts(orders) == ["这次多查了 3 列，不影响这个数", "这次数的是行数，口径对 amount 求和"]
+    assert orders["diff"]["fallback"] is False
+    # 整题：订单数不来自口径，不等价
+    whole = ledger_labels.whole_of({c: label(row, c) for c in COLUMNS}, {})
+    assert whole["equivalent"] is False and whole["counts"]["caliber"] == 1
+    assert whole["reason"] == "有的数和相关口径不一致，不来自口径"
+    # 只引用销售额一格（不逐格）：整条按引用的列投影就证明了，照旧
+    single = compare(ANSWER, COLUMNS, {"销售额"})
+    assert single["status"] == "proven" and "cells" not in single
+
+
+def test_a_number_proven_against_only_one_of_two_calibers_is_labelled_with_that_one():
+    paid = caliber(workflow_id="wf-paid")
+    everything = caliber(WEEKLY.replace("status = 'paid' AND ", ""), workflow_id="wf-all")
+    row = compare(ANSWER, COLUMNS, set(COLUMNS), [paid, everything])
+    sales = label(row, "销售额")
+    assert sales["kind"] == "caliber" and sales["caliber"] == "wf-paid@1"
+    cands = {c["caliber"]: c for c in row["cells"]["销售额"]["candidates"]}
+    assert cands["wf-paid@1"]["status"] == "proven" and cands["wf-all@1"]["status"] == "related"
+    assert any("status = 'paid'" in t for t in texts(cands["wf-all@1"]))
+    # 反过来：含退款的写法只对得上另一份
+    refunds = ANSWER.replace("status = 'paid' AND ", "")
+    other = label(compare(refunds, COLUMNS, set(COLUMNS), [paid, everything]), "销售额")
+    assert other["kind"] == "caliber" and other["caliber"] == "wf-all@1"
+
+
+def test_a_caliber_with_several_metrics_is_paired_column_by_column():
+    two = caliber(WEEKLY.replace("SUM(amount) AS gmv", "SUM(amount) AS gmv, COUNT(*) AS n"),
+                  [{"id": "gmv", "name": "销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"},
+                   {"id": "n", "name": "订单数", "expression": "cell(nodes.fetch, 0, 'n')"}])
+    sql = f"SELECT COUNT(*) AS 单数, SUM(amount) AS 销售额, MAX(paid_date) AS 末日 FROM orders WHERE {WINDOW}"
+    row = compare(sql, ["单数", "销售额", "末日"], {"单数", "销售额", "末日"}, [two])
+    assert row["status"] == "related"
+    assert (label(row, "单数")["kind"], label(row, "单数")["column"]) == ("caliber", "n")
+    assert (label(row, "销售额")["kind"], label(row, "销售额")["column"]) == ("caliber", "gmv")
+    assert label(row, "末日")["kind"] == "related"
+
+
+def test_a_metric_over_two_cells_uses_the_cell_proofs_only_on_a_one_row_query():
+    two = caliber(WEEKLY.replace("SUM(amount) AS gmv", "SUM(amount) AS gmv, COUNT(*) AS n"),
+                  [{"id": "aov", "name": "客单价", "expression": "cell(nodes.fetch, 0, 'gmv') / cell(nodes.fetch, 0, 'n')"}])
+    sql = f"SELECT SUM(amount) AS 销售额, COUNT(*) AS 订单数, MAX(paid_date) AS 末日 FROM orders WHERE {WINDOW}"
+    row = compare(sql, ["销售额", "订单数", "末日"], {"销售额", "订单数", "末日"}, [two])
+    inputs = [{"path": f"cell(nodes.q, 0, '{c}')", "artifact": "q1", "via": "tool_cell", "locator": {"row": 0, "column": c}}
+              for c in ("销售额", "订单数")]
+    card = {"metrics": [{"id": "aov", "expression": "cell(nodes.q, 0, '销售额') / cell(nodes.q, 0, '订单数')",
+                         "inputs": inputs}]}
+    by_key = {two.key: {"expressions": list(two.expressions)}}
+    got = ledger_labels.metric_label(card, "aov", {"q1": row}, by_key)
+    assert got["kind"] == "caliber" and got["metric"] == "aov", got
+    # 同一条查询要是多行的，每一列各自对得上不等于每一行的两列一起对得上：不拿逐格的结论合起来用
+    multi = {**row, "one_row": False}
+    assert ledger_labels.metric_label(card, "aov", {"q1": multi}, by_key)["kind"] == "related"
+
+
+# --------------------------------------------------------------------------
+# 不错给信任：逐格投影照样守上次的投影规则
+# --------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("sql", [
+    f"SELECT paid_date, SUM(amount) AS 销售额, COUNT(*) AS 订单数 FROM orders WHERE {WINDOW} GROUP BY paid_date",
+    f"SELECT DISTINCT amount AS 销售额, paid_date AS 订单数, id FROM orders WHERE {WINDOW}",
+], ids=["extra-group-by", "distinct"])
+def test_cells_are_not_projected_where_the_projection_rules_say_no(sql):
+    columns = ["paid_date", "销售额", "订单数"] if "GROUP BY" in sql else ["销售额", "订单数", "id"]
+    row = compare(sql, columns, {"销售额", "订单数"})
+    assert row["status"] == "related"
+    for name in ("销售额", "订单数"):
+        cell = row["cells"][name]
+        assert cell["status"] == "related" and all("projection" not in c for c in cell["candidates"]), name
+        assert cell["candidates"] == row["candidates"], "投影不了的格照用整条查询的结论"
+        assert label(row, name)["kind"] == "related"
+
+
+# --------------------------------------------------------------------------
+# 真跑一遍真实例子：标签、整题、出具入口、证据片段的 ledger 键
+# --------------------------------------------------------------------------
+
+
+@pytest.fixture
+async def engine_up(tmp_path):
+    await run_manager.setup()
+    await week_source(tmp_path)
+    yield
+    await ledger_match.drain()
+    await run_manager.shutdown()
+
+
+async def by_column(run_id: str, labels: dict) -> dict[str, dict]:
+    """标签按它引用的列排开（报告文档里每个片段引用的是哪一列）。"""
+    from app.api.evidence import _reports, _sealed
+    from app.engine.evidence import iter_units
+
+    [report] = _reports(await _sealed(run_id))
+    out = {}
+    for _, unit in iter_units(report.doc):
+        for seg in unit.get("segments") or []:
+            if seg.get("id") in labels:
+                out[seg["cite"]["locator"]["column"]] = {**labels[seg["id"]], "_segment": seg["id"]}
+    return out
+
+
+async def test_last_weeks_sales_labels_only_the_sales_number_as_from_the_caliber(engine_up, monkeypatch):
+    graph = S.caliber_graph(WEEKLY)
+    wf = await S.publish(graph=graph)
+    row = await S.explore(monkeypatch, ANSWER, origin="chat", text=(
+        "上周销售额 [[v:Q1.r0.销售额]] 元，共 [[v:Q1.r0.订单数]] 单（[[v:Q1.r0.区间首日]] 至 [[v:Q1.r0.区间末日]]）。"))
+    content = (await store.latest(row.id)).content
+    assert content["algo"] == store.ALGO == "4"
+    [query] = content["queries"]
+    assert query["status"] == "related" and set(query["cells"]) == set(COLUMNS)
+    [labels] = [r["labels"] for r in content["reports"].values()]
+    cols = await by_column(row.id, labels)
+    assert cols["销售额"]["kind"] == "caliber" and cols["销售额"]["caliber"] == f"{wf}@1"
+    assert cols["销售额"]["binding"] == {"start": "2026-10-05", "end": "2026-10-11"}
+    assert {c: cols[c]["kind"] for c in COLUMNS[1:]} == {c: "related" for c in COLUMNS[1:]}
+    whole = content["whole"]
+    assert whole["equivalent"] is False and whole["counts"] == {"caliber": 1, "related": 3, "reasoning": 0,
+                                                                 "unknown": 0}
+    # 工件 meta：逐格证明了的口径也记上（召回按旧版本反查只读它）
+    from sqlalchemy import select
+
+    from app.db.base import SessionLocal
+    from app.db.models import Artifact
+
+    async with SessionLocal() as session:
+        meta = (await session.execute(select(Artifact.meta).where(
+            Artifact.id == (await store.latest(row.id)).artifact, Artifact.run_id == row.id))).scalar_one()
+    assert meta["calibers"] == [f"{wf}@1"]
+
+    from app.main import app
+
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        body = (await client.get(f"/api/runs/{row.id}/ledger")).json()
+        assert not [a for a in body["actions"] if a["kind"] == "issue"], "整题不等价：不给按口径出具的入口（拍板 L1）"
+        assert body["issued"] == [], "不自动出具"
+        [q] = body["queries"]
+        assert q["cells"]["销售额"]["status"] == "proven" and q["cells"]["订单数"]["status"] == "related"
+        shown = (await by_column(row.id, body["reports"]["write"]["labels"]))
+        assert shown["销售额"]["kind"] == "caliber" and shown["销售额"]["workflow_name"] == "周销售"
+        seg = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cols['销售额']['_segment']}")).json()
+        assert seg["ledger"]["label"]["kind"] == "caliber"
+        seg = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cols['订单数']['_segment']}")).json()
+        assert seg["ledger"]["label"]["kind"] == "related"
+        where = seg["ledger"]["label"]["related"][0]["diff"]["items"]
+        assert [i["text"] for i in where] == ["这次多查了 3 列，不影响这个数", "这次数的是行数，口径对 amount 求和"]
+
+
+# --------------------------------------------------------------------------
+# 评测逐题比 SQL：取和期望配上的那一格
+# --------------------------------------------------------------------------
+
+
+async def test_the_per_case_sql_match_takes_the_cell_that_matched_the_expectation(engine_up, monkeypatch):
+    from sqlalchemy import select
+
+    from app.db.base import SessionLocal
+    from app.db.models import DataSource
+    from app.eval import score
+
+    run = await S.explore(monkeypatch, ANSWER, text="上周销售额 [[v:Q1.r0.销售额]] 元，共 [[v:Q1.r0.订单数]] 单。")
+    answer = await score.read_answer(run.id)
+    async with SessionLocal() as session:
+        sid = (await session.execute(select(DataSource.id).where(DataSource.name == S.CURRENT["source"]))).scalar()
+    fixed = "SELECT SUM(amount) AS gmv FROM orders WHERE " + WINDOW
+    case = {"gold": {"kind": "sql", "datasource_id": sid, "sql": "x"}, "expectations": []}
+    gold = {"status": "ok", "before": {"rendered": {"sql": fixed}, "sets": []}}
+    # 没给配上的格：逐格证明了的那一格
+    match = await score.sql_match(case, gold, answer)
+    assert match["status"] == "proven" and match["column"] == "销售额", match
+    # 期望配上的是订单数那一格：取那一格的结论，差在哪按那一格说
+    query = answer.cited[0]
+    match = await score.sql_match(case, gold, answer, cells=[{"artifact": query, "column": "订单数"}])
+    assert match["status"] == "related" and match["column"] == "订单数" and match["aspects"] == ["aggregate"], match

backend/tests/test_ledger_compose.py

@@ -253,7 +253,7 @@ async def test_an_answer_built_with_use_caliber_is_labelled_composed(client, mon
     await ledger_match.drain()
     recorded = await store.latest(row.id)
     content = recorded.content
-    assert content["algo"] == store.ALGO == "3"
+    assert content["algo"] == store.ALGO == "4"
     [query] = content["queries"]
     assert query["status"] == "composed" and query["node_id"] == "c1_fetch"
     [cand] = query["candidates"]

backend/tests/test_ledger_projection.py

@@ -130,7 +130,10 @@ def test_without_a_projection_every_pair_is_exactly_the_old_result():
         (f"SELECT SUM(amount) AS gmv, COUNT(*) AS n FROM orders WHERE {WINDOW}", ["gmv"], {"gmv"}),  # 列数对不上
     ]
     for sql, columns, cited in cases:
-        assert compare(sql, columns, cited) == compare(sql, columns, None), sql
+        row = compare(sql, columns, cited)
+        # 逐格比（algo 4）是另加的 cells、one_row 两个键：整条查询这一层的结论和候选照旧一字不差
+        assert {k: v for k, v in row.items() if k not in ("cells", "one_row")} == compare(sql, columns, None), sql
+        assert ("cells" in row) == (len(cited) >= 2 and row["status"] == "related"), sql
 
 
 # --------------------------------------------------------------------------
@@ -185,7 +188,7 @@ async def test_last_weeks_sales_with_extra_columns_is_whole_equivalent(engine_up
     row = await S.explore(monkeypatch, sql, text="上周销售额 [[v:Q1.r0.销售额]]，共 [[v:Q1.r0.单数]] 单。")
     recorded = await store.latest(row.id)
     content = recorded.content
-    assert content["algo"] == "3"
+    assert content["algo"] == store.ALGO
     [q] = content["queries"]
     assert q["status"] == "related", "销售额、单数两列都被引用，口径只有销售额：投影之后列数对不上，照旧不等价"
 

frontend/src/typesLedger.ts

@@ -82,6 +82,11 @@ export interface LedgerQuery {
   reason?: string
   note?: string
   candidates: (LedgerRelated & { status?: string; how?: string })[]
+  /**
+   * 逐格比过的（服务端 ledger/match._cells，algo 4）：被引用的每一列各自的结论和候选。整条（按全部引用列投影）没证明、
+   * 报告引用了这条查询好几列时才有；每个数的标签已经按它那一格标好，界面照标签画，不拿它重算
+   */
+  cells?: Record<string, { status: string; reason?: string; candidates: (LedgerRelated & { status?: string; how?: string })[] }>
 }
 
 /** 整题判断（主报告）：所有带标签的数都来自同一口径的同一版、参数一致时 equivalent */

scripts/check-ledger.mjs

@@ -24,7 +24,8 @@
 import { readFileSync } from 'node:fs'
 import { chromium } from '../frontend/node_modules/playwright-core/index.mjs'
 import {
-  ISSUE, QUEUE, RECONCILE, RECONCILE_VALUES, WHOLE, WHOLE_LABELS, copiedMetricDetail, ledgerOf, metricDetail, reconciledOf,
+  CELL_LABELS, CELL_WHOLE, ISSUE, QUEUE, RECONCILE, RECONCILE_VALUES, WHOLE, WHOLE_LABELS, copiedMetricDetail, ledgerOf,
+  metricDetail, reconciledOf,
 } from './ledger-fixtures.mjs'
 
 const WEB = process.env.AGENTLAB_WEB ?? 'http://localhost:5273'
@@ -180,6 +181,29 @@ await section('「口径」一步的三种情况：来自口径、相关口径
   check('没有运行时报错', errors.length === 0, errors.join(' | '))
 })
 
+// ---------------------------------------------------------------- 逐格判
+
+await section('同一条查询的几个数逐格标：一格来自口径、一格只是相关，整题不等价不给出具', async () => {
+  const { page, errors } = await open('/ui-harness.html?evidence=1', {
+    reply: evidenceReply(() => ledgerOf(RUN, { labels: CELL_LABELS, whole: CELL_WHOLE, actions: [] })) })
+  await page.waitForSelector('#evidence-wide [data-ev-caliber-of="s6"]', { timeout: 10000 })
+  check('正文的小标记按数挂：来自口径的那一格挂上，相关的那一格不挂', await page.locator('#evidence-wide [data-ev-caliber-of="s6"]').count() === 1
+    && await page.locator('#evidence-wide [data-ev-caliber-of="s10"]').count() === 0)
+  await openSeg(page, 's6')
+  await page.locator('[data-evidence-panel] [data-ev-ledger="caliber"]').waitFor({ timeout: 5000 })
+  const caliber = await panelPart(page).innerText()
+  check('销售额那一格：来自口径、等价已证明、按口径的哪一列、参数', caliber.includes('来自口径「周报口径」v2（受管）')
+    && caliber.includes('等价已证明') && caliber.includes('gmv') && caliber.includes('start = 2026-10-05'), caliber)
+  check('……整题不等价：不给按口径出具', await page.locator('[data-evidence-panel] [data-ledger-issue]').count() === 0)
+  await openSeg(page, 's10')
+  await page.locator('[data-evidence-panel] [data-ev-ledger="related"]').waitFor({ timeout: 5000 })
+  const related = await panelPart(page).innerText()
+  check('同一条查询的订单数那一格：相关口径，差在哪先说多查了几列、再说计数对求和', related.includes('相关口径「周报口径」v2')
+    && related.includes('这次多查了 3 列，不影响这个数') && related.includes('这次数的是行数，口径对 amount 求和')
+    && !related.includes('未能归纳'), related)
+  check('没有运行时报错', errors.length === 0, errors.join(' | '))
+})
+
 // ---------------------------------------------------------------- 对账
 
 await section('口径对账：核对参数 → 发起 → 选对照的指标 → 三个数各自点得开', async () => {

scripts/ledger-fixtures.mjs

@@ -94,6 +94,24 @@ export const WHOLE = { equivalent: true, caliber: WEEKLY.caliber, workflow_id: W
   level_text: '受管', workflow_name: '周报口径', how: 'independent', binding: [{ field: 'week', value: '2026-W37' }],
   missing: [], missing_required: [], counts: { caliber: 3, related: 0, reasoning: 0, unknown: 0 } }
 
+/**
+ * 逐格判（服务端 ledger/match._cells，algo 4）的真实例子：同一条查询（q-gmv）报告引用了好几列，s6 销售额那一格证明了
+ * 等价于周报口径，s10 订单数那一格只是相关（差在哪先说多查了几列、再说计数对求和），整题不等价、没有出具入口
+ */
+export const CELL_DIFF = { items: [
+  { aspect: 'projection', a: null, b: null, text: '这次多查了 3 列，不影响这个数' },
+  { aspect: 'aggregate', a: 'COUNT(*)', b: 'SUM(amount)', text: '这次数的是行数，口径对 amount 求和' }],
+  fallback: false, text_diff: [] }
+export const CELL_LABELS = {
+  s6: caliberLabel(WEEKLY, 'independent', { binding: [{ field: 'start', value: '2026-10-05' }, { field: 'end', value: '2026-10-11' }],
+                                            query: 'q-gmv', node_id: 'fetch', column: 'gmv' }),
+  s10: { kind: 'related', query: 'q-gmv', related: [{ caliber: WEEKLY.caliber, workflow_id: WEEKLY.workflow_id, version: 2,
+                                                       level: 'governed', level_text: '受管', workflow_name: '周报口径',
+                                                       node_id: 'fetch', comparable: true, diff: CELL_DIFF }] },
+}
+export const CELL_WHOLE = { equivalent: false, reason: '有的数和相关口径不一致，不来自口径',
+  counts: { caliber: 1, related: 1, reasoning: 0, unknown: 0 } }
+
 /** 对过的一次账（单行：一组对照）。status 可以换 */
 export const reconciledOf = (status = 'succeeded') => ({
   run_id: 'run-rec-1', status, query: 'q-orders', caliber: ORDERS.caliber, workflow_id: ORDERS.workflow_id, version: 1,