给评测留一个按引用的列投影的入口（referenced_of），顺手折行、补模块说明

- ledger/match.referenced_of(sealed)：这次运行封存的报告引用了哪条查询的哪几列，正好是 compare_queries 的 referenced。
  评测逐题比 SQL（eval/score.sql_match，另一组在改）传它，就和账本一样按引用的列投影，不必自己再拼一遍。compute 和它
  共用读指标集的那一段（_card_loader）。
- data/sql_semantics 模块说明的「几处共用这里」补上模板绑定、按引用的列投影、同一个形状三处。
- 本轮几处新代码折到 120 列以内（只换行，不改逻辑）。

测试：tests/test_ledger_projection.py 的整题等价那条另核 referenced_of 交回的就是被引用的那一列。

backend/app/data/sql_semantics.py

@@ -14,6 +14,10 @@ A / B，只差一个方面、又对得上数据目录里的事实（时间归属
   （规范形式一字不差）才返回空列表。
 - facets：把一条查询拆成表、连接、WHERE 合取式、时间条件、分组、聚合、算出来的列、HAVING、取哪几行（给调用方取
   结构用；CTE、子查询按位置编号，见 _Statement）。facets 不受下面等价判定的改动影响。
+- bind_template / match_template：口径模板的参数绑定与等价证明（口径账本，见那一节）。
+- project_select / project_template：按引用的列投影——去掉最外层选择列表里没被引用、去掉也不改变被引用那几列取值的
+  选择项，认不准的一律不投影（见那一节）。投影本身不判等价，投影后的 SQL 照样交给上面的判等价。
+- sql_shape：同一个形状（认证队列的确定性重复）：条件的先后、交换律成立处两边的先后、字面量的值都不算（见那一节）。
 
 **等价怎么判。** 过去几轮按方面归纳任意 SQL，每轮修好一批又冒出一批（还回退过），收不拢；现在先判等价、再说差在哪。
 两边各自规范化，规范化的每一步都只做保持语义的变换，同一级的两个规范形式一字不差（比语法树逐个节点的类型和参数，
@@ -9245,8 +9249,9 @@ def _shape_print(node: Any, read: str | None) -> tuple[str, list[Any]]:
             and not _quantified(node.expression):
         left, right = _shape_print(node.this, read), _shape_print(node.expression, read)
         name = type(node).__name__
-        if (_fixed(node.this) and not _fixed(node.expression)) or (
-                _fixed(node.this) == _fixed(node.expression) and (left[0], value_key(left[1])) > (right[0], value_key(right[1]))):
+        later = (left[0], value_key(left[1])) > (right[0], value_key(right[1]))
+        fixed_left, fixed_right = _fixed(node.this), _fixed(node.expression)
+        if (fixed_left and not fixed_right) or (fixed_left == fixed_right and later):
             left, right, name = right, left, _FLIP[name]
         return f"{name}({left[0]},{right[0]})", left[1] + right[1]
     if isinstance(node, exp.Expr) and type(node).__name__ in _SYMMETRIC:

backend/app/engine/lineage.py

@@ -97,8 +97,9 @@ def metric_copy(path: str) -> tuple[str, str, int] | None:
     if not (isinstance(tree, ast.Attribute) and tree.attr == "value" and isinstance(tree.value, ast.Subscript)):
         return None
     index, metrics = tree.value.slice, tree.value.value
-    if not (isinstance(metrics, ast.Attribute) and metrics.attr == "metrics" and isinstance(metrics.value, ast.Attribute)
-            and isinstance(metrics.value.value, ast.Name) and metrics.value.value.id in ("vars", "nodes")):
+    if not (isinstance(metrics, ast.Attribute) and metrics.attr == "metrics"
+            and isinstance(metrics.value, ast.Attribute) and isinstance(metrics.value.value, ast.Name)
+            and metrics.value.value.id in ("vars", "nodes")):
         return None
     if not (isinstance(index, ast.Constant) and type(index.value) is int and index.value >= 0):
         return None

backend/app/engine/nodes/metrics.py

@@ -656,8 +656,10 @@ def _input_of(path: str, value: Any, state: GraphState, ctx: NodeContext,
         # 沙箱算出来的数喂给口径卡：记下它自认是「取数」还是「计算」，治理规则按它判
         entry["role"] = str(producer.config.get("evidence_role") or "compute")
     entry["status"] = "ok" if value is not None else "missing"
-    if producer is not None and str(producer.type) == "metrics" and (up := _upstream(path, tctx or template_context(state))):
-        entry["upstream"] = up
+    if producer is not None and str(producer.type) == "metrics":
+        # 转抄另一张口径卡的指标：跨卡往下追（engine/lineage.metric_copy）
+        if (up := _upstream(path, tctx or template_context(state))) is not None:
+            entry["upstream"] = up
     if producer is not None and str(producer.type) == "agent" and field:
         cited = _agent_field(state, producer.id, root if head else "", field)
         if cited is not None:

backend/app/ledger/certify.py

@@ -111,7 +111,8 @@ class Shape:
 
     def params(self, canonical_positions: Sequence[int]) -> list[int]:
         """规范形里的位置 → literals 里的位置（template_of 要的）；对不上的不要。"""
-        return sorted(p for i in canonical_positions if i < len(self.positions) and (p := self.positions[i]) is not None)
+        found = (self.positions[i] for i in canonical_positions if i < len(self.positions))
+        return sorted(p for p in found if p is not None)
 
 
 def shape_of(snap: Any, kinds: dict[str, str | None]) -> Shape | None:
@@ -258,7 +259,8 @@ async def _dismissed_members(session: Any) -> list[set[str]]:
 
 def _cluster_dismissed(cluster: Any, dismissed: set[str], members: list[set[str]]) -> bool:
     """这一类点过「不是口径」没有：类的 id 记过；以前按种子问题原文算的 id 记过；或者过半的成员是某一次记下的。"""
-    if f"cluster:{cluster.id}" in dismissed or item_id("cluster", " ".join(cluster.members[0].text.split())) in dismissed:
+    legacy = item_id("cluster", " ".join(cluster.members[0].text.split()))
+    if f"cluster:{cluster.id}" in dismissed or legacy in dismissed:
         return True
     keys = set(cluster.keys)
     return any(len(keys & seen) * 2 > len(keys) for seen in members)

backend/app/ledger/labels.py

@@ -431,5 +431,5 @@ 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", "referenced_columns",
-           "related_view", "view_label", "whole_of"]
+           "card_expressions", "card_reads", "cell_label", "diff_view", "label_doc", "metric_label",
+           "referenced_columns", "related_view", "view_label", "whole_of"]

