合入复核修复（证据接入与入口）：跨段指标不当两期查、证据接口重渲染与组装同一个函数、一次请求同一份结论只复算一次、比率闭合核对、代入式按点名的段、写作目录分析段限额、核对页写方法叫法、运行对比只认做完的执行、读快照进线程、四个工具与合并查询的真算子测试、夹具清源

backend/app/api/evidence.py

@@ -104,6 +104,7 @@ from app.db.base import SessionLocal
 from app.db.models import DataSource, Run, RunEvent, Workflow
 from app.engine import analysis_claims, direct_select, recompute
 from app.engine.evidence import (
+    CELL_DISPLAY,
     ENTITY_KINDS,
     GUESS_NOTE,
     GUESS_SCHEMA,
@@ -116,6 +117,7 @@ from app.engine.evidence import (
     analysis_recheck,
     build_catalog,
     closest_entities,
+    doc_cell_display,
     find_segment,
     guess_sources,
     input_eid,
@@ -126,6 +128,7 @@ from app.engine.evidence import (
     query_entry_fields,
     render_cell,
     render_metric,
+    resolve_ref,
     schema_column,
     schema_table,
     sql_tables,
@@ -221,6 +224,9 @@ class _Sealed:
     #: 节点 id → node.started 上记的名字、类型。子工作流里的节点（「子工作流节点 id/…」）不在这次运行的图上，名字从这里取
     labels: dict[str, str] = field(default_factory=dict)
     types: dict[str, str] = field(default_factory=dict)
+    #: 这一次请求里复算过的分析结论：结论工件 id → 复算的结论（_analysis_recheck）。导出页里同一份结论被判定语、具名的数、
+    #: 整表的每一格、分析范围各引一处，各接一条链，复算只做一次。只活在这一次请求里：下一次请求重新取、重新验
+    rechecks: dict[str, dict[str, Any]] = field(default_factory=dict)
 
     @property
     def sealed(self) -> bool:
@@ -1000,11 +1006,16 @@ _OP_SETTINGS: dict[str, tuple[str, ...]] = {
     "differ": ("differ_alpha", "min_n", "min_expected"),
     "comparable": (),
 }
-#: 输入的角色（analysis.contract.AnalysisInput.role）给人看的叫法
+#: 输入的角色（analysis.contract.AnalysisInput.role）给人看的叫法；比率拆解的 headline 是整体比率，不是总数
 _INPUT_ROLES = {"data": "数据", "base": "基期", "target": "本期", "headline_base": "基期总数",
                 "headline_target": "本期总数", "parent": "上一层拆解"}
-#: 闭合核对：headline 的那一格对着结论里的哪个总数（细案 01 §2.1）
-_HEADLINE_TOTALS = {"headline_base": "base_total", "headline_target": "target_total"}
+_RATIO_ROLES = {"headline_base": "基期整体比率", "headline_target": "本期整体比率"}
+#: 闭合核对：headline 的那一格对着结论里的哪个数（细案 01 §2.1、§2.2）。可加拆解对各段之和（base_total / target_total），
+#: 比率拆解对整体比率，即各段分子之和 ÷ 分母之和（r_base_total / r_target_total）
+_HEADLINE_TOTALS = {
+    "additive": {"headline_base": "base_total", "headline_target": "target_total"},
+    "ratio": {"headline_base": "r_base_total", "headline_target": "r_target_total"},
+}
 #: 计算一步最多给结果表的几行（前 table_top_k 段，加上被引用的那一行前后各 WINDOW 行）
 MAX_ANALYSIS_ROWS = MAX_WINDOW_ROWS
 
@@ -1041,17 +1052,22 @@ def _analysis_cited(cite: dict[str, Any], result: dict[str, Any] | None) -> dict
     return {"what": "phrase"}
 
 
-def _analysis_rendered(cite: dict[str, Any], result: dict[str, Any]) -> str | None:
-    """按封存的结论重新渲染这一处引用（判定语按文本格、具名标量和结果表的格按单元格的规则）。渲染不出来是 None。"""
-    try:
-        cited = _analysis_cited(cite, result)
-        if cited["what"] == "phrase":
-            return render_cell(result.get("phrase"), kind="text") if isinstance(result.get("phrase"), str) else None
-        if cited["what"] == "value":
-            return render_cell(cited["raw"], cite.get("conv"))
-        return render_cell(cited["raw"], cite.get("conv"), plain=cite.get("plain") is True)
-    except RenderError:
-        return None
+def _analysis_rendered(seg: dict[str, Any], cite: dict[str, Any], entry: dict[str, Any], result: dict[str, Any],
+                       doc: dict[str, Any]) -> str | None:
+    """按封存的结论把这一处引用重新解析一遍，交回渲染出来的字（渲染不出来是 None）。和组装、出口复核同一个函数
+    （evidence.resolve_ref，按文档记下的那一版单元格显示规则）：判定语按文本格，具名标量按单元格的规则，结果表的格照结果表
+    记的列类型（文本码值 "1203" 不加千分位）、显示规则（ISO 日期时间的 T 换成空格）、标识列不分千位。片段上没记引用的
+    （口径卡指标上的可比性、分析范围里的编号）按判定语渲染。"""
+    ref = seg.get("ref")
+    if not isinstance(ref, str) or not ref.strip():
+        phrase = result.get("phrase")
+        try:
+            return render_cell(phrase, kind="text") if isinstance(phrase, str) else None
+        except RenderError:
+            return None
+    alias = str(cite.get("alias") or entry.get("alias") or "")
+    again = resolve_ref(ref, {alias: {**entry, "result": result}}, display=doc_cell_display(doc) or CELL_DISPLAY)
+    return again.get("rendered") if again.get("status") == "resolved" else None
 
 
 async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[str, Any], doc: dict[str, Any],
@@ -1082,7 +1098,7 @@ async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict
     in_ledger = sealed.trusted and (str(entry.get("node_id") or ""), artifact) in sealed.ledger \
         if chained is None else sealed.trusted and chained
     eid = make_eid(LEDGER_KIND, artifact or None, {})
-    rendered = _analysis_rendered(cite, result) if result is not None else None
+    rendered = _analysis_rendered(seg, cite, entry, result, doc) if result is not None else None
     judged: dict[str, Any] = {
         "step": "analysis", "alias": alias, "op": entry.get("op"), "mode": entry.get("mode"),
         "method_label": method_label(entry.get("op"), entry.get("mode")),
@@ -1122,7 +1138,7 @@ async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict
     inputs, steps = await _analysis_inputs(alias, result, catalog, doc, sealed, masks, queries=queries)
     hidden = {str(c).lower() for q in steps for c in q.get("masked") or []}
     calc = _analysis_calc(alias, result, judged["cited"], hidden)
-    again = await _analysis_recheck(result, sealed)
+    again = await _analysis_recheck(result, sealed, artifact)
     recheck = {"step": "analysis_recompute", "alias": alias, **again}
     return [judged, method, calc, *inputs, *steps, recheck, seal]
 
@@ -1170,11 +1186,15 @@ async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str
                            sealed: _Sealed, masks: _Masks, *, queries: bool = True
                            ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
     """输入：每份一步（analysis_input），再给每份输入快照一个查询步骤（_query_steps，高亮用到的列；整份用到的给前
-    MAX_WINDOW_ROWS 行；queries 为假时不给）。headline 的那一格另核对和结论里的总数闭不闭合（照实写判定是不是
-    unclosed，不在这里重算容差）；下钻的上一层落到报告目录的哪个 A。"""
+    MAX_WINDOW_ROWS 行；queries 为假时不给）。headline 的那一格另核对和结论里的数闭不闭合（_HEADLINE_TOTALS：可加拆解是
+    各段之和，比率拆解是整体比率；closure.mode 写明是哪一种，照实写判定是不是 unclosed，不在这里重算容差）；下钻的上一层
+    落到报告目录的哪个 A。"""
     values = result.get("values") if isinstance(result.get("values"), dict) else {}
     verdict = result.get("verdict") if isinstance(result.get("verdict"), dict) else {}
     unclosed = verdict.get("code") == "unclosed"
+    mode = str(result.get("mode") or "")
+    totals = _HEADLINE_TOTALS.get(mode, {}) if result.get("op") == "decompose" else {}
+    roles = {**_INPUT_ROLES, **(_RATIO_ROLES if mode == "ratio" else {})}
     steps: list[dict[str, Any]] = []
     cells: dict[str, list[tuple[Any, Any]]] = {}
     for item in result.get("inputs") or []:
@@ -1184,7 +1204,7 @@ async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str
         role = str(item.get("role") or "")
         step: dict[str, Any] = {
             "step": "analysis_input", "of": alias, "ref": item.get("ref"), "role": role,
-            "role_label": _INPUT_ROLES.get(role, role or "输入"), "artifact": artifact or None,
+            "role_label": roles.get(role, role or "输入"), "artifact": artifact or None,
             **{k: item[k] for k in ("columns", "rows", "row", "column", "value") if k in item},
         }
         if role == "parent":
@@ -1197,10 +1217,9 @@ async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str
         query = sealed.queries.get(artifact)
         step.update(query=_catalog_alias_of(catalog, artifact), node_id=(query or {}).get("node_id"),
                     sealed=sealed.trusted and query is not None)
-        if role in _HEADLINE_TOTALS and _HEADLINE_TOTALS[role] in values:
-            total = values[_HEADLINE_TOTALS[role]]
+        if role in totals and totals[role] in values:
             with contextlib.suppress(RenderError):
-                step["closure"] = {"total_key": _HEADLINE_TOTALS[role], "total": render_cell(total),
+                step["closure"] = {"mode": mode, "total_key": totals[role], "total": render_cell(values[totals[role]]),
                                    "headline": render_cell(item.get("value")), "closed": not unclosed}
         steps.append(step)
         if artifact and query is not None:
@@ -1222,10 +1241,21 @@ async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str
     return steps, found
 
 
-async def _analysis_recheck(result: dict[str, Any], sealed: _Sealed) -> dict[str, Any]:
+async def _analysis_recheck(result: dict[str, Any], sealed: _Sealed, artifact: str | None = None) -> dict[str, Any]:
     """复算（evidence.analysis_recheck）：输入快照交工件原内容，下钻的上一层交那份结论，都按 sha256 取回、复验哈希，在线程里
     复算、比较。只认封存范围内的——输入快照要是这次运行交回过的查询（sealed.queries），上一层要是台账记过的分析结论；
-    不在范围内、哈希对不上的一律当取不回（复算照实说取不回哪一份）。"""
+    不在范围内、哈希对不上的一律当取不回（复算照实说取不回哪一份）。
+
+    artifact 是这份结论的工件 id（内容寻址）：一次请求里同一份结论只复算一次，结果记在 sealed.rechecks。"""
+    if artifact and artifact in sealed.rechecks:
+        return dict(sealed.rechecks[artifact])
+    again = await _recheck_now(result, sealed)
+    if artifact:
+        sealed.rechecks[artifact] = again
+    return dict(again)
+
+
+async def _recheck_now(result: dict[str, Any], sealed: _Sealed) -> dict[str, Any]:
     snapshots = {str(i["artifact"]) for i in result.get("inputs") or []
                  if isinstance(i, dict) and i.get("artifact") and i.get("role") != "parent"}
     parents = {str(i["artifact"]) for i in result.get("inputs") or []
@@ -3260,22 +3290,24 @@ def scope_aliases(doc: dict[str, Any]) -> list[str]:
     return list(dict.fromkeys(out))
 
 
-async def _scope_chains(doc: dict[str, Any], sealed: _Sealed, report_node: str) -> dict[str, list[dict[str, Any]]]:
-    """分析范围里每个编号的出处链（和点开正文里的判定语同一条 _chain，按那条结论的判定语认）：{编号: 链}。目录里没有的
-    编号不给（文档里写着、目录里却没有，只能是文档有问题，范围那一项点不开）。"""
+def _scope_cite_key(alias: str) -> tuple:
+    """分析范围里一个编号的引用键：和正文里引用判定语（[[a:A1]]）的片段同一个键（_export_cite_key），共用一条链。"""
+    return (LEDGER_KIND, alias, "resolved", None, None, None, None, None, None)
+
+
+async def _scope_chain(alias: str, doc: dict[str, Any], sealed: _Sealed, report_node: str) -> list[dict[str, Any]] | None:
+    """分析范围里一个编号的出处链（和点开正文里的判定语同一条 _chain，按那条结论的判定语认）。目录里没有的编号为 None
+    （文档里写着、目录里却没有，只能是文档有问题，范围那一项点不开）。"""
     catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {}
-    out: dict[str, list[dict[str, Any]]] = {}
-    for alias in scope_aliases(doc):
-        entry = catalog.get(alias)
-        if not isinstance(entry, dict) or entry.get("kind") != LEDGER_KIND:
-            continue
-        result, _ = _analysis_of(str(entry.get("artifact") or ""))
-        seg = {"id": SCOPE_REF.format(alias=alias), "kind": LEDGER_KIND, "state": "deterministic",
-               "text": render_cell(result.get("phrase"), kind="text") if result and result.get("phrase") else alias,
-               "cite": {"alias": alias, "kind": LEDGER_KIND, "status": "resolved", "locator": {},
-                        "eid": entry.get("eid")}}
-        out[alias] = await _chain(seg, doc, sealed, report_node)
-    return out
+    entry = catalog.get(alias)
+    if not isinstance(entry, dict) or entry.get("kind") != LEDGER_KIND:
+        return None
+    result, _ = _analysis_of(str(entry.get("artifact") or ""))
+    seg = {"id": SCOPE_REF.format(alias=alias), "kind": LEDGER_KIND, "state": "deterministic",
+           "text": render_cell(result.get("phrase"), kind="text") if result and result.get("phrase") else alias,
+           "cite": {"alias": alias, "kind": LEDGER_KIND, "status": "resolved", "locator": {},
+                    "eid": entry.get("eid")}}
+    return await _chain(seg, doc, sealed, report_node)
 
 
 async def _export_withheld(sealed: _Sealed, doc: dict[str, Any]) -> frozenset[str]:
@@ -3383,14 +3415,17 @@ async def export_report_html(run_id: str, report: str | None = None) -> Response
             refs[sid] = str(ck)
             if key is not None:
                 memo[key] = str(ck)
-    # 系统写的「分析范围」：每个编号点得开，链和点开正文里的判定语是同一条（引用键相同的共用）
-    for alias, chain in (await _scope_chains(doc, sealed, chosen.node_id)).items():
-        sid = SCOPE_REF.format(alias=alias)
-        ck = (LEDGER_KIND, alias, "resolved", None, None, None, None, None, None)
+    # 系统写的「分析范围」：每个编号点得开，链和点开正文里的判定语是同一条（引用键相同的共用：正文已经引了判定语的，
+    # 不再建一遍）
+    for alias in scope_aliases(doc):
+        ck = _scope_cite_key(alias)
         if ck not in memo:
+            chain = await _scope_chain(alias, doc, sealed, chosen.node_id)
+            if chain is None:
+                continue
             memo[ck] = str(ck)
             chains[str(ck)] = chain
-        refs[sid] = memo[ck]
+        refs[SCOPE_REF.format(alias=alias)] = memo[ck]
     async with SessionLocal() as session:
         run = await session.get(Run, run_id)
         if run is None:      # 两次读之间被删掉了
@@ -4013,18 +4048,22 @@ def _scope_catalog(sealed: _Sealed, report_node: str) -> dict[str, Any]:
 
     doc.catalog 只留了正文用到的实体，给「是不是想写…」找候选不够；运行状态（checkpoint）不在封存
     范围里，也不能用。图读不出来、找不到报告节点时退回全部封存台账。
+
+    台账里的分析结论不进：候选只看表和字段，用不到结论的内容，进了目录就要把每份结论读盘、复验一遍哈希（build_catalog
+    给 A 条目装整份结论）。查询、表和字段的编号不受影响（分析结论另编 A）。
     """
     from app.engine.nodes.report import report_catalog
     from app.engine.schema import GraphSpec
 
+    entries = [e for e in sealed.entries if e.get("kind") != LEDGER_KIND]
     try:
         spec = GraphSpec.model_validate(sealed.graph)
         node = spec.node_map().get(report_node)
     except Exception:  # noqa: BLE001 - 老运行的图可能已经不合现在的格式，退回不分范围
         node = None
     if node is None:
-        return build_catalog(nodes={}, ledger=sealed.entries)
-    return report_catalog({"evidence": sealed.entries, "nodes": {}, "input": {}}, spec, node)  # type: ignore[arg-type]
+        return build_catalog(nodes={}, ledger=entries)
+    return report_catalog({"evidence": entries, "nodes": {}, "input": {}}, spec, node)  # type: ignore[arg-type]
 
 
 def _suspicious_step(seg: dict[str, Any], cite: dict[str, Any], sealed: _Sealed, report_node: str) -> dict[str, Any]:

backend/app/bundle/build.py

@@ -63,7 +63,7 @@ from pathlib import Path
 from typing import Any
 from urllib.parse import quote
 
-from app.analysis.contract import ARTIFACT_KIND, LEDGER_KIND, NODE_TYPE as ANALYSIS_NODE, is_result
+from app.analysis.contract import ARTIFACT_KIND, LEDGER_KIND, NODE_TYPE as ANALYSIS_NODE, is_result, method_label
 from app.core.artifact_store import canonical_json, content_hash
 from app.core.events import EventType
 from app.engine.evidence import analysis_recheck, iter_units, query_entry_fields
@@ -1161,8 +1161,9 @@ class _Artifacts:
 def _attach_analysis(artifacts: _Artifacts, store: _Store, item: dict[str, Any] | None, artifact: str,
                      owner: str | None, cited_queries: list[str], depth: int = 0) -> None:
     """随文件附带的分析结论：它的输入快照（种类 query，遮罩照常认）、下钻的上一层结论（同样附带它的输入，最多
-    _PARENT_DEPTH 层）一并列进清单，再把服务端的复算结论记在这份结论的条目上（recheck）。只给 id 的结论、被改过的
-    （哈希对不上）不顺着它去附带别的：被改过的结论说输入是哪几份，不能替它圆。"""
+    _PARENT_DEPTH 层）一并列进清单，再把服务端的复算结论记在这份结论的条目上（recheck），连同方法的叫法（method_label，
+    contract.method_label，和导出页同一份：核对页只显示，不在页面里另起一套叫法）。只给 id 的结论、被改过的（哈希对不上）
+    不顺着它去附带别的：被改过的结论说输入是哪几份，不能替它圆。"""
     if item is None or "b64" not in item or not store.intact(artifact) or "recheck" in item:
         return
     result = store.content(artifact)
@@ -1176,6 +1177,7 @@ def _attach_analysis(artifacts: _Artifacts, store: _Store, item: dict[str, Any]
             continue
         artifacts.add(input_id, "query", owner)
         cited_queries.append(input_id)
+    item["method_label"] = method_label(result.get("op"), result.get("mode"))
     item["recheck"] = _recheck(result, store)
 
 

backend/app/bundle/verifier.js

@@ -1983,6 +1983,14 @@
     return parts
   }
 
+  /** 方法与版本那一行：服务端在工件条目上带的叫法（method_label，和导出页同一份「变化拆解（可加） · decompose.additive@1」）
+   *  加方法 id；条目上没带叫法的只写 id */
+  function methodText(art, result) {
+    const label = art && isObj(art.a) && typeof art.a.method_label === 'string' ? art.a.method_label : ''
+    const id = String(result.method || '—')
+    return label ? `${label} · ${id}` : id
+  }
+
   /** 口径卡指标上的可比性（口径卡执行时自动做的可比性检查）：卡上记的判定和原因；那份结论随文件附带、核对通过的，再给
    *  判定标准、方法与版本、各项检查和导出时服务端的复算。没附带的照卡上的记录写，说明为什么没有更多 */
   function comparabilityBlock(c) {
@@ -2002,7 +2010,7 @@
     }
     parts.push(kv([
       result.criterion ? [T.dCriterion, String(result.criterion)] : null,
-      [T.dMethod, String(result.method || '—')],
+      [T.dMethod, methodText(art, result)],
     ]))
     const table = isObj(result.table) ? result.table : null
     const columns = table && Array.isArray(table.columns) ? table.columns.map(String) : []
@@ -2030,7 +2038,7 @@
       parts.push(kv([
         [T.dPhrase, String(result.phrase || '—')],
         result.criterion ? [T.dCriterion, String(result.criterion)] : null,
-        [T.dMethod, String(result.method || '—')],
+        [T.dMethod, methodText(art, result)],
       ]))
       const caveats = (Array.isArray(result.caveats) ? result.caveats : []).filter(isObj).map((c) => String(c.text || ''))
       if (caveats.length) parts.push(h('p', { class: 'sub' }, T.dCaveats), h('ul', { class: 'items' }, caveats.map((c) => h('li', {}, c))))

backend/app/engine/compare.py

@@ -336,17 +336,22 @@ def _queries(side: Side) -> list[_Query]:
 
 
 def _analysis_items(side: Side) -> dict[str, tuple[dict[str, dict[str, Any]], int]]:
-    """封存范围内每张分析卡各项最后一次的结论（analysis.end，engine/nodes/analysis.py 每项一条）：{卡: ({项: 事件数据},
-    执行了几次)}，项按记下的先后。循环里执行了几次的，同一项后一次的盖过前一次。"""
+    """封存范围内每张分析卡各项最后一次做完的结论（analysis.end，engine/nodes/analysis.py 每项一条）：{卡: ({项: 事件数据},
+    执行了几次)}，项按记下的先后。只认后面跟着这张卡 node.finished 的那几次执行：执行到一半失败的（前几项已经发了
+    analysis.end，后面一项让卡失败），那次的判定不是这次运行的结果，不列，也不算执行次数。循环里执行了几次的，同一项后
+    一次的盖过前一次。"""
     out: dict[str, tuple[dict[str, dict[str, Any]], int]] = {}
+    pending: dict[str, dict[str, dict[str, Any]]] = {}       # 卡 → 这一次执行里已经做完的项（还没见到 node.finished）
     for _, etype, node_id, data in side.events:
         if not node_id or not isinstance(data, dict):
             continue
-        if etype == EventType.ANALYSIS_END and isinstance(data.get("item"), str):
+        if etype == EventType.NODE_STARTED:
+            pending.pop(node_id, None)
+        elif etype == EventType.ANALYSIS_END and isinstance(data.get("item"), str):
+            pending.setdefault(node_id, {})[data["item"]] = data
+        elif etype == EventType.NODE_FINISHED and node_id in pending:
             items, runs = out.setdefault(node_id, ({}, 0))
-            items[data["item"]] = data
-        elif etype == EventType.NODE_FINISHED and node_id in out:
-            items, runs = out[node_id]
+            items.update(pending.pop(node_id))
             out[node_id] = (items, runs + 1)
     return out
 

backend/app/engine/evidence.py

@@ -2106,8 +2106,11 @@ def resolve_support(ref: str, catalog: dict[str, Any], *, loader: Loader | _Snap
 
 
 def resolve_ref(ref: str, catalog: dict[str, Any], *, cells_allowed: bool = True,
-                loader: Loader | _Snapshots | None = None) -> dict[str, Any]:
-    """片段上记的 ref（m:gmv|万、i:week、v:Q1.r0.amount）→ Citation。证据接口点开片段时用它重新解析。"""
+                loader: Loader | _Snapshots | None = None, display: int | None = None) -> dict[str, Any]:
+    """片段上记的 ref（m:gmv|万、i:week、v:Q1.r0.amount）→ Citation。证据接口点开片段时用它重新解析。display 是按哪一版
+    单元格显示规则渲染（文档记下的那一版，doc_cell_display；复核按它），不给按 loader 的（缺省当前版）。"""
+    if display is not None:
+        loader = _snapshots(loader, display=display)
     return resolve_marker(_parse_ref(ref), catalog, cells_allowed=cells_allowed, loader=loader)
 
 
@@ -4057,11 +4060,18 @@ def _analysis_sources(entry: dict[str, Any], catalog: dict[str, Any]) -> str:
                     f"，另有 {missing} 份输入不在这份目录里" if missing else ""])
 
 
+#: 写作目录里一条分析结论的几种样子，从全到简（_analysis_prompt 的 level）：full 连同结果表的前 table_top_k 段；compact 不列
+#: 各段，结果表只写一句；brief 只留判定语和具名的数；head 只列编号、方法和来自哪几个查询
+ANALYSIS_LEVELS = ("full", "compact", "brief", "head")
+
+
 def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, *,
-                     table: bool = True) -> list[str]:
+                     level: str = "full") -> list[str]:
     """一条分析结论在写作目录里的样子：编号、方法、来自哪几个查询，判定语（[[a:A1]] 渲染出来的那一句）、判定标准、具名的数
-    （可引用的键，值按渲染规则写）、附带说明、输入查询的 SQL 检查问题、算不了的原因；table 为真时再列结果表的前
-    table_top_k 段，其余写「其余 n 段」。agent 写的假设不列：它不是结论，只在证据面板显示。
+    （可引用的键，值按渲染规则写）、附带说明、输入查询的 SQL 检查问题、算不了的原因；level 为 full 时再列结果表的前
+    table_top_k 段，其余写「其余 n 段」。目录放不下时往简里降（ANALYSIS_LEVELS）：compact 结果表只写一句；brief 只留
+    判定语和具名的数（判定标准、附带说明、SQL 检查的提醒、算不了的原因那句不列，判定语里本来就写着）；head 只列编号那一行。
+    agent 写的假设不列：它不是结论，只在证据面板显示。
 
     受管级别的正式出具、契约没开「单元格引用」时，Agent 现场做的分析只列判定语和判定标准，写明只能当依据（[[see:A1]]），
     不列数和表：列出来的数写作者只能照抄成没有出处的数。"""
@@ -4070,8 +4080,11 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna
     result, why = snaps.analysis(alias, entry)
     if result is None:
         return [f"{head}（{why}，不要引用）"]
+    if level == "head":
+        return [head]
+    brief = level == "brief"
     lines = [head]
-    if note := _sql_check_note(entry.get("artifact"), snaps, "这次分析的输入查询"):
+    if not brief and (note := _sql_check_note(entry.get("artifact"), snaps, "这次分析的输入查询")):
         lines.append(note)
     phrase = _clean(str(result.get("phrase") or ""))
     see_only = not cells_allowed and not analysis_card(entry)
@@ -4080,13 +4093,15 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna
                      f"[[see:{alias}]] 作依据）")
     else:
         lines.append(f"  判定语 [[a:{alias}]] = {phrase}" if phrase else f"  （没有判定语，不要写 [[a:{alias}]]）")
