问数据建图用量不再是下限；评测通知链到评测详情；评测接口补界面要的几样

用量（第 1 波遗留）：查清楚了，丢的不是浏览器那边——问数据页一直读到 SSE 结束、usage 那一条收得到——而是服务端生成过程
（api/copilot_gen._iter_ops）读到 done、look、ask 那一行就关掉模型的流，流末尾那一段才报的用量（Anthropic 的输出 token、
OpenAI 兼容系的 stream_usage）一次都拿不到，每次建图都记成「估不出」，这一轮的花费多半是下限。改成交出那一行之后先把流读完
再关（_drain，内容丢掉）：照约定那一行就是最后一行，读完只多收用量那一段；模型在那一行之后还往下写超过 DRAIN_CHARS 字闲话的
当场关流、照实记成没报全（不为闲话付钱，也不当成报全了）。评测的作答（app/eval/answer.Driver）走同一个 answer_ops，
用量跟着报全，和问数据页同一个口径。test_usage_meter：原先断言「输出 token 为 0、记成估不出」的那条改成报全；另加 look 后、
闲话超限、评测和问数据页收到同一份用量三条；去掉修复回跑，三条都抓得住。providers/usage.py 的说明跟着改。

通知：notify_eval 的链接从定时任务的触发历史改成这次评测的详情页 /evals/runs/{id}（评测页这一波上线）；测试设上对外访问地址、
断言链接（原来的断言在没设地址时对空列表恒真）。

评测接口（只加不改，app/api/evals.py）：题集列表和详情带 latest（最近一次评测）、trend（同一变体最近几次评完的准确率，旧的
在前，换了模型、目录的不连成一条线）、schedules（目标是它的定时任务）；评测详情带 actions（停止、只重试没做完的、重新评分、
钉住 / 取消基线能不能做，和接口拒收的条件同一份）和没做完的项数；POST /suites/{id}/cases/preview 写 SQL 定期望的预览
（同期望的 SQL 探针，交回列和前 50 行，遮罩列照保存时那一道拒收、没过守卫的指明是哪一格，都不回 422）；GET /signals
「口径命中与重复度」的判据和分档、窗口趋势（数只出自 ledger/stats.compute_stats）、成员不少于 3 的问题类（只出自
chat/questions，评测运行不进）。为什么放服务端：界面层不放逻辑，统计、能不能做都由服务端给。tests/test_evals_ui_api.py。

backend/app/api/copilot_gen.py

@@ -41,6 +41,11 @@ _ASK_OPS = ("ask", "bp_ask")
 _STOP_OPS = ("done", "look", *_ASK_OPS)
 #: 不占次数的查看回话（need 没写、次数用完）连着来几次就不再接着写：模型在原地打转，再要下去只是花钱
 _IDLE_LOOKS = 2
+#: 读到停下的那一行（done、look、ask）之后，最多再收多少字。照约定那一行就是这一次输出的最后一行，后面只剩流末尾那一段
+#: 用量（Anthropic 的输出 token、OpenAI 兼容系开了 stream_usage 的用量都在最后一段才报）：读完它，这一次的花费才报得全，
+#: 问数据页这一轮的花费、评测按角色的成本才不是下限。超过这么多字说明模型还在往下写闲话：不为它付钱，当场关流，这一次的
+#: 用量记成没报全（providers/usage.py 照实记估不出，不当成报全了）
+DRAIN_CHARS = 400
 
 
 # ==========================================================================
@@ -72,9 +77,24 @@ class CopilotIncomplete(Exception):
         self.detail = detail
 
 
+async def _drain(stream: Any, already: int) -> None:
+    """停下的那一行之后把模型的流读完（内容一概不要），只为收到流末尾才报的用量；读到超过 DRAIN_CHARS 字（模型还在写闲话）
+    就不读了，交回调用方关流。already 是缓冲里那一行后面已经收到的字数。"""
+    from app.engine.state import message_text as _text
+    from app.engine.state import thinking_text
+
+    seen = already
+    async for chunk in stream:
+        seen += len(_text(chunk)) + len(thinking_text(chunk) or "")
+        if seen > DRAIN_CHARS:
+            return
+
+
 async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None = None):
     """流式读模型输出，按行切出操作。done、look、ask 之后即停——后面就算有闲话也不要了；look 之后停下也就是
-    「读到这一行就停止本次输出」，模型那条流随之关掉。
+    「读到这一行就停止本次输出」。停下之后不马上关流：交出那一行、调用方接着要下一条时先把流读完（_drain，内容丢掉，最多
+    DRAIN_CHARS 字），流末尾那一段用量才收得到——以前读到这一行就关流，输出 token 一个都拿不到，问数据这一轮的花费多半是下限
+    （第 1 波遗留）。
 
     除了操作，还把模型的思考原样转出去（thinking_text：Claude 的 thinking 块、OpenAI 兼容系的 reasoning_content），
     不把非 JSON 行当成思考。lines 给了的话，认出来的操作行原样记进去：接着写时作为 assistant 消息交回模型。
@@ -101,6 +121,7 @@ async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None
                         lines.append(line.strip())
                     yield op
                     if op.get("op") in _STOP_OPS:
+                        await _drain(stream, len(buffer))
                         return
             cut = _cut_off(chunk) or cut
     op = _parse_op_line(buffer)

backend/app/api/evals.py

@@ -24,6 +24,17 @@
   GET /evaluations/{id}/compare?base=（和上一次或点名的那一次比：指纹差异、逐题转移和归因）；GET /pairs/{pair_id}（A/B 配对表和
   精确 McNemar）；POST /suites/{id}/cleanup（按保留规则清理更早的明细）。「口径命中与重复度」的 E1–E4 直接用口径账本的
   GET /api/ledger/stats（总方案 X2：只有一份实现）。
+
+第 4 波评测页（细案 01 第 18 项）缺的几样，补在这里（统计、能不能做都由这里给，界面只显示）：
+- GET /suites 每个题集另带 latest（最近一次评测的状态和准确率）、trend（同一变体最近几次评完的准确率，旧的在前）、schedules
+  （目标是这个题集的定时任务）——列表页不再为每个题集再取一遍历次评测；
+- GET /evaluations/{id} 另带 actions：停止、只重试没做完的、重新评分、钉住 / 取消基线现在能不能做（和那几个接口拒收的条件
+  同一份），以及 open（没做完的项数）；
+- POST /suites/{id}/cases/preview：写 SQL 定期望时的第一步——按此刻渲染、过只读守卫、跑一次和期望同一条路的 SQL 探针，交回
+  结果的列和前几行（界面上点选格或列）；用到遮罩列的照保存时那一道拒收、照实写原因（不回 422）；
+- GET /signals：「口径命中与重复度」页签的判据（细案 copilot-look-and-caliber-ledger §3.8 的参考线：E1 ≥30% / <10%，
+  E3 ≥20%）、当前窗口和往前几个窗口各自落在哪一档、成员不少于 3 的问题类（可以加入题集或提取模板）。数都出自
+  ledger/stats.compute_stats 和 chat/questions（全站各一份），这里只按窗口调它们、按判据分档。
 """
 from __future__ import annotations
 
