research(rectification): Jev 意图分类离线对照，结论不可接

来源 C 900 条 + jev-1.13.0 双跑。高置信错误 7%、点选题 answer_class 65%。不改线上分类器。

scripts/research/jev_intent_probe.py

@@ -0,0 +1,679 @@
+#!/usr/bin/env python3
+"""Run Jev (and optional current-model) on the frozen T0 samples.
+
+Offline. Pins `jev-1.13.0`. Reads TYPESAFE_API_KEY from the environment only.
+Source B rows are aggregated and never written as raw text.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
+from typing import Any, Mapping, Sequence
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from scripts.research.jev_intent_questions import (  # noqa: E402
+    JEV_MODEL,
+    build_state,
+    enforce_combo,
+    parse_jev_answers,
+    sdk_questions,
+)
+
+SAMPLES_DIR = ROOT / "scripts" / "research" / "jev_intent_samples"
+CACHE_DIR = Path(r"G:\Ferti\Jyotisha\.cache\jev_intent")
+REPORT_JSON = ROOT / "docs" / "research" / "jev_intent_2026_09_19.json"
+REPORT_MD = ROOT / "docs" / "research" / "jev_intent_2026_09_19.md"
+
+
+def load_jsonl(path: Path) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    if not path.is_file():
+        return rows
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if line.strip():
+            rows.append(json.loads(line))
+    return rows
+
+
+def gold_of(sample: Mapping[str, Any]) -> dict[str, Any]:
+    gold = sample.get("gold") or {}
+    return enforce_combo(gold.get("intent"), gold.get("answer_class"), gold.get("has_new_dated_event"))
+
+
+def call_jev(sample: Mapping[str, Any], *, timeout: float = 60.0) -> dict[str, Any]:
+    from typesafe_sdk import TypeSafeClient
+
+    started = time.perf_counter()
+    with TypeSafeClient() as client:
+        response = client.system_one(
+            state=build_state(sample),
+            questions=sdk_questions(sample),
+            model=JEV_MODEL,
+            timeout=timeout,
+        )
+    elapsed_ms = int((time.perf_counter() - started) * 1000)
+    payload = response.model_dump() if hasattr(response, "model_dump") else dict(response)
+    answers = payload.get("answers") or getattr(response, "answers", {}) or {}
+    if hasattr(answers, "items") and answers and not isinstance(next(iter(answers.values()), None), dict):
+        # SDK objects → dicts
+        converted = {}
+        for key, value in answers.items():
+            converted[key] = value.model_dump() if hasattr(value, "model_dump") else {
+                "type": getattr(value, "type", None),
+                "choice": getattr(value, "choice", None),
+                "probabilities": getattr(value, "probabilities", None),
+                "confidence": getattr(value, "confidence", None),
+                "noul": getattr(value, "noul", None),
+            }
+        answers = converted
+    parsed = parse_jev_answers(answers)
+    usage = payload.get("usage") or {}
+    return {
+        "ok": True,
+        "unavailable": False,
+        "model": payload.get("model") or JEV_MODEL,
+        "intent": parsed["intent"],
+        "answer_class": parsed["answer_class"],
+        "has_new_dated_event": parsed["has_new_dated_event"],
+        "confidence": parsed["confidence"],
+        "raw": parsed.get("raw"),
+        "input_tokens": usage.get("input_tokens"),
+        "output_tokens": usage.get("output_tokens"),
+        "elapsed_ms": elapsed_ms,
+    }
+
+
+def call_jev_retry(sample: Mapping[str, Any], *, retries: int = 4) -> dict[str, Any]:
+    last_error = ""
+    delay = 1.0
+    for attempt in range(retries):
+        try:
+            return call_jev(sample)
+        except Exception as exc:  # noqa: BLE001
+            last_error = type(exc).__name__
+            name = type(exc).__name__
+            if "RateLimit" in name or "Overload" in name or "Timeout" in name:
+                time.sleep(delay)
+                delay = min(delay * 2, 16)
+                continue
+            if "Authentication" in name or "Permission" in name:
+                break
+            time.sleep(delay)
+            delay = min(delay * 2, 16)
+    return {
+        "ok": False,
+        "unavailable": True,
+        "model": JEV_MODEL,
+        "intent": None,
+        "answer_class": None,
+        "has_new_dated_event": None,
+        "confidence": None,
+        "raw": None,
+        "input_tokens": None,
+        "output_tokens": None,
+        "elapsed_ms": None,
+        "error": last_error,
+    }
+
+
+def match_fields(pred: Mapping[str, Any] | None, gold: Mapping[str, Any]) -> dict[str, bool]:
+    if not pred or pred.get("unavailable"):
+        return {"intent": False, "answer_class": False, "has_new_dated_event": False, "all": False}
+    intent_ok = pred.get("intent") == gold.get("intent")
+    class_needed = gold.get("intent") == "answer_current_focus"
+    class_ok = (pred.get("answer_class") == gold.get("answer_class")) if class_needed else True
+    dated_ok = bool(pred.get("has_new_dated_event")) == bool(gold.get("has_new_dated_event"))
+    return {
+        "intent": intent_ok,
+        "answer_class": class_ok,
+        "has_new_dated_event": dated_ok,
+        "all": intent_ok and class_ok and dated_ok,
+    }
+
+
+def same_output(a: Mapping[str, Any] | None, b: Mapping[str, Any] | None) -> bool:
+    if not a or not b:
+        return False
+    if a.get("unavailable") or b.get("unavailable"):
+        return a.get("unavailable") is True and b.get("unavailable") is True
+    return (
+        a.get("intent") == b.get("intent")
+        and a.get("answer_class") == b.get("answer_class")
+        and bool(a.get("has_new_dated_event")) == bool(b.get("has_new_dated_event"))
+    )
+
+
+def percentile(values: Sequence[float], p: float) -> float | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    if len(ordered) == 1:
+        return float(ordered[0])
+    idx = int(round((len(ordered) - 1) * p))
+    return float(ordered[idx])
+
+
+def layer_metrics(rows: Sequence[Mapping[str, Any]], *, pred_key: str) -> dict[str, Any]:
+    n = len(rows)
+    unavailable = sum(1 for row in rows if (row.get(pred_key) or {}).get("unavailable"))
+    intent_ok = 0
+    class_ok = 0
+    class_n = 0
+    dated_ok = 0
+    all_ok = 0
+    high_conf_err = 0
+    low_conf = 0
+    wrong = 0
+    wrong_low = 0
+    no_unsure = 0
+    confidences: list[float] = []
+    elapsed: list[float] = []
+    tokens: list[float] = []
+    for row in rows:
+        gold = row["gold"]
+        pred = row.get(pred_key) or {}
+        flags = match_fields(pred, gold)
+        if pred.get("unavailable"):
+            wrong += 1
+            continue
+        intent_ok += int(flags["intent"])
+        if gold.get("intent") == "answer_current_focus":
+            class_n += 1
+            class_ok += int(flags["answer_class"])
+        dated_ok += int(flags["has_new_dated_event"])
+        all_ok += int(flags["all"])
+        conf = pred.get("confidence")
+        if isinstance(conf, (int, float)):
+            confidences.append(float(conf))
+            if conf >= 0.8 and not flags["all"]:
+                high_conf_err += 1
+            if conf < 0.5:
+                low_conf += 1
+            if not flags["all"]:
+                wrong += 1
+                if conf < 0.5:
+                    wrong_low += 1
+        elif not flags["all"]:
+            wrong += 1
+        if (
+            gold.get("intent") == "answer_current_focus"
+            and gold.get("answer_class") in {"no", "unsure"}
+            and pred.get("answer_class") in {"no", "unsure"}
+            and pred.get("answer_class") != gold.get("answer_class")
+        ):
+            no_unsure += 1
+        if isinstance(pred.get("elapsed_ms"), (int, float)):
+            elapsed.append(float(pred["elapsed_ms"]))
+        if isinstance(pred.get("input_tokens"), (int, float)):
+            tokens.append(float(pred["input_tokens"]))
+    denom = max(n, 1)
+    return {
+        "n": n,
+        "unavailable": unavailable,
+        "intent_acc": intent_ok / denom,
+        "answer_class_acc": (class_ok / class_n) if class_n else None,
+        "answer_class_n": class_n,
+        "dated_acc": dated_ok / denom,
+        "all_acc": all_ok / denom,
+        "high_conf_error_rate": high_conf_err / denom,
+        "low_conf_coverage": low_conf / denom,
+        "low_conf_recall": (wrong_low / wrong) if wrong else None,
+        "no_vs_unsure": no_unsure,
+        "median_ms": percentile(elapsed, 0.5),
+        "p95_ms": percentile(elapsed, 0.95),
+        "mean_input_tokens": (sum(tokens) / len(tokens)) if tokens else None,
+        "cost_usd": ((sum(tokens) / 1_000_000) * 0.042) if tokens else None,
+    }
+
+
+def self_consistency(rows: Sequence[Mapping[str, Any]], a: str, b: str) -> float | None:
+    if not rows:
+        return None
+    hits = sum(1 for row in rows if same_output(row.get(a), row.get(b)))
+    return hits / len(rows)
+
+
+def curve_theta(rows: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
+    """Pick θ on the confidence–accuracy curve for Jev-first fallback."""
+    points = []
+    for theta in (0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9):
+        used = 0
+        correct = 0
+        fallback = 0
+        high_err = 0
+        for row in rows:
+            pred = row.get("jev_1") or {}
+            gold = row["gold"]
+            conf = pred.get("confidence")
+            if pred.get("unavailable") or not isinstance(conf, (int, float)) or conf < theta:
+                fallback += 1
+                continue
+            used += 1
+            flags = match_fields(pred, gold)
+            correct += int(flags["all"])
+            if not flags["all"] and conf >= 0.8:
+                high_err += 1
+        points.append({
+            "theta": theta,
+            "jev_share": used / max(len(rows), 1),
+            "fallback_share": fallback / max(len(rows), 1),
+            "jev_acc": (correct / used) if used else None,
+            "high_conf_errors_in_used": high_err,
+        })
+    # Prefer θ whose used-set accuracy is high and fallback is not everything.
+    ranked = [p for p in points if (p["jev_acc"] or 0) >= 0.85 and p["jev_share"] >= 0.2]
+    recommended = ranked[0]["theta"] if ranked else 0.8
+    if ranked:
+        recommended = max(ranked, key=lambda p: (p["jev_acc"] or 0, p["jev_share"]))["theta"]
+    return {"points": points, "recommended_theta": recommended}
+
+
+def error_portrait(rows: Sequence[Mapping[str, Any]], *, limit: int = 3) -> dict[str, list[dict[str, Any]]]:
+    buckets = {"cjk_colloquial": [], "literal": [], "other": []}
+    for row in rows:
+        gold = row["gold"]
+        pred = row.get("jev_1") or {}
+        current = row.get("current_1") or {}
+        jev_flags = match_fields(pred, gold)
+        cur_flags = match_fields(current, gold) if current else {"all": False}
+        if jev_flags["all"] or not current or not cur_flags["all"]:
+            if jev_flags["all"] or pred.get("unavailable"):
+                continue
+            # still portrait Jev-wrong even if current didn't run
+        message = str(row.get("user_message") or "")
+        rewritten = message
+        if len(rewritten) > 40:
+            rewritten = rewritten[:18] + "…" + rewritten[-12:]
+        kind = "other"
+        if any(token in message for token in ("嗯", "吧", "咋", "噻", "emmm", "hahaha", "额")):
+            kind = "cjk_colloquial"
+        elif gold.get("intent") != pred.get("intent"):
+            kind = "literal"
+        if len(buckets[kind]) >= limit:
+            continue
+        buckets[kind].append({
+            "id": row.get("id"),
+            "layer": row.get("layer"),
+            "rewritten_message": rewritten,
+            "gold": gold,
+            "jev": {
+                "intent": pred.get("intent"),
+                "answer_class": pred.get("answer_class"),
+                "has_new_dated_event": pred.get("has_new_dated_event"),
+                "confidence": pred.get("confidence"),
+            },
+        })
+    return buckets
+
+
+def run_batch(
+    samples: Sequence[Mapping[str, Any]],
+    *,
+    workers: int,
+    run_id: str,
+    cache: dict[str, Any],
+) -> None:
+    pending = []
+    for sample in samples:
+        key = f"{run_id}:{sample['id']}"
+        if key in cache:
+            sample[run_id] = cache[key]
+        else:
+            pending.append(sample)
+    if not pending:
+        return
+    print(f"{run_id}: {len(pending)} calls, {len(samples) - len(pending)} cached", flush=True)
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        futures = {pool.submit(call_jev_retry, sample): sample for sample in pending}
+        done = 0
+        for future in as_completed(futures):
+            sample = futures[future]
+            result = future.result()
+            sample[run_id] = result
+            cache[f"{run_id}:{sample['id']}"] = result
+            done += 1
+            if done % 50 == 0 or done == len(pending):
+                print(f"  {run_id} {done}/{len(pending)}", flush=True)
+
+
+def load_cache() -> dict[str, Any]:
+    path = CACHE_DIR / "jev_runs.json"
+    if not path.is_file():
+        return {}
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError:
+        return {}
+
+
+def save_cache(cache: dict[str, Any]) -> None:
+    CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    path = CACHE_DIR / "jev_runs.json"
+    path.write_text(json.dumps(cache, ensure_ascii=False), encoding="utf-8")
+
+
+def flatten_for_report(sample: Mapping[str, Any], *, include_message: bool) -> dict[str, Any]:
+    row = {
+        "id": sample.get("id"),
+        "source": sample.get("source"),
+        "layer": sample.get("layer"),
+        "gold": sample.get("gold"),
+        "jev_1": sample.get("jev_1"),
+        "jev_2": sample.get("jev_2"),
+        "current_1": sample.get("current_1"),
+        "current_2": sample.get("current_2"),
+    }
+    if include_message:
+        row["user_message"] = sample.get("user_message")
+        row["persona"] = sample.get("persona")
+    return row
+
+
+def write_markdown(report: Mapping[str, Any]) -> None:
+    layers = report["metrics"]["jev_by_layer"]
+    current = report["metrics"].get("current_by_layer") or {}
+    conclusion = report["conclusion"]
+    lines = [
+        "# TypeSafe Jev 接管校正意图分类 · 离线对照（2026-09-19）",
+        "",
+        f"- 任务：`docs/tasks/TASK-rectification-jev-intent-classifier-research-20260919.md`",
+        f"- 基线：`origin/staging` @ `{report['meta']['baseline']}`",
+        f"- Jev 模型：`{JEV_MODEL}`（不用 jev-latest）",
+        f"- 生成器 / 复核：`{report['meta']['generator']}` / `{report['meta']['reviewer']}`",
+        f"- 真值 sha256：synthetic `{report['meta']['sha256']['synthetic']}`；simulated `{report['meta']['sha256']['simulated']}`；disputed `{report['meta']['sha256']['disputed']}`",
+        f"- 来源 B：{report['meta']['source_b_n']} 条（本地，不提交原文）",
+        "",
+        "## 结论",
+        "",
+        f"**{conclusion['verdict']}**。{conclusion['reason']}",
+        "",
+        f"若接：{conclusion['if_connect']}",
+        "",
+        "## T3 指标（Jev，来源 C）",
+        "",
+        "| 层 | n | intent | answer_class | dated | 高置信错误 | 低置信覆盖 | 低置信召回 | no/unsure 互判 | 自洽率 | 中位 ms | P95 ms | 次均 input tok |",
+        "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+    ]
+    for layer in ("choice", "collect", "none"):
+        m = layers[layer]
+        cons = report["metrics"]["jev_self_consistency"].get(layer)
+        def pct(value: float | None) -> str:
+            return "—" if value is None else f"{value:.1%}"
+        def ms(value: float | None) -> str:
+            return "—" if value is None else f"{value:.0f}"
+        lines.append(
+            f"| {layer} | {m['n']} | {pct(m['intent_acc'])} | {pct(m['answer_class_acc'])} | {pct(m['dated_acc'])} | "
+            f"{pct(m['high_conf_error_rate'])} | {pct(m['low_conf_coverage'])} | {pct(m['low_conf_recall'])} | "
+            f"{m['no_vs_unsure']} | {pct(cons)} | {ms(m['median_ms'])} | {ms(m['p95_ms'])} | "
+            f"{(m['mean_input_tokens'] or 0):.0f} |"
+        )
+    lines += [
+        "",
+        "## 现行模型对照",
+        "",
+    ]
+    source_a_m = report["metrics"].get("source_a")
+    if source_a_m:
+        lines += [
+            "",
+            f"来源 A（测试夹具，n={source_a_m['n']}）intent {source_a_m['intent_acc']:.1%}，answer_class {source_a_m['answer_class_acc'] or 0:.1%}，高置信错误 {source_a_m['high_conf_error_rate']:.1%}。",
+        ]
+    if report["meta"].get("current_model"):
+        lines.append(f"模型：`{report['meta']['current_model']}`。")
+        lines.append("")
+        lines.append("| 层 | n | intent | answer_class | dated | 自洽率 | 中位 ms |")
+        lines.append("| --- | ---: | ---: | ---: | ---: | ---: | ---: |")
+        for layer in ("choice", "collect", "none"):
+            m = current.get(layer) or {}
+            cons = (report["metrics"].get("current_self_consistency") or {}).get(layer)
+            def pct(value: float | None) -> str:
+                return "—" if value is None else f"{value:.1%}"
+            lines.append(
+                f"| {layer} | {m.get('n', 0)} | {pct(m.get('intent_acc'))} | {pct(m.get('answer_class_acc'))} | "
+                f"{pct(m.get('dated_acc'))} | {pct(cons)} | {m.get('median_ms') or 0:.0f} |"
+            )
+    else:
+        lines.append(report["meta"].get("current_model_note") or "未跑现行模型。")
+    lines += [
+        "",
+        "## 代表性检验（来源 B vs 来源 C）",
+        "",
+        report["metrics"]["representativeness"]["note"],
+        "",
+        "## 置信度–准确率曲线与 θ",
+        "",
+        f"推荐 θ = {report['metrics']['theta']['recommended_theta']}。点：",
+        "",
+        "```json",
+        json.dumps(report["metrics"]["theta"]["points"], ensure_ascii=False, indent=2),
+        "```",
+        "",
+        "## 错例画像（改写后，无法对应真实会话）",
+        "",
+    ]
+    portrait = report["metrics"]["error_portrait"]
+    for kind, rows in portrait.items():
+        lines.append(f"### {kind}")
+        lines.append("")
+        if not rows:
+            lines.append("无。")
+            lines.append("")
+            continue
+        for item in rows:
+            lines.append(
+                f"- `{item['id']}` / {item['layer']} / 「{item['rewritten_message']}」"
+                f" gold={item['gold']} jev={item['jev']}"
+            )
+        lines.append("")
+    lines += [
+        "## T1 判据对照",
+        "",
+        "见 `scripts/research/jev_intent_questions.py` 的 `CRITERION_MAP`。丢掉的语义：生产提示里的「通常」「不要按 A/B/C/D 猜」「不要按关键词表」；结构不变量改由 `enforce_combo` 强制。",
+        "",
+        "| 现行提示原句 | Jev 落点 | 丢失 |",
+        "| --- | --- | --- |",
+    ]
+    from scripts.research.jev_intent_questions import CRITERION_MAP
+    for row in CRITERION_MAP:
+        lost = row["lost"].replace("|", "\\|") if row["lost"] else "—"
+        lines.append(f"| {row['production']} | {row['jev']} | {lost} |")
+    lines += [
+        "",
+        "## 回退",
+        "",
+        "任何上线方案必须保留回退到现行会话模型的路径。官方限流会动态调整。",
+        "",
+    ]
+    REPORT_MD.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def decide_verdict(report: dict[str, Any]) -> dict[str, str]:
+    layers = report["metrics"]["jev_by_layer"]
+    current = report["metrics"].get("current_by_layer") or {}
+    theta = report["metrics"]["theta"]["recommended_theta"]
+    source_b_n = report["meta"]["source_b_n"]
+    high_err = max(layers[layer]["high_conf_error_rate"] for layer in layers)
+    low_recall = [layers[layer]["low_conf_recall"] for layer in layers if layers[layer]["low_conf_recall"] is not None]
+    recall_ok = bool(low_recall) and all(v is not None and v >= 0.60 for v in low_recall)
+    choice_class = layers.get("choice", {}).get("answer_class_acc")
+    if source_b_n < 30:
+        represent = "来源 B 不足 30 条，代表性检验只报数、不判定。"
+    else:
+        represent = report["metrics"]["representativeness"]["note"]
+    absolute_fail: list[str] = []
+    if high_err > 0.03:
+        absolute_fail.append(f"高置信错误率 {high_err:.1%} > 3%")
+    if not recall_ok:
+        absolute_fail.append(f"低置信召回 {low_recall} < 60%（错了却仍高置信）")
+    if choice_class is not None and choice_class < 0.80:
+        absolute_fail.append(f"点选题 answer_class 准确率 {choice_class:.1%}（写库字段）")
+    if absolute_fail:
+        return {
+            "verdict": "不可接",
+            "reason": (
+                "来源 C 上 Jev 的绝对门槛未过：" + "；".join(absolute_fail) + "。"
+                + represent
+                + (" 现行模型对照未跑，相对 −3pp 门槛无法计算。" if not current else "")
+            ),
+            "if_connect": (
+                "不接。现行分类器继续用会话选定的贵模型。"
+                "若还要观察中文口语，只允许 (b) 影子双跑只记日志，不得按 confidence 写库。"
+                f"曲线上 θ={theta} 时高置信错误仍未清零。"
+            ),
+        }
+    if not current:
+        return {
+            "verdict": "缺数据",
+            "reason": (
+                "绝对门槛未破，但本机没有会话模型凭据，现行分类器对照未跑，"
+                "无法检验「各层 ≥ 现行 − 3 个百分点」。"
+                f"{represent}"
+            ),
+            "if_connect": (
+                f"不得上线。若只做影子，用 (b) 双跑只记日志。"
+                f"若将来现行对照过门，优先 (a) Jev 先判、confidence < {theta} 回退现行模型。"
+            ),
+        }
+    gaps = []
+    for layer in ("choice", "collect", "none"):
+        jev = layers[layer]["intent_acc"]
+        cur = (current.get(layer) or {}).get("intent_acc")
+        if cur is None:
+            continue
+        gaps.append((layer, jev, cur, jev - (cur - 0.03)))
+    failed = [g for g in gaps if g[3] < 0]
+    if failed:
+        return {
+            "verdict": "不可接",
+            "reason": f"相对门槛未过：失败层={failed}。{represent}",
+            "if_connect": "不接。现行分类器继续用会话选定的贵模型。",
+        }
+    return {
+        "verdict": "可接",
+        "reason": f"来源 C 三层 intent 均 ≥ 现行 − 3pp，高置信错误 ≤3%，低置信召回 ≥60%。{represent}",
+        "if_connect": f"(a) Jev 先判，confidence < {theta} 回退现行模型。必须保留回退路径。",
+    }
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--workers", type=int, default=8)
+    parser.add_argument("--limit", type=int, default=0)
+    parser.add_argument("--skip-second", action="store_true")
+    parser.add_argument("--second-fraction", type=float, default=1.0)
+    args = parser.parse_args(argv)
+    if not os.environ.get("TYPESAFE_API_KEY"):
+        print("TYPESAFE_API_KEY missing", file=sys.stderr)
+        return 2
+    synthetic = load_jsonl(SAMPLES_DIR / "synthetic.jsonl")
+    simulated = load_jsonl(SAMPLES_DIR / "simulated.jsonl")
+    source_b = load_jsonl(CACHE_DIR / "source_b.jsonl")
+    samples = [row for row in synthetic + simulated if row.get("source") in {"A", "C"}]
+    if args.limit:
+        samples = samples[: args.limit]
+    cache = load_cache()
+    try:
+        run_batch(samples, workers=args.workers, run_id="jev_1", cache=cache)
+        save_cache(cache)
+        second = samples
+        if args.skip_second:
+            second = []
+        elif args.second_fraction < 1:
+            n = max(1, int(len(samples) * args.second_fraction))
+            second = [row for row in samples if row.get("source") == "C"][:n]
+        if second:
+            run_batch(second, workers=args.workers, run_id="jev_2", cache=cache)
+            save_cache(cache)
+        if source_b:
+            run_batch(source_b, workers=args.workers, run_id="jev_1", cache=cache)
+            save_cache(cache)
+    finally:
+        save_cache(cache)
+
+    source_c = [row for row in samples if row.get("source") == "C"]
+    source_a = [row for row in samples if row.get("source") == "A"]
+    by_layer = {layer: [row for row in source_c if row.get("layer") == layer] for layer in ("choice", "collect", "none")}
+    jev_metrics = {layer: layer_metrics(rows, pred_key="jev_1") for layer, rows in by_layer.items()}
+    jev_cons = {layer: self_consistency(rows, "jev_1", "jev_2") for layer, rows in by_layer.items()}
+    source_a_metrics = layer_metrics(source_a, pred_key="jev_1") if source_a else None
+    source_b_metrics = layer_metrics(source_b, pred_key="jev_1") if source_b else None
+    represent_note = "来源 B 不足 30 条，代表性检验只报数、不判定。"
+    if source_b_metrics and source_b_metrics["n"] >= 30:
+        deltas = []
+        for layer in ("choice", "collect", "none"):
+            # source B may not be layered the same; compare overall intent acc
+            pass
+        delta = abs((source_b_metrics["intent_acc"] or 0) - (
+            sum(jev_metrics[layer]["intent_acc"] * jev_metrics[layer]["n"] for layer in jev_metrics)
+            / max(sum(jev_metrics[layer]["n"] for layer in jev_metrics), 1)
+        ))
+        if delta > 0.10:
+            represent_note = f"来源 B 与来源 C 的 intent 准确率相差 {delta:.1%} > 10pp，模拟语料不代表真人，来源 C 门槛结论降为缺数据。"
+        else:
+            represent_note = f"来源 B 与来源 C intent 准确率相差 {delta:.1%}，未超过 10pp。"
+    sha = {
+        "synthetic": __import__("hashlib").sha256((SAMPLES_DIR / "synthetic.jsonl").read_bytes()).hexdigest(),
+        "simulated": __import__("hashlib").sha256((SAMPLES_DIR / "simulated.jsonl").read_bytes()).hexdigest(),
+        "disputed": __import__("hashlib").sha256((SAMPLES_DIR / "disputed.jsonl").read_bytes()).hexdigest()
+        if (SAMPLES_DIR / "disputed.jsonl").is_file() else "",
+    }
+    report: dict[str, Any] = {
+        "meta": {
+            "baseline": "d9722d56",
+            "jev_model": JEV_MODEL,
+            "generator": "agent-template-v1",
+            "reviewer": "agent-rule-v1",
+            "sha256": sha,
+            "source_b_n": len(source_b),
+            "current_model": None,
+            "current_model_note": (
+                "本机无会话模型目录凭据（模型 key 在数据库加密配置里）。"
+                "现行 `classifyRectificationTurnIntent` 对照未跑。"
+            ),
+        },
+        "metrics": {
+            "jev_by_layer": jev_metrics,
+            "jev_self_consistency": jev_cons,
+            "current_by_layer": None,
+            "source_a": source_a_metrics,
+            "source_b": source_b_metrics,
+            "representativeness": {"note": represent_note},
+            "theta": curve_theta([row for row in samples if row.get("source") == "C"]),
+            "error_portrait": error_portrait(samples),
+        },
+        "rows": [flatten_for_report(row, include_message=True) for row in samples],
+        "source_b_aggregate": source_b_metrics,
+    }
+    report["conclusion"] = decide_verdict(report)
+    REPORT_JSON.parent.mkdir(parents=True, exist_ok=True)
+    REPORT_JSON.write_text(json.dumps({
+        **report,
+        "rows": [flatten_for_report(row, include_message=row.get("source") != "B") for row in samples],
+    }, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    write_markdown(report)
+    print(json.dumps({
+        "conclusion": report["conclusion"],
+        "jev_by_layer": {k: {
+            "n": v["n"],
+            "intent": v["intent_acc"],
+            "high_conf_err": v["high_conf_error_rate"],
+            "low_conf_recall": v["low_conf_recall"],
+            "unavailable": v["unavailable"],
+        } for k, v in jev_metrics.items()},
+        "self_consistency": jev_cons,
+        "source_b_n": len(source_b),
+    }, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())