-    if result.get("status") == "refused":
+    if brief:
+        pass
+    elif result.get("status") == "refused":
         lines.append(f"  算不了：{_clean(str(result.get('reason') or '没有写原因'))}。这条结论不能当「异常」「主要来自」"
                      "「呈上升趋势」这类判断的依据，只能照实写算不了的原因")
     elif criterion := _clean(str(result.get("criterion") or "")):
         lines.append(f"  判定标准：{criterion}")
     caveats = [_clean(str(c.get("text") or "")) for c in result.get("caveats") or [] if isinstance(c, dict)]
-    if caveats := [c for c in caveats if c]:
+    if not brief and (caveats := [c for c in caveats if c]):
         lines.append(f"  附带说明：{'；'.join(caveats)}")
     if see_only:
         return lines
@@ -4100,6 +4115,8 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna
     if shown:
         more = f"，另有 {len(values) - _PROMPT_VALUES} 个" if len(values) > _PROMPT_VALUES else ""
         lines.append(f"  具名的数（写 [[a:{alias}.键]]）：{'，'.join(shown)}{more}")
+    if brief:
+        return lines
     snapshot, why = snaps.table(alias, entry)
     if why is not None:
         return lines
@@ -4107,7 +4124,7 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna
     labels = snapshot.get("labels") or {}
     rows = snapshot["rows"]
     named = "、".join(f"{c}（{_clean(str(labels[c]))}）" if labels.get(c) else c for c in columns)
-    if not table:
+    if level != "full":
         lines.append(f"  结果表共 {len(rows)} 段，列 {named}（这里没列出各段；要列出就单独一行写 [[table:{alias}]]）")
         return lines
     settings = result.get("settings") if isinstance(result.get("settings"), dict) else {}
@@ -4135,12 +4152,13 @@ _TOP_K = 10
 
 @dataclass
 class _AnalysisLines:
-    """写作目录里「分析结论」一段：每条结论两种样子——full 连同结果表的前 table_top_k 段，compact 只有判定、判定标准、
-    具名的数和说明。放得下时全按 full；放不下时先按 compact 列出，再按顺序补上结果表，排在查询的行前面（_trimmed_catalog）。"""
+    """写作目录里「分析结论」一段：每条结论四种样子（ANALYSIS_LEVELS，从全到简）。放得下时全按 full；放不下时这一段最多占
+    预算的 1/_PROMPT_ANALYSIS_SHARE：先按 compact 列出，还超就从后往前降成 brief、再降成 head，然后按顺序补上结果表，排在
+    查询的行前面（_trimmed_catalog）。"""
 
     order: list[str]
-    full: dict[str, list[str]]
-    compact: dict[str, list[str]]
+    #: 样子 → {编号: 那几行}
+    levels: dict[str, dict[str, list[str]]]
     #: 有 Agent 现场做的分析只能当依据（受管级别的出具、契约没开「单元格引用」）
     see_only: bool = False
 
@@ -4148,15 +4166,19 @@ class _AnalysisLines:
     def of(cls, catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool) -> "_AnalysisLines":
         entries = [e for e in catalog.values() if isinstance(e, dict) and e.get("kind") == LEDGER_KIND]
         return cls([e["alias"] for e in entries],
-                   {e["alias"]: _analysis_prompt(e, catalog, snaps, cells_allowed) for e in entries},
-                   {e["alias"]: _analysis_prompt(e, catalog, snaps, cells_allowed, table=False) for e in entries},
+                   {level: {e["alias"]: _analysis_prompt(e, catalog, snaps, cells_allowed, level=level)
+                            for e in entries} for level in ANALYSIS_LEVELS},
                    see_only=not cells_allowed and any(not analysis_card(e) for e in entries))
 
+    def shape(self, alias: str, level: str) -> list[str]:
+        return self.levels[level][alias]
+
     def lines(self, tier: dict[str, str] | None = None) -> list[str]:
-        """这一段的文字：tier 里记成 compact 的按 compact，其余按 full。没有分析结论时是空的（目录和以前一字不差）。"""
+        """这一段的文字：每条按 tier 里记的样子，没记的按 full。没有分析结论时是空的（目录和以前一字不差）。"""
         if not self.order:
             return []
-        shapes = [self.compact[a] if (tier or {}).get(a) == "compact" else self.full[a] for a in self.order]
+        shapes = [self.shape(a, level if (level := (tier or {}).get(a)) in ANALYSIS_LEVELS else "full")
+                  for a in self.order]
         head = [f"\n{ANALYSIS_PROMPT}", *([ANALYSIS_SEE_ONLY_PROMPT] if self.see_only else [])]
         return [*head, *(line for shape in shapes for line in shape)]
 
@@ -4176,6 +4198,8 @@ CATALOG_MAX_CHARS = 160_000
 _PROMPT_COMPACT_ROWS = 3
 #: 放不下时表和字段、知识库检索两节合起来最多占预算的几分之一（剩下的给查询）
 _PROMPT_TAIL_SHARE = 4
+#: 放不下时分析结论一段最多占预算的几分之一：几十份结论、每份上千段时，不能把查询挤得只剩一行摘要
+_PROMPT_ANALYSIS_SHARE = 2
 _PROMPT_CUT = "\n…（目录太长，后面的省略了）"
 
 
@@ -4451,16 +4475,17 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str],
                      pinned: list[str], analyses: _AnalysisLines | None = None) -> WriterCatalog:
     """放不下时的目录。以前从头截断：查询按编号排，前面的节点占满预算，后面的节点一次都不露面。
 
-    1. 口径卡指标、运行输入、结构化字段一律列出（第一优先）；分析结论的判定、判定标准、具名的数、说明也一律列出
+    1. 口径卡指标、运行输入、结构化字段一律列出（第一优先）；分析结论的判定、判定标准、具名的数、说明也列出，只是这一段
+       最多占预算的 1/_PROMPT_ANALYSIS_SHARE：超了就按结论的先后从后往前降，先只留判定语和具名的数，再只列编号
     2. 表和字段、知识库检索两节合起来最多占预算的 1/_PROMPT_TAIL_SHARE
-    3. 分析结论的结果表（前 table_top_k 段）先补：判定是从原始明细算出来的结论，比原始明细更值得留
+    3. 分析结论的结果表（前 table_top_k 段）先补（不超过那个上限）：判定是从原始明细算出来的结论，比原始明细更值得留
     4. 每次查询先只占一行摘要；结构化字段引用到的查询先展开（pinned），再按产出它的节点分组、轮流展开：
        第一轮每次列前 _PROMPT_COMPACT_ROWS 行，第二轮补到 _PROMPT_ROWS 行，放不下为止
     5. 开头写明目录有省略、各来源省略了几次，以及没展开的查询怎么用（OMITTED_RULES）
     最后整份量一次：超了就把最后展开的退回去（先退查询，最后才退分析结论的结果表）；指标和字段本身就超（极少见）
     才从尾部截断。
     """
-    analyses = analyses or _AnalysisLines([], {}, {})
+    analyses = analyses or _AnalysisLines([], {})
     if tail and len("\n".join(tail)) > budget // _PROMPT_TAIL_SHARE:
         tail = _cut_lines(tail, budget // _PROMPT_TAIL_SHARE, "…（表和字段、知识库检索太长，后面的省略了）")
     tables: dict[str, list[str]] = {}
@@ -4493,9 +4518,7 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str],
         notice = (f"注意：证据太多，这份目录有省略（篇幅上限约 {budget} 字）。共 {len(queries)} 次查询："
                   f"{n['full']} 次列出前 {_PROMPT_ROWS} 行，{n['compact']} 次只列前 {_PROMPT_COMPACT_ROWS} 行，"
                   f"{n['omitted']} 次只列编号和摘要。各来源：{per}。")
