虚构库 retail_trends 与分析题集：26 周的零售订单（截止昨天、固定种子），相对上一个整周埋了八个效应；题集 analysis_suite.json 每个效应一题，期望由平台的分析方法在期望 SQL 上算

- tests/fixtures/acceptance/retail_trends.py（python -m tests.fixtures.acceptance.retail_trends <db> [--today]，数据源名 retail_trends，不动 retail.py）：华东线下客单价降到七成带来的销售额下降（主要来自华东）；三成线下订单转到线上造成的退款率辛普森（各渠道都降、整体升，结构为主）；订单数的常规波动；取消单激增；线上销售额连涨十周以上；线上销售额的变化来自订单数；西南一家小店的退款率看着高但样本太少；本周没过完（和上周不可比，周一时本周还没有数据、那一题期望算不了）。订单数按店按周定好、金额在窄区间随机、退款按渠道比例挑，各效应不靠运气
- tests/fixtures/eval/analysis_suite.json：八道题，问法用使用者的说法（上周销售额为什么下降？退款率为什么上升了？上周订单数下降算不算异常？……），各配一个贡献或判定期望，订单数、取消单两题另配一个不必需的「一个数」
- 测试：同一个今天逐字节相同；四个不同的今天（周三、周日、周六、跨年）上用 app.analysis.run 按题集的期望 SQL 核对八个效应；周一那一题照实算不了；走接口导入题集、逐题试算，分析期望都算得出、判定和埋的效应一致；真运行的算期望（作答前后对得上、阈值按设置快照、参数对不上照实说）、读作答运行里的分析结论和引用、重评老结果、从头到尾评一题（助手剧本搭查库 → 分析卡 → 报告，对且可追溯、逐题 E1 等价已证明）

backend/tests/fixtures/acceptance/retail_trends.py