backend/app/ledger/match.py

@@ -513,13 +513,10 @@ def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline:
     return out
 
 
-def compute(sealed: Any, calibers: list[Caliber], kinds: dict[str, str | None], *, timeout: float) -> dict[str, Any]:
-    """一次运行的比对工件内容（见 ledger/store 的模块说明）。在线程里调：读快照、比 SQL、标数都是同步的活。"""
-    from app.api.evidence import _choose, _reports
+def _card_loader(sealed: Any) -> Any:
+    """按工件读封存范围内的指标集（不在封存台账里的、读不出的为 None）。"""
     from app.core.artifact_store import load
 
-    deadline = time.monotonic() + timeout
-    queries = read_queries(sealed, kinds)
     sealed_cards = {artifact for _, artifact in sealed.ledger}
 
     def load_card(artifact: str) -> dict[str, Any] | None:
@@ -531,6 +528,25 @@ def compute(sealed: Any, calibers: list[Caliber], kinds: dict[str, str | None],
             return None
         return card if isinstance(card, dict) else None
 
+    return load_card
+
+
+def referenced_of(sealed: Any) -> dict[str, set[str]]:
+    """这次运行封存的报告引用了哪条查询的哪几列（labels.referenced_columns，只读哈希核对一致的报告）：compare_queries 的
+    referenced。评测逐题比 SQL 时传它，就和账本一样按引用的列投影。在线程里调（读盘）。"""
+    from app.api.evidence import _reports
+
+    docs = [r.doc for r in _reports(sealed) if r.doc is not None and r.hash_ok is True]
+    return ledger_labels.referenced_columns(docs, _card_loader(sealed))
+
+
+def compute(sealed: Any, calibers: list[Caliber], kinds: dict[str, str | None], *, timeout: float) -> dict[str, Any]:
+    """一次运行的比对工件内容（见 ledger/store 的模块说明）。在线程里调：读快照、比 SQL、标数都是同步的活。"""
+    from app.api.evidence import _choose, _reports
+
+    deadline = time.monotonic() + timeout
+    queries = read_queries(sealed, kinds)
+    load_card = _card_loader(sealed)
     found = _reports(sealed)
     readable = [report for report in found if report.doc is not None and report.hash_ok is True]
     if not sealed.trusted:
@@ -743,7 +759,7 @@ async def shutdown() -> None:
     await asyncio.gather(*tasks, return_exceptions=True)
 
 
-__all__ = ["BACKFILL_DAYS", "CONCURRENCY", "EXPLORE_ORIGINS", "EXTRA_COLUMNS", "MAX_PROJECTIONS", "TIMEOUT_S", "Caliber",
-           "Fetch", "Query", "backfill", "caliber_of", "compare_queries", "composed_nodes", "compute", "drain",
-           "ineligible", "load_calibers", "match_run", "pending", "read_queries", "schedule", "shutdown",
-           "start_backfill", "wanted"]
+__all__ = ["BACKFILL_DAYS", "CONCURRENCY", "EXPLORE_ORIGINS", "EXTRA_COLUMNS", "MAX_PROJECTIONS", "TIMEOUT_S",
+           "Caliber", "Fetch", "Query", "backfill", "caliber_of", "compare_queries", "composed_nodes", "compute",
+           "drain", "ineligible", "load_calibers", "match_run", "pending", "read_queries", "referenced_of",
+           "schedule", "shutdown", "start_backfill", "wanted"]

backend/tests/test_ledger_projection.py

@@ -193,6 +193,10 @@ async def test_last_weeks_sales_with_extra_columns_is_whole_equivalent(engine_up
     content = (await store.latest(row.id)).content
     [q] = content["queries"]
     assert q["status"] == "proven"
+    # 评测逐题比 SQL 时拿同一份「引用了哪几列」（referenced_of），和账本一样投影
+    from app.api.evidence import _sealed
+
+    assert ledger_match.referenced_of(await _sealed(row.id)) == {q["artifact"]: {"销售额"}}
     [cand] = q["candidates"]
     assert cand["projection"]["dropped"] == 3 and cand["columns"] == [{"index": 0, "ours": "销售额", "caliber": "gmv"}]
     [labels] = [r["labels"] for r in content["reports"].values()]