-        if bare := sum(1 for alias in analyses.order if tier[alias] == "compact"):
-            notice += (f"另有 {len(analyses.order)} 条分析结论，判定和具名的数都列出了，其中 {bare} 条没列出结果表的各段"
-                       "（要列出就单独一行写整表引用，系统照结果表生成）。")
+        notice += _analysis_notice(Counter(tier[alias] for alias in analyses.order), len(analyses.order))
         body = [*head, notice, OMITTED_RULES, *first, *analyses.lines(tier), *rules]
         for r, entries in zip(stats, groups.values()):
             body.append(f"\n来自「{r['title']}」的查询（共 {r['total']} 次，展开 {r['full'] + r['compact']} 次）：")
@@ -4510,14 +4533,24 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str],
     def size(lines: list[str]) -> int:
         return sum(len(line) + 1 for line in lines)
 
+    # 分析结论一段最多占预算的 1/_PROMPT_ANALYSIS_SHARE：超了按结论的先后从后往前降，先只留判定语和具名的数，再只列编号
+    cap = budget // _PROMPT_ANALYSIS_SHARE
+    for level in ("brief", "head"):
+        for alias in reversed(analyses.order):
+            if size(analyses.lines(tier)) <= cap:
+                break
+            tier[alias] = level
     # 省略情况那句里的数字会变长几位：留一点余量，最后再整份量一次
     remaining = budget - len(assemble()) - 8 * (len(groups) + 3)
     upgraded: list[tuple[str, str]] = []
+    listed = size(analyses.lines(tier))
     for alias in analyses.order:
-        # 分析结论的结果表先补，排在查询的行前面；整份量下来超了，最后才退回它们
-        if (delta := size(analyses.full[alias]) - size(analyses.compact[alias])) <= remaining:
+        # 分析结论的结果表先补，排在查询的行前面（这一段照样不超过上限）；整份量下来超了，最后才退回它们
+        delta = size(analyses.shape(alias, "full")) - size(analyses.shape(alias, "compact"))
+        if tier[alias] == "compact" and delta <= remaining and listed + delta <= cap:
             tier[alias] = "full"
             remaining -= delta
+            listed += delta
             upgraded.append((alias, "compact"))
     for alias in pinned:
         if tier.get(alias) == "brief" and (delta := size(compact[alias]) - size(brief[alias])) <= remaining:
@@ -4551,6 +4584,29 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str],
                          omitted=[e["alias"] for e in queries if tier[e["alias"]] == "brief"], empty_sources=empty)
 
 
+#: 目录有省略、分析结论只是结果表没列各段时，开头那句里接的说明（给模型看）
+ANALYSIS_TABLES_OMITTED_PROMPT = ("另有 {total} 条分析结论，判定和具名的数都列出了，其中 {bare} 条没列出结果表的各段"
+                                  "（要列出就单独一行写整表引用，系统照结果表生成）。")
+#: 分析结论还降到了只留判定语和具名的数、只列编号时接的说明（给模型看）：各样子各有几条（ANALYSIS_LEVEL_PROMPT_WORDS），
+#: 以及没列出的怎么引用
+ANALYSIS_TRIMMED_PROMPT = ("另有 {total} 条分析结论，这一段最多占篇幅的一半：{parts}。没列出的也能引用：判定语单独写 "
+                           "[[a:编号]]，系统照结论渲染；要列出结果表就单独一行写整表引用，系统照结果表生成；没列出的具名的数"
+                           "不要编造。")
+ANALYSIS_LEVEL_PROMPT_WORDS = (("full", "列出了结果表的前几段"), ("compact", "没列出结果表的各段"),
+                               ("brief", "只列了判定语和具名的数（判定标准、附带说明没列）"), ("head", "只列了编号"))
+
+
+def _analysis_notice(levels: Counter, total: int) -> str:
+    """目录有省略时开头那句里分析结论的情况：都列全了的不说；只是结果表没列各段的照旧一句；还降到了只留判定语和具名的数、
+    只列编号的，照实写各有几条，以及没列出的怎么引用。"""
+    if not total or levels["full"] == total:
+        return ""
+    if not levels["brief"] and not levels["head"]:
+        return ANALYSIS_TABLES_OMITTED_PROMPT.format(total=total, bare=levels["compact"])
+    parts = "，".join(f"{levels[level]} 条{word}" for level, word in ANALYSIS_LEVEL_PROMPT_WORDS if levels[level])
+    return ANALYSIS_TRIMMED_PROMPT.format(total=total, parts=parts)
+
+
 def catalog_prompt(catalog: dict[str, Any], *, budget: int = DEFAULT_CATALOG_BUDGET, cells_allowed: bool = True,
                    loader: Loader | None = None, fields: list[dict[str, Any]] | None = None,
                    titles: dict[str, str] | None = None) -> str:
@@ -4748,7 +4804,11 @@ _DECOMPOSE_NUMBERS = ("base", "target", "delta", "pct", "share", "contrib", "fla
 
 def _decompose_row(result: dict[str, Any]) -> tuple[str, list[Any]] | None:
     """变化拆解的结果表里判定点名的那一段：(这一段的叫法, 那一行)。段按维度列（params.by；乘积模式是 factor 一列）对
-    verdict.subjects 的第一项，对不上就取第一行（结果表按变化的绝对值从大到小排）。叫法是维度值用「·」连起来。"""
+    verdict.subjects 的第一项，对不上就取第一行（结果表按变化的绝对值从大到小排）。维度值和叫法都按算子写段的同一个函数
+    （app.analysis.phrases.key_text：空值写「（空）」、布尔写「是 / 否」，两列用「·」连）：subjects 就是这么写的，拿原值的
+    str 去比，空值、布尔维度的段对不上，代入式会退回第一行。"""
+    from app.analysis.phrases import key_text
+
     table = result.get("table") if isinstance(result.get("table"), dict) else {}
     rows = [r for r in table.get("rows") or [] if isinstance(r, list)]
     columns = [str(c) for c in table.get("columns") or []]
@@ -4765,8 +4825,8 @@ def _decompose_row(result: dict[str, Any]) -> tuple[str, list[Any]] | None:
     picked = rows[0]
     if subjects and at and isinstance(subjects[0], list):
         wanted = [str(v) for v in subjects[0]]
-        picked = next((row for row in rows if [str(row[i]) for i in at if i < len(row)] == wanted), rows[0])
-    name = "·".join(str(picked[i]) for i in at if i < len(picked) and picked[i] is not None)
+        picked = next((row for row in rows if [key_text([row[i]]) for i in at if i < len(row)] == wanted), rows[0])
+    name = key_text([picked[i] for i in at if i < len(picked)]) if at else ""
     return name or "这一段", picked
 
 

backend/app/engine/nodes/metrics.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import ast
+import asyncio
 import json
 import math
 import re
@@ -56,8 +57,9 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
     取数的那条 SQL 对照数据目录查出 error 级问题的（查询快照的 checks，见 data/sqlcheck.py：一对多关联后重复计算、
     存量跨期加总），同样值照算，标 sql_check_failed 并写明原因，出具契约据此降档（issuance.sql_check_gaps）。
 
-    一个指标的输入落在两份不同的查询快照上、两份 SQL 同形只差字面量（本周对上周）时，自动做一次可比性检查（分析算子
-    comparable），结论落成工件、记在指标的 comparability 上（_comparability）。做不成不挡口径卡，照实记一条日志。
+    一个指标的输入落在两份不同的查询快照上、两份 SQL 同形只差字面量、不同的字面量里有日期或期次（本周对上周）时，自动做
+    一次可比性检查（分析算子 comparable），结论落成工件、记在指标的 comparability 上（_comparability）。只差地区、门店号这类
+    字面量的（华东对华南）是两段不是两期，不查。做不成不挡口径卡，照实记一条日志。
     """
     definitions = ctx.cfg("metrics", []) or []
     caliber_raw, caliber_version = ctx.cfg("caliber", "") or ctx.node.title, str(ctx.cfg("caliber_version", "") or "v1")
@@ -475,11 +477,15 @@ def _rendered(value: Any, definition: dict[str, Any], display: dict[str, Any]) -
 # 可比性检查（docs/design/analysis-round/00-总方案.md §3.4 第 3 条）
 # --------------------------------------------------------------------------
 
-#: 可比性检查没做成（探查失败、算子出错、还没有算子……）时那条日志的 code
+#: 可比性检查没做成（快照读不出来、探查失败、算子出错、算不了……）时那条日志的 code
 COMPARABILITY_SKIPPED = "metric_comparability_skipped"
 #: 那条日志的说法
 COMPARABILITY_SKIPPED_TEXT = "指标「{name}」用到的两次查询没做成可比性检查：{why}。指标照常计算，两期能不能直接比请自行核对"
-_DAY = re.compile(r"^(\d{4})-?(\d{2})-?(\d{2})(?:[ T].*)?$")
+#: 期次的几种写法（_period_key，排先后用）：月 2026-09、202609，周 2026-W41，季 2026Q3、2026-Q3，年 2026
+_MONTH_TEXT = re.compile(r"^(\d{4})-?(0[1-9]|1[0-2])$")
+_WEEK_TEXT = re.compile(r"^(\d{4})-?[Ww](\d{1,2})$")
+_QUARTER_TEXT = re.compile(r"^(\d{4})-?[Qq]([1-4])$")
+_YEAR_TEXT = re.compile(r"^(?:19|20|21)\d{2}$")
 
 
 class _Comparing:
@@ -494,13 +500,15 @@ class _Comparing:
 async def _comparability(metric_id: str, name: str, inputs: list[dict[str, Any]], ctx: NodeContext,
                          comparing: _Comparing, snapshots: dict[str, Any]) -> dict[str, Any] | None:
     """指标的可比性（contract.MetricComparability：结论工件、判定、判定语、原因）。只在这种时候做：输入里取自查询快照的格
-    （cell() 取的格、按位置取的格、Agent 按出处交来的字段）恰好落在两份不同的快照上，两份 SQL 同形、只差字面量，而且查的是
-    同一个数据源——本周对上周。SQL 结构不同的（退款额 / 销售额）不是同一个量的两期，不查。
+    （cell() 取的格、按位置取的格、Agent 按出处交来的字段）恰好落在两份不同的快照上，两份 SQL 同形、只差字面量，查的是同一个
+    数据源，而且不同的字面量里至少有一处两边都是日期或期次（_dated_pairs：时间条件的界、像日期的常量）——本周对上周。SQL 结构不同的
+    （退款额 / 销售额）不是同一个量的两期，不查；只差地区、门店号这类字面量的（华东对华南、门店 1203 对 1204）是两段，也
+    不查。日期和别的筛选都不同的照查，算子判不可比（两边除了日期，筛选的还不一样）。
 
-    哪份是基期：两边第一处不同的字面量是日期（或同样写法的期次）的，早的是基期；认不出的，表达式里先出现的那份算本期
-    （「(本周 - 上周) / 上周」的写法）。
+    哪份是基期：不同的日期、期次里第一处两边写法相同、排得出先后的（真正的日期，或同一种写法的月、周、季、年），早的是基期；
+    排不出的（按运行当天现算的界 date('now', '-7 days')），表达式里先出现的那份算本期（「(本周 - 上周) / 上周」的写法）。
 
-    做不成（快照读不出来、探查失败、算子出错、算不了、还没有算子）不挡口径卡：照实记一条日志（COMPARABILITY_SKIPPED），
+    做不成（快照读不出来、探查失败、算子出错、算不了）不挡口径卡：照实记一条日志（COMPARABILITY_SKIPPED），
     这个指标不写 comparability。同一对快照只查一次。"""
     from app.engine.lineage import SNAPSHOT_VIAS
 
@@ -526,7 +534,8 @@ async def _comparability(metric_id: str, name: str, inputs: list[dict[str, Any]]
 
 
 class _NotPeriods(Exception):
-    """两份快照不是同一个量的两期（SQL 结构不同、数据源不同、字面量一样、没有 SQL）：不查，也不记日志。"""
+    """两份快照不是同一个量的两期（SQL 结构不同、数据源不同、字面量一样、不同的字面量里没有日期或期次、没有 SQL）：不查，
+    也不记日志。"""
 
 
 class _Skipped(Exception):
@@ -551,7 +560,8 @@ async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]],
     for artifact in pair:
         if artifact not in snapshots:
             try:
-                snapshots[artifact] = artifact_store.load(artifact)
+                # 快照可能有几 MB：读文件、复验哈希放进线程，不占着事件循环（同 tools/analysis.prepare）
+                snapshots[artifact] = await asyncio.to_thread(artifact_store.load, artifact)
             except (ValueError, OSError):
                 snapshots[artifact] = None
         if not isinstance(snapshots[artifact], dict):
@@ -565,7 +575,10 @@ async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]],
     norms = [normalize_sql(q, dialect) for q in sqls]
     if not norms[0].same_shape(norms[1]) or norms[0].literals == norms[1].literals:
         raise _NotPeriods()
-    first, second = _period_order(norms[0].literals, norms[1].literals)
+    dated = _dated_pairs(sqls[0], sqls[1], dialect)
+    if not dated:
+        raise _NotPeriods()                       # 不同的字面量里没有日期或期次：比的是两段，不是两期
+    first, second = _period_order(dated)
     order = (pair[first], pair[second])           # (基期, 本期)
     refs: dict[str, str] = {}
     for role, artifact in zip(("base", "target"), order):
@@ -608,20 +621,55 @@ async def _dialect_of(source: str, comparing: _Comparing) -> str | None:
     return comparing.kinds[source]
 
 
-def _period_order(a: tuple[str, ...], b: tuple[str, ...]) -> tuple[int, int]:
-    """两份同形 SQL 的字面量 → (基期的下标, 本期的下标)。第一处不同、两边都认得出的日期（YYYY-MM-DD、YYYYMMDD，带不带
-    时间）或同样长短的期次写法（2026-W41、202610）定先后，早的是基期；都认不出的，表达式里先出现的那份（下标 0）是本期。"""
-    for x, y in zip(a, b):
-        if x == y:
-            continue
-        left, right = x.strip("'\""), y.strip("'\"")
-        dx, dy = _DAY.match(left), _DAY.match(right)
-        if dx and dy:
-            if dx.groups() != dy.groups():
-                return (0, 1) if dx.groups() < dy.groups() else (1, 0)
-            continue
-        if len(left) == len(right) and re.fullmatch(r"[\dW\-/Qq]+", left) and re.fullmatch(r"[\dW\-/Qq]+", right):
-            return (0, 1) if left < right else (1, 0)
+def _dated_pairs(a: str, b: str, dialect: str | None) -> list[tuple[str, str]]:
+    """两份同形 SQL 里不同的字面量中，两边都是日期或期次的那几对：[(左边的值, 右边的值)]（不带引号），按语法树里的先后。
+    一对都没有（只差地区、门店号这类字面量），或者解析不了、两边的字面量对不上的，为空。
+
+    日期、期次的认法和可比性算子「同口径」一项同一套（app/analysis/comparable 的 _undated）：data/coverage 认出的时间条件的
+    界（整个界的表达式：day BETWEEN '2026-10-12' AND …、month = '2026-09'、yr = 2026、week = '2026-W41'，按运行当天现算的
+    date('now', '-7 days') 里的字面量也算），加上别处像日期的常量（coverage.literal_bound 认得出的真正的日期）。两边同形，
+    语法树一样，字面量按同一个先后一一对上。"""
+    from sqlglot import exp
+
+    from app.data import coverage
+
+    sides: list[list[tuple[Any, bool]]] = []
+    for sql in (a, b):
+        tree = coverage.parse_sql(sql, dialect)
+        if tree is None:
+            return []
+        kept = {id(n) for flt in coverage.filters(tree) for _, bound, _ in flt.conds for n in bound.walk()}
+        sides.append([(lit, id(lit) in kept or coverage.literal_bound(lit)[1] is not None)
+                      for lit in tree.find_all(exp.Literal)])
+    if len(sides[0]) != len(sides[1]):
+        return []
+    return [(str(x.this), str(y.this)) for (x, x_dated), (y, y_dated) in zip(*sides)
+            if x_dated and y_dated and (x.this, x.is_string) != (y.this, y.is_string)]
+
+
+def _period_key(text: str) -> tuple[str, tuple[int, ...]] | None:
+    """一个日期、期次字面量（去掉引号）排先后用的 (写法, 键)：真正的日期（YYYY-MM-DD、YYYYMMDD，带不带时间）按那一天，
+    月、周、季、年按 (年, 第几)。认不出的（-7 days 这类偏移量、序号）为 None：只有写法相同的两个才比先后。"""
+    from app.data import coverage
+
+    day = coverage.as_day(text)
+    if day is not None:
+        return "day", (day.day.toordinal(),)
+    for shape, pattern in (("month", _MONTH_TEXT), ("week", _WEEK_TEXT), ("quarter", _QUARTER_TEXT)):
+        if found := pattern.match(text):
+            return shape, (int(found.group(1)), int(found.group(2)))
+    if _YEAR_TEXT.match(text):
+        return "year", (int(text),)
+    return None
+
+
+def _period_order(pairs: list[tuple[str, str]]) -> tuple[int, int]:
+    """两份同形 SQL 里不同的日期、期次（_dated_pairs）→ (基期的下标, 本期的下标)。第一对两边写法相同、排得出先后的
+    （_period_key）定先后，早的是基期；都排不出的，表达式里先出现的那份（下标 0）是本期。"""
+    for x, y in pairs:
+        kx, ky = _period_key(x.strip()), _period_key(y.strip())
+        if kx is not None and ky is not None and kx[0] == ky[0] and kx[1] != ky[1]:
+            return (0, 1) if kx[1] < ky[1] else (1, 0)
     return 1, 0
 
 