@@ -0,0 +1,170 @@
+"""分析题用的虚构零售库（docs/design/analysis-round/00-总方案.md §3.8、D14）。店名、金额都是编的，和任何真实业务无关。
+
+    python -m tests.fixtures.acceptance.retail_trends <输出 .db 路径> [--today YYYY-MM-DD]
+
+生成「今天」之前 26 周的订单（截止昨天），随机数种子固定：同一个「今天」生成的库逐字节一样。和 retail.py（retail_demo）分开：
+那边的测试断言了它的比例、地区排序，在里面埋效应会改掉后面全部的随机序列，60 天也不够放回 8 周以上的历史里看。
+
+相对「上一个整周」（上周一到上周日，下面叫「上周」，它前一周叫「上上周」）埋了这些效应，各放在不互相干扰的指标或段上：
+
+1. 销售额（已支付的金额）上周比上上周下降，主要来自华东：华东线下上周的客单价降到七成，订单数不变。
+2. 退款率的辛普森：线上、线下各自的退款率上周都比上上周低，整体却升了——上周起三成线下订单转到了线上，线上的退款率本来
+   就高（结构变化为主）。
+3. 订单数（不含取消）的常规波动：线下每周有 ±5% 的起伏，上周比上上周少一点，在常规波动范围内，不该叫异常。
+4. 取消单激增：平常每周二十来单，上周是三倍多，明显超出常规波动。
+5. 线上销售额持续上升：线上订单每周多 4%（上周又加上转过来的那三成），最近十几周周周都涨。
+6. 线上销售额上周的变化主要来自订单数，客单价没变。
+7. 西南只有一家小店，近四周的退款率看着比华东高，样本太少，差异不足以下结论。
+8. 本周还没过完（截止昨天）：拿本周和上周比，不可比。今天是周一时本周还没有数据，那一题的期望算不了。
+
+订单数每家店每周按定好的数生成（不逐笔抽随机），金额在很窄的区间里随机，退款单按每周每个渠道定好的比例挑：各效应不靠运气
+成立。tests/test_retail_trends_fixture.py 在几个不同的「今天」上用平台的算子，按题集的期望 SQL 逐个核对。
+
+评测的分析题集（tests/fixtures/eval/analysis_suite.json）按数据源名 SOURCE_NAME 对上这个库：生成之后登记成这个名字的只读
+数据源，再导入题集（POST /api/evals/suites/import），题目的期望都按这个库算。
+"""
+from __future__ import annotations
+
+import argparse
+import random
+import sqlite3
+from datetime import date, timedelta
+from pathlib import Path
+
+WEEKS = 26
+DAYS = WEEKS * 7
+SEED = 20261021
+#: 分析题集里写的数据源名
+SOURCE_NAME = "retail_trends"
+ONLINE, OFFLINE = "线上", "线下"
+
+#: (门店 id, 店名, 地区)：西南只有一家小店（第 7 条效应的小样本）
+STORES = [
+    (1, "青禾一号店", "华东"), (2, "青禾二号店", "华东"), (3, "青禾三号店", "华东"),
+    (4, "青禾四号店", "华南"), (5, "青禾五号店", "华南"), (6, "青禾六号店", "华南"),
+    (7, "青禾七号店", "华北"), (8, "青禾八号店", "华北"),
+    (9, "青禾九号店", "西南"),
+]
+SMALL_STORE = 9
+#: 每家店每天的线下订单数（平常的水平）
+OFFLINE_DAILY = {1: 16, 2: 14, 3: 12, 4: 14, 5: 12, 6: 10, 7: 12, 8: 10, 9: 1}
+#: 每家店每天的线上订单数（上周的水平；往前每周少 4%）
+ONLINE_DAILY = {1: 7, 2: 6, 3: 5, 4: 6, 5: 5, 6: 5, 7: 5, 8: 4, 9: 0.5}
+ONLINE_GROWTH = 1.04
+#: 线下每周的起伏（±）；上周比上上周再少这么多
+OFFLINE_SWING = 0.05
+LAST_WEEK_DIP = 0.975
+#: 上周起线下订单转到线上的比例
+SHIFT = 0.3
+#: 金额的区间（元）；华东线下上周的客单价乘这个数
+PRICE = (140.0, 180.0)
+EAST_DROP = 0.7
+#: 各渠道的退款率：上上周及以前、上周起
+REFUND_RATE = {ONLINE: (0.10, 0.09), OFFLINE: (0.04, 0.034)}
+#: 每周的取消单数和起伏；上周的倍数
+CANCELLED_WEEKLY = 24
+CANCEL_SWING = 0.15
+CANCEL_SPIKE = 3.2
+
+SCHEMA = """
+CREATE TABLE stores (id INTEGER PRIMARY KEY, name TEXT NOT NULL, region TEXT NOT NULL);
+CREATE TABLE orders (
+  id INTEGER PRIMARY KEY,
+  store_id INTEGER NOT NULL REFERENCES stores(id),
+  order_date DATE NOT NULL,
+  channel TEXT NOT NULL,
+  amount REAL NOT NULL,
+  status TEXT NOT NULL
+);
+"""
+
+
+def _spread(count: int, days: list[date]) -> list[date]:
+    """一周里的 count 单按天摊开：尽量平均，多出来的几单从这一周的头几天起各加一单。"""
+    if not days or count <= 0:
+        return []
+    base, extra = divmod(count, len(days))
+    return [d for i, d in enumerate(days) for _ in range(base + (1 if i < extra else 0))]
+
+
+def build(path: Path | str, today: date | None = None) -> Path:
+    """生成到 path（已存在就覆盖）。today 缺省取运行这一天；订单截止 today 的前一天。"""
+    path = Path(path)
+    if path.exists():
+        path.unlink()
+    today = today or date.today()
+    rng = random.Random(SEED)
+    first, last = today - timedelta(days=DAYS), today - timedelta(days=1)
+    monday = today - timedelta(days=today.weekday())
+    # 第 j 周：本周 j = 0、上周 j = 1……最早的那一周可能不完整。每周的起伏先抽好（个数和「今天」是星期几无关）
+    oldest = (monday - (first - timedelta(days=first.weekday()))).days // 7
+    swing = {j: 1 + rng.uniform(-OFFLINE_SWING, OFFLINE_SWING) for j in range(oldest, -1, -1)}
+    cancel = {j: 1 + rng.uniform(-CANCEL_SWING, CANCEL_SWING) for j in range(oldest, -1, -1)}
+    swing[1] = swing[2] * LAST_WEEK_DIP
+    cancel[1] = CANCEL_SPIKE
+
+    def days_of(j: int) -> list[date]:
+        start = monday - timedelta(days=7 * j)
+        return [d for d in (start + timedelta(days=k) for k in range(7)) if first <= d <= last]
+
+    def price(store: int, channel: str, j: int) -> float:
+        amount = rng.uniform(*PRICE)
+        if j == 1 and channel == OFFLINE and _REGION[store] == "华东":
+            amount *= EAST_DROP
+        return round(amount, 2)
+
+    rows: list[list] = []
+    for j in range(oldest, -1, -1):
+        days = days_of(j)
+        if not days:
+            continue
+        share = len(days) / 7
+        week: dict[str, list[list]] = {ONLINE: [], OFFLINE: []}
+        for store, _, _ in STORES:
+            online = round(7 * share * ONLINE_DAILY[store] * ONLINE_GROWTH ** (1 - j))
+            offline = round(7 * share * OFFLINE_DAILY[store] * swing[j])
+            if j <= 1:
+                moved = round(offline * SHIFT)
+                online, offline = online + moved, offline - moved
+            for channel, count in ((ONLINE, online), (OFFLINE, offline)):
+                for day in _spread(count, days):
+                    row = [store, day.isoformat(), channel, price(store, channel, j), "paid"]
+                    week[channel].append(row)
+                    rows.append(row)
+        # 退款：每个渠道按这一周的比例挑（西南那家小店另算：每周一单线下退款）
+        for channel, pool in week.items():
+            usual = [r for r in pool if r[0] != SMALL_STORE]
+            rate = REFUND_RATE[channel][1 if j <= 1 else 0]
+            for row in rng.sample(usual, round(len(usual) * rate)):
+                row[4] = "refunded"
+        small = [r for r in week[OFFLINE] if r[0] == SMALL_STORE]
+        if small:
+            small[0][4] = "refunded"
+        # 取消单：另外的单（不占上面的订单数），店、渠道、日子随机
+        for _ in range(round(CANCELLED_WEEKLY * share * cancel[j])):
+            store = rng.choice([s for s, _, _ in STORES if s != SMALL_STORE])
+            channel = OFFLINE if rng.random() < 0.6 else ONLINE
+            rows.append([store, rng.choice(days).isoformat(), channel, price(store, channel, j), "cancelled"])
+    rows.sort(key=lambda r: (r[1], r[0], r[2], r[4], r[3]))
+    db = sqlite3.connect(path)
+    db.executescript(SCHEMA)
+    db.executemany("INSERT INTO stores VALUES (?, ?, ?)", STORES)
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?)", [(i, *r) for i, r in enumerate(rows, start=1)])
+    db.commit()
+    db.close()
+    return path
+
+
+_REGION = {store: region for store, _, region in STORES}
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="生成分析题用的虚构零售库")
+    parser.add_argument("path")
+    parser.add_argument("--today", type=date.fromisoformat, default=None)
+    args = parser.parse_args()
+    print(build(args.path, args.today))
+
+
+if __name__ == "__main__":
+    main()

backend/tests/fixtures/eval/analysis_suite.json