@@ -37,7 +48,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.coded import CodedHTTPException
 from app.api.deps import actor_of
-from app.db.base import get_session, new_id, utcnow
+from app.db.base import SessionLocal, get_session, new_id, utcnow
 from app.db.models_eval import EvalCase, EvalCatalogSnapshot, EvalRun, EvalSuite
 from app.eval import cases as case_rules
 from app.eval import gold as gold_rules
@@ -223,8 +234,70 @@ async def _insert_case(session: AsyncSession, suite: EvalSuite, fields: dict[str
 # ==========================================================================
 
 
+#: 列表里每个题集最多看最近几次评测（最近一次、趋势都从这几次里取）；趋势最多几个点
+OVERVIEW_RUNS = 30
+TREND_POINTS = 8
+
+
+def _rate_of(run: EvalRun) -> dict[str, Any] | None:
+    """一次评测总表里的准确率（stats.rate 的形状：passed、n、rate、ci）；还没有总表的为 None。"""
+    acc = (run.summary or {}).get("accuracy")
+    return dict(acc) if isinstance(acc, dict) else None
+
+
+async def _overview(session: AsyncSession, suite_ids: list[str]) -> dict[str, dict[str, Any]]:
+    """每个题集：最近一次评测（不论结束没有）、同一变体最近几次评完的准确率（趋势，旧的在前）、目标是它的定时任务。
+
+    趋势只连同一变体（fingerprint.variant_key，和上一次比挑基准同一个认法）的几次：换了模型、换了目录的那几次不连成一条线，
+    免得把「换了配置」读成「变好了 / 变差了」。A/B 的几臂各是一次评测，同一变体的那一臂才进趋势。"""
+    from app.db.models import Schedule
+    from app.eval import runner
+    from app.scheduler import cron
+
+    out: dict[str, dict[str, Any]] = {sid: {"latest": None, "trend": [], "schedules": []} for sid in suite_ids}
+    if not suite_ids:
+        return out
+    ranked = select(EvalRun, func.row_number().over(partition_by=EvalRun.suite_id,
+                                                    order_by=EvalRun.created_at.desc()).label("n")) \
+        .where(EvalRun.suite_id.in_(suite_ids)).subquery()
+    from sqlalchemy.orm import aliased
+
+    run_alias = aliased(EvalRun, ranked)
+    rows = list((await session.execute(select(run_alias).where(ranked.c.n <= OVERVIEW_RUNS)
+                                       .order_by(run_alias.created_at.desc()))).scalars())
+    await runner.reconcile_stale(session, rows)
+    by_suite: dict[str, list[EvalRun]] = {}
+    for row in rows:
+        by_suite.setdefault(row.suite_id, []).append(row)
+    for sid, runs in by_suite.items():
+        latest = runs[0]
+        out[sid]["latest"] = {"id": latest.id, "status": latest.status, "running": runner.running(latest),
+                              "arm": latest.arm, "pair_id": latest.pair_id, "trigger": latest.trigger,
+                              "created_at": _iso(latest.created_at), "finished_at": _iso(latest.finished_at),
+                              "counts": dict(latest.counts or {}), "accuracy": _rate_of(latest),
+                              "traced_accuracy": dict((latest.summary or {}).get("traced_accuracy") or {}) or None}
+        finished = [r for r in runs if r.status in ("succeeded", "partial") and _rate_of(r) is not None]
+        if finished:
+            key = (finished[0].fingerprint or {}).get("variant_key")
+            same = [r for r in finished if (r.fingerprint or {}).get("variant_key") == key][:TREND_POINTS]
+            out[sid]["trend"] = [{"id": r.id, "created_at": _iso(r.created_at), "arm": r.arm, **(_rate_of(r) or {})}
+                                 for r in reversed(same)]
+    schedules = (await session.execute(select(Schedule).where(
+        Schedule.target_kind == "eval_suite", Schedule.target_id.in_(suite_ids))
+        .order_by(Schedule.created_at))).scalars()
+    for row in schedules:
+        try:
+            text = cron.describe(row.cron, row.timezone)
+        except cron.CronError:
+            text = f"按 cron 表达式 {row.cron}（{row.timezone}）"
+        out[row.target_id]["schedules"].append({"id": row.id, "name": row.name, "enabled": row.enabled,
+                                                "description_text": text, "next_run_at": _iso(row.next_run_at)})
+    return out
+
+
 @router.get("/suites")
 async def list_suites(archived: bool = False, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """题集（归档的不列，archived=true 时一起列）。每个题集另带 latest、trend、schedules（_overview）。"""
     query = select(EvalSuite).order_by(EvalSuite.updated_at.desc())
     if not archived:
         query = query.where(EvalSuite.archived.is_(False))
@@ -239,7 +312,8 @@ async def list_suites(archived: bool = False, session: AsyncSession = Depends(ge
             counts[suite_id]["cases"] += n
             if enabled:
                 counts[suite_id]["enabled"] += n
-    return {"suites": [_suite_out(r, counts=counts[r.id], site=site) for r in rows]}
+    overview = await _overview(session, [r.id for r in rows])
+    return {"suites": [{**_suite_out(r, counts=counts[r.id], site=site), **overview[r.id]} for r in rows]}
 
 
 @router.post("/suites", status_code=201)
@@ -266,6 +340,7 @@ async def get_suite(suite_id: str, session: AsyncSession = Depends(get_session))
     active = [c for c in out if not c["archived"]]
     return {**_suite_out(suite, counts={"cases": len(active), "enabled": sum(1 for c in active if c["enabled"])},
                          site=await _site(session)),
+            **(await _overview(session, [suite_id]))[suite_id],
             "cases": out, "pending_upgrades": sum(1 for c in active if (c["upgrade"] or {}).get("pending"))}
 
 
@@ -342,6 +417,62 @@ async def trial_unsaved(suite_id: str, request: Request, session: AsyncSession =
     return {"trial": (await _trial(session, suite, fields, actor_of(x_actor))).to_dict()}
 
 
+#: 预览最多交回几行（界面上点选格或列用；多行期望的取值仍由试算按整份结果算）
+PREVIEW_ROWS = 50
+
+
+@router.post("/suites/{suite_id}/cases/preview")
+async def preview_sql(suite_id: str, request: Request, session: AsyncSession = Depends(get_session),
+                      x_actor: str | None = Header(default=None)) -> dict[str, Any]:
+    """写 SQL 定期望的第一步：{datasource_id, sql}（SQL 可以带 {{ schedule.* }}）。按此刻渲染、过只读守卫、跑一次和期望同一条路
+    的 SQL 探针（裸图探索运行，来源 eval），交回结果的列和前 PREVIEW_ROWS 行，界面上据此点选格或列、再去试算。
+
+    结果里有遮罩列、取不到结果、运行失败、超时的，照保存时那一道拒收、照实写原因（status 不是 ok，reason、code 同试算）——不回
+    422：这是在写题的途中，界面把原因落在 SQL 那一格下面。{status, reason, code, columns, rows, row_count, truncated, run_id,
+    rendered}。"""
+    from app.core import artifact_store
+    from app.db.models import Artifact
+
+    suite = await _suite(session, suite_id)
+    body = await _body(request)
+    probe = {"question": "（预览）", "gold": {"kind": "sql", "datasource_id": body.get("datasource_id"),
+                                            "sql": body.get("sql")},
+             # 只为跑一次探针：「没有数据」这一种期望不取任何格，取不到值时下面照样交回结果
+             "expectations": [{"id": "preview", "kind": "none"}]}
+    try:
+        fields = normalize_case(probe)
+        await resolve_case(session, fields, tz=suite.timezone)
+    except CaseInvalid as e:
+        # 没选数据源、SQL 空着、变量写错、没过只读守卫：照实写原因、指明是哪一格
+        return {"status": "invalid", "reason": e.message, "code": e.code, "field": e.field, "columns": [], "rows": [],
+                "row_count": None, "truncated": False, "run_id": None, "rendered": {}}
+    outcome = await _trial(session, suite, fields, actor_of(x_actor))
+    out: dict[str, Any] = {"status": outcome.status, "reason": outcome.reason, "code": outcome.code, "columns": [],
+                           "rows": [], "row_count": None, "truncated": False,
+                           "run_id": outcome.run_ids[0] if outcome.run_ids else None,
+                           "rendered": dict(outcome.rendered or {})}
+    if outcome.status not in ("ok", "no_value") or not outcome.run_ids:
+        return out
+    async with SessionLocal() as read:
+        refs = (await read.execute(select(Artifact.id).where(
+            Artifact.run_id == outcome.run_ids[0], Artifact.kind == "query_snapshot",
+            Artifact.node_id == gold_rules.QUERY_NODE).order_by(Artifact.created_at.desc()))).scalars().all()
+    snapshot = None
+    for ref in refs:
+        try:
+            snapshot = artifact_store.load(ref)
+        except ValueError:
+            snapshot = None
+        if isinstance(snapshot, dict):
+            break
+    if not isinstance(snapshot, dict):
+        return {**out, "status": "failed", "reason": "预览的运行没有留下查询结果", "code": gold_rules.GOLD_FAILED}
+    rows = [list(r) if isinstance(r, (list, tuple)) else [r] for r in snapshot.get("rows") or []]
+    return {**out, "status": "ok", "reason": None, "code": None, "columns": [str(c) for c in snapshot.get("columns") or []],
+            "rows": rows[:PREVIEW_ROWS], "row_count": snapshot.get("row_count", len(rows)),
+            "truncated": bool(snapshot.get("truncated")) or len(rows) > PREVIEW_ROWS, "shown": min(len(rows), PREVIEW_ROWS)}
+
+
 @router.patch("/suites/{suite_id}/cases/{case_id}")
 async def update_case(suite_id: str, case_id: str, request: Request, session: AsyncSession = Depends(get_session),
                       x_actor: str | None = Header(default=None)) -> dict[str, Any]:
@@ -859,7 +990,29 @@ async def get_evaluation(evaluation_id: str, session: AsyncSession = Depends(get
                                      .where(EvalResult.eval_run_id == row.id, EvalResult.status == "running")
                                      .order_by(EvalResult.seq))).all()
     return {**_evaluation_out(row),
-            "current": [{"case_seq": seq, "repeat": rep, "arm": arm, "doing": doing} for seq, rep, arm, doing in running]}
+            "current": [{"case_seq": seq, "repeat": rep, "arm": arm, "doing": doing} for seq, rep, arm, doing in running],
+            "actions": await _actions(session, row)}
+
+
+async def _actions(session: AsyncSession, row: EvalRun) -> dict[str, Any]:
+    """这次评测现在能做哪几件事（界面照它摆按钮，不自己判）：和 stop / retry / rescore / pin 几个接口拒收的条件同一份。
+    open 是这次评测（A/B 的话几臂合起来）还没做完的项数：停止、中断、出错、到了上限没开评的，重试就是把它们接着做。"""
+    from app.core import job_runtime as runtime
+    from app.db.models_eval import EvalResult
+    from app.eval import runner
+
+    active = row.status in runtime.ACTIVE
+    leader = runner.leader_of(row)
+    arms = list((await session.execute(select(EvalRun.id).where(
+        (EvalRun.id == leader) | (EvalRun.pair_id == leader)))).scalars())
+    open_items = int(await session.scalar(select(func.count()).select_from(EvalResult).where(
+        EvalResult.eval_run_id.in_(arms), EvalResult.status != "ok")) or 0)
+    suite = await session.get(EvalSuite, row.suite_id)
+    cleaned = row.cleaned_at is not None
+    finished = row.status in ("succeeded", "partial")
+    return {"stop": active, "retry": not active and not cleaned and open_items > 0 and suite is not None
+            and not suite.archived, "rescore": not active and not cleaned,
+            "pin": finished and not row.pinned, "unpin": bool(row.pinned), "open": open_items, "cleaned": cleaned}
 
 
 @router.get("/evaluations/{evaluation_id}/results")
@@ -997,6 +1150,97 @@ async def get_pair(pair_id: str, session: AsyncSession = Depends(get_session)) -
     return {"pair_id": pair_id, "evaluations": [_evaluation_out(r) for r in rows], "table": table, "cleaned": cleaned}
 
 
+# ==========================================================================
+# 口径命中与重复度（E1、E3）：判据、分档、窗口趋势、成员不少于 3 的问题类
+# ==========================================================================
+
+#: 判据（docs/design/copilot-look-and-caliber-ledger.md §3.8，已修订为持续统计的参考线）：E1 等价命中率 ≥30% 值得按口径走、
+#: <10% 先只做对账、介于两者之间先加强规范化；E3 问题重复度 ≥20% 沉淀有价值。只作参考，不卡门
+E1_ROUTE, E1_LOW, E3_WORTH = 0.30, 0.10, 0.20
+SIGNAL_CRITERIA = {
+    "e1": [{"band": "route", "at_least": E1_ROUTE, "text": "等价命中率 ≥ 30%：值得按已治理的口径走"},
+           {"band": "normalize", "at_least": E1_LOW, "text": "10%–30%：先加强规范化，让写法不同的等价 SQL 认得出"},
+           {"band": "reconcile", "at_least": 0.0, "text": "< 10%：先只做口径对账"}],
+    "e3": [{"band": "worth", "at_least": E3_WORTH, "text": "问题重复度 ≥ 20%：把反复问的问题沉淀成口径有价值"},
+           {"band": "not_yet", "at_least": 0.0, "text": "< 20%：反复问的问题还不多"}],
+    "note": "参考线出自口径账本的设计文档，只作参考，不卡门",
+}
+SIGNAL_DAYS_MAX = 365
+SIGNAL_POINTS_MAX = 12
+#: 每类最多列几个问题
+CLUSTER_MEMBERS = 10
+
+
+def _e1_band(rate: float | None) -> str | None:
+    if rate is None:
+        return None
+    return "route" if rate >= E1_ROUTE else "normalize" if rate >= E1_LOW else "reconcile"
+
+
+def _e3_band(share: float | None) -> str | None:
+    if share is None:
+        return None
+    return "worth" if share >= E3_WORTH else "not_yet"
+
+
+def _point(stats: dict[str, Any], since: Any, until: Any) -> dict[str, Any]:
+    """一个窗口的 E1、E3（取自 ledger/stats.compute_stats 的结果，不另算）和各自落在哪一档。"""
+    e1, e3 = stats.get("e1") or {}, stats.get("e3") or {}
+    rate = (e1.get("rates") or {}).get("proven")
+    share = e3.get("cluster_share")
+    return {"since": _iso(since), "until": _iso(until),
+            "e1": {"rate": rate, "n": int(e1.get("with_candidates") or 0), "queries": int(e1.get("queries") or 0),
+                   "band": _e1_band(rate)},
+            "e3": {"share": share, "n": int(e3.get("questions") or 0), "lower_bound": e3.get("repeated_share"),
+                   "band": _e3_band(share)}}
+
+
+@router.get("/signals")
+async def signals(days: int = 30, points: int = 6, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """「口径命中与重复度」页签：判据、最近 days 天和往前 points - 1 个同样长的窗口各自的 E1、E3（趋势，旧的在前），以及最近
+    days 天里成员不少于 3 的问题类。
+
+    E1、E3 的数只出自口径账本的 compute_stats（总方案 X2，全站一份；评测运行不计入，见 engine/origin.counted_runs），问题类只出自
+    chat/questions（和 E3、认证队列同一份：只取问题原文、不含 history、排除系统派生的运行）。界面上的详细分布另调
+    GET /api/ledger/stats。每一类给代表问题、最近几个成员和最近一个挂着运行的成员（「提取模板」就从那次运行提取）。"""
+    from datetime import timedelta
+
+    from app.api.deps import current_principal
+    from app.chat.questions import cluster_questions, embedder, recent_questions
+    from app.ledger.stats import QUESTION_LIMIT, REPEAT_RUNS, compute_stats
+
+    if not 1 <= days <= SIGNAL_DAYS_MAX:
+        raise CodedHTTPException(422, f"窗口需要是 1 到 {SIGNAL_DAYS_MAX} 天", case_rules.CASE_INVALID, field="days")
+    if not 1 <= points <= SIGNAL_POINTS_MAX:
+        raise CodedHTTPException(422, f"趋势需要是 1 到 {SIGNAL_POINTS_MAX} 个窗口", case_rules.CASE_INVALID,
+                                 field="points")
+    principal = current_principal()
+    now = utcnow()
+    span = timedelta(days=days)
+    trend = []
+    for i in reversed(range(points)):
+        until = now - span * i
+        since = until - span
+        trend.append(_point(await compute_stats(session, since=since, until=until, principal=principal), since, until))
+    since = now - span
+    questions = [q for q in await recent_questions(session, principal, limit=QUESTION_LIMIT)
+                 if q.at is None or q.at >= since]
+    clusters = []
+    for c in await cluster_questions(questions) if questions else []:
+        if c.count < REPEAT_RUNS:
+            continue
+        with_run = next((m for m in c.members if m.run_id), None)
+        clusters.append({"count": c.count, "representative": c.members[0].text[:500],
+                         "first_at": _iso(c.members[-1].at), "last_at": _iso(c.members[0].at),
+                         "members": [{"text": m.text[:500], "at": _iso(m.at), "run_id": m.run_id, "turn_id": m.turn_id}
+                                     for m in c.members[:CLUSTER_MEMBERS]],
+                         "run_id": with_run.run_id if with_run else None})
+    clusters.sort(key=lambda c: c["count"], reverse=True)
+    return {"window": {"since": _iso(since), "until": _iso(now), "days": days}, "criteria": SIGNAL_CRITERIA,
+            "current": trend[-1], "trend": trend, "clusters": clusters, "questions": len(questions),
+            "threshold": REPEAT_RUNS, "embedder": embedder()}
+
+
 @router.post("/suites/{suite_id}/cleanup")
 async def cleanup_suite(suite_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
     """按保留规则清理这个题集更早的评测明细（最近 keep_runs 次、钉住的基线、A/B 的另几臂留着；总表永久保留）。"""

backend/app/notify/service.py

@@ -463,7 +463,8 @@ async def notify_source_changed(source_id: str) -> int:
 async def notify_eval(eval_run_id: str, event: str, *, passed: int, graded: int, regressed: int) -> int:
     """定时评测的通知（eval.regressed 有一道题从通过变成不通过、eval.failed 没评完：失败、到了上限、止损）。只推订阅了它的
     全局渠道。消息里只有题集名、状态、通过几题、退步几题、评测编号和链接（细案 01 做法 15、总方案 X6）：不放期望值、答案里的数、
-    题目原文。链接去发起它的那条定时任务的触发历史（评测页在第 4 波）。"""
+    题目原文。链接去这次评测的详情页（/evals/runs/{评测 id}：进度、逐题结果、和上一次比；第 4 波评测页上线之前是定时任务的
+    触发历史），退步的那几题在详情页上点得开。"""
     from app.db.models_eval import EvalRun, EvalSuite
 
     async with SessionLocal() as session:
@@ -485,7 +486,7 @@ async def notify_eval(eval_run_id: str, event: str, *, passed: int, graded: int,
     if run.arm:
         lines.append(("对照", run.arm[:40]))
     lines.append(("评测", f"#{eval_run_id[:6]}"))
-    links = [("查看定时任务", f"{base}/schedules/{run.schedule_id}?tab=history")] if base and run.schedule_id else []
+    links = [("查看评测", f"{base}/evals/runs/{eval_run_id}")] if base else []
     message = {"event": event, "status": run.status, "ok": False,
                "title": f"定时评测「{name}」{'有退步' if regressed_event else '没有评完'}", "lines": lines, "links": links,
                "subject": {"eval_run_id": eval_run_id, "suite_id": run.suite_id, "schedule_id": run.schedule_id}}

backend/app/providers/usage.py

@@ -13,9 +13,10 @@ async generator：with 块跨着 yield 的话，变量在两次 yield 之间留
 
 **金额。** 按 providers/catalog.estimate_cost 算，和运行的 llm.end 一个口径。估不出的调用记 unpriced_calls：模型不在价格
 目录里，或者用量没报全而目录价又不是 0。估不出的不当成 0 元，调用方据此写「≥」或「金额无法估算」。用量没报全有两种：
-回复没带用量（有的网关流式不给）；读到一半就不读了——生成过程读到 done、look、ask 那一行就关掉模型的流（后面的闲话不要、
-也不再为它付钱），用量在流的末尾才报全（Anthropic 的输出 token 在最后一条，OpenAI 兼容系开了 stream_usage 也在最后），
-关掉的那一次只拿得到已经报了的部分。这一次照样计次数、计已报的 token，记成估不出，不悄悄少算。
+回复没带用量（有的网关流式不给）；读到一半就不读了——用量在流的末尾才报全（Anthropic 的输出 token 在最后一条，OpenAI
+兼容系开了 stream_usage 也在最后）。生成过程读到 done、look、ask 那一行之后照样把流读完再关（api/copilot_gen._drain），
+流末尾那一段用量收得到；只有模型在那一行之后还往下写超过 DRAIN_CHARS 字的闲话时才当场关流（不为闲话付钱），连接断了也一样，
+这一次只拿得到已经报了的部分：照样计次数、计已报的 token，记成估不出，不悄悄少算。
 
 模型名按发起调用时的参数取（invocation_params 的 model / model_name，就是接入里配的模型 id，价格目录按它查），取不到再
 看回复里的 model_name。
@@ -85,7 +86,7 @@ class Meter(BaseCallbackHandler):
                 row["unreported"] += 1
 
     def on_llm_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None:
-        """流读到一半停了（生成过程读到 done / look / ask 就关流，或者连接断了）：LangChain 照样交来已经收到的那几段。
+        """流读到一半停了（生成过程在最后一行之后读到太多闲话就关流，或者连接断了）：LangChain 照样交来已经收到的那几段。
         有收到东西的照计一次、记已报的 token，算用量没报全；一段都没收到的（请求刚发出去就被拒了）不计。"""
         from langchain_core.outputs import ChatGenerationChunk
 

backend/tests/test_eval_schedule.py

@@ -73,6 +73,34 @@ def received(monkeypatch):
     return got
 
 
+async def _public_url(url: str):
+    """临时设对外访问地址（通知里的链接按它拼），交回复原的函数：别的测试不受影响。"""
+    from app.core import public_url
+    from app.db.models import Setting
+
+    async with SessionLocal() as session:
+        row = await session.get(Setting, "notify")
+        before = dict(row.value or {}) if row is not None else None
+        if row is None:
+            session.add(Setting(key="notify", value={"public_base_url": url}))
+        else:
+            row.value = {**(row.value or {}), "public_base_url": url}
+        await session.commit()
+        await public_url.refresh(session)
+
+    async def restore() -> None:
+        async with SessionLocal() as session:
+            row = await session.get(Setting, "notify")
+            if before is None and row is not None:
+                await session.delete(row)
+            elif row is not None:
+                row.value = before
+            await session.commit()
+            await public_url.refresh(session)
+
+    return restore
+
+
 def stub(monkeypatch, *, delay: float = 0.0, path: str = "asked"):
     async def drive(self):
         await asyncio.sleep(delay)
@@ -212,8 +240,12 @@ async def test_an_unfinished_scheduled_evaluation_pushes_failed(client, monkeypa
     suite_id = await es.make_suite("none", cases=ASK * 4, name="虚构失败题集", budget={"max_parallel": 1})
     body = await _schedule(client, suite_id)
     await _channel(["eval.failed"])
-    fire = await service.fire_once(body["id"], utcnow().replace(microsecond=0), "manual")
-    await es.wait_eval(fire.eval_run_id)
+    restore = await _public_url("https://agentlab.example.com")
+    try:
+        fire = await service.fire_once(body["id"], utcnow().replace(microsecond=0), "manual")
+        await es.wait_eval(fire.eval_run_id)
+    finally:
+        await restore()
     async with SessionLocal() as session:
         run = await session.get(EvalRun, fire.eval_run_id)
         skipped = (await session.execute(select(EvalResult).where(
@@ -223,8 +255,9 @@ async def test_an_unfinished_scheduled_evaluation_pushes_failed(client, monkeypa
     message = json.loads(request.content)
     assert message["event"] == "eval.failed" and message["title"] == "定时评测「虚构失败题集」没有评完"
     assert "通过：0 / 3 题" in message["text"]
-    # 链接去那条定时任务的触发历史（设了对外访问地址才有链接）
-    assert all("/schedules/" in link["url"] for link in message["links"])
+    # 链接去这次评测的详情页（第 4 波评测页上线之前去定时任务的触发历史）；设了对外访问地址才有链接
+    assert [link["url"] for link in message["links"]] == [f"https://agentlab.example.com/evals/runs/{fire.eval_run_id}"], \
+        message["links"]
 
 
 def test_the_eval_events_say_they_are_for_scheduled_evaluations():

backend/tests/test_evals_ui_api.py

@@ -0,0 +1,202 @@
+"""评测页（第 4 波，细案 01 第 18 项）在 app/api/evals.py 里补的几样：统计、能不能做都由服务端给，界面只显示。
+
+- 题集列表每个题集带最近一次评测、同一变体最近几次评完的准确率（旧的在前，换了变体的不连成一条线）、目标是它的定时任务；
+- 评测详情带 actions：只重试没做完的、重新评分、钉住 / 取消基线现在能不能做，和接口拒收的条件同一份；
+- 写 SQL 定期望的预览：交回结果的列和前几行；用到遮罩列的照保存时那一道拒收，没过只读守卫的指明是哪一格（都不回 422）；
+- 「口径命中与重复度」：判据、窗口趋势（出自 ledger/stats.compute_stats）、成员不少于 3 的问题类（出自 chat/questions），
+  评测运行不进问题类。
+
+库、表、题集名、问法都是虚构的。
+"""
+from __future__ import annotations
+
+import uuid
+from datetime import timedelta
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import delete
+
+from app.db.base import SessionLocal, utcnow
+from app.db.models import Conversation, ConversationTurn, Run, Schedule
+from app.db.models_eval import EvalResult, EvalRun
+from app.engine.runner import run_manager
+from app.main import app
+from tests import eval_samples as es
+from tests.fixtures import dual_check_shop as shop
+from tests.fixtures.sources import drop_source
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+@pytest.fixture
+async def source(tmp_path):
+    name, sid = await shop.add_source(shop.build(tmp_path / "shop.db"))
+    yield {"name": name, "id": sid}
+    await drop_source(sid)
+
+
+def _summary(passed: int, n: int) -> dict:
+    return {"accuracy": {"passed": passed, "n": n, "rate": round(passed / n, 4) if n else None, "ci": None},
+            "traced_accuracy": {"passed": passed, "n": n, "rate": round(passed / n, 4) if n else None, "ci": None}}
+
+
+async def _evaluation(suite_id: str, *, status: str, key: str, passed: int, n: int, minutes_ago: int,
+                      arm: str | None = None, pinned: bool = False, cleaned: bool = False) -> str:
+    async with SessionLocal() as session:
+        row = EvalRun(suite_id=suite_id, status=status, trigger="manual", variant={}, resolved_models={},
+                      fingerprint={"variant_key": key}, summary=_summary(passed, n) if status != "queued" else {},
+                      counts={"total": n, "done": n}, arm=arm, pinned=pinned,
+                      cleaned_at=utcnow() if cleaned else None)
+        session.add(row)
+        await session.flush()
+        row.created_at = utcnow() - timedelta(minutes=minutes_ago)
+        await session.commit()
+        return row.id
+
+
+async def test_the_suite_list_carries_the_latest_run_a_trend_of_the_same_variant_and_its_schedules(client, source):
+    suite_id = await es.make_suite(source["id"], name=f"虚构题集_{uuid.uuid4().hex[:6]}")
+    old = await _evaluation(suite_id, status="succeeded", key="A", passed=1, n=4, minutes_ago=50)
+    other = await _evaluation(suite_id, status="succeeded", key="B", passed=4, n=4, minutes_ago=40)
+    failed = await _evaluation(suite_id, status="failed", key="A", passed=0, n=4, minutes_ago=30)
+    mid = await _evaluation(suite_id, status="partial", key="A", passed=2, n=4, minutes_ago=20)
+    new = await _evaluation(suite_id, status="succeeded", key="A", passed=3, n=4, minutes_ago=10)
+    async with SessionLocal() as session:
+        sched = Schedule(name="虚构每周回归", workflow_id="", workflow_name="虚构题集", target_kind="eval_suite",
+                         target_id=suite_id, cron="0 9 * * 1", timezone="Asia/Shanghai", enabled=True)
+        session.add(sched)
+        await session.commit()
+        sched_id = sched.id
+    try:
+        listed = {s["id"]: s for s in (await client.get("/api/evals/suites")).json()["suites"]}
+        got = listed[suite_id]
+        assert got["latest"]["id"] == new and got["latest"]["accuracy"]["rate"] == 0.75
+        # 趋势只连同一变体、评完了的：B 那一次、失败的那一次都不进；旧的在前
+        assert [p["id"] for p in got["trend"]] == [old, mid, new], got["trend"]
+        assert [p["rate"] for p in got["trend"]] == [0.25, 0.5, 0.75]
+        assert other not in [p["id"] for p in got["trend"]] and failed not in [p["id"] for p in got["trend"]]
+        [schedule] = got["schedules"]
+        assert schedule["id"] == sched_id and schedule["enabled"] is True and "09:00" in schedule["description_text"]
+        detail = (await client.get(f"/api/evals/suites/{suite_id}")).json()
+        assert detail["latest"]["id"] == new and [s["id"] for s in detail["schedules"]] == [sched_id]
+    finally:
+        async with SessionLocal() as session:
+            await session.execute(delete(Schedule).where(Schedule.id == sched_id))
+            await session.commit()
+        await es.drop_suites()
+
+
+async def test_the_evaluation_says_what_can_be_done_now(client, source):
+    suite_id = await es.make_suite(source["id"], name=f"虚构题集_{uuid.uuid4().hex[:6]}")
+    try:
+        run_id = await _evaluation(suite_id, status="partial", key="A", passed=1, n=2, minutes_ago=5)
+        async with SessionLocal() as session:
+            session.add(EvalResult(eval_run_id=run_id, seq=1, case_id=None, case_seq=1, case_hash="h", repeat=0,
+                                   status="ok", verdict="pass"))
+            session.add(EvalResult(eval_run_id=run_id, seq=2, case_id=None, case_seq=2, case_hash="h", repeat=0,
+                                   status="cancelled", error="评测已停止"))
+            await session.commit()
+        actions = (await client.get(f"/api/evals/evaluations/{run_id}")).json()["actions"]
+        assert actions == {"stop": False, "retry": True, "rescore": True, "pin": True, "unpin": False, "open": 1,
+                           "cleaned": False}
+        pinned = await client.post(f"/api/evals/evaluations/{run_id}/pin", json={"pinned": True})
+        assert pinned.status_code == 200
+        actions = (await client.get(f"/api/evals/evaluations/{run_id}")).json()["actions"]
+        assert actions["pin"] is False and actions["unpin"] is True
+
+        done = await _evaluation(suite_id, status="succeeded", key="A", passed=2, n=2, minutes_ago=4, cleaned=True)
+        actions = (await client.get(f"/api/evals/evaluations/{done}")).json()["actions"]
+        # 明细清理了的：不能重试、不能重评（和接口拒收的条件一致）
+        assert actions["retry"] is False and actions["rescore"] is False and actions["cleaned"] is True
+        assert (await client.post(f"/api/evals/evaluations/{done}/rescore")).status_code == 409
+    finally:
+        await es.drop_suites()
+
+
+async def test_previewing_an_expected_sql_gives_columns_and_rows(client, source):
+    made = await client.post("/api/evals/suites", json={"name": f"虚构题集_{uuid.uuid4().hex[:6]}"})
+    suite_id = made.json()["id"]
+    try:
+        body = {"datasource_id": source["id"],
+                "sql": "SELECT status AS 状态, SUM(amount) AS gmv FROM orders GROUP BY status ORDER BY status"}
+        got = (await client.post(f"/api/evals/suites/{suite_id}/cases/preview", json=body)).json()
+        assert got["status"] == "ok" and got["columns"] == ["状态", "gmv"], got
+        assert got["rows"] and all(len(r) == 2 for r in got["rows"]) and got["run_id"]
+        assert got["rendered"]["sql"].startswith("SELECT status")
+        # 没过只读守卫：指明是 SQL 那一格，不回 422
+        res = await client.post(f"/api/evals/suites/{suite_id}/cases/preview",
+                                json={"datasource_id": source["id"], "sql": "DELETE FROM orders"})
+        assert res.status_code == 200 and res.json()["status"] == "invalid" and res.json()["field"] == "gold.sql"
+        res = await client.post(f"/api/evals/suites/{suite_id}/cases/preview", json={"sql": "SELECT 1"})
+        assert res.json()["status"] == "invalid" and res.json()["field"] == "gold.datasource_id"
+    finally:
+        await es.drop_suites()
+
+
+async def test_a_preview_touching_a_masked_column_is_refused(client, tmp_path):
+    name, sid = await shop.add_source(shop.build(tmp_path / "masked.db"), options={"mask_columns": ["region"]})
+    made = await client.post("/api/evals/suites", json={"name": f"虚构题集_{uuid.uuid4().hex[:6]}"})
+    try:
+        got = (await client.post(f"/api/evals/suites/{made.json()['id']}/cases/preview", json={
+            "datasource_id": sid, "sql": "SELECT region AS 地区, SUM(amount) AS gmv FROM orders GROUP BY region"})).json()
+        assert got["status"] == "refused" and got["code"] == "eval_gold_masked" and "遮罩" in got["reason"], got
+        assert got["rows"] == [] and got["columns"] == [], "遮罩列的值一格都不能交出去"
+    finally:
+        await es.drop_suites()
+        await drop_source(sid)
+
+
+async def _asked(question: str, n: int, *, origin: str = "chat") -> list[str]:
+    async with SessionLocal() as session:
+        conv = Conversation(kind="chat", title="虚构对话")
+        session.add(conv)
+        await session.flush()
+        ids = []
+        for i in range(n):
+            run = Run(workflow_name="问数据", status="succeeded", graph={}, input={"question": question},
+                      origin_kind=origin, run_class="exploratory")
+            session.add(run)
+            await session.flush()
+            session.add(ConversationTurn(conversation_id=conv.id, seq=i, question=question, run_id=run.id,
+                                         status="done"))
+            ids.append(run.id)
+        await session.commit()
+        return ids
+
+
+async def test_signals_give_criteria_trend_and_repeated_question_clusters(client):
+    question = f"虚构门店 {uuid.uuid4().hex[:6]} 上周的销售额是多少？"
+    runs = await _asked(question, 3)
+    eval_question = f"虚构评测题 {uuid.uuid4().hex[:6]} 本月退款率？"
+    await _asked(eval_question, 3, origin="eval")
+    got = (await client.get("/api/evals/signals", params={"days": 30, "points": 3})).json()
+    assert [c["band"] for c in got["criteria"]["e1"]] == ["route", "normalize", "reconcile"]
+    assert got["criteria"]["e3"][0]["at_least"] == 0.2
+    assert len(got["trend"]) == 3 and got["current"] == got["trend"][-1]
+    assert got["trend"][0]["until"] <= got["trend"][1]["since"] or got["trend"][0]["until"] == got["trend"][1]["since"]
+    assert got["current"]["e3"]["n"] >= 3 and got["current"]["e3"]["band"] in ("worth", "not_yet")
+    mine = [c for c in got["clusters"] if c["representative"] == question]
+    assert mine and mine[0]["count"] == 3 and mine[0]["run_id"] in runs, got["clusters"]
+    assert not [c for c in got["clusters"] if c["representative"] == eval_question], "评测运行不进问题类"
+    assert got["threshold"] == 3 and got["embedder"]
+    assert (await client.get("/api/evals/signals", params={"days": 0})).status_code == 422
+
+
+def test_bands_follow_the_reference_lines():
+    from app.api import evals
+
+    assert [evals._e1_band(x) for x in (None, 0.05, 0.1, 0.29, 0.3)] == [None, "reconcile", "normalize", "normalize",
+                                                                           "route"]
+    assert [evals._e3_band(x) for x in (None, 0.19, 0.2)] == [None, "not_yet", "worth"]

backend/tests/test_usage_meter.py

@@ -5,7 +5,10 @@
   不当成 0 元；
 - metered 只计生成过程自己的调用（含它在一步里起的任务）；调用方在两次 yield 之间做的事（发起运行）不计——运行的用量
   另有 llm.end，包进来就重复了；嵌套时内层并进外层；
-- answer_ops 末尾的 usage 记下这一轮建图的每一次调用（起草、自查修正各一次）；复核的结果带 usage。
+- answer_ops 末尾的 usage 记下这一轮建图的每一次调用（起草、自查修正各一次）；复核的结果带 usage；
+- 读到 done / look / ask 那一行之后照样把模型的流读完再关（读完只为拿到流末尾才报的用量）：输出 token 报全了、金额估得出，
+  不再是下限；模型在那一行之后还往下写一大段闲话的，读到上限就当场关流、照实记成没报全；
+- 评测的作答（app/eval/answer.Driver）记的助手用量就是问数据页收到的那条 usage：同一个口径。
 
 模型都是剧本，不调真接口；价格用目录里登记过的模型 id，名字之外都是虚构的。
 """
@@ -190,9 +193,19 @@ async def client():
         yield c
 
 
+async def _sse(client, body: dict) -> list[dict]:
+    events = []
+    async with client.stream("POST", "/api/copilot/generate-stream", json=body) as r:
+        async for line in r.aiter_lines():
+            if line.startswith("data: "):
+                events.append(json.loads(line[6:]))
+    return events
+
+
 async def test_answer_ops_ends_with_the_usage_of_every_build_call(client, monkeypatch):
-    """起草、自查修正各一次。生成过程读到 done 就关流（后面的闲话不要），流末尾才报的输出 token 拿不到：这两次照样计次数、
-    计已报的输入 token，记成估不出（unpriced_calls），金额只是估得出的那部分——不悄悄少算。"""
+    """起草、自查修正各一次。读到 done 那一行之后照样把模型的流读完（模型照约定 done 就是最后一行，读完只多收流末尾那一段
+    用量）：输出 token 报全了、金额估得出，这一轮的花费不再是下限（第 1 波的遗留：以前读到 done 就关流，输出 token 一个都
+    拿不到，两次都记成估不出）。"""
     import app.api.copilot as copilot
 
     model = Scripted(model=PRICED, replies=[GRAPH, FIX], tokens=(3000, 500))
@@ -201,19 +214,96 @@ async def test_answer_ops_ends_with_the_usage_of_every_build_call(client, monkey
         return model, PRICED
 
     monkeypatch.setattr(copilot, "get_chat_model", _model)
-    events = []
-    async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "数一数", "intent": "answer"}) as r:
-        async for line in r.aiter_lines():
-            if line.startswith("data: "):
-                events.append(json.loads(line[6:]))
+    events = await _sse(client, {"instruction": "数一数", "intent": "answer"})
     assert [e["op"] for e in events if e["op"] == "check"] and events[-2]["op"] == "final"
     spent = events[-1]
     assert spent["op"] == "usage", spent
     assert model.calls == 2, spent  # 起草一次、自查修正一次
-    assert spent["by_model"][PRICED]["calls"] == 2, spent
-    assert spent["by_model"][PRICED]["input_tokens"] == 6000 and spent["by_model"][PRICED]["output_tokens"] == 0
-    assert spent["cost_usd"] == pytest.approx(estimate_cost(PRICED, 6000, 0), abs=1e-6)
-    assert spent["unpriced_calls"] == 2, "关流前没报全的两次记成估不出"
+    row = spent["by_model"][PRICED]
+    assert row["calls"] == 2, spent
+    assert (row["input_tokens"], row["output_tokens"]) == (6000, 1000), "流末尾才报的输出 token 也收到了"
+    assert spent["unpriced_calls"] == 0, "读完了的流用量报全了，不记估不出"
+    assert spent["cost_usd"] == pytest.approx(estimate_cost(PRICED, 6000, 1000), abs=1e-6)
+
+
+async def test_a_look_line_also_waits_for_the_usage_at_the_end_of_the_stream(client, monkeypatch):
+    """读到 look 停下查看、ask 等用户回答：同样把这一次的流读完再关，查看前那一次的输出 token 不丢。"""
+    import app.api.copilot as copilot
+
+    looked = _ops({"op": "look", "kind": "table", "id": "不存在的表", "need": "看看有哪些列"})
+    model = Scripted(model=PRICED, replies=[looked, _ops({"op": "ask", "question": "要按哪一天算？"})], tokens=(1200, 80))
+
+    async def _model(*_a, **_k):
+        return model, PRICED
+
+    monkeypatch.setattr(copilot, "get_chat_model", _model)
+    events = await _sse(client, {"instruction": "看一下再问", "intent": "answer"})
+    spent = events[-1]
+    assert spent["op"] == "usage" and model.calls == 2, [e["op"] for e in events]
+    row = spent["by_model"][PRICED]
+    assert (row["input_tokens"], row["output_tokens"], row["unpriced_calls"]) == (2400, 160, 0), spent
+
+
+class Chatty(Scripted):
+    """done 那一行之后还往下写一大段闲话（不守约定），用量照样在最后一段才报。"""
+
+    chatter: int = 40
+
+    async def _astream(self, messages, stop=None, run_manager=None, **kwargs):
+        lines = self._next()
+        inp, out = self.tokens
+        for i, line in enumerate(lines):
+            meta = {"input_tokens": inp, "output_tokens": 0, "total_tokens": inp} if i == 0 else None
+            yield ChatGenerationChunk(message=AIMessageChunk(content=line + "\n", **({"usage_metadata": meta} if meta
+                                                                                     else {})))
+        for _ in range(self.chatter):
+            yield ChatGenerationChunk(message=AIMessageChunk(content="这是一段写在最后一行之后、没人要的闲话。\n"))
+        yield ChatGenerationChunk(message=AIMessageChunk(content="", usage_metadata={
+            "input_tokens": 0, "output_tokens": out, "total_tokens": out}))
+
+
+async def test_chatter_after_the_last_line_is_cut_off_and_recorded_as_not_fully_reported(client, monkeypatch):
+    """读完流只为拿用量，不为闲话付钱：最后一行之后又收到超过上限的字就当场关流。这一次的用量没报全，照实记成估不出，不当成
+    报全了，也不悄悄当成 0 元。"""
+    import app.api.copilot as copilot
+    from app.api import copilot_gen
+
+    model = Chatty(model=PRICED, replies=[_ops({"op": "ask", "question": "要按哪一天算？"})], tokens=(500, 300))
+
+    async def _model(*_a, **_k):
+        return model, PRICED
+
+    monkeypatch.setattr(copilot, "get_chat_model", _model)
+    assert model.chatter * len("这是一段写在最后一行之后、没人要的闲话。\n") > copilot_gen.DRAIN_CHARS
+    events = await _sse(client, {"instruction": "按天看销售额", "intent": "answer"})
+    assert events[-1]["op"] == "usage" and model.calls == 1, [e["op"] for e in events]
+    row = events[-1]["by_model"][PRICED]
+    assert (row["calls"], row["input_tokens"], row["output_tokens"], row["unpriced_calls"]) == (1, 500, 0, 1), row
+
+
+async def test_the_eval_driver_records_the_same_usage_the_chat_page_receives(client, monkeypatch):
+    """评测的作答（app/eval/answer.Driver._generate → _take）收下的助手用量，就是问数据页经 SSE 收到的那条 usage：同一个
+    生成过程（answer_ops）、同一份计量，读完流之后两边都是报全了的数。"""
+    import app.api.copilot as copilot
+    from app.eval import answer
+
+    replies = [_ops({"op": "ask", "question": "要按哪一天算？"})]
+    model = Scripted(model=PRICED, replies=replies, tokens=(700, 90))
+
+    async def _model(*_a, **_k):
+        return model, PRICED
+
+    monkeypatch.setattr(copilot, "get_chat_model", _model)
+    chat = (await _sse(client, {"instruction": "按天看销售额", "intent": "answer"}))[-1]
+    assert chat["op"] == "usage"
+    driver = answer.Driver("按天看销售额", datasource_ids=[], ctx=answer.TurnContext(
+        suite_id="s", suite_name="虚构题集", eval_run_id="e", case_id=None, case_seq=1, actor=None))
+    ops, refused = await driver._generate("按天看销售额", None)
+    assert refused is None and any(o.get("op") == "ask" for o in ops)
+    driver._take(ops)
+    got = driver.turn.usage["assistant"]
+    assert got["by_model"] == chat["by_model"] and got["cost_usd"] == chat["cost_usd"], (got, chat)
+    assert got["by_model"][PRICED]["output_tokens"] == 90 and got["unpriced_calls"] == 0
 
 
 async def test_a_build_on_an_unpriced_model_says_it_cannot_be_priced(client, monkeypatch):