backend/app/report_export/runtime.js

@@ -95,8 +95,10 @@
     calc: '计算', dValues: '具名的数', dFormulas: '关键数的代入式', dResultTable: '结果表',
     resultRows: (shown, total) => `列出 ${shown} 行，共 ${total} 行`,
     analysisInput: '输入', dRole: '角色', dRef: '调用里写的名字', dColumnsUsed: '用到的列', dRowsUsed: '用到的行',
-    allRows: '全部行', dHeadline: '总数那一格', dClosure: '闭合核对',
+    allRows: '全部行', dHeadline: '总数那一格', dHeadlineRatio: '整体比率那一格', dClosure: '闭合核对',
     closureOk: (h, t) => `总数 ${h} 与各段之和 ${t} 闭合`, closureBad: (h, t) => `总数 ${h} 与各段之和 ${t} 对不上`,
+    ratioClosureOk: (h, t) => `整体比率 ${h} 与各段分子之和 ÷ 分母之和 ${t} 闭合`,
+    ratioClosureBad: (h, t) => `整体比率 ${h} 与各段分子之和 ÷ 分母之和 ${t} 对不上`,
     dParent: '下钻自', recompute: '复算', seal: '封存', inputOfAnalysis: (alias) => `分析结论 ${alias} 的输入`,
     inLedger: '封存范围内的台账记过这份分析结论', notInLedger: '封存范围内的台账里没有这份分析结论',
     viaCard: '记在口径卡的指标集里，随卡封存（不进台账）',
@@ -541,10 +543,13 @@
     return h('section', { class: 'part', 'data-step': 'analysis_calc' }, h('h3', {}, T.calc), h('div', { class: 'part-body' }, ...body))
   }
 
-  /** 分析结论 · 输入：角色、落到哪个 Q、用到的列和行、总数那一格和闭合核对、下钻的上一层 */
+  /** 分析结论 · 输入：角色、落到哪个 Q、用到的列和行、总数（比率拆解是整体比率）那一格和闭合核对、下钻的上一层 */
   function analysisInputPart(step) {
     const rows = Array.isArray(step.rows) ? step.rows.map((r) => `第 ${Number(r) + 1} 行`).join('、') : step.rows === 'all' ? T.allRows : null
     const closure = isObj(step.closure) ? step.closure : null
+    const ratio = closure !== null && closure.mode === 'ratio'
+    const closed = closure && (ratio ? (closure.closed ? T.ratioClosureOk : T.ratioClosureBad)
+      : (closure.closed ? T.closureOk : T.closureBad))
     const body = [kv([
       [T.dRole, String(step.role_label || step.role || NONE)],
       step.ref ? [T.dRef, h('code', {}, String(step.ref))] : null,
@@ -553,8 +558,8 @@
       step.node_id ? [T.dNode, nodeLabel(step.node_id)] : null,
       Array.isArray(step.columns) && step.columns.length ? [T.dColumnsUsed, step.columns.join('、')] : null,
       rows ? [T.dRowsUsed, rows] : null,
-      'value' in step ? [T.dHeadline, `${step.column ? `第 ${Number(step.row) + 1} 行「${step.column}」 = ` : ''}${valueText(step.value)}`] : null,
-      closure ? [T.dClosure, closure.closed ? T.closureOk(closure.headline, closure.total) : T.closureBad(closure.headline, closure.total)] : null,
+      'value' in step ? [ratio ? T.dHeadlineRatio : T.dHeadline, `${step.column ? `第 ${Number(step.row) + 1} 行「${step.column}」 = ` : ''}${valueText(step.value)}`] : null,
+      closure ? [T.dClosure, closed(closure.headline, closure.total)] : null,
       step.phrase ? [T.dPhrase, String(step.phrase)] : null,
     ])]
     const bits = integrity(step)

backend/app/tools/analysis.py

@@ -67,8 +67,9 @@ class AnalysisScope:
 
     queries：编号 Qn → 输入快照（contract.InputSnapshot；记台账时先放交回模型的那份 payload，分析前 prepare 换成工件库里
     那份完整快照，补上遮罩列）。results：编号 An → 整份结论；artifacts：An → 结论的工件 id。snapshot：发起时的分析设置快照
-    （没有为 None，现读设置）。mask_cache：快照工件 → 遮罩列；source_cache：数据源和表 → 目录、查询用的源；probe_cache：
-    数据源名 → coverage.Cache（一次节点执行里各读、各查一次，只在工具执行时用，不影响编号）。"""
+    （没有为 None，现读设置）。mask_cache：快照工件 → 遮罩列；source_cache：数据源和表 → 目录、查询用的源；tables_cache：
+    SQL → 它用到的表（每次分析调用都要按本节点全部查询认表，同一条 SQL 只解析一次）；probe_cache：数据源名 →
+    coverage.Cache（一次节点执行里各读、各查一次，只在工具执行时用，不影响编号）。"""
 
     node_id: str
     snapshot: dict[str, Any] | None = None
@@ -77,6 +78,7 @@ class AnalysisScope:
     artifacts: dict[str, str] = field(default_factory=dict)
     mask_cache: dict[str, list[str]] = field(default_factory=dict)
     source_cache: dict[str, Any] = field(default_factory=dict)
+    tables_cache: dict[str, tuple[str, ...]] = field(default_factory=dict)
     probe_cache: dict[str, Any] = field(default_factory=dict)
     #: 已经换成工件库里那份完整快照的编号（prepare 每份只取回一次）
     prepared: set[str] = field(default_factory=set)
@@ -198,27 +200,33 @@ async def prepare(scope: AnalysisScope) -> None:
 # --------------------------------------------------------------------------
 
 
-def _tables_of(snapshots: Iterable[Mapping[str, Any]]) -> dict[str, set[str]]:
-    """数据源名 → 这几份快照的 SQL 里用到的表（evidence.sql_tables，去掉 schema 前缀前的写法）。合并查询不是数据源。"""
+def _tables_of(snapshots: Iterable[Mapping[str, Any]],
+               memo: dict[str, tuple[str, ...]] | None = None) -> dict[str, set[str]]:
+    """数据源名 → 这几份快照的 SQL 里用到的表（evidence.sql_tables，去掉 schema 前缀前的写法）。合并查询不是数据源。
+    memo 是 SQL → 表的缓存（调用方的，同一个节点里分析好几次，每条 SQL 只解析一次）。"""
     from app.engine.evidence import sql_tables
     from app.engine.merge_query import MERGE_SOURCE
 
+    memo = memo if memo is not None else {}
     out: dict[str, set[str]] = {}
     for snap in snapshots:
         name = str(snap.get("source") or "")
         if not name or name == MERGE_SOURCE:
             continue
-        try:
-            tables = sql_tables(str(snap.get("sql") or ""))
-        except Exception:  # noqa: BLE001 - 认不出表就只是少给一点目录
-            tables = []
-        out.setdefault(name, set()).update(str(t) for t in tables)
+        sql = str(snap.get("sql") or "")
+        if sql not in memo:
+            try:
+                memo[sql] = tuple(str(t) for t in sql_tables(sql))
+            except Exception:  # noqa: BLE001 - 认不出表就只是少给一点目录
+                memo[sql] = ()
+        out.setdefault(name, set()).update(memo[sql])
     return out
 
 
 async def context_of(snapshots: Iterable[Mapping[str, Any]], *, tool_ctx: Any,
                      cache: dict[str, Any] | None = None,
-                     probe_cache: dict[str, Any] | None = None) -> contract.Context:
+                     probe_cache: dict[str, Any] | None = None,
+                     tables_cache: dict[str, tuple[str, ...]] | None = None) -> contract.Context:
     """算子要的运行期依赖（contract.Context）。每次分析调用一份（探查器有总时限，一次调用建一个）。
 
     - today：运行当天，服务端时区（算子据此判本期完不完整）
@@ -229,15 +237,16 @@ async def context_of(snapshots: Iterable[Mapping[str, Any]], *, tool_ctx: Any,
       遮罩列按数据源设的不探，探不成由算子退回剖析事实；一个源都解析不出来时为 None
 
     tool_ctx 是查询用的那个 ToolContext（固定的数据版本、目录视图都从它取）。cache 是调用方的缓存（同一个节点里分析好几次，
-    目录、数据源读一次）；probe_cache 是数据源名 → coverage.Cache（同一张表的范围一个节点里只查一次）。读不到的源当没有：
-    目录、剖析、探查都是附加的，不能挡住分析。"""
+    目录、数据源读一次）；tables_cache 是 SQL → 用到的表（同一条 SQL 一个节点里只解析一次：agent 每次分析都按本节点全部
+    查询认表）；probe_cache 是数据源名 → coverage.Cache（同一张表的范围一个节点里只查一次）。读不到的源当没有：目录、剖析、
+    探查都是附加的，不能挡住分析。"""
     from app.analysis.probe import profile_facts
 
     cache = cache if cache is not None else {}
     catalogs: dict[str, Any] = {}
     facts: dict[str, Any] = {}
     sources: dict[str, Any] = {}
-    for name, tables in _tables_of(snapshots).items():
+    for name, tables in _tables_of(snapshots, tables_cache).items():
         key = f"{name}\n" + "\n".join(sorted(tables))
         if key not in cache:
             cache[key] = await _source_context(name, tables, tool_ctx)

backend/app/tools/builtin/analysis.py

@@ -194,7 +194,7 @@ async def run_tool(op: str, ctx: ToolContext, args: BaseModel) -> dict[str, Any]
     request = request_of(op, args)
     await prepare(scope)
     context = await context_of(list(scope.queries.values()), tool_ctx=ctx, cache=scope.source_cache,
-                               probe_cache=scope.probe_cache)
+                               probe_cache=scope.probe_cache, tables_cache=scope.tables_cache)
     result = await analysis_ops.run(request, scope.resolve, settings=settings, context=context, parents=scope.parent)
     artifact = await store(result, run_id=ctx.run_id, node_id=scope.node_id)
     return {"analysis_artifact": artifact, "result": result}

backend/tests/conftest.py

@@ -111,6 +111,24 @@ def pytest_collection_modifyitems(config, items) -> None:
             item.add_marker(skip)
 
 
+@pytest.fixture
+async def scratch_source():
+    """测试里建的 SQLite 数据源：source_id = await scratch_source("shop", 库文件路径, options={...})。同名的旧源先删掉（连同
+    它的数据目录、剖析和现查缓存，tests/fixtures/sources.sqlite_source），测试结束时再删一遍、清掉现查缓存（forget_source）。
+    分析相关的用例都建名叫 shop 的源：别的用例留下的数据目录、5 分钟内现查过的表范围会改掉这次的判定。"""
+    from tests.fixtures.sources import forget_source, sqlite_source
+
+    made: list[str] = []
+
+    async def make(name: str, path, **kw) -> str:
+        made.append(await sqlite_source(name, path, **kw))
+        return made[-1]
+
+    yield make
+    for source_id in made:
+        await forget_source(source_id)
+
+
 @pytest.fixture(scope="session")
 def scenic_db(tmp_path_factory) -> str:
     """合成的景区业务库（tests/fixtures/catalog/scenic.py）：50 张表、真外键与只能靠命名推断的关系并存。

backend/tests/fixtures/analysis_ops.py

@@ -1,4 +1,5 @@
-"""测试用的分析算子替身（第 1 波各包并行：W1-a 的 app.analysis.run 落地前，入口的测试不等它）。
+"""测试用的分析算子替身：入口（agent 分析工具、分析卡、口径卡的可比性挂钩）的测试用它断言入口交给算子的东西、算子交回后
+入口做的事，不受算子的判定细节牵连。接真算子的端到端见 tests/test_analysis_e2e*.py。
 
 FakeOps 换掉模块上的 app.analysis.run（入口都按模块取 run），按请求交回 app.analysis.testing.fake_result 的假结论，
 并记下每次调用拿到的请求、按名字解析到的输入快照、阈值、运行期依赖、上一层结论——测试据此断言入口交进去的东西。

backend/tests/fixtures/sources.py

@@ -3,9 +3,15 @@
 数据源表在每个测试进程的测试库里是共用的：留下来的源会出现在助手的数据源清单里（50 张表的合成库还会
 触发挑表、多一次模型调用），别的用例的结果就跟着执行顺序变。阶段 1B 就是这样被绊过一次。
 删数据源时它的数据目录随外键级联删除（测试库开着 PRAGMA foreign_keys=ON）。
+
+分析相关的用例都建名叫 shop 的源（sqlite_source / forget_source，tests/conftest.py 的 scratch_source 夹具按它们建、按它们删）：
+别的用例留下的数据目录（金额标成了流量，窗口长度不同就从有条件可比变成不可比）、coverage 的现查缓存里 5 分钟内按同一个
+源 id 记下的表范围（另一份库的日期范围），都会改掉这次的判定。所以建之前先把同名的旧源连同这两样删干净，用完再删一遍。
 """
 from __future__ import annotations
 
+from typing import Any
+
 from sqlalchemy import delete, select
 
 from app.data.engine import engines
@@ -26,6 +32,42 @@ async def drop_source(source_id: str | None = None, *, name: str | None = None)
     await engines.invalidate(source_id)
 
 
+async def sqlite_source(name: str, path: Any, *, options: dict[str, Any] | None = None, schema: bool = False) -> str:
+    """建一个指向这份 SQLite 库文件的只读数据源（启用），交回 id。同名的旧源先删掉（forget_source：连同它的数据目录、剖析
+    和现查缓存），这次的源是新的 id。schema 为真时顺手探好表结构（要写数据目录的用例用）。用完调 forget_source。"""
+    await forget_source(name=name)
+    async with SessionLocal() as session:
+        row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True,
+                         options=dict(options or {}))
+        session.add(row)
+        await session.flush()
+        if schema:
+            from app.data.introspect import introspect
+
+            row.schema_cache = await introspect(row)
+        await session.commit()
+        source_id = row.id
+    await engines.invalidate(source_id)
+    return source_id
+
+
+async def forget_source(source_id: str | None = None, *, name: str | None = None) -> None:
+    """按 id 或名字删一个数据源（drop_source：数据目录、剖析随外键级联删除，连接池作废），再清掉 coverage 现查缓存里它的
+    表范围（data/coverage 的 _LIVE，按数据源 id 记、5 分钟有效：同一个进程里后面的用例会拿到这份旧范围）。不存在就什么都
+    不做。"""
+    from app.data import coverage
+
+    if source_id is None:
+        async with SessionLocal() as session:
+            source_id = (await session.execute(
+                select(DataSource.id).where(DataSource.name == name))).scalar_one_or_none()
+        if source_id is None:
+            return
+    await drop_source(source_id)
+    for key in [k for k in coverage._LIVE if k and k[0] == source_id]:
+        coverage._LIVE.pop(key, None)
+
+
 async def introspect_now(client, source_id: str, *, expect: int = 202) -> dict:
     """探查结构并等它做完（探查是后台任务：POST /introspect 回 202 和任务，next-round 03 A1），返回数据源此刻的样子
     （DataSourceOut）。expect 不是 202 时只发请求、断言状态码，返回响应体。"""

backend/tests/test_analysis_bundle.py

@@ -106,6 +106,8 @@ def test_the_analysis_and_its_input_travel_with_the_file_and_verify():
     # 导出时服务端真的复算了一遍：按封存的输入快照（这份文件里哈希对得上的原字节）重算，逐字段相同
     assert analysis_item["recheck"] == {"status": "same", "method": "decompose.additive@1",
                                         "text": "复算一致：按封存的输入重新计算，结论逐项相同"}
+    # 方法的叫法由服务端带上（和导出页同一份），核对页只显示「变化拆解（可加） · decompose.additive@1」
+    assert analysis_item["method_label"] == "变化拆解（可加）"
     [query] = [x for x in bundle["artifacts"] if x["kind"] == "query"]
     assert "b64" in query
 
@@ -259,7 +261,7 @@ def test_the_comparability_on_a_card_travels_with_the_card_and_is_chained_throug
     assert "b64" in item(bundle, card_id)
     comparability = item(bundle, found)
     assert comparability["kind"] == ARTIFACT_KIND and "b64" in comparability and comparability["node_id"] == "card"
-    assert comparability["recheck"]["status"] == "same"
+    assert comparability["recheck"]["status"] == "same" and comparability["method_label"] == "可比性检查"
     got = verdicts(bundle)
     assert all(r == "pass" for key in ("events", "manifest", "artifacts", "chain") for r in got[key]), got["why"]
     assert {w["kind"] for w in got["why"]["chain"]} >= {"metric_set", ARTIFACT_KIND, "query"}

backend/tests/test_analysis_card.py

@@ -1,7 +1,8 @@
 """分析卡（节点类型 analysis，docs/design/analysis-round/00-总方案.md §3.2）：画图时的校验、执行、产出的变量、受管规则、
 登记处（叫法、变量表、助手查看、目录影响面、升级提醒）。运行对比按项对齐见 tests/test_analysis_compare.py。
 