@@ -0,0 +1,209 @@
+{
+  "format": "agentlab-eval-suite",
+  "version": 1,
+  "exported_at": "2026-10-21T00:00:00+00:00",
+  "suite": {
+    "name": "零售分析题集",
+    "description": "配 tests/fixtures/acceptance/retail_trends.py 的虚构零售库（数据源名 retail_trends）：上周销售额下降主要来自华东、退款率的结构变化（辛普森）、订单数的常规波动、取消单激增、线上销售额持续上升、线上销售额的变化来自订单数、小样本门店的退款率、本周和上周不可比。期望的来源、判定都由平台的分析方法在期望 SQL 的结果上算，不靠人填。用于分析能力的回归和真实模型验收；店名、金额都是虚构的",
+    "timezone": "Asia/Shanghai",
+    "budget": {}
+  },
+  "cases": [
+    {
+      "seq": 1,
+      "kind": "ask",
+      "question": "上周销售额为什么下降？",
+      "datasources": ["retail_trends"],
+      "tags": ["分析", "贡献", "按地区拆"],
+      "gold": {
+        "kind": "sql",
+        "datasource": "retail_trends",
+        "sql": "SELECT s.region AS 地区, date(o.order_date, 'weekday 0', '-6 days') AS 周, ROUND(SUM(o.amount), 2) AS 销售额 FROM orders o JOIN stores s ON s.id = o.store_id WHERE o.status = 'paid' AND o.order_date BETWEEN date('{{ schedule.last_week_start }}', '-7 days') AND '{{ schedule.last_week_end }}' GROUP BY 地区, 周 ORDER BY 地区, 周"
+      },
+      "expectations": [
+        {
+          "id": "source",
+          "kind": "contribution",
+          "mode": "additive",
+          "params": {"by": ["地区"], "period": "周", "measure": "销售额", "label": "销售额"},
+          "top_n": 1,
+          "label": "上周销售额的下降主要来自哪个地区"
+        }
+      ],
+      "enabled": true
+    },
+    {
+      "seq": 2,
+      "kind": "ask",
+      "question": "退款率为什么上升了？",
+      "datasources": ["retail_trends"],
+      "tags": ["分析", "贡献", "结构变化"],
+      "gold": {
+        "kind": "sql",
+        "datasource": "retail_trends",
+        "sql": "SELECT channel AS 渠道, date(order_date, 'weekday 0', '-6 days') AS 周, SUM(status = 'refunded') AS 退款单数, COUNT(*) AS 订单数 FROM orders WHERE status <> 'cancelled' AND order_date BETWEEN date('{{ schedule.last_week_start }}', '-7 days') AND '{{ schedule.last_week_end }}' GROUP BY 渠道, 周 ORDER BY 渠道, 周"
+      },
+      "expectations": [
+        {
+          "id": "mix",
+          "kind": "contribution",
+          "mode": "ratio",
+          "params": {"by": ["渠道"], "period": "周", "numerator": "退款单数", "denominator": "订单数", "label": "退款率"},
+          "top_n": 1,
+          "label": "上周退款率的上升来自结构变化还是各渠道自身"
+        }
+      ],
+      "enabled": true
+    },
+    {
+      "seq": 3,
+      "kind": "ask",
+      "question": "上周订单数下降算不算异常？",
+      "datasources": ["retail_trends"],
+      "tags": ["分析", "判定", "常规波动"],
+      "gold": {
+        "kind": "sql",
+        "datasource": "retail_trends",
+        "sql": "SELECT date(order_date, 'weekday 0', '-6 days') AS 周, COUNT(*) AS 订单数 FROM orders WHERE status <> 'cancelled' AND order_date BETWEEN date('{{ schedule.this_week_start }}', '-175 days') AND '{{ schedule.last_week_end }}' GROUP BY 周 ORDER BY 周"
+      },
+      "expectations": [
+        {
+          "id": "band",
+          "kind": "verdict",
+          "op": "baseline",
+          "params": {"period": "周", "measure": "订单数", "label": "订单数"},
+          "label": "上周订单数的变化在不在常规波动范围内"
+        },
+        {
+          "id": "orders",
+          "kind": "scalar",
+          "column": "订单数",
+          "row": 24,
+          "required": false,
+          "label": "上周的订单数"
+        }
+      ],
+      "enabled": true
+    },
+    {
+      "seq": 4,
+      "kind": "ask",
+      "question": "上周取消的订单是不是突然变多了？",
+      "datasources": ["retail_trends"],
+      "tags": ["分析", "判定", "异常"],
+      "gold": {
+        "kind": "sql",
+        "datasource": "retail_trends",
+        "sql": "SELECT date(order_date, 'weekday 0', '-6 days') AS 周, COUNT(*) AS 取消单数 FROM orders WHERE status = 'cancelled' AND order_date BETWEEN date('{{ schedule.this_week_start }}', '-175 days') AND '{{ schedule.last_week_end }}' GROUP BY 周 ORDER BY 周"
+      },
+      "expectations": [
+        {
+          "id": "spike",
+          "kind": "verdict",
+          "op": "baseline",
+          "params": {"period": "周", "measure": "取消单数", "label": "取消单数"},
+          "label": "上周取消单数是不是明显超出常规波动"
+        },
+        {
+          "id": "cancelled",
+          "kind": "scalar",
+          "column": "取消单数",
+          "row": 24,
+          "required": false,
+          "label": "上周的取消单数"
+        }
+      ],
+      "enabled": true
+    },
+    {
+      "seq": 5,
+      "kind": "ask",
+      "question": "线上销售额最近是什么走势？",
+      "datasources": ["retail_trends"],
+      "tags": ["分析", "判定", "趋势"],
+      "gold": {
+        "kind": "sql",
+        "datasource": "retail_trends",
+        "sql": "SELECT date(order_date, 'weekday 0', '-6 days') AS 周, ROUND(SUM(amount), 2) AS 线上销售额 FROM orders WHERE status = 'paid' AND channel = '线上' AND order_date BETWEEN date('{{ schedule.this_week_start }}', '-84 days') AND '{{ schedule.last_week_end }}' GROUP BY 周 ORDER BY 周"
+      },
+      "expectations": [
+        {
+          "id": "trend",
+          "kind": "verdict",
+          "op": "trend",
+          "params": {"period": "周", "measure": "线上销售额", "label": "线上销售额"},
+          "label": "最近 12 周线上销售额的走势"
+        }
+      ],
+      "enabled": true
+    },
+    {
+      "seq": 6,
+      "kind": "ask",
+      "question": "上周线上销售额涨了，是订单多了还是客单价高了？",
+      "datasources": ["retail_trends"],
+      "tags": ["分析", "贡献", "因子"],
+      "gold": {
+        "kind": "sql",
+        "datasource": "retail_trends",
+        "sql": "SELECT date(order_date, 'weekday 0', '-6 days') AS 周, SUM(amount) AS 线上销售额, COUNT(*) AS 订单数, SUM(amount) * 1.0 / COUNT(*) AS 客单价 FROM orders WHERE status = 'paid' AND channel = '线上' AND order_date BETWEEN date('{{ schedule.last_week_start }}', '-7 days') AND '{{ schedule.last_week_end }}' GROUP BY 周 ORDER BY 周"
+      },
+      "expectations": [
+        {
+          "id": "factor",
+          "kind": "contribution",
+          "mode": "factor",
+          "params": {"period": "周", "product": "线上销售额", "factors": ["订单数", "客单价"], "label": "线上销售额"},
+          "top_n": 1,
+          "label": "上周线上销售额的变化主要来自订单数还是客单价"
+        }
+      ],
+      "enabled": true
+    },
+    {
+      "seq": 7,
+      "kind": "ask",
+      "question": "最近四周，西南店的退款率是不是比华东高？",
+      "datasources": ["retail_trends"],
+      "tags": ["分析", "判定", "小样本"],
+      "gold": {
+        "kind": "sql",
+        "datasource": "retail_trends",
+        "sql": "SELECT s.region AS 地区, SUM(o.status = 'refunded') AS 退款单数, COUNT(*) AS 订单数 FROM orders o JOIN stores s ON s.id = o.store_id WHERE o.status <> 'cancelled' AND s.region IN ('西南', '华东') AND o.order_date BETWEEN date('{{ schedule.last_week_start }}', '-21 days') AND '{{ schedule.last_week_end }}' GROUP BY 地区 ORDER BY 地区"
+      },
+      "expectations": [
+        {
+          "id": "differ",
+          "kind": "verdict",
+          "op": "differ",
+          "mode": "rate",
+          "params": {"by": ["地区"], "pairs": [["西南", "华东"]], "numerator": "退款单数", "denominator": "订单数", "label": "退款率"},
+          "label": "西南和华东退款率的差异站不站得住"
+        }
+      ],
+      "enabled": true
+    },
+    {
+      "seq": 8,
+      "kind": "ask",
+      "question": "这周和上周比卖得怎么样？",
+      "datasources": ["retail_trends"],
+      "tags": ["分析", "判定", "可比性"],
+      "gold": {
+        "kind": "sql",
+        "datasource": "retail_trends",
+        "sql": "SELECT date(order_date, 'weekday 0', '-6 days') AS 周, COUNT(*) AS 订单数, ROUND(SUM(CASE WHEN status = 'paid' THEN amount ELSE 0 END), 2) AS 销售额 FROM orders WHERE status <> 'cancelled' AND order_date >= '{{ schedule.last_week_start }}' GROUP BY 周 ORDER BY 周"
+      },
+      "expectations": [
+        {
+          "id": "comparable",
+          "kind": "verdict",
+          "op": "comparable",
+          "params": {"period": "周", "base": "{{ schedule.last_week_start }}", "target": "{{ schedule.this_week_start }}", "measure": "订单数"},
+          "label": "本周和上周能不能直接比"
+        }
+      ],
+      "enabled": true
+    }
+  ]
+}