-算子（app.analysis.run）换成按请求交回假结论的替身（tests/fixtures/analysis_ops.FakeOps）。库是虚构的零售库 shop。
+算子（app.analysis.run）换成按请求交回假结论的替身（tests/fixtures/analysis_ops.FakeOps）；样例里的下钻那一项另用真算子
+跑一遍（替身不核参数）。库是虚构的零售库 shop：orders(region, week, amount, channel)，华东、华南、华北两周。
 """
 from __future__ import annotations
 
@@ -14,9 +15,8 @@ from sqlalchemy import select
 from app.analysis import contract
 from app.analysis.contract import CallError
 from app.core import artifact_store
-from app.data.engine import engines
 from app.db.base import SessionLocal
-from app.db.models import DataSource, Run, RunEvent
+from app.db.models import Run, RunEvent
 from app.engine.runner import run_manager
 from app.engine.schema import GraphSpec, NodeType, validate_graph
 from tests.fixtures.analysis_ops import FakeOps
@@ -24,16 +24,22 @@ from tests.fixtures.analysis_ops import FakeOps
 TOOL = "db_query__shop"
 WEEKLY = "SELECT week, SUM(amount) AS gmv FROM orders GROUP BY week ORDER BY week"
 REGION = "SELECT region, week, SUM(amount) AS gmv FROM orders GROUP BY region, week ORDER BY region, week"
-ROWS = [(1, "华东", "2026-W41", 820.0), (2, "华东", "2026-W42", 700.0), (3, "华南", "2026-W41", 700.0),
-        (4, "华南", "2026-W42", 666.0), (5, "华北", "2026-W41", 600.0), (6, "华北", "2026-W42", 560.0)]
+#: 各地区两周的销售额：华东 820 → 700（线上 500 → 420、线下 320 → 280），华南 700 → 666，华北 600 → 560
+ROWS = [(1, "华东", "2026-W41", 500.0, "线上"), (2, "华东", "2026-W41", 320.0, "线下"), (3, "华东", "2026-W42", 420.0, "线上"),
+        (4, "华东", "2026-W42", 280.0, "线下"), (5, "华南", "2026-W41", 700.0, "线上"), (6, "华南", "2026-W42", 666.0, "线上"),
+        (7, "华北", "2026-W41", 600.0, "线下"), (8, "华北", "2026-W42", 560.0, "线下")]
+#: 华东按渠道（DRILL 的输入）：两个渠道之和 820 → 700，和 WHERE 里华东那一段闭合
+EAST = ("SELECT channel, week, SUM(amount) AS gmv FROM orders WHERE region = '华东' GROUP BY channel, week "
+        "ORDER BY channel, week")
 WOW = {"id": "wow", "op": "baseline", "from": "weekly", "params": {"period": "week", "measure": "gmv",
                                                                      "label": "销售额"}}
 WHERE = {"id": "where", "op": "decompose", "mode": "additive", "from": "by_region",
          "params": {"by": ["region"], "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv",
                     "headline": {"base": "weekly.r0.gmv", "target": "weekly.r1.gmv"}, "label": "销售额"},
          "hypothesis": "下降是不是集中在某个地区"}
-DRILL = {"id": "east", "op": "decompose", "mode": "additive", "from": "by_region",
-         "params": {"by": ["week"], "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv",
+#: 下钻：WHERE 里华东那一段按渠道再拆一层，读华东按渠道的那次查询（drill_graph 里的 east_q）
+DRILL = {"id": "east", "op": "decompose", "mode": "additive", "from": "east_q",
+         "params": {"by": ["channel"], "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv",
                     "parent": {"of": "where", "key": ["华东"]}}}
 
 
@@ -44,30 +50,20 @@ async def engine_up():
     await run_manager.shutdown()
 
 
-async def _shop(tmp_path, options=None):
+async def _shop(tmp_path, scratch_source, options=None):
+    """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，测试结束时再删）。"""
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
-    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, week TEXT, amount REAL);")
-    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", ROWS)
+    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, week TEXT, amount REAL, channel TEXT);")
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", ROWS)
     db.commit()
     db.close()
-    async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, options or {}
-        await session.commit()
-        source_id = row.id
-    await engines.invalidate(source_id)
-    return source_id
+    return await scratch_source("shop", path, options=options)
 
 
 @pytest.fixture
-async def shop(tmp_path):
-    source_id = await _shop(tmp_path)
-    yield
-    await engines.invalidate(source_id)
+async def shop(tmp_path, scratch_source):
+    return await _shop(tmp_path, scratch_source)
 
 
 def node(nid, ntype, **config):
@@ -90,6 +86,12 @@ def card_graph(items=(WOW, WHERE), *, extra_nodes=(), extra_edges=(), **card):
     return {"nodes": nodes, "edges": edges}
 
 
+def drill_graph(items=(WOW, WHERE, DRILL), **card):
+    """带下钻那一项的分析卡：华东按渠道的那次查询（east_q）接在 by_region 后面、分析卡前面。"""
+    return card_graph(items=items, extra_nodes=[node("east_q", "tool", tool=TOOL, args={"sql": EAST})],
+                      extra_edges=[edge("by_region", "east_q"), edge("east_q", "card")], **card)
+
+
 def card_issues(graph, level="error"):
     return [i for i in validate_graph(GraphSpec.model_validate(graph)).issues
             if i.node_id == "card" and (level is None or i.level == level)]
@@ -139,7 +141,7 @@ def test_node_type_and_labels():
 
 
 def test_a_well_formed_card_validates():
-    assert card_issues(card_graph(items=(WOW, WHERE, DRILL), overrides={"main_share": 0.6}), level=None) == []
+    assert card_issues(drill_graph(overrides={"main_share": 0.6}), level=None) == []
 
 
 @pytest.mark.parametrize("items, field, code, why", [
@@ -174,9 +176,9 @@ def test_a_well_formed_card_validates():
      "items[1].params.parent", "analysis.params_invalid", "没有「parent」"),
 ])
 def test_item_problems(items, field, code, why):
-    found = [i for i in card_issues(card_graph(items=items)) if i.field == field]
+    found = [i for i in card_issues(drill_graph(items=items)) if i.field == field]
     assert found and found[0].code == code and why in found[0].message, [i.message for i in card_issues(
-        card_graph(items=items))]
+        drill_graph(items=items))]
 
 
 @pytest.mark.parametrize("extra, why", [
@@ -235,8 +237,7 @@ def test_two_snapshot_items_need_both_nodes():
 
 async def test_the_card_runs_each_item_and_records_it(monkeypatch, shop):
     ops = FakeOps(monkeypatch, verdicts=["outside_band", "driver", "driver"], peek=["wow", "where", "east"])
-    row = await run_graph(card_graph(items=(WOW, WHERE, DRILL), overrides={"main_share": 0.6}),
-                          options={contract.SETTINGS_KEY: {"k_warn": 2.5}})
+    row = await run_graph(drill_graph(overrides={"main_share": 0.6}), options={contract.SETTINGS_KEY: {"k_warn": 2.5}})
     assert row.status == "succeeded", row.error
 
     # 算子拿到的：请求、按上游节点 id 解析的快照（只认台账里的那几份）、阈值（缺省 ← 发起时的快照 ← 卡上的覆盖）
@@ -280,6 +281,25 @@ async def test_the_card_runs_each_item_and_records_it(monkeypatch, shop):
     assert row.output["判定"] == results["wow"]["phrase"]
 
 
+async def test_the_drill_item_holds_up_with_the_real_operator(shop):
+    """上面的 DRILL 换真算子也成立（替身不核参数）：华东按渠道拆，两个渠道之和和上一项里华东那一段闭合，主要来自线上；上一层
+    按存下的工件认（parent 的哈希就是 where 那一项的结论工件），按封存的工件复算逐字段相同。"""
+    import app.analysis as analysis_ops
+
+    row = await run_graph(drill_graph(overrides={"main_share": 0.6}))
+    assert row.status == "succeeded", row.error
+    _, output = await output_of(row.id, "card")
+    results = output["results"]
+    assert results["where"]["verdict"] == "driver" and "主要来自华东" in results["where"]["phrase"]
+    east = artifact_store.load(results["east"]["artifact"])
+    assert east["status"] == "ok" and east["verdict"] == {"code": "driver", "subjects": [["线上"]], "direction": "down"}
+    assert east["values"]["base_total"] == "820" and east["values"]["target_total"] == "700"
+    [parent] = [i for i in east["inputs"] if i.get("role") == "parent"]
+    assert parent["artifact"] == results["where"]["artifact"]
+    again = analysis_ops.recompute(east, artifact_store.load, load_result=artifact_store.load)
+    assert contract.same(east, again)
+
+
 async def test_a_refused_item_does_not_fail_the_card(monkeypatch, shop):
     FakeOps(monkeypatch, statuses=["refused", "ok"])
     row = await run_graph(card_graph())
@@ -320,15 +340,12 @@ async def test_inputs_that_are_not_query_snapshots_fail_clearly(monkeypatch, sho
     assert row.status == "failed" and "不是存成快照的查询结果" in row.error
 
 
-async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path):
-    source_id = await _shop(tmp_path, {"mask_columns": ["region"]})
-    try:
-        ops = FakeOps(monkeypatch)
-        row = await run_graph(card_graph(items=[WHERE]))
-        assert row.status == "succeeded", row.error
-        assert "region" in ops.calls[0]["snapshots"]["by_region"]["masked"]
-    finally:
-        await engines.invalidate(source_id)
+async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path, scratch_source):
+    await _shop(tmp_path, scratch_source, {"mask_columns": ["region"]})
+    ops = FakeOps(monkeypatch)
+    row = await run_graph(card_graph(items=[WHERE]))
+    assert row.status == "succeeded", row.error
+    assert "region" in ops.calls[0]["snapshots"]["by_region"]["masked"]
 
 
 # --------------------------------------------------------------------------

backend/tests/test_analysis_compare.py

@@ -40,3 +40,24 @@ def test_compare_aligns_analysis_items():
     assert wow["change"] == "phrase"
     # 没有分析卡的两次运行：analyses 是空的
     assert compare.compare(week("r5").side(), week("r6").side())["analyses"] == []
+
+
+def test_a_card_that_failed_half_way_lists_only_what_a_finished_execution_decided():
+    """分析卡执行到一半失败：前几项已经发了 analysis.end，后一项让卡失败（node.failed，没有 node.finished）。那次执行的判定
+    不是这次运行的结果：只认后面跟着 node.finished 的执行，失败的那次不列、也不算执行次数。"""
+    base = _with_card(week("r7"), [("wow", "within_band", "在常规波动范围内")])
+    failed = week("r8")
+    failed.add("node.started", "insight", {"node_type": "analysis"})
+    failed.add("analysis.end", "insight", {"item": "wow", "op": "baseline", "mode": None, "status": "ok",
+                                           "verdict": "outside_band", "direction": "down", "phrase": "超出常规波动范围",
+                                           "analysis_artifact": "wow-failed"})
+    failed.add("node.failed", "insight", {"error": "分析项「where」（变化拆解）：比率不能按段相加"})
+    [wow] = compare.compare(base.side(), failed.side())["analyses"]
+    assert wow["change"] == "removed" and wow["target"] is None and wow["base"]["verdict"] == "within_band"
+
+    # 失败后重来一次做完了：按做完的那次对比，执行次数只算做完的
+    _with_card(failed, [("wow", "far_outside_band", "明显超出常规波动范围")])
+    result = compare.compare(base.side(), failed.side())
+    [wow] = result["analyses"]
+    assert wow["change"] == "verdict" and wow["target"]["verdict"] == "far_outside_band"
+    assert not [n for n in result["notes"] if "执行了" in n]

backend/tests/test_analysis_e2e.py

@@ -1,5 +1,6 @@
-"""分析入口接真算子跑一遍（不用替身）：agent 分析工具（含下钻）、分析卡（波动基线 + 变化拆解）、口径卡的可比性挂钩、
-审批恢复时的重放。断言工件、台账、交回模型的那一段、节点产出，并用 analysis.recompute 按封存的工件复算、逐字段相同。
+"""分析入口接真算子跑一遍（不用替身）：agent 的五个分析工具（变化拆解含下钻，波动基线、趋势、差异检验、可比性检查各一次）、
+分析卡（波动基线 + 变化拆解；输入是合并查询的一次）、口径卡的可比性挂钩、审批恢复时的重放。断言工件、台账、交回模型的那一段、
+节点产出，并用 analysis.recompute 按封存的工件复算、逐字段相同。
 
 虚构零售库 shop：orders(day, region, channel, amount)，2026-03-02 起 12 个整周，每周每个地区、渠道两笔（周三、周日），
 最后一周华东掉三成。模型换成脚本（MockChatModel._decide），不调用任何模型。
@@ -19,13 +20,13 @@ import app.analysis as analysis_ops
 from app.analysis import contract
 from app.core import artifact_store
 from app.core.artifact_store import canonical_json, content_hash
-from app.data.engine import engines
 from app.db.base import SessionLocal
-from app.db.models import DataSource, Run, RunEvent
+from app.db.models import Run, RunEvent
 from app.engine.runner import run_manager
 
 TOOL = "db_query__shop"
 DECOMPOSE = contract.TOOL_NAMES["decompose"]
+BASELINE, TREND, DIFFER, COMPARABLE = (contract.TOOL_NAMES[op] for op in ("baseline", "trend", "differ", "comparable"))
 WEEK = "date(day, '-6 days', 'weekday 1')"
 BY_REGION = (f"SELECT region, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' "
              "GROUP BY region, week ORDER BY region, week")
@@ -64,24 +65,15 @@ async def engine_up():
 
 
 @pytest.fixture(autouse=True)
-async def shop(tmp_path):
+async def shop(tmp_path, scratch_source):
+    """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，用完再删）。"""
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
     db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, channel TEXT, amount INTEGER);")
     db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", _orders())
     db.commit()
     db.close()
-    async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
-        await session.commit()
-        source_id = row.id
-    await engines.invalidate(source_id)
-    yield
-    await engines.invalidate(source_id)
+    return await scratch_source("shop", path)
 
 
 def node(nid, ntype, **config):
@@ -150,9 +142,9 @@ async def events(run_id, etype=None, node_id=None):
         return [e.data for e in (await session.execute(q.order_by(RunEvent.seq))).scalars()]
 
 
-def agent_graph(approval="never"):
+def agent_graph(approval="never", tools=(TOOL, DECOMPOSE)):
     return chain(node("start", "input"),
-                 node("bot", "agent", prompt="销售额为什么下降", tools=[TOOL, DECOMPOSE], approval=approval),
+                 node("bot", "agent", prompt="销售额为什么下降", tools=list(tools), approval=approval),
                  node("out", "output", fields=[{"name": "answer", "value": "{{ nodes.bot.text }}"}]))
 
 
@@ -211,6 +203,77 @@ async def test_an_agent_decomposes_and_drills_down_with_the_real_operator(monkey
     recomputes(a2)
 
 
+async def agent_analysis(monkeypatch, tool, calls):
+    """Agent 依次查库、最后调一次分析工具（真算子）：交回 (那份结论, 交回模型的那一段的 JSON)。顺带断言台账记了这一条（带
+    call_id）、交回模型的编号是「【分析 A1】」、判定语和结论一字不差，并按封存的工件复算、逐字段相同。"""
+    script = Script(monkeypatch, calls)
+    run = await run_manager.start(graph=agent_graph(tools=(TOOL, tool)), input_payload={})
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+    ends = await events(row.id, "tool.end", "bot")
+    artifact = ends[-1]["analysis_artifact"]
+    stored = artifact_store.load(artifact)
+    [finished] = await events(row.id, "node.finished", "bot")
+    [entry] = [e for e in finished["evidence"] if e["kind"] == "analysis"]
+    assert entry == contract.ledger_entry(stored, node_id="bot", exec_no=1, artifact=artifact,
+                                          call_id=f"call_{len(calls) - 1}")
+    assert set(entry["inputs"]) == {e["query_artifact"] for e in ends[:-1]}
+    head, body = script.replies()[-1].split("\n", 1)
+    assert head == "【分析 A1】" and json.loads(body)["phrase"] == stored["phrase"]
+    recomputes(stored)
+    return stored, json.loads(body)
+
+
+def _live(stored):
+    """结论里现查的表范围：按这次运行的数据源探 orders.day。"""
+    return [(p["table"], p["column"], p["basis"], p["max"]) for p in stored["probes"]]
+
+
+async def test_an_agent_puts_the_week_into_its_history_with_the_real_operator(monkeypatch):
+    stored, body = await agent_analysis(monkeypatch, BASELINE, [
+        (TOOL, {"sql": WEEKLY}),
+        (BASELINE, {"query": "Q1", "period": "week", "measure": "gmv", "label": "销售额", "hypothesis": "这周是不是异常"})])
+    assert stored["op"] == "baseline" and stored["status"] == "ok" and stored["hypothesis"] == "这周是不是异常"
+    assert stored["verdict"] == {"code": "far_outside_band", "direction": "down", "subjects": []}
+    assert stored["phrase"].startswith("销售额本期变化 -10.6%，明显超出过去 10 周的常规波动范围")
+    assert "record_low" in {c["code"] for c in stored["caveats"]} and body["verdict"] == "far_outside_band"
+    assert _live(stored) == [("orders", "day", "live", "2026-05-24")]      # 本期完不完整按现查的表范围判
+
+
+async def test_an_agent_asks_for_a_trend_with_the_real_operator(monkeypatch):
+    stored, body = await agent_analysis(monkeypatch, TREND, [
+        (TOOL, {"sql": WEEKLY}), (TREND, {"query": "Q1", "period": "week", "measure": "gmv", "label": "销售额"})])
+    assert stored["op"] == "trend" and stored["status"] == "ok"
+    assert stored["verdict"] == {"code": "no_trend", "direction": None, "subjects": []}
+    assert stored["phrase"] == "过去 12 周销售额没有明显的上升或下降趋势" and stored["values"]["n"] == "12"
+    assert _live(stored) == [("orders", "day", "live", "2026-05-24")] and body["values"] == stored["values"]
+
+
+async def test_an_agent_tests_a_difference_with_the_real_operator(monkeypatch):
+    rows = "SELECT region, amount FROM orders WHERE region IN ('华东', '华南') ORDER BY region, day"
+    stored, body = await agent_analysis(monkeypatch, DIFFER, [
+        (TOOL, {"sql": rows}),
+        (DIFFER, {"mode": "rows", "query": "Q1", "by": ["region"], "value": "amount", "label": "单笔金额"})])
+    assert stored["op"] == "differ" and stored["mode"] == "rows" and stored["status"] == "ok"
+    assert stored["verdict"] == {"code": "differs", "direction": "up", "subjects": [["华东"], ["华南"]]}
+    assert stored["phrase"] == "华东的单笔金额（196.83）高于华南（152），差异站得住"
+    assert stored["values"]["n_a"] == stored["values"]["n_b"] == "48" and stored["values"]["test"] == "welch"
+    assert body["subjects"] == [["华东"], ["华南"]]
+
+
+async def test_an_agent_checks_two_weeks_are_comparable_with_the_real_operator(monkeypatch):
+    stored, body = await agent_analysis(monkeypatch, COMPARABLE, [
+        (TOOL, {"sql": _week_sql("2026-05-18", "2026-05-24")}), (TOOL, {"sql": _week_sql("2026-05-11", "2026-05-17")}),
+        (COMPARABLE, {"base_query": "Q2", "target_query": "Q1", "measure": "gmv"})])
+    assert stored["op"] == "comparable" and stored["status"] == "ok" and stored["phrase"] == "两期可比"
+    assert stored["params"]["base_query"] == "Q2" and stored["params"]["target_query"] == "Q1"
+    values = stored["values"]
+    assert (values["base_start"], values["base_end"], values["target_start"], values["target_end"]) == (
+        "2026-05-11", "2026-05-17", "2026-05-18", "2026-05-24")
+    assert {values[k] for k in ("same_query", "window", "complete", "weekdays", "data")} == {"pass"}
+    assert _live(stored) == [("orders", "day", "live", "2026-05-24")] and body["verdict"] == "comparable"
+
+
 async def test_replaying_after_approvals_does_not_recompute(monkeypatch):
     counted = Counting(monkeypatch)
     script = Script(monkeypatch, AGENT_CALLS)
@@ -275,6 +338,31 @@ async def test_an_analysis_card_runs_the_real_operators():
     recomputes(decomposed)
 
 
+async def test_an_analysis_card_reads_a_merged_query():
+    """分析卡的输入是合并查询（merge 节点）：按合并查询节点的做法取它的快照，输入写明 sqlite 方言（合并查询在 SQLite 里
+    执行），真算子照常拆、复算逐字段相同。"""
+    where = {"id": "where", "op": "decompose", "mode": "additive", "from": "merged",
+             "params": {"by": ["region"], **TWO_WEEKS, "label": "销售额"}}
+    graph = chain(node("start", "input"), node("by_region", "tool", tool=TOOL, args={"sql": BY_REGION}),
+                  node("merged", "merge", inputs={"a": "by_region"}, sql="SELECT region, week, gmv FROM a"),
+                  node("card", "analysis", items=[where]),
+                  node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.results.where.phrase }}"}]))
+    run = await run_manager.start(graph=graph, input_payload={})
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+    [finished] = await events(row.id, "node.finished", "card")
+    results = artifact_store.load(finished["artifact"])["results"]
+    stored = artifact_store.load(results["where"]["artifact"])
+    assert stored["verdict"]["code"] == "driver" and stored["verdict"]["subjects"] == [["华东"]]
+    assert row.output["r"] == stored["phrase"] and "主要来自华东" in stored["phrase"]
+    [given] = stored["inputs"]
+    [merged] = await events(row.id, "node.finished", "merged")
+    assert (given["ref"], given["role"], given["dialect"]) == ("merged", "data", "sqlite")
+    assert given["artifact"] == merged["evidence"][0]["artifact"]
+    assert [(e["kind"], e["item"]) for e in finished["evidence"]] == [("analysis", "where")]
+    recomputes(stored)
+
+
 # --------------------------------------------------------------------------
 # 口径卡的可比性挂钩：真 comparable
 # --------------------------------------------------------------------------
@@ -312,3 +400,86 @@ async def test_a_caliber_card_checks_two_periods_with_the_real_operator(this_wee
     assert any(p.get("table") == "orders" and p.get("basis") == "live" for p in stored["probes"])
     recomputes(stored)
     assert not [e for e in await events(row.id, "log", "card") if e.get("code") == "metric_comparability_skipped"]
+
+
+def _region_sql(region: str, start: str, end: str) -> str:
+    return f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{region}' AND day BETWEEN '{start}' AND '{end}'"
+
+
+@pytest.mark.parametrize("south_week, verdict", [
+    (("2026-05-18", "2026-05-24"), None),             # 同一周、只差地区：是两段不是两期，不查
+    (("2026-05-11", "2026-05-17"), "incomparable"),   # 日期和地区都不同：两边除了日期，筛选的还不一样
+])
+async def test_a_ratio_between_two_regions_is_not_a_period_comparison(south_week, verdict):
+    """华东对华南的倍数：两份同形 SQL 只差地区字面量，不做可比性检查，带契约的出口照常正式出具（以前判成两期不可比、降档）。
+    日期和地区都不同的照查，真算子判不可比，出口记缺口降档。"""
+    metrics = [{"id": "times", "name": "华东对华南的倍数", "decimals": 4,
+                "expression": "cell(nodes.east, 0, 'gmv') / cell(nodes.south, 0, 'gmv')"}]
+    graph = chain(node("start", "input"),
+                  node("east", "tool", tool=TOOL, args={"sql": _region_sql("华东", "2026-05-18", "2026-05-24")}),
+                  node("south", "tool", tool=TOOL, args={"sql": _region_sql("华南", *south_week)}),
+                  node("card", "metrics", caliber="周报口径", metrics=metrics),
+                  node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}],
+                       contract={"metrics_from": ["card"], "required": ["times"]}))
+    run = await run_manager.start(graph=graph, input_payload={})
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+    [finished] = await events(row.id, "node.finished", "card")
+    [metric] = artifact_store.load(finished["artifact"])["metrics"]
+    issuance = row.output["_issuance"]
+    assert not [e for e in await events(row.id, "log", "card") if e.get("code") == "metric_comparability_skipped"]
+    if verdict is None:
+        assert "comparability" not in metric
+        assert issuance["tier"] == "formal" and not issuance.get("gaps") and not issuance.get("comparability")
+        return
+    found = metric["comparability"]
+    assert found["verdict"] == verdict and found["reasons"] == ["两边除了日期，筛选的还不一样"]
+    assert issuance["tier"] == "degraded"
+    assert issuance["gaps"] == ["指标「华东对华南的倍数」两期不可比：两边除了日期，筛选的还不一样"]
+    recomputes(artifact_store.load(found["artifact"]))
+
+
+# --------------------------------------------------------------------------
+# 测试夹具：复用名叫 shop 的源，前后都删干净
+# --------------------------------------------------------------------------
+
+
+async def test_a_scratch_source_starts_clean_and_forgets_its_probes(tmp_path):
+    """分析相关的用例都建名叫 shop 的源（tests/conftest.py 的 scratch_source）：别的用例留下的数据目录、coverage 的现查缓存里
+    按同一个源 id 记下的表范围（5 分钟有效）会改掉这次的判定。sqlite_source 建的源是新的 id，同名的旧源连同数据目录、现查
+    缓存先删掉；forget_source 删源（数据目录随外键级联删除）、清掉现查缓存里它的表范围。"""
+    from app.analysis.probe import make_prober
+    from app.data import catalog as data_catalog
+    from app.data import coverage
+    from app.data.catalog import make_item
+    from app.db.models import CatalogNote, DataSource
+    from tests.fixtures.sources import forget_source, sqlite_source
+
+    path = tmp_path / "other.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE orders (day TEXT, amount INTEGER); INSERT INTO orders VALUES ('2026-01-05', 1);")
+    db.close()
+
+    async def leave_traces(source_id):
+        async with SessionLocal() as session:
+            await data_catalog.write_entry(session, source_id, "orders", {"columns": {"amount": {
+                "measure": make_item("flow", "human")}}}, if_version=0, actor=None)
+            row = await session.get(DataSource, source_id)
+        probe = await make_prober({"scratch": row})("scratch", "orders", "day")
+        assert probe["max"] == "2026-01-05" and probe["basis"] == "live"
+        assert any(k[0] == source_id for k in coverage._LIVE)
+
+    async def traces(source_id):
+        async with SessionLocal() as session:
+            notes = (await session.execute(select(CatalogNote).where(CatalogNote.source_id == source_id))).scalars().all()
+            row = await session.get(DataSource, source_id)
+        return {"row": row is not None, "catalog": len(notes), "probes": sum(k[0] == source_id for k in coverage._LIVE)}
+
+    old = await sqlite_source("scratch", path)
+    await leave_traces(old)
+    new = await sqlite_source("scratch", path)
+    assert new != old and await traces(old) == {"row": False, "catalog": 0, "probes": 0}
+    assert await traces(new) == {"row": True, "catalog": 0, "probes": 0}
+    await leave_traces(new)
+    await forget_source(new)
+    assert await traces(new) == {"row": False, "catalog": 0, "probes": 0}

backend/tests/test_analysis_e2e_claims.py

@@ -26,15 +26,12 @@ from app.analysis import contract
 from app.core import artifact_store
 from app.data import catalog as data_catalog
 from app.data.catalog import make_item
-from app.data.engine import engines
-from app.data.introspect import introspect
 from app.db.base import SessionLocal
-from app.db.models import DataSource, Run, RunEvent
+from app.db.models import Run, RunEvent
 from app.engine import analysis_claims as ac
 from app.engine.evidence import iter_segments
 from app.engine.runner import run_manager
 from app.main import app
-from tests.fixtures.sources import drop_source
 
 TOOL = "db_query__shop"
 DECOMPOSE = contract.TOOL_NAMES["decompose"]
@@ -74,7 +71,7 @@ async def engine_up():
 
 
 @pytest.fixture(autouse=True)
-async def shop(tmp_path):
+async def shop(tmp_path, scratch_source):
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
     db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, region TEXT, channel TEXT, amount INTEGER);")
@@ -84,25 +81,14 @@ async def shop(tmp_path):
     db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", [(i, w, r, c, a) for i, (r, w, c, a) in enumerate(rows, 1)])
     db.commit()
     db.close()
+    # scratch_source 建的是新源（同名的旧源连同数据目录、现查缓存先删掉），测试结束时删掉、数据目录随外键级联删除：同一个
+    # 测试进程里别的用例也用名叫 shop 的源，留着这份目录会改掉它们的判定（比如金额标成了流量，窗口长度不同就从有条件可比
+    # 变成不可比）
+    source_id = await scratch_source("shop", path, schema=True)
     async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
-        await session.commit()
-        row.schema_cache = await introspect(row)
-        await session.commit()
         for table, notes in NOTES.items():
-            current = await data_catalog.read_entry(session, row.id, table)
-            await data_catalog.write_entry(session, row.id, table, notes, if_version=current.version if current else 0,
-                                           actor=None)
-        source_id = row.id
-    await engines.invalidate(source_id)
-    yield
-    # 删掉这个源（数据目录随外键级联删除）：同一个测试进程里别的用例也用名叫 shop 的源，留着这份目录会改掉它们的判定
-    # （比如金额标成了流量，窗口长度不同就从有条件可比变成不可比）
-    await drop_source(source_id)
+            await data_catalog.write_entry(session, source_id, table, notes, if_version=0, actor=None)
+    return source_id
 
 
 def _node(nid, ntype, **config):

backend/tests/test_analysis_evidence.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import asyncio
+import re
 from concurrent.futures import ThreadPoolExecutor
 from datetime import date
 from typing import Any
@@ -26,6 +27,7 @@ from app.analysis.contract import CallError, effective_settings, ledger_entry
 from app.analysis.testing import fake_result
 from app.core.artifact_store import canonical_json, content_hash
 from app.engine.evidence import (
+    ANALYSIS_LEVELS,
     ANALYSIS_CELLS_REASON,
     StreamRenderer,
     analysis_formulas,
@@ -473,6 +475,65 @@ def test_trimmed_catalogs_keep_the_judgments_and_fill_result_tables_before_query
     assert "另有 1 条分析结论，判定和具名的数都列出了，其中 1 条没列出结果表的各段" in tight.text
 
 
+def _analysis_part(text: str) -> str:
+    """写作目录里「分析结论」一段（从那段开头的说明到查询结果那段之前）。"""
+    start = text.index("\n分析结论（")
+    end = text.find("\n查询结果（", start)
+    return text[start:end if end >= 0 else len(text)]
+
+
+def _many_analyses(n: int, queries: int = 30) -> tuple[Store, dict[str, Any]]:
+    """n 份分析结论（A1…An，内容各不相同）、1 + queries 次查询的写作目录。"""
+    store, ledger, q, _ = setup()
+    for i in range(n - 1):
+        res = fake_result(inputs=[{"ref": "Q2", "artifact": q, "role": "data", "columns": ["region", "week", "gmv"],
+                                   "rows": "all"}], hypothesis=f"假设 {i}")
+        ledger.append(ledger_entry(res, node_id="an", exec_no=1, artifact=store.put(res), item=f"i{i}"))
+    many = [{"kind": "query", "node_id": "fetch", "exec": i + 2, "artifact": store.put({**SNAPSHOT, "sql": f"-- {i}"}),
+             "tool": "db_query__shop", "source": "shop", "columns": SNAPSHOT["columns"], "rows": 8}
+            for i in range(queries)]
+    return store, build_catalog(nodes={}, ledger=[*ledger, *many], loader=store.load)
+
+
+@pytest.mark.parametrize("n, first, last, words", [
+    # 先截结果表的段，还超就从后往前只留判定语和具名的数
+    (20, "compact", "brief", ("条没列出结果表的各段", "条只列了判定语和具名的数")),
+    # 都只留判定语和具名的数还超：再从后往前只列编号
+    (80, "brief", "head", ("条只列了判定语和具名的数", "条只列了编号")),
+])
+def test_the_analysis_part_of_a_trimmed_catalog_takes_at_most_half_of_the_budget(n, first, last, words):
+    """几十份结论时，分析结论一段最多占预算的一半：按结论的先后从后往前降，先截结果表的段、再只留判定语和具名的数、再只列
+    编号，开头照实写各有几条、没列的照样能引用；剩下的篇幅给查询展开。以前这一段没有上限，40 份结论、每份上千段时在 51.2k
+    的预算里占了 50.6k，查询只剩一行摘要。"""
+    store, catalog = _many_analyses(n)
+    budget = 12_000
+    writer = writer_catalog(catalog, loader=store.load, budget=budget)
+    assert writer.trimmed and len(writer.text) <= budget
+    part = _analysis_part(writer.text)
+    assert len(part) <= budget // 2
+    blocks = [b.strip("\n") for b in re.split(r"\n(?=- A\d+：)", part)[1:]]
+    assert len(blocks) == n                                  # 每条结论都还在，编号连着
+
+    def level(block: str) -> str:
+        if "\n" not in block:
+            return "head"
+        if "判定标准：" not in block:
+            return "brief"
+        return "full" if "\n  r0：" in block else "compact"
+
+    assert level(blocks[0]) == first and level(blocks[-1]) == last
+    assert blocks[-1].startswith(f"- A{n}：变化拆解（可加），来自 Q1")
+    for block in blocks:
+        if level(block) == "brief":
+            assert "判定语 [[a:" in block and "具名的数（写 [[a:" in block and "附带说明" not in block
+    order = [ANALYSIS_LEVELS.index(level(b)) for b in blocks]
+    assert order == sorted(order)                            # 越往后越简
+    notice = next(line for line in writer.text.split("\n") if line.startswith("注意："))
+    assert f"另有 {n} 条分析结论，这一段最多占篇幅的一半：" in notice and all(w in notice for w in words), notice
+    assert "没列出的也能引用：判定语单独写 [[a:编号]]" in notice
+    assert writer.queries == 31 and writer.compact > 0      # 查询照样有展开的
+
+
 def test_metric_comparability_is_spoken_in_the_writer_catalog():
     card = {"kind": "metric_set", "caliber": "周报", "caliber_version": "v1", "artifact": "c" * 64, "metrics": [
         {"id": "wow", "name": "环比", "value": -0.087, "unit": "", "comparability": {
@@ -601,6 +662,25 @@ def test_formulas_substitute_the_key_numbers():
     assert analysis_formulas(fake_result(values={})) == []
 
 
+@pytest.mark.parametrize("keys, named", [((None, "华南", "华北"), "（空）"), ((True, False, None), "是")])
+def test_the_formula_names_the_segment_the_verdict_names(keys, named):
+    """代入式里点名的那一段按判定的 subjects 找，维度值按算子写段的同一个函数比（空值写「（空）」、布尔写「是 / 否」）。
+    这里判定点名的段不在结果表第一行（第一行是反方向变化最大的那段）：以前拿原值的 str 比，对不上就退回第一行，代入式写成了
+    第一行那段。"""
+    store = Store()
+    driver, opposite, other = keys                         # 变化：driver -220、opposite +250、other -180，总共 -150
+    snap = {"columns": ["region", "week", "gmv"], "column_types": {"region": "text", "week": "text", "gmv": "number"},
+            "rows": [[driver, "W41", 1000], [driver, "W42", 780], [opposite, "W41", 500], [opposite, "W42", 750],
+                     [other, "W41", 600], [other, "W42", 420]],
+            "sql": "SELECT region, week, SUM(gmv) AS gmv FROM orders GROUP BY region, week", "source": "shop",
+            "truncated": False}
+    result = real_result(store.put(snap), snap)
+    assert result["verdict"]["code"] == "driver" and result["verdict"]["subjects"] == [[named]]
+    assert [r[0] for r in result["table"]["rows"]] == [opposite, driver, other]     # 按变化的绝对值排
+    texts = {f["label"]: f["text"] for f in analysis_formulas(result)}
+    assert texts["占比"] == f"{named}的占比 = {named}的变化 ÷ 总变化 = -220 ÷ （-150） = 146.7%"
+
+
 def test_formulas_for_factor_ratio_and_baseline():
     store = Store()
     factor_snap = {"columns": ["week", "gmv", "orders", "aov"],

backend/tests/test_analysis_evidence_api.py

@@ -178,6 +178,51 @@ async def test_named_values_and_result_cells_point_at_what_they_cite(client):
     assert cell["note"] == "该值由系统从分析结论的结果表中取出、按固定规则显示，未经模型转写"
 
 
+#: 按门店拆的快照：门店码是文本（"1203"），渲染时不加千分位
+STORES: dict[str, Any] = {
+    "columns": ["store", "week", "gmv"], "column_types": {"store": "text", "week": "text", "gmv": "number"},
+    "rows": [["1203", "W41", 820], ["1203", "W42", 700], ["1305", "W41", 600], ["1305", "W42", 560],
+             ["2201", "W41", 700], ["2201", "W42", 666]],
+    "sql": "SELECT store, week, SUM(gmv) AS gmv FROM orders GROUP BY store, week ORDER BY store, week",
+    "source": "shop", "truncated": False, "row_count": 6,
+}
+#: 按时段拆的快照：时段是 ISO 日期时间（带 T），文档按 2 版显示规则写成空格
+SLOTS: dict[str, Any] = {
+    "columns": ["slot", "week", "gmv"], "column_types": {"slot": "text", "week": "text", "gmv": "number"},
+    "rows": [["2026-10-12T08:00:00", "W41", 820], ["2026-10-12T08:00:00", "W42", 700],
+             ["2026-10-12T20:00:00", "W41", 600], ["2026-10-12T20:00:00", "W42", 560]],
+    "sql": "SELECT slot, week, SUM(gmv) AS gmv FROM orders GROUP BY slot, week ORDER BY slot, week",
+    "source": "shop", "truncated": False, "row_count": 4,
+}
+
+
+@pytest.mark.parametrize("snapshot, by, shown", [
+    (STORES, "store", "1203"),
+    (SLOTS, "slot", "2026-10-12 08:00:00"),
+])
+async def test_a_result_cell_rerenders_the_way_it_was_composed(client, monkeypatch, snapshot, by, shown):
+    """证据接口按组装时同一套规则重新渲染结果表的格（evidence.resolve_ref，按文档记下的单元格显示规则）：文本码值照结果表
+    记的列类型不加千分位，ISO 日期时间的 T 换成空格。以前不看列类型和显示规则，渲染成「1,203」「2026-10-12T08:00:00」，
+    render_ok 为假，证据面板和导出页报假的「与按结论渲染的结果不一致」。"""
+    import tests.test_analysis_evidence_api as here
+
+    monkeypatch.setattr(here, "SNAPSHOT", snapshot)
+    q = await artifact_store.put_json(snapshot, kind="query_snapshot")
+    result = await analyzed({"Q2": (snapshot, q)}, by=[by])
+    ref = f"v:A1.r0.{by}"
+    made = await sealed_run(f"{by} [[{ref}]] 变化 [[v:A1.r0.delta]]。[[see:A1]]", result=result)
+    segment = next(s for s in iter_segments(made["doc"]) if s.get("ref") == ref)
+    assert segment["text"] == shown
+    body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{segment['id']}")).json()
+    judged = body["chain"][0]
+    assert judged["rendered"] == shown and judged["render_ok"] is True
+    assert judged["cited"] == {"what": "cell", "row": 0, "column": by, "raw": snapshot["rows"][0][0]}
+    # 导出页的数据块是同一条链
+    data = data_block((await client.get(f"/api/runs/{made['run']}/report.html")).text)
+    chain = data["chains"][data["refs"][segment["id"]]]
+    assert chain[0]["step"] == "analysis" and chain[0]["render_ok"] is True
+
+
 async def test_the_recompute_gets_the_sealed_inputs_by_hash(client, monkeypatch):
     """复算的 load 交工件原内容（不加出处、不改列的类型），只认封存范围内交回过的查询；别的一律当取不回。"""
     seen: dict[str, Any] = {}
@@ -242,6 +287,25 @@ async def test_a_tampered_result_is_said_and_not_shown(client):
     assert seal["hash_ok"] is False and seal["sealed"] is False and body["seal"]["covered"] is False
 
 
+def test_suspicious_name_hints_do_not_read_the_analysis_results(monkeypatch):
+    """可疑实体找「最接近的名字」时按封存的台账重建目录（_scope_catalog）：分析结论不进，不读盘、不复验哈希（候选只看表和
+    字段，用不到结论的内容）。以前每点开一个可疑的名字，就把这次运行的分析结论全读一遍。"""
+    from app.api import evidence as api
+    from app.engine import evidence as engine_evidence
+
+    read: list[str] = []
+    monkeypatch.setattr(engine_evidence, "load_artifact", lambda artifact: read.append(artifact))
+    q, a = "1" * 64, "2" * 64
+    entries = [_query_entry(SNAPSHOT, q), {"kind": "analysis", "node_id": "an", "exec": 1, "artifact": a,
+                                           "op": "decompose", "mode": "additive", "item": "drop", "inputs": [q]}]
+    graph = {**GRAPH, "edges": [{"source": "fetch", "target": "an"}, {"source": "an", "target": "write"}]}
+    sealed = api._Sealed(run_id="r", graph=graph, seal={"sealed": True, "ok": True}, events=[], entries=entries)
+    for report in ("write", "nope"):
+        catalog = api._scope_catalog(sealed, report)
+        assert "Q1" in catalog and not [e for e in catalog.values() if e.get("kind") == "analysis"]
+    assert read == []
+
+
 async def test_agent_analyses_are_marked_as_such(client):
     made = await sealed_run(TEXT, item=None, call_id="call-9")
     body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json()
@@ -260,12 +324,51 @@ async def test_headline_inputs_show_the_closure(client):
     body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json()
     inputs = [s for s in body["chain"] if s["step"] == "analysis_input"]
     assert [s["role"] for s in inputs] == ["headline_base", "headline_target", "data"]
-    assert inputs[0]["closure"] == {"total_key": "base_total", "total": "2,220", "headline": "2,220", "closed": True}
-    assert inputs[1]["closure"] == {"total_key": "target_total", "total": "2,026", "headline": "2,026", "closed": True}
+    assert inputs[0]["closure"] == {"mode": "additive", "total_key": "base_total", "total": "2,220", "headline": "2,220",
+                                    "closed": True}
+    assert inputs[1]["closure"] == {"mode": "additive", "total_key": "target_total", "total": "2,026",
+                                    "headline": "2,026", "closed": True}
+    assert [s["role_label"] for s in inputs[:2]] == ["基期总数", "本期总数"]
     assert inputs[1]["query"] == "Q2" and inputs[1]["sealed"] is True and inputs[2]["query"] == "Q1"
     assert next(s for s in body["chain"] if s["step"] == "analysis_recompute")["status"] == "same"
 
 
+async def ratio_run() -> dict[str, Any]:
+    """一次封存的运行：按渠道拆退款率（比率拆解），给了整体比率的两格（headline，0.035 → 0.053），报告引用判定语。"""
+    rates = {"columns": ["channel", "week", "refunds", "orders"],
+             "column_types": {"channel": "text", "week": "text", "refunds": "number", "orders": "number"},
+             "rows": [["线上", "W41", 50, 1000], ["线上", "W42", 80, 1000], ["线下", "W41", 20, 1000],
+                      ["线下", "W42", 26, 1000]],
+             "sql": "SELECT channel, week, SUM(refund_flag) AS refunds, COUNT(*) AS orders FROM orders GROUP BY 1, 2",
+             "source": "shop", "truncated": False, "row_count": 4}
+    head = {"columns": ["week", "rate"], "column_types": {"week": "text", "rate": "number"},
+            "rows": [["W41", "0.035"], ["W42", "0.053"]],
+            "sql": "SELECT week, SUM(refund_flag) * 1.0 / COUNT(*) AS rate FROM orders GROUP BY week",
+            "source": "shop", "truncated": False, "row_count": 2}
+    r, h = (artifact_store.content_hash(artifact_store.canonical_json(c)) for c in (rates, head))
+    result = await analyzed({"Q2": (rates, r), "Q5": (head, h)}, {"op": "decompose", "mode": "ratio", "params": {}},
+                            query="Q2", by=["channel"], period="week", base="W41", target="W42", numerator="refunds",
+                            denominator="orders", label="退款率", headline={"base": "Q5.r0.rate", "target": "Q5.r1.rate"})
+    assert result["status"] == "ok" and result["verdict"]["code"] != "unclosed"
+    return await sealed_run("[[a:A1]]。", result=result, queries=[rates, head])
+
+
+async def test_a_ratio_headline_closes_against_the_overall_ratio(client):
+    """比率拆解给了整体比率的两格（headline）：闭合核对对着结论里的整体比率，即各段分子之和 ÷ 分母之和（r_base_total、
+    r_target_total），角色写「基期整体比率」「本期整体比率」，closure.mode 写明是比率（导出页照它说「各段分子之和 ÷ 分母
+    之和」，不说「各段之和」）。以前只认可加拆解的总数键，比率拆解的 headline 不显示闭合核对。"""
+    made = await ratio_run()
+    body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json()
+    inputs = {s["role"]: s for s in body["chain"] if s["step"] == "analysis_input"}
+    assert inputs["headline_base"]["closure"] == {"mode": "ratio", "total_key": "r_base_total", "total": "0.035",
+                                                  "headline": "0.035", "closed": True}
+    assert inputs["headline_target"]["closure"] == {"mode": "ratio", "total_key": "r_target_total", "total": "0.053",
+                                                    "headline": "0.053", "closed": True}
+    assert inputs["headline_base"]["role_label"] == "基期整体比率"
+    assert inputs["headline_target"]["role_label"] == "本期整体比率"
+    assert next(s for s in body["chain"] if s["step"] == "analysis_recompute")["status"] == "same"
+
+
 async def test_a_drill_down_points_at_its_parent_and_recomputes_from_it(client):
     """下钻（华东按渠道再拆）：输入里有上一层结论（parent），落到报告目录的 A1；复算按 sha256 取回台账记过的上一层。"""
     q = artifact_store.content_hash(artifact_store.canonical_json(SNAPSHOT))
@@ -434,6 +537,36 @@ async def test_the_html_export_carries_the_analysis_chain_and_the_scope(client):
     assert f'data-seg="{phrase}"' in html and 'data-kind="analysis"' in html
 
 
+async def test_the_export_recomputes_each_conclusion_once(client, monkeypatch):
+    """导出页里同一份结论被判定语、具名的数、整表的每一格、句末依据、分析范围各引一处，引用键不同、各接一条链，复算只做
+    一次（按结论工件记在这一次请求里）；分析范围的编号和正文的判定语同一个引用键，不再另建一条链。以前每条链各复算一次，
+    1000 段的拆解导出一次复算三十多次。"""
+    calls: list[str] = []
+    real = analysis.recompute
+
+    def counting(result, load, *, load_result=None):
+        calls.append(str(result.get("method")))
+        return real(result, load, load_result=load_result)
+
+    monkeypatch.setattr(analysis, "recompute", counting)
+    text = ("[[a:A1]]。华东占 [[a:A1.top_share|pct]]。[[see:A1]]\n\n"
+            "[[table:A1 cols=region,delta,share rows=0-3]]")
+    made = await sealed_run(text, doc_extra=SCOPE)
+    data = data_block((await client.get(f"/api/runs/{made['run']}/report.html")).text)
+    assert calls == ["decompose.additive@1"]
+    cells = [s["id"] for s in iter_segments(made["doc"]) if str(s.get("ref") or "").startswith("v:A1.")]
+    assert len(cells) == 12
+    keys = {data["refs"][sid] for sid in [*cells, seg_id(made["doc"], "a:A1"), seg_id(made["doc"], "a:A1.top_share|pct")]}
+    assert len(keys) == 14                                  # 每处引用各一条链
+    for key in keys:
+        [again] = [s for s in data["chains"][key] if s["step"] == "analysis_recompute"]
+        assert again["status"] == "same"
+    assert data["refs"]["scope:A1"] == data["refs"][seg_id(made["doc"], "a:A1")]
+    # 下一次请求重新复算（缓存只活在一次请求里）
+    await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")
+    assert len(calls) == 2
+
+
 async def test_docs_without_a_scope_render_no_scope(client):
     made = await sealed_run(TEXT)
     html = (await client.get(f"/api/runs/{made['run']}/report.html")).text

backend/tests/test_analysis_pages.py

@@ -20,7 +20,9 @@ import app.bundle
 import app.report_export
 from app.main import app as asgi
 from tests.test_analysis_bundle import bundle_of, card_bundle
-from tests.test_analysis_evidence_api import LAST_WEEK, SCOPE, TEXT, THIS_WEEK, comparability_card, sealed_run, seg_id
+from tests.test_analysis_evidence_api import (
+    LAST_WEEK, SCOPE, TEXT, THIS_WEEK, comparability_card, ratio_run, sealed_run, seg_id,
+)
 
 NODE = shutil.which("node")
 pytestmark = pytest.mark.skipif(NODE is None, reason="没有 node，无法执行页面上的脚本")
@@ -155,6 +157,23 @@ async def test_the_export_page_opens_the_analysis_steps_and_the_scope_numbers():
     assert out["scope:A1"]["steps"] == opened["steps"]
 
 
+async def test_the_export_page_speaks_of_the_overall_ratio_for_a_ratio_decomposition():
+    """比率拆解的总数那一格是整体比率：导出页写「整体比率那一格」「整体比率 x 与各段分子之和 ÷ 分母之和 y 闭合」，不说
+    「各段之和」。可加拆解照旧（上一个用例的页面里没有「整体比率」）。"""
+    made = await ratio_run()
+    async with AsyncClient(transport=ASGITransport(app=asgi), base_url="http://test") as client:
+        html = (await client.get(f"/api/runs/{made['run']}/report.html")).text
+    data = re.search(r'<script type="application/json" id="report-data">(.*?)</script>', html, re.S).group(1)
+    phrase = seg_id(made["doc"], "a:A1")
+    out = run_node(RUN_RUNTIME, RUNTIME, {"data": data, "segs": [
+        {"seg": phrase, "kind": "analysis", "state": "deterministic", "text": made["result"]["phrase"]}]})
+    body = out[phrase]["body"]
+    for text in ("基期整体比率", "整体比率那一格", "整体比率 0.035 与各段分子之和 ÷ 分母之和 0.035 闭合",
+                 "整体比率 0.053 与各段分子之和 ÷ 分母之和 0.053 闭合"):
+        assert text in body, text
+    assert "各段之和 0.0" not in body
+
+
 async def test_the_bundle_page_shows_the_analysis_detail_and_the_scope():
     bundle, _, _ = bundle_of(doc_extra=SCOPE)
     out = run_node(RUN_VERIFIER, VERIFIER, {"data": json.dumps(bundle), "seal": bundle["meta"]["manifest_hash"]})
@@ -164,7 +183,8 @@ async def test_the_bundle_page_shows_the_analysis_detail_and_the_scope():
     assert "没拆渠道" in scope and "这次做了 2 次差异检验，几次检验之间没有按次数校正" in scope
     phrase = next(text for text in out["opened"].values() if "出处分析结论 A1 的判定语" in text)
     assert "核对结果✓ 通过" in phrase and "分析结论里的值「销售额下降 194（-8.7%）" in phrase
-    assert "判定语销售额下降 194（-8.7%）" in phrase and "方法与版本decompose.additive@1" in phrase
+    # 方法写服务端带来的叫法和 id（和导出页同一个写法），不只写 id
+    assert "判定语销售额下降 194（-8.7%）" in phrase and "方法与版本变化拆解（可加） · decompose.additive@1" in phrase
     assert "输入数据 · Q1" in phrase                       # 输入的角色写叫法，不写 data
     assert "复算一致：按封存的输入重新计算，结论逐项相同（导出时服务端按封存的输入复算的结论；本页只核哈希" in phrase
     value = next(text for text in out["opened"].values() if "具名的数「top_share」" in text)
@@ -196,7 +216,7 @@ def test_the_bundle_page_shows_the_comparability_in_the_metric_detail():
     out = run_node(RUN_VERIFIER, VERIFIER, {"data": json.dumps(bundle), "seal": bundle["meta"]["manifest_hash"]})
     assert out["verified"] == ""
     [detail] = out["opened"].values()
-    for text in ("两期可比性", "判定两期可比", "判定标准只差日期", "方法与版本comparable@1", "各项检查", "同口径",
+    for text in ("两期可比性", "判定两期可比", "判定标准只差日期", "方法与版本可比性检查 · comparable@1", "各项检查", "同口径",
                  "复算一致：按封存的输入重新计算，结论逐项相同（导出时服务端按封存的输入复算的结论"):
         assert text in detail, text
 

backend/tests/test_analysis_tools.py

@@ -1,8 +1,9 @@
 """agent 分析工具（docs/design/analysis-round/00-总方案.md §3.2）：参数、方法说明、查询容器、台账、编号、下钻、次数上限、
 遮罩列、续跑。
 
-算子（app.analysis.run）还没落地：换成按请求交回假结论的替身（tests/fixtures/analysis_ops.FakeOps），断言入口交给算子的
-东西、算子交回后入口做的事。模型换成脚本（MockChatModel._decide），不调用任何模型。库是虚构的零售库 shop。
+算子（app.analysis.run）换成按请求交回假结论的替身（tests/fixtures/analysis_ops.FakeOps），断言入口交给算子的东西、算子
+交回后入口做的事；五个工具接真算子的端到端见 tests/test_analysis_e2e.py。模型换成脚本（MockChatModel._decide），不调用任何
+模型。库是虚构的零售库 shop。
 """
 from __future__ import annotations
 
@@ -18,9 +19,8 @@ from app.analysis import contract
 from app.analysis.contract import CallError
 from app.core import artifact_store
 from app.core.artifact_store import canonical_json, content_hash
-from app.data.engine import engines
 from app.db.base import SessionLocal
-from app.db.models import Artifact, DataSource, Run, RunEvent
+from app.db.models import Artifact, Run, RunEvent
 from app.engine.runner import run_manager
 from app.tools.analysis import CALL_ERROR_HEAD, REFUSED_CODE
 from app.tools.builtin.analysis import ANALYSIS_METHOD_PROMPT, ARGS, METHOD_GUIDE, NO_SCOPE
@@ -45,30 +45,20 @@ async def engine_up():
     await run_manager.shutdown()
 
 
-async def _shop(tmp_path, options=None):
+async def _shop(tmp_path, scratch_source, options=None):
+    """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，测试结束时再删）。"""
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
     db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, week TEXT, amount REAL, phone TEXT);")
     db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", ROWS)
     db.commit()
     db.close()
-    async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, options or {}
-        await session.commit()
-        source_id = row.id
-    await engines.invalidate(source_id)
-    return source_id
+    return await scratch_source("shop", path, options=options)
 
 
 @pytest.fixture
-async def shop(tmp_path):
-    source_id = await _shop(tmp_path)
-    yield
-    await engines.invalidate(source_id)
+async def shop(tmp_path, scratch_source):
+    return await _shop(tmp_path, scratch_source)
 
 
 class Script:
@@ -230,6 +220,30 @@ def test_input_snapshots_carry_the_payload_as_is():
     assert merged["dialect"] == "sqlite" and merged["masked"] == []
 
 
+async def test_each_query_is_parsed_once_per_node_for_its_tables(monkeypatch):
+    """agent 每次分析都按本节点全部查询认表（目录、剖析、探查都按表给，tools/analysis.context_of）：同一条 SQL 一个节点里只
+    解析一次（AnalysisScope.tables_cache）。以前每次调用都把本节点的查询从头解析一遍。"""
+    from app.engine import evidence
+    from app.tools.analysis import AnalysisScope, SCOPE_KEY, input_snapshot
+    from app.tools.builtin.analysis import DecomposeArgs, run_tool
+    from app.tools.registry import ToolContext
+
+    FakeOps(monkeypatch)
+    parsed: list[str] = []
+    real = evidence.sql_tables
+    monkeypatch.setattr(evidence, "sql_tables", lambda sql: parsed.append(sql) or real(sql))
+    scope = AnalysisScope(node_id="bot", snapshot={})
+    for alias, sql in (("Q1", SQL), ("Q2", MASKED_SQL)):
+        content = {"columns": ["region", "week", "gmv"], "rows": [["华东", "2026-W41", 820.0]], "sql": sql,
+                   "source": "nowhere", "truncated": False}
+        artifact = await artifact_store.put_json(content, kind="query_snapshot")
+        scope.add_query(alias, input_snapshot(content, ref=alias, artifact=artifact, node_id="bot"))
+    ctx = ToolContext(run_id="r", node_id="bot", extra={SCOPE_KEY: scope})
+    for _ in range(3):
+        await run_tool("decompose", ctx, DecomposeArgs(**DECOMPOSE_ARGS))
+    assert parsed == [SQL, MASKED_SQL] and scope.tables_cache[SQL] == ("orders",)
+
+
 # --------------------------------------------------------------------------
 # 方法说明：绑了分析工具的 Agent 注入一次，模型调用节点、没绑的 Agent 不注入
 # --------------------------------------------------------------------------
@@ -383,31 +397,25 @@ async def test_the_number_of_analyses_per_node_is_capped(monkeypatch, shop):
     assert ops.calls[0]["settings"]["max_calls"] == 1
 
 
-async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path):
-    source_id = await _shop(tmp_path, {"mask_columns": ["phone"]})
-    try:
-        ops = FakeOps(monkeypatch)
-        Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["tel"]})])
-        await run_graph(agent_graph())
-        masked = ops.calls[0]["snapshots"]["Q1"]["masked"]
-        # 按 SQL 认出改了名字的遮罩列（phone AS tel）排在前面，数据源设的列名也在
-        assert masked[0] == "tel" and "phone" in masked
-    finally:
-        await engines.invalidate(source_id)
+async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path, scratch_source):
+    await _shop(tmp_path, scratch_source, {"mask_columns": ["phone"]})
+    ops = FakeOps(monkeypatch)
+    Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["tel"]})])
+    await run_graph(agent_graph())
+    masked = ops.calls[0]["snapshots"]["Q1"]["masked"]
+    # 按 SQL 认出改了名字的遮罩列（phone AS tel）排在前面，数据源设的列名也在
+    assert masked[0] == "tel" and "phone" in masked
 
 
-async def test_mask_results_runs_hand_over_the_masked_snapshot(monkeypatch, tmp_path):
-    source_id = await _shop(tmp_path, {"mask_columns": ["phone"]})
-    try:
-        ops = FakeOps(monkeypatch)
-        Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["region"]})])
-        await run_graph(agent_graph(), options={"mask_results": True, "read_only_data": True})
-        snap = ops.calls[0]["snapshots"]["Q1"]
-        assert "tel" in snap["masked"]
-        tel = snap["columns"].index("tel")
-        assert {r[tel] for r in snap["rows"]} == {"已遮罩"}, "遮罩运行里算子拿到的是遮过的快照"
-    finally:
-        await engines.invalidate(source_id)
+async def test_mask_results_runs_hand_over_the_masked_snapshot(monkeypatch, tmp_path, scratch_source):
+    await _shop(tmp_path, scratch_source, {"mask_columns": ["phone"]})
+    ops = FakeOps(monkeypatch)
+    Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["region"]})])
+    await run_graph(agent_graph(), options={"mask_results": True, "read_only_data": True})
+    snap = ops.calls[0]["snapshots"]["Q1"]
+    assert "tel" in snap["masked"]
+    tel = snap["columns"].index("tel")
+    assert {r[tel] for r in snap["rows"]} == {"已遮罩"}, "遮罩运行里算子拿到的是遮过的快照"
 
 
 # --------------------------------------------------------------------------

backend/tests/test_contract_analysis_claims.py

@@ -17,9 +17,8 @@ from httpx import ASGITransport, AsyncClient
 from langchain_core.messages import AIMessage
 from sqlalchemy import select
 
-from app.data.engine import engines
 from app.db.base import SessionLocal
-from app.db.models import DataSource, Run, RunEvent, Workflow, WorkflowVersion
+from app.db.models import Run, RunEvent, Workflow, WorkflowVersion
 from app.engine import analysis_claims as ac
 from app.engine import compiler
 from app.engine.nodes.io import _analysis_claims, _comparability
@@ -271,25 +270,15 @@ async def test_a_governed_formal_run_counts_the_untraced_cited_metric(client, mo
 
 
 @pytest.fixture
-async def shop(tmp_path):
+async def shop(tmp_path, scratch_source):
+    """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，测试结束时再删）。"""
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
     db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, amount REAL);")
     db.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(1, "2026-W37", 100.5), (2, "2026-W37", 200.0)])
     db.commit()
     db.close()
-    async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        else:
-            row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
-        await session.commit()
-        source_id = row.id
-    await engines.invalidate(source_id)
-    yield
-    await engines.invalidate(source_id)
+    return await scratch_source("shop", path)
 
 
 async def test_a_metric_read_from_a_query_cell_is_traced(client, monkeypatch, shop):

backend/tests/test_metric_comparability.py

@@ -4,7 +4,8 @@
 - 同形只差日期 → 有 comparability，早的那期是基期（和表达式里谁先写无关）；同一对快照只查一次；原因按算子的
   analysis.comparable_reasons 拆
 - SQL 结构不同（退款额 / 销售额）、两次是同一条 SQL → 不查、不记日志
-- 算子出错、算不了、算子没有实现（NotImplementedError）→ 不挡口径卡，记一条 metric_comparability_skipped
+- 不同的字面量里没有日期、期次（只差地区、只差门店号）→ 是两段不是两期，不查、不记日志；日期和地区都不同的照查
+- 算子出错（调用错误、意料之外的异常）、算不了 → 不挡口径卡，记一条 metric_comparability_skipped
 
 这里的算子是替身（tests/fixtures/analysis_ops.FakeOps）；真算子跑一遍的见 tests/test_analysis_e2e.py。
 """