backend/tests/test_eval_analysis_runs.py

@@ -0,0 +1,315 @@
+"""分析期望走真运行（app/eval/gold.py、score.py、runner.py；docs/design/analysis-round/00-总方案.md §3.8）。库是虚构的
+retail_trends（tests/fixtures/acceptance/retail_trends.py，按今天生成）；写作模型、助手都换成剧本，不调任何模型。
+
+- 算期望：期望 SQL 的探针运行封存的那份快照上，用平台的算子（真算子、真现查）算出判定；作答前后两次期望照样对得上（运行当天、
+  现查时刻、输入快照的工件不比）；阈值按交进来的分析设置快照；参数和结果对不上的照实说、整题没有可用的值。
+- 读答案：作答运行台账里的分析结论、报告怎么引用了它（判定语、结果表的格、句末依据）、有判断词的句子；逐题 E1 按分析结论的
+  输入查询比。
+- 重评老结果：明细里没有分析结论那一项的照样重评（从封存重新读）。
+- 从头到尾：助手剧本搭「查库 → 分析卡 → 报告」，报告引用判定语；评测给这一题「对且可追溯」，逐题 E1 等价已证明，作答运行和
+  算期望用的是同一份分析设置快照（记进指纹）。
+"""
+from __future__ import annotations
+
+import json
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import pytest
+from langchain_core.messages import AIMessageChunk
+from sqlalchemy import select
+
+from app.core import artifact_store
+from app.db.base import SessionLocal
+from app.db.models import Run
+from app.db.models_eval import EvalCase, EvalResult, EvalRun
+from app.engine.runner import run_manager
+from app.eval import cases, gold, judgements, runner, score
+from app.eval.gold import Context
+from tests import eval_samples as es
+from tests.fixtures.acceptance import retail_trends
+
+TZ = "Asia/Shanghai"
+SUITE = json.loads((Path(__file__).parent / "fixtures" / "eval" / "analysis_suite.json").read_text(encoding="utf-8"))
+#: 题集里的第 1 题（上周销售额下降主要来自华东）和第 3 题（订单数的常规波动）
+REGION = next(c for c in SUITE["cases"] if c["seq"] == 1)
+ORDERS = next(c for c in SUITE["cases"] if c["seq"] == 3)
+
+
+@pytest.fixture(autouse=True)
+async def engine_up(monkeypatch):
+    from app.eval import snapshots
+
+    monkeypatch.setattr(snapshots, "MAX_SOURCES", 10_000)
+    await run_manager.setup()
+    yield
+    await runner.manager.shutdown()
+    await es.drop_suites()
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def source(tmp_path, scratch_source):
+    path = retail_trends.build(tmp_path / "trends.db", date.today())
+    sid = await scratch_source(retail_trends.SOURCE_NAME, path, schema=True)
+    return {"id": sid, "name": retail_trends.SOURCE_NAME, "path": path}
+
+
+def now() -> datetime:
+    return datetime.now(ZoneInfo(TZ))
+
+
+def ctx(**kw) -> Context:
+    return Context(as_of=now(), tz=TZ, actor="评测测试", suite_id="s1", suite_name="虚构分析题集", **kw)
+
+
+def case_of(raw: dict, source_id: str, *, expectations: list[dict] | None = None, sql: str | None = None) -> dict:
+    fields = cases.normalize_case({"question": raw["question"], "datasource_ids": [source_id],
+                                   "gold": {"kind": "sql", "datasource_id": source_id, "sql": sql or raw["gold"]["sql"]},
+                                   "expectations": expectations or raw["expectations"]})
+    return {**fields, "id": None, "seq": raw["seq"]}
+
+
+def rendered(sql: str) -> str:
+    return cases.render(sql, cases.variable_values(now(), TZ))
+
+
+# --------------------------------------------------------------------------
+# 算期望
+# --------------------------------------------------------------------------
+
+
+async def test_the_gold_runs_the_platform_analysis_on_the_probe_snapshot(source):
+    import sqlite3
+
+    case = case_of(ORDERS, source["id"])
+    before = await gold.compute_gold(case, ctx())
+    assert before.ok, before.reason
+    band, orders = before.sets[0].values["band"], before.sets[0].values["orders"]
+    assert (band["kind"], band["status"], band["judgement"]) == ("verdict", "ok", "normal"), band
+    assert band["phrase"].startswith("订单数本期变化") and "常规波动范围内" in band["phrase"]
+    result = band["result"]
+    assert result["inputs"][0]["ref"] == judgements.GOLD_REF and result["inputs"][0]["artifact"] == before.sets[0].snapshot
+    assert result["params"]["today"] == date.today().isoformat()
+    assert result["settings"]["k_warn"] == 2.0, "没有设置快照（试算）时现读设置，缺省 2 倍"
+    # 本期齐不齐按现查的表范围判（SQL 里按日期算出来的界也在库里算一次），都记进结论
+    assert [(p["table"], p["column"], p["basis"]) for p in result["probes"] if "table" in p] == \
+        [("orders", "order_date", "live")], result["probes"]
+    assert all(p.get("at") for p in result["probes"])
+    with sqlite3.connect(source["path"]) as db:
+        direct = db.execute(rendered(ORDERS["gold"]["sql"])).fetchall()
+    assert orders["value"] == direct[24][1] and len(direct) == 25
+
+    # 作答后再算一次：快照的工件（记着查询用时）、现查时刻都不同，期望照样对得上
+    after = await gold.compute_gold(case, ctx(role="gold_after"))
+    assert after.sets[0].snapshot != before.sets[0].snapshot
+    assert gold.settle(before, after).status == "ok"
+
+
+async def test_the_settings_snapshot_decides_the_thresholds(source):
+    case = case_of(ORDERS, source["id"])
+    outcome = await gold.compute_gold(case, ctx(analysis={"min_history": 30, "max_history": 52}))
+    assert outcome.ok, outcome.reason
+    band = outcome.sets[0].values["band"]
+    assert band["status"] == "refused" and band["reason_code"] == "history_short", band
+    assert band["result"]["settings"]["min_history"] == 30
+    graded = score.score_case(case, gold.settle(outcome, outcome).to_dict(), score.Answer(run_id="r"), path="ran")
+    assert graded.scores[0]["outcome"] == "ungradable" and graded.scores[0]["reason"].startswith("期望算不了：")
+
+
+async def test_parameters_that_do_not_fit_the_sql_are_said_plainly(source):
+    wrong_column = case_of(REGION, source["id"], expectations=[
+        {"kind": "contribution", "mode": "additive", "params": {"by": ["门店"], "period": "周", "measure": "销售额"}}])
+    outcome = await gold.compute_gold(wrong_column, ctx())
+    assert outcome.status == "no_value" and "期望 SQL 里没有「门店」这一列" in outcome.reason, outcome.reason
+    three_weeks = REGION["gold"]["sql"].replace("'-7 days'", "'-14 days'")
+    outcome = await gold.compute_gold(case_of(REGION, source["id"], sql=three_weeks), ctx())
+    assert outcome.status == "no_value" and "有 3 期" in outcome.reason, outcome.reason
+
+
+# --------------------------------------------------------------------------
+# 读答案、重评老结果
+# --------------------------------------------------------------------------
+
+REPORT = ("## 上周销售额\n\n上周销售额下降：[[a:A1]]。华东的变化是 [[v:A1.r0.delta]]。\n\n"
+          "下降主要来自华东，其余地区变化不大。[[see:A1]]")
+
+
+def _node(nid: str, ntype: str, **config) -> dict:
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+
+def _card_item() -> dict:
+    values = cases.variable_values(now(), TZ)
+    target = values["last_week_start"]
+    base = (date.fromisoformat(target) - timedelta(days=7)).isoformat()
+    return {"id": "where", "op": "decompose", "mode": "additive", "from": "q",
+            "params": {"by": ["地区"], "period": "周", "base": base, "target": target, "measure": "销售额",
+                       "label": "销售额"}}
+
+
+def answer_graph() -> dict:
+    nodes = [_node("start", "input"),
+             _node("q", "tool", tool=f"db_query__{retail_trends.SOURCE_NAME}", args={"sql": rendered(REGION["gold"]["sql"])}),
+             _node("card", "analysis", items=[_card_item()]),
+             _node("write", "report", instructions="写上周的周报"),
+             _node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}])]
+    ids = [n["id"] for n in nodes]
+    return {"nodes": nodes, "edges": [{"source": a, "target": b} for a, b in zip(ids, ids[1:])]}
+
+
+async def _answer_run(monkeypatch) -> Run:
+    from tests.test_analysis_card import wait
+
+    es.use_report(monkeypatch, REPORT)
+    started = await run_manager.start(graph=answer_graph(), input_payload={})
+    row = await wait(started.id)
+    assert row.status == "succeeded", row.error
+    return row
+
+
+async def test_reading_the_analyses_of_the_answer_run(monkeypatch, source):
+    row = await _answer_run(monkeypatch)
+    answer = await score.read_answer(row.id)
+    assert answer.problem is None and answer.report == "write"
+    [record] = answer.analyses
+    assert (record["alias"], record["op"], record["mode"], record["judgement"]) == ("A1", "decompose", "additive",
+                                                                                     "source")
+    assert record["subjects"][0] == "华东" and record["direction"] == "down"
+    assert record["cited"] == ["phrase", "cell", "see"], record
+    # 判定语不是答案里的数；结果表的那一格是，维度是那一段（华东）
+    [cell] = answer.values
+    assert cell["kind"] == "cell" and cell["dims"] == {"地区": "华东"} and cell["artifact"] == record["artifact"]
+    # 逐题 E1 拿分析结论的输入查询比
+    [query] = answer.queries
+    assert answer.cited == [query["artifact"]] and record["inputs"] == [query["artifact"]]
+    [sentence] = answer.sentences
+    assert sentence["analyses"] == ["A1"] and "主要来自华东" in sentence["text"]
+
+    case = case_of(REGION, source["id"])
+    before = await gold.compute_gold(case, ctx())
+    settled = gold.settle(before, before).to_dict()
+    graded = score.score_case(case, settled, answer, path="ran")
+    assert graded.verdict == "pass" and graded.traced, graded.scores
+    assert graded.scores[0]["found"]["alias"] == "A1"
+    match = await score.sql_match(case, settled, answer)
+    assert match["status"] == "proven" and match["basis"] == "cited", match
+
+
+async def test_rescoring_old_results_reads_the_analyses_from_the_seal(monkeypatch, source):
+    from app.eval import rescore
+    from app.eval.cases import case_hash
+
+    row = await _answer_run(monkeypatch)
+    case = case_of(REGION, source["id"])
+    settled = gold.settle(*(2 * [await gold.compute_gold(case, ctx())])).to_dict()
+    old = score.Answer(run_id=row.id, text="老的答案").to_dict()
+    old.pop("analyses")
+    old.pop("sentences")
+    details = await artifact_store.put_json({"format": 1, "turn": {"reply": None}, "answer": old}, kind="eval_result")
+    suite_id = await es.make_suite(source["id"], cases=[{k: case[k] for k in ("question", "datasource_ids", "gold",
+                                                                                "expectations")}])
+    async with SessionLocal() as session:
+        stored = (await session.execute(select(EvalCase).where(EvalCase.suite_id == suite_id))).scalar_one()
+        ev = EvalRun(suite_id=suite_id, status="succeeded", fingerprint={"scorer_version": 2})
+        session.add(ev)
+        await session.flush()
+        session.add(EvalResult(eval_run_id=ev.id, case_id=stored.id, case_hash=case_hash(case), status="ok",
+                               path="ran", run_id=row.id, verdict="fail", details_artifact=details, gold=settled,
+                               scores=[{"id": "source", "kind": "contribution", "required": True, "outcome": "wrong"}]))
+        await session.commit()
+        ev_id = ev.id
+    assert await rescore.rescore(ev_id) == 1
+    async with SessionLocal() as session:
+        result = (await session.execute(select(EvalResult).where(EvalResult.eval_run_id == ev_id))).scalar_one()
+        assert result.verdict == "pass" and result.scores[0]["outcome"] == "traced", result.scores
+        assert (await session.get(EvalRun, ev_id)).fingerprint["scorer_version"] == score.SCORER_VERSION == 3
+        assert (await session.get(EvalRun, ev_id)).summary["outcomes_by_kind"] == {"contribution": {"traced": 1}}
+
+
+# --------------------------------------------------------------------------
+# 从头到尾
+# --------------------------------------------------------------------------
+
+
+def _line(op: dict) -> AIMessageChunk:
+    return AIMessageChunk(content=json.dumps(op, ensure_ascii=False) + "\n")
+
+
+class AnalysisCopilot:
+    """助手剧本：直接搭「入口 → 查库 → 分析卡 → 报告 → 出口」并运行；上线前自查直接过。"""
+
+    async def astream(self, messages):
+        if "上线前自查" in messages[1][1]:
+            yield _line({"op": "done"})
+            return
+        yield _line({"op": "plan", "summary": "按地区拆上周销售额的变化"})
+        graph = answer_graph()
+        for n in graph["nodes"]:
+            yield _line({"op": "add_node", "node": {"id": n["id"], "type": n["type"], "label": n["data"]["label"],
+                                                    "config": n["data"]["config"]}})
+        for e in graph["edges"]:
+            yield _line({"op": "add_edge", "edge": e})
+        yield _line({"op": "done", "explanation": "拆一下", "run": True})
+
+
+async def test_an_analysis_case_end_to_end(monkeypatch, source):
+    from app.api.settings import analysis_settings
+
+    es.use_copilot(monkeypatch, AnalysisCopilot())
+    es.use_report(monkeypatch, REPORT)
+    case = case_of(REGION, source["id"])
+    suite_id = await es.make_suite(source["id"], cases=[{k: case[k] for k in ("question", "datasource_ids", "gold",
+                                                                                "expectations")}])
+    from app.db.models_eval import EvalSuite
+    from app.eval.variants import normalize_variant
+
+    async with SessionLocal() as session:
+        suite = await session.get(EvalSuite, suite_id)
+        [ev] = await runner.start(session, suite, [normalize_variant({})], actor="评测测试")
+        expected_settings = await analysis_settings(session)
+    await es.wait_eval(ev.id)
+    async with SessionLocal() as session:
+        ev = await session.get(EvalRun, ev.id)
+        [item] = (await session.execute(select(EvalResult).where(EvalResult.eval_run_id == ev.id))).scalars()
+        answer_run = await session.get(Run, item.run_id)
+    assert ev.status == "succeeded", (ev.error, item.error)
+    assert item.path == "ran" and item.verdict == "pass", (item.error, item.scores)
+    [scored] = item.scores
+    assert scored["outcome"] == "traced" and scored["want"]["subjects"] == ["华东"], scored
+    assert item.sql_match["status"] == "proven", item.sql_match
+    # 算期望和作答用同一份分析设置快照，记进指纹
+    assert ev.fingerprint["analysis"] == expected_settings
+    assert answer_run.options["analysis"] == expected_settings
+    assert item.gold["before"]["sets"][0]["values"]["source"]["result"]["settings"]["main_share"] == \
+        expected_settings["main_share"]
+    assert ev.summary["outcomes_by_kind"] == {"contribution": {"traced": 1}}
+    assert ev.fingerprint["scorer_version"] == 3
+
+
+async def test_suites_without_analysis_expectations_take_no_settings_snapshot(monkeypatch, tmp_path):
+    from tests.fixtures.sources import drop_source
+
+    es.use_copilot(monkeypatch, es.Copilot("none"))
+    name, sid = await es.make_source(tmp_path)
+    try:
+        es.use_copilot(monkeypatch, es.Copilot(name))
+        es.use_report(monkeypatch)
+        suite_id = await es.make_suite(sid)
+        from app.db.models_eval import EvalSuite
+        from app.eval.variants import normalize_variant
+
+        async with SessionLocal() as session:
+            suite = await session.get(EvalSuite, suite_id)
+            [ev] = await runner.start(session, suite, [normalize_variant({})], actor="评测测试")
+        await es.wait_eval(ev.id)
+        async with SessionLocal() as session:
+            ev = await session.get(EvalRun, ev.id)
+            [item] = (await session.execute(select(EvalResult).where(EvalResult.eval_run_id == ev.id))).scalars()
+            answer_run = await session.get(Run, item.run_id)
+        assert item.verdict == "pass", item.error
+        assert "analysis" not in ev.fingerprint and "analysis" not in (answer_run.options or {}), \
+            "别的题集的指纹、作答运行的选项和以前一样"
+    finally:
+        await drop_source(sid)