@@ -18,9 +19,8 @@ from sqlalchemy import select
 
 from app.analysis.contract import CallError
 from app.core import artifact_store
-from app.data.engine import engines
 from app.db.base import SessionLocal
-from app.db.models import DataSource, Run, RunEvent
+from app.db.models import Run, RunEvent
 from app.engine.nodes.metrics import COMPARABILITY_SKIPPED
 from app.engine.runner import run_manager
 from tests.fixtures.analysis_ops import FakeOps
@@ -40,26 +40,17 @@ async def engine_up():
 
 
 @pytest.fixture(autouse=True)
-async def shop(tmp_path):
+async def shop(tmp_path, scratch_source):
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
-    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, amount REAL, refund REAL);")
-    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", [
-        (1, "2026-10-06", "华东", 500.0, 0.0), (2, "2026-10-08", "华南", 400.0, 20.0),
-        (3, "2026-10-13", "华东", 450.0, 0.0), (4, "2026-10-15", "华北", 300.0, 15.0)])
+    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, store INTEGER, amount REAL, "
+                     "refund REAL);")
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?)", [
+        (1, "2026-10-06", "华东", 1203, 500.0, 0.0), (2, "2026-10-08", "华南", 1204, 400.0, 20.0),
+        (3, "2026-10-13", "华东", 1203, 450.0, 0.0), (4, "2026-10-15", "华北", 1204, 300.0, 15.0)])
     db.commit()
     db.close()