backend/tests/test_eval_analysis_suite.py

@@ -0,0 +1,62 @@
+"""仓库里附的分析题集（tests/fixtures/eval/analysis_suite.json）：在虚构库 retail_trends（按今天生成）上导入，每道题都试算
+通过，分析期望都算得出来、判定和库里埋的效应一致（导入走接口：按数据源名对上、逐题过校验、试算一次）。今天是周一时本周还
+没有数据，「这周和上周比」那一题的期望算不了（照实记，这一项不能评分），导入照样成功。"""
+from __future__ import annotations
+
+import json
+from datetime import date
+from pathlib import Path
+from urllib.parse import quote
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.engine.runner import run_manager
+from app.eval import judgements
+from app.main import app
+from tests import eval_samples as es
+from tests.fixtures.acceptance import retail_trends
+
+SUITE_FILE = Path(__file__).parent / "fixtures" / "eval" / "analysis_suite.json"
+ACTOR = {"X-Actor": quote("评测员甲")}
+#: 每道题的分析期望该算出的判定类（库里埋的效应，见 retail_trends 的说明）
+WANTED = {1: "source", 2: "mix", 3: "normal", 4: "anomaly", 5: "rising", 6: "source", 7: "no_evidence",
+          8: "incomparable"}
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await es.drop_suites()
+    await run_manager.shutdown()
+
+
+async def test_the_analysis_suite_imports_and_every_expectation_computes(tmp_path, scratch_source):
+    path = retail_trends.build(tmp_path / "trends.db", date.today())
+    sid = await scratch_source(retail_trends.SOURCE_NAME, path, schema=True)
+    content = json.loads(SUITE_FILE.read_text(encoding="utf-8"))
+    assert {c["seq"] for c in content["cases"]} == set(WANTED)
+    assert all(c["gold"]["datasource"] == retail_trends.SOURCE_NAME for c in content["cases"])
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        res = await client.post("/api/evals/suites/import", json={"content": content}, headers=ACTOR)
+        assert res.status_code == 201, res.text
+        assert res.json()["imported"] == len(WANTED) and res.json()["skipped"] == []
+        suite = res.json()["suite"]
+        assert all(c["gold"]["datasource_id"] == sid for c in suite["cases"])
+        monday = date.today().weekday() == 0
+        for case in suite["cases"]:
+            assert case["trial"]["status"] == "ok", (case["seq"], case["trial"])
+            trial = (await client.post(f"/api/evals/suites/{suite['id']}/cases/{case['id']}/trial")).json()["trial"]
+            assert trial["status"] == "ok", (case["seq"], trial["reason"])
+            [values] = [s["values"] for s in trial["sets"]]
+            found = [v for v in values.values() if v["kind"] in judgements.ANALYSIS_KINDS]
+            assert len(found) == 1, (case["seq"], values)
+            if monday and case["seq"] == 8:
+                assert found[0]["status"] == "refused", found[0]
+                continue
+            assert found[0]["status"] == "ok" and found[0]["judgement"] == WANTED[case["seq"]], (case["seq"], found[0])
+            # 期望值里没有人填的数：来源、判定都是算子在期望 SQL 上算的，判定语由系统渲染
+            assert found[0]["phrase"] and found[0]["result"]["inputs"][0]["ref"] == judgements.GOLD_REF
+            for exp in case["expectations"]:
+                assert not set(exp) & set(judgements.JUDGEMENTS) and "subjects" not in exp