-    async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
-        await session.commit()
-        source_id = row.id
-    await engines.invalidate(source_id)
-    yield
-    await engines.invalidate(source_id)
+    return await scratch_source("shop", path)
 
 
 def node(nid, ntype, **config):
@@ -150,9 +141,59 @@ async def test_different_structures_and_the_same_query_are_not_checked(monkeypat
     assert not [log for log in [*logs, *logs2] if log.get("code") == COMPARABILITY_SKIPPED]
 
 
+async def test_two_segments_of_the_same_week_are_not_checked(monkeypatch):
+    """只差地区（华东对华北的倍数）、只差门店号的两份查询是两段，不是两期：不查、不记日志。日期和地区都不同的照查，两期按
+    日期排先后（不同的那一对里第一处排得出的）。"""
+    ops = FakeOps(monkeypatch)
+    ratio = "cell(nodes.this_week, 0, 'gmv') / cell(nodes.last_week, 0, 'gmv')"
+    east, north = (f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{r}' AND day BETWEEN '2026-10-12' AND "
+                   "'2026-10-18'" for r in ("华东", "华北"))
+    _, card, logs = await run_card(graph([{"id": "times", "name": "华东对华北的倍数", "expression": ratio}],
+                                         this=east, last=north))
+    assert "comparability" not in by_id(card)["times"] and ops.calls == []
+    one, two = (f"SELECT SUM(amount) AS gmv FROM orders WHERE store = {n}" for n in (1203, 1204))
+    _, card, logs2 = await run_card(graph([{"id": "stores", "name": "两家门店之比", "expression": ratio}],
+                                          this=one, last=two))
+    assert "comparability" not in by_id(card)["stores"] and ops.calls == []
+    assert not [log for log in [*logs, *logs2] if log.get("code") == COMPARABILITY_SKIPPED]
+
+    south_last = "SELECT SUM(amount) AS gmv FROM orders WHERE region = '华南' AND day BETWEEN '2026-10-05' AND '2026-10-11'"
+    _, card, _ = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW}], this=east, last=south_last))
+    [call] = ops.calls
+    assert call["request"]["params"] == {"base_query": "last_week", "target_query": "this_week"}
+    assert by_id(card)["wow"]["comparability"]["verdict"] == "comparable"
+
+
+@pytest.mark.parametrize("this, last, pairs, order", [
+    ("region = '华东'", "region = '华南'", [], None),
+    ("store = 1203", "store = 1204", [], None),
+    ("amount > 100", "amount > 200", [], None),
+    ("day BETWEEN '2026-10-12' AND '2026-10-18'", "day BETWEEN '2026-10-05' AND '2026-10-11'",
+     [("2026-10-12", "2026-10-05"), ("2026-10-18", "2026-10-11")], (1, 0)),
+    ("day BETWEEN '2026-10-05' AND '2026-10-11'", "day BETWEEN '2026-10-12' AND '2026-10-18'",
+     [("2026-10-05", "2026-10-12"), ("2026-10-11", "2026-10-18")], (0, 1)),
+    ("region = '华东' AND day >= '2026-10-12'", "region = '华南' AND day >= '2026-10-05'",
+     [("2026-10-12", "2026-10-05")], (1, 0)),
+    ("month = '2026-09'", "month = '2026-10'", [("2026-09", "2026-10")], (0, 1)),
+    ("week = '2026-W41'", "week = '2026-W40'", [("2026-W41", "2026-W40")], (1, 0)),
+    ("ym = 202609", "ym = 202608", [("202609", "202608")], (1, 0)),
+    ("yr = 2025", "yr = 2026", [("2025", "2026")], (0, 1)),
+    # 按运行当天现算的界：排不出先后，表达式里先出现的那份算本期
+    ("day >= date('now', '-7 days')", "day >= date('now', '-14 days')", [("-7 days", "-14 days")], (1, 0)),
+])
+def test_which_literals_are_periods(this, last, pairs, order):
+    from app.engine.nodes.metrics import _dated_pairs, _period_order
+
+    found = _dated_pairs(f"SELECT SUM(amount) AS gmv FROM orders WHERE {this}",
+                         f"SELECT SUM(amount) AS gmv FROM orders WHERE {last}", "sqlite")
+    assert found == pairs
+    if order is not None:
+        assert _period_order(found) == order
+
+
 async def test_an_operator_that_is_not_there_does_not_block_the_card(monkeypatch):
-    """算子没有实现（合并前的骨架抛 NotImplementedError）：口径卡照常算，照实记一条没做成。"""
-    FakeOps(monkeypatch, errors=[NotImplementedError("analysis.run 还没有实现")])
+    """算子抛出意料之外的异常（这里是 NotImplementedError）：口径卡照常算，照实记一条没做成。"""
+    FakeOps(monkeypatch, errors=[NotImplementedError("这种写法还没有实现")])
     _, card, logs = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW, "decimals": 4}]))
     wow = by_id(card)["wow"]
     assert "comparability" not in wow and wow["value"] == pytest.approx((450 + 300 - 900) / 900, abs=1e-4)