backend/tests/test_retail_trends_fixture.py

@@ -0,0 +1,105 @@
+"""分析题用的虚构零售库（tests/fixtures/acceptance/retail_trends.py）：同一个「今天」生成的库一样；埋的八个效应在题集
+（tests/fixtures/eval/analysis_suite.json）的期望 SQL 上用平台的算子（app.analysis.run，缺省阈值）算出来真的成立，换几个
+「今天」也成立（星期几不同、跨年）。表的日期范围按库里的实际最小、最大值现查（同探查器交回的形状）。"""
+from __future__ import annotations
+
+import json
+import sqlite3
+from datetime import date, datetime
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import pytest
+
+import app.analysis as analysis
+from app.analysis.contract import DEFAULTS
+from app.eval import cases, judgements
+from tests.fixtures.acceptance import retail_trends
+
+SUITE = json.loads((Path(__file__).parent / "fixtures" / "eval" / "analysis_suite.json").read_text(encoding="utf-8"))
+#: 几个不同的「今天」：周三、周日、周六，和跨年的那一周
+TODAYS = [date(2026, 10, 21), date(2026, 10, 25), date(2026, 7, 4), date(2027, 1, 6)]
+
+
+def _rows(path):
+    with sqlite3.connect(path) as db:
+        return db.execute("SELECT * FROM orders ORDER BY id").fetchall()
+
+
+async def _judge(path: Path, today: date) -> dict[int, dict]:
+    """题集每道题的分析期望在这个库上算出来的期望值：{题号: judgements.expected}。"""
+    db = sqlite3.connect(path)
+    as_of = datetime(today.year, today.month, today.day, 9, 0, tzinfo=ZoneInfo("Asia/Shanghai"))
+    variables = cases.variable_values(as_of, "Asia/Shanghai")
+
+    async def probe(source: str, table: str, column: str) -> dict:
+        lo, hi = db.execute(f"SELECT MIN({column}), MAX({column}) FROM {table}").fetchone()
+        return {"source": source, "table": table, "column": column, "min": lo, "max": hi, "basis": "live",
+                "at": as_of.isoformat()}
+
+    out: dict[int, dict] = {}
+    for raw in SUITE["cases"]:
+        fields = cases.normalize_case({"question": raw["question"], "expectations": raw["expectations"],
+                                       "gold": {"kind": "sql", "datasource_id": "x", "sql": raw["gold"]["sql"]}})
+        sql = cases.render(raw["gold"]["sql"], variables)
+        cursor = db.execute(sql)
+        snapshot = {"columns": [d[0] for d in cursor.description], "rows": [list(r) for r in cursor.fetchall()],
+                    "sql": sql, "source": retail_trends.SOURCE_NAME, "dialect": "sqlite", "masked": []}
+        [exp] = [e for e in fields["expectations"] if e["kind"] in judgements.ANALYSIS_KINDS]
+        request = judgements.request_of(exp, variables, snapshot)
+        result = await analysis.run(request, lambda name: snapshot if name == judgements.GOLD_REF else None,
+                                    settings=DEFAULTS, context={"today": today, "probe": probe}, refuse_data_errors=True)
+        out[raw["seq"]] = judgements.expected(exp, result)
+    db.close()
+    return out
+
+
+def _caveats(value: dict) -> set[str]:
+    return {c["code"] for c in value["result"].get("caveats") or []}
+
+
+def test_the_same_day_builds_the_same_database(tmp_path):
+    a = retail_trends.build(tmp_path / "a.db", TODAYS[0])
+    b = retail_trends.build(tmp_path / "b.db", TODAYS[0])
+    assert a.read_bytes() == b.read_bytes() and len(_rows(a)) > 20_000
+    with sqlite3.connect(a) as db:
+        first, last = db.execute("SELECT MIN(order_date), MAX(order_date) FROM orders").fetchone()
+        stores = db.execute("SELECT COUNT(*) FROM stores WHERE region = '西南'").fetchone()[0]
+    assert (date.fromisoformat(last) - TODAYS[0]).days == -1, "截止昨天"
+    assert (TODAYS[0] - date.fromisoformat(first)).days == retail_trends.DAYS and stores == 1
+
+
+@pytest.mark.parametrize("today", TODAYS, ids=[d.isoformat() for d in TODAYS])
+async def test_the_effects_hold_on_the_suite_sql(tmp_path, today):
+    found = await _judge(retail_trends.build(tmp_path / "trends.db", today), today)
+    assert all(v["status"] == "ok" for v in found.values()), {k: v.get("reason") for k, v in found.items()}
+    # 1 上周销售额下降，主要来自华东
+    assert (found[1]["judgement"], found[1]["subjects"], found[1]["direction"]) == ("source", ["华东"], "down")
+    # 2 退款率上升来自结构变化，各渠道自身的退款率都在降（辛普森）
+    assert (found[2]["judgement"], found[2]["direction"]) == ("mix", "up") and "simpson" in _caveats(found[2])
+    rows = found[2]["result"]["table"]["rows"]
+    columns = found[2]["result"]["table"]["columns"]
+    r_base, r_target = columns.index("r_base"), columns.index("r_target")
+    assert all(float(r[r_target]) < float(r[r_base]) for r in rows), rows
+    # 3 订单数在常规波动范围内（不该叫异常）
+    assert found[3]["judgement"] == "normal" and found[3]["result"]["verdict"]["code"] == "within_band"
+    # 4 取消单明显超出常规波动，往上
+    assert (found[4]["judgement"], found[4]["direction"], found[4]["verdict"]) == ("anomaly", "up", "far_outside_band")
+    # 5 线上销售额呈上升趋势，末尾连涨十周以上
+    values = found[5]["result"]["values"]
+    assert found[5]["judgement"] == "rising" and int(values["run_length"]) >= 10 and values["run_direction"] == "up"
+    # 6 线上销售额的变化主要来自订单数，不是客单价
+    assert (found[6]["judgement"], found[6]["subjects"], found[6]["direction"]) == ("source", ["订单数"], "up")
+    # 7 西南（一家小店）看着比华东高，但样本太少，差异不足以下结论
+    assert found[7]["judgement"] == "no_evidence" and found[7]["subjects"] == ["西南", "华东"]
+    a_value, b_value = (float(found[7]["result"]["values"][k]) for k in ("a", "b"))
+    assert a_value > b_value and "small_sample" in _caveats(found[7])
+    # 8 本周还没过完：和上周不可比
+    assert found[8]["judgement"] == "incomparable" and "还差" in found[8]["phrase"]
+
+
+async def test_on_a_monday_this_week_has_no_data_yet(tmp_path):
+    monday = date(2026, 10, 26)
+    found = await _judge(retail_trends.build(tmp_path / "trends.db", monday), monday)
+    assert found[8]["status"] == "refused" and found[8]["reason_code"] == "input_unavailable", found[8]
+    assert all(found[seq]["status"] == "ok" for seq in range(1, 8)), "别的效应照样成立"