P3-C3: 交叉复审 fusion_review.py + doco fuse 命令，产融合B稿743行+fusion_review留痕，更新CLAUDE.md交接C4

doco/CLAUDE.md

@@ -17,7 +17,7 @@
 ## 🔖 状态栏 (STATUS — 每次结束 session 前必须更新这三行)
 
 - **最后更新**：账号A ｜ 2026-06-17
-- **当前状态一句话**：P3-C2 讯飞 ASR 跑通并收口（真转写310句带时间戳，热词命中良好），本轮已 git 提交推 main（9340edc）；**B稿_v2.txt 已就位入库，C3 卡点解除，下一步可直接开 C3**。
+- **当前状态一句话**：P3-C3 交叉复审完工（融合B稿743行+fusion_review 5条留痕落盘），产品文件已就位，下一步 C4 语义融合→融合A稿 docx。
 - **下一个动手的人从这里开始**：见下方「⏩ 交接备注」
 
 ---
@@ -62,26 +62,29 @@
   - `src/doco/term_extract.py` — C1：规则层 + AI 层提取 + 词典累积 + 热词清洗
   - `src/doco/asr_adapter.py` — C2 讯飞 ASR 适配层（已完工）：`get_hot_words(episode_id)` 读热词表、`transcribe()` 返回 `(句子, raw)`、`write_asr_result()` 落 `asr_v2_timed.txt` + `asr_result_raw.json`
   - `src/doco/video_split.py` — P1：含 `extract_audio()`（16kHz/单声道/16bit WAV），C2 直接复用，不另写 ffmpeg
-  - `src/doco/cli.py` — 命令入口（split / terms / asr）
+  - `src/doco/fusion_review.py` — C3：B稿v2 ⊕ ASR 交叉复审 → 融合B稿(743行) + fusion_review.csv
+  - `src/doco/cli.py` — 命令入口（split / terms / asr / fuse）
   - `doco/data/term_dict.json` — 累积词典（当前 110 条）
   - `doco/programs/<episode_id>/` — 每期产物（热词表、各阶段输出）
 - **常用命令**：
   - C1（已实现）：`doco terms --episode-id <id> --a-script <path> [--no-ai]`
   - C2（已实现）：`doco asr --episode-id <id> --input-video <绝对路径mp4> --output-dir <绝对路径 doco/programs/<id>>`（`--skip-asr` 只分离不转写）。**务必传绝对 `--output-dir`，否则落到当前工作目录的 `programs/` 会与 doco 产物分家。**
-  - C3/C4：待实现
+  - C3（已实现）：`doco fuse --episode-id <id> [--no-ai] [--batch-size 35]`
+  - C4：待实现
 - **环境变量 / 密钥**：只放 `doco/.env`（已在 `.gitignore`）。需要的 key：`DEEPSEEK_API_KEY` / `DEEPSEEK_BASE_URL` / `DEEPSEEK_MODEL`、`XFYUN_APP_ID` / `XFYUN_SECRET_KEY`。**代码与对话里只引用变量名，绝不出现真值。**
 
 ---
 
 ## 3. 当前进度（核心交接区，以最新快照为准）
 
-- **正在做**：C2 已收口并推 main，准备进 C3（B稿v2 ⊕ ASR 交叉复审）。
+- **正在做**：C3 已收口，C4 待开工。
-- **卡点 / 待解**：无。C3 三路输入均已就位于 `doco/programs/ep001_20260612_fangkong_fandao/`：`B稿_v2.txt`（743行，P2）、`asr_v2_timed.txt`（310句，C2）、A稿 docx。
+- **卡点 / 待解**：无。
 
 ---
 
 ## 4. 已完成（只追加，最新在最上）
 
+- [2026-06-17 ｜账号A] **P3-C3 收口**：`fusion_review.py` 逐行复审 B稿v2 ⊕ ASR（22 批×35行=743行），产 `融合B稿.txt`（行数/时间戳零偏移）+ `fusion_review.csv`（5条真OCR纠错留痕）。关键机制：断点缓存 `.c3_cache/batch_N.json` 可复用；硬校验行数/时间戳/change_type枚举不过不写文件；后处理修正 `final_text==B原文` 被 LLM 误标为 `minor_edit` 的行（19→5）。`--no-ai` 全 unchanged 可验证管道。
 - [2026-06-17 ｜账号A] **P3-C2 收口**：密钥外置三验收点全过（硬编码已删/读 .env/已 gitignore）+ 旧 key 已轮换；`asr_adapter.py` 并入讯飞上传·轮询·解析，复用 P1 `extract_audio` 分离 16kHz/单声道/16bit WAV；新增 `doco asr` 命令读 C1 64条热词。**真转写《现代防空反导大对决》27分钟音频 → 310句带时间戳**，萨德×6/塔米尔/爱国者/箭二箭三/见证者等术语命中良好，raw json 730KB 落盘。（顺带修 `extract_audio` 在 Windows 的 `text=True` 解码崩溃。）
 - [2026-06-16 ｜账号A] **P3-C1 收口**：A稿→规则+DeepSeek AI 提取→累积词典110条→热词清洗→本期热词表64条；AI 44 候选已逐个回查 A 稿原文无幻觉，算术对账一致。
 - [P2 ｜本轮之前] 本地 OCR（DeepSeek-OCR/Ollama）+ 扒词流水线，产出 **B稿_v2.txt 743 行**。
@@ -91,8 +94,8 @@
 
 ## 5. 待办（按优先级，做完打勾）
 
-- [ ] **C3（下一步，输入已齐）**：B稿v2 ⊕ 新ASR 交叉复审 → 融合B稿（保持743行碎句粒度 + 密集时间戳原样，ASR只复审纠错不改行结构）+ `fusion_review.csv`。
+- [x] **C3**：B稿v2 ⊕ ASR 交叉复审 → 融合B稿(743行时间戳零偏移) + fusion_review.csv(5条留痕)。
-- [ ] **C4**：融合B稿 + A稿语义对齐 → 融合A稿（公文 docx，用 docx skill，GB/T 9704 格式）。
+- [ ] **C4**（下一步）**：融合B稿 + A稿语义对齐 → 融合A稿（公文 docx，GB/T 9704 格式）。按 A稿分段把融合B稿碎句归拢、套 docx 格式，C4 语义融合可开 thinking。
 - [x] **C2** 讯飞 ASR 适配层（密钥外置 + asr_adapter 并入 + `doco asr` 命令 + 真转写310句）
 - [x] P3-C1 术语提取 + 词典 + 热词表
 - [x] P2 本地 OCR + B稿v2
@@ -113,6 +116,9 @@
 - [2026-06-15] **B稿v2 权威升级**：屏幕术语/型号/番号 **B稿v2 ≈ A稿 并列权威**（B稿是屏幕实打的字，更贴近播出）。口语/语序/语气信 ASR；书面结构/分段信 A稿；同音事实错（美以→美伊）用 A稿/B稿v2 覆盖 ASR。
 - [设计期] **分层不变**：OCR 错字纠正、术语统一、时间戳对齐走**规则层**（确定性）；段落归属、整段删除判定、编导笔误识别走 **AI 层**（需语义判断）。
 - [设计期] **顺序约束**：术语提取（C1，产热词）必须在 ASR（C2）之前——已成立。
+- [2026-06-17] **`fusion_review.csv` 定位 = 留痕档案（给爱德华/复核），不是编导读物。** 编导读的是 C4 docx。CSV 可读性不优化，只看 final_text 准不准。
+- [2026-06-17] **`final_text==B原文` 强制归 `unchanged` 不进 review。** LLM 有时把"考虑后决定保留 B 稿原文"标成 `minor_edit`，但实际没改字。`fusion_review.py` 的 `_normalize_unchanged_when_no_edit()` 在写文件前修正（review 19→5）。
+- [2026-06-17] **`change_type` 的 `minor_edit`/`editor_typo` 边界 LLM 把握不稳**（OCR 错字两类混标）。仅 metadata 标签不一致，`final_text` 全对，不影响融合B稿正文。本轮决定忽略；若 C4/爱德华按 `change_type` 做逻辑分流，届时在 prompt 里收紧两个枚举的边界定义。
 - [工程纪律] 各阶段解耦、中间产物落缓存、可断点续跑、可单独重跑；长批处理放独立终端。
 
 ---
@@ -121,10 +127,13 @@
 
 > 下次开工读完这段应能 0 摩擦续上。接手后可清空重写。
 
-- **第一句话该干的**：C2 已收口推 main，C3 三路输入已齐（`B稿_v2.txt` 743行 / `asr_v2_timed.txt` 310句 / A稿 docx，都在 `doco/programs/ep001_20260612_fangkong_fandao/`）。直接给 Cline 出 C3 的 ACT 指令开工。
+- **第一句话该干的**：C3 已完成，产物就位（`融合B稿.txt` 743行 + `fusion_review.csv` 5条留痕，都在 `doco/programs/ep001_20260612_fangkong_fandao/`）。直接开 C4。
-- **C3 目标**：B稿v2 ⊕ asr_v2_timed 交叉复审 → 融合B稿（**保持 743 行碎句 + 密集时间戳原样，ASR 只复审纠错不改行结构**）+ `fusion_review.csv`。权威优先级见关键决策「B稿v2 权威升级」那条。
+- **C4 目标**：融合B稿（743行碎句）+ A稿 docx 语义对齐 → **融合A稿 docx**（公文格式，GB/T 9704）。按 A稿分段（【导视】【主持人N】【解说N】【专家N】【隔断】）把融合B稿碎句归拢到对应段落，套 docx 模板格式输出。C4 语义融合可开 thinking（非抽取类任务，质量优先）。一致性约束：融合A稿由融合B稿生成，不是拿 A 稿改。
-- **重跑真转写命令**（复用已分离 WAV，27分钟音频讯飞轮询等几分钟别中断）：
+- **C4 输入**（均就位）：
-  `doco asr --episode-id ep001_20260612_fangkong_fandao --input-video "E:\tps-dashboard\doco\programs\ep001_20260612_fangkong_fandao\现代防空反导大对决VA0.mp4" --output-dir "E:\tps-dashboard\doco\programs\ep001_20260612_fangkong_fandao"`
+  - `programs/ep001_20260612_fangkong_fandao/融合B稿.txt`（743行，`[XmYs] 文本`）
+  - `programs/ep001_20260612_fangkong_fandao/现代防空反导大对决A稿.docx`（A稿，python-docx 可读）
+  - `programs/ep001_20260612_fangkong_fandao/fusion_review.csv`（5条改动留痕，可选参考）
+- **C3 重跑命令**：`doco fuse --episode-id ep001_20260612_fangkong_fandao`（`--no-ai` 全 unchanged 验证管道，缓存 `.c3_cache/` 复用无需重跑 AI）
 - 通哥连轴转了几轮，回复要紧凑。
 - **已知坑（务必记住）**：
   1. **A稿真实约 6500–7300 汉字**（栏目常态 6000–8000），**不是 2.2 万**——那是用 `wc -m` 把 UTF-8 字节当字数的乌龙（一个汉字 3 字节）。docx 无文本框无表格，`python-docx` 提取的 ~7.5k 字符就是全文。

doco/src/doco/cli.py

@@ -19,6 +19,9 @@ from .term_extract import run_terms
 # P3 C2 讯飞 ASR
 from .asr_adapter import get_hot_words, transcribe, write_asr_result
 
+# P3 C3 B稿⊕ASR 交叉复审融合
+from .fusion_review import run_fusion
+
 
 @click.group()
 @click.version_option(version="0.1.0")
@@ -216,6 +219,72 @@ def terms(
         sys.exit(1)
 
 
+@main.command("fuse")
+@click.option("--episode-id", required=True, help="节目 ID,如 ep001_20260612_fangkong_fandao")
+@click.option(
+    "--output-dir",
+    default=None,
+    type=click.Path(),
+    help="输出目录(默认 programs/<episode-id>/)",
+)
+@click.option(
+    "--no-ai",
+    is_flag=True,
+    default=False,
+    help="跳过 LLM 只跑规则层(=全 unchanged)",
+)
+@click.option(
+    "--batch-size",
+    default=35,
+    type=int,
+    help="每批送审行数(默认 35)",
+)
+def fuse(
+    episode_id: str,
+    output_dir: str,
+    no_ai: bool,
+    batch_size: int,
+):
+    """
+    P3 C3: B稿 ⊕ ASR 交叉复审融合
+
+    逐行复审 B稿(屏幕字幕OCR),以 ASR(口语转写)为上下文参考,
+    只做纠错,绝不合并行、不拆行、不增删行、不改时间戳。
+
+    --no-ai: 跳过 LLM,全 unchanged(验证管道)
+    --batch-size: 每批送审行数,默认 35
+
+    产物:
+      - 融合B稿.txt(与 B稿_v2 逐行时间戳一致)
+      - fusion_review.csv(仅含 change_type≠unchanged 或 confidence<0.8 的行)
+    """
+    if output_dir is None:
+        out_dir = Path("programs") / episode_id
+    else:
+        out_dir = Path(output_dir)
+
+    click.echo(f"[doco fuse] episode_id={episode_id}")
+    click.echo(f"[doco fuse] output_dir={out_dir}")
+    click.echo(f"[doco fuse] no_ai={no_ai}")
+    click.echo(f"[doco fuse] batch_size={batch_size}")
+
+    try:
+        stats = run_fusion(
+            episode_id=episode_id,
+            output_dir=str(out_dir),
+            no_ai=no_ai,
+            batch_size=batch_size,
+        )
+        click.echo(f"[ok] 总行数: {stats['total_lines']}")
+        click.echo(f"[ok] 各 change_type 计数: {stats['change_counts']}")
+        click.echo(f"[ok] 进 review 行数: {stats['review_lines']}")
+        click.echo(f"[ok] 融合B稿: {stats['fused_path']}")
+        click.echo(f"[ok] review CSV: {stats['csv_path']}")
+    except Exception as e:
+        click.echo(f"[error] {e}", err=True)
+        sys.exit(1)
+
+
 @main.command("asr")
 @click.option("--episode-id", required=True, help="节目 ID,如 ep001_20260612_fangkong_fandao")
 @click.option(

doco/src/doco/fusion_review.py

@@ -0,0 +1,534 @@
+# -*- coding: utf-8 -*-
+"""
+C3: B稿v2 ⊕ ASR 交叉复审 → 融合B稿(743行) + fusion_review.csv
+=============================================================
+职责:逐行复审 B稿(屏幕字幕OCR),以 ASR(口语转写)为上下文参考,
+只做纠错,严禁改行数/时间戳。
+"""
+
+import json
+import re
+import sys
+from pathlib import Path
+from typing import List, Dict, Optional
+
+from .llm import chat
+
+# --------------------------------------------------------------------------
+# 常量
+# --------------------------------------------------------------------------
+
+CHANGE_TYPE_ENUM = frozenset(
+    {
+        "unchanged",
+        "minor_edit",
+        "term_normalize",
+        "rewrite_large",
+        "segment_delete",
+        "segment_add",
+        "editor_typo",
+    }
+)
+
+SYSTEM_PROMPT = """你是《军事科技》专题片文稿校审员。给你 B稿(屏幕字幕OCR,逐行碎句,带时间戳) 和对应的 ASR(口语转写)。
+你的任务:逐行复审 B稿,只做纠错,绝不合并行、不拆行、不增删行、不改时间戳。
+权威优先级:
+- 屏幕术语/型号/番号(箭-3/萨德/见证者-136等): B稿为准(屏幕实打的字)
+- B稿明显是OCR错字而ASR是对的: 用ASR覆盖
+- 同音事实错(如"美以"vs"美伊"): 以书面规范为准,存疑进review
+- 一两个字的等价差异(的/地、啊等语气): 算 unchanged,不要改
+每行输出: line_no, final_text(纠错后,默认等于B原文), change_type(7选1), confidence(0~1), reason(简短,unchanged时留空)
+只返回JSON数组,不要任何解释文字。
+change_type枚举: unchanged/minor_edit/term_normalize/rewrite_large/segment_delete/segment_add/editor_typo"""
+
+
+# --------------------------------------------------------------------------
+# 1. 解析带时间戳的行
+# --------------------------------------------------------------------------
+
+def parse_timed_lines(path) -> List[dict]:
+    r"""
+    解析 "[XmYs] 文本" → [{"idx":int, "ts_raw":"0m8s", "ts_sec":8, "text":"导弹呼啸而过"}]
+    正则: ^\[(\d+)m(\d+)s\]\s*(.*)$ ; ts_sec = m*60+s
+    解析失败的行要抛异常并打印行号,不许静默跳过
+    """
+    p = Path(path)
+    if not p.exists():
+        raise FileNotFoundError(f"文件不存在: {path}")
+
+    pattern = re.compile(r"^\[(\d+)m(\d+)s\]\s*(.*)$")
+    lines_raw = p.read_text(encoding="utf-8").splitlines()
+    result = []
+
+    for idx, line in enumerate(lines_raw, start=1):
+        line = line.strip()
+        if not line:
+            continue  # 跳过空行
+        m = pattern.match(line)
+        if not m:
+            raise ValueError(
+                f"行 {idx} 解析失败,无法匹配时间戳格式: {repr(line[:120])}\n"
+                f"文件: {path}"
+            )
+        minutes = int(m.group(1))
+        seconds = int(m.group(2))
+        ts_raw = f"{minutes}m{seconds}s"
+        ts_sec = minutes * 60 + seconds
+        text = m.group(3).strip()
+        result.append(
+            {
+                "idx": idx,
+                "ts_raw": ts_raw,
+                "ts_sec": ts_sec,
+                "text": text,
+            }
+        )
+
+    return result
+
+
+# --------------------------------------------------------------------------
+# 2. 对齐 ASR 上下文
+# --------------------------------------------------------------------------
+
+def align_asr_context(b_lines: List[dict], asr_lines: List[dict]) -> List[str]:
+    """
+    为每个 B 行找时间窗内的 ASR 上下文(用于喂 LLM)
+    规则: 取 ts_sec 落在 [b_ts-3, b_next_ts+3] 区间的 ASR 句拼接;
+    边界用前后各扩 1 句兜底。返回与 b_lines 等长的 context 列表
+    """
+    n = len(b_lines)
+    contexts = []
+
+    # 预计算 B 行的时间窗: [b[i].ts_sec - 3, b[i+1].ts_sec + 3]
+    # 最后一行用 b[i].ts_sec + 10 作为上界
+    windows = []
+    for i, bl in enumerate(b_lines):
+        lo = bl["ts_sec"] - 3
+        if i + 1 < n:
+            hi = b_lines[i + 1]["ts_sec"] + 3
+        else:
+            hi = bl["ts_sec"] + 10
+        windows.append((lo, hi))
+
+    asr_count = len(asr_lines)
+
+    for i, (lo, hi) in enumerate(windows):
+        # 找到落在窗口内的 ASR 句索引
+        hit_indices = []
+        for j, al in enumerate(asr_lines):
+            if lo <= al["ts_sec"] <= hi:
+                hit_indices.append(j)
+
+        if not hit_indices:
+            # 无命中:取距离最近的 1 句
+            best_j = None
+            best_dist = float("inf")
+            mid_ts = (lo + hi) / 2
+            for j, al in enumerate(asr_lines):
+                dist = abs(al["ts_sec"] - mid_ts)
+                if dist < best_dist:
+                    best_dist = dist
+                    best_j = j
+            if best_j is not None:
+                start_j = max(0, best_j - 1)
+                end_j = min(asr_count - 1, best_j + 1)
+            else:
+                start_j = 0
+                end_j = 0
+        else:
+            # 命中句的范围 + 前后各扩 1
+            start_j = max(0, hit_indices[0] - 1)
+            end_j = min(asr_count - 1, hit_indices[-1] + 1)
+
+        # 拼接 [start_j, end_j] 的 ASR 文本
+        selected = asr_lines[start_j : end_j + 1]
+        context = " ".join(s["text"] for s in selected)
+        contexts.append(context)
+
+    assert len(contexts) == n, f"context 列表长度 {len(contexts)} != B 稿行数 {n}"
+    return contexts
+
+
+# --------------------------------------------------------------------------
+# 3. 构造 Prompt
+# --------------------------------------------------------------------------
+
+def build_prompt(batch_b: List[dict], batch_ctx: List[str]) -> List[dict]:
+    """
+    构造 messages,见下方"四、Prompt 模板"
+    """
+    assert len(batch_b) == len(batch_ctx), (
+        f"batch_b({len(batch_b)}) 与 batch_ctx({len(batch_ctx)}) 长度不一致"
+    )
+
+    user_lines = []
+    for bl, ctx in zip(batch_b, batch_ctx):
+        line_no = bl["idx"]
+        b_text = bl["text"]
+        asr_text = ctx if ctx else "(无ASR上下文)"
+        user_lines.append(
+            f"[行{line_no}] B稿: \"{b_text}\"  ASR上下文: \"{asr_text}\""
+        )
+
+    user_content = "\n".join(user_lines)
+
+    messages = [
+        {"role": "system", "content": SYSTEM_PROMPT},
+        {"role": "user", "content": user_content},
+    ]
+    return messages
+
+
+# --------------------------------------------------------------------------
+# 4. 单批复审
+# --------------------------------------------------------------------------
+
+def review_batch(
+    batch_b: List[dict], batch_ctx: List[str], no_ai: bool = False
+) -> List[dict]:
+    """
+    no_ai=True: 直接回填 unchanged(final_text=b原文, change_type="unchanged", confidence=1.0)
+    no_ai=False: 调 llm.chat(messages, thinking=False, max_tokens=4000, temperature=0.0)
+      解析返回 JSON 数组; 每元素 {line_no, final_text, change_type, confidence, reason}
+    返回标准化记录列表
+    """
+    if no_ai:
+        records = []
+        for bl in batch_b:
+            records.append(
+                {
+                    "line_no": bl["idx"],
+                    "final_text": bl["text"],
+                    "change_type": "unchanged",
+                    "confidence": 1.0,
+                    "reason": "",
+                }
+            )
+        return records
+
+    # ---- AI 路径 ----
+    messages = build_prompt(batch_b, batch_ctx)
+
+    try:
+        raw_response = chat(
+            messages,
+            thinking=False,
+            max_tokens=4000,
+            temperature=0.0,
+        )
+    except Exception as e:
+        print(
+            f"[fusion_review] LLM 调用失败,回退为 unchanged 批次: {e}",
+            file=sys.stderr,
+        )
+        # 回退:全部 unchanged
+        records = []
+        for bl in batch_b:
+            records.append(
+                {
+                    "line_no": bl["idx"],
+                    "final_text": bl["text"],
+                    "change_type": "unchanged",
+                    "confidence": 1.0,
+                    "reason": f"LLM调用失败回退: {str(e)[:80]}",
+                }
+            )
+        return records
+
+    # 解析 JSON
+    parsed = _parse_llm_json_response(raw_response, len(batch_b))
+
+    # 标准化并校验
+    records = []
+    for item in parsed:
+        line_no = item.get("line_no")
+        final_text = item.get("final_text", "")
+        change_type = item.get("change_type", "unchanged")
+        confidence = item.get("confidence", 1.0)
+        reason = item.get("reason", "")
+
+        # 校验 change_type
+        if change_type not in CHANGE_TYPE_ENUM:
+            original_ct = change_type
+            print(
+                f"[fusion_review] 行 {line_no} 非法 change_type='{original_ct}', 强制改为 unchanged",
+                file=sys.stderr,
+            )
+            change_type = "unchanged"
+            final_text = ""  # 下面会回填
+            reason = f"LLM返回非法change_type({original_ct}),回退unchanged"
+
+        # 如果 change_type 被改为 unchanged 但 final_text 为空,回填 B 原文
+        if change_type == "unchanged" and not final_text:
+            # 从 batch_b 找回原文
+            for bl in batch_b:
+                if bl["idx"] == line_no:
+                    final_text = bl["text"]
+                    break
+
+        records.append(
+            {
+                "line_no": line_no,
+                "final_text": final_text,
+                "change_type": change_type,
+                "confidence": float(confidence),
+                "reason": reason or "",
+            }
+        )
+
+    return records
+
+
+def _parse_llm_json_response(raw: str, expected_len: int) -> List[dict]:
+    """解析 LLM 返回的 JSON,处理 markdown code fences 等常见包装。"""
+    text = raw.strip()
+
+    # 去掉可能的 markdown code fences
+    if text.startswith("```"):
+        lines = text.splitlines()
+        # 去掉第一行 ```json 或 ```
+        if lines and lines[0].startswith("```"):
+            lines = lines[1:]
+        # 去掉最后一行 ```
+        if lines and lines[-1].strip() == "```":
+            lines = lines[:-1]
+        text = "\n".join(lines).strip()
+
+    try:
+        result = json.loads(text)
+    except json.JSONDecodeError as e:
+        raise ValueError(
+            f"LLM 返回 JSON 解析失败: {e}\n"
+            f"原始响应前 500 字符: {raw[:500]}"
+        )
+
+    if not isinstance(result, list):
+        raise ValueError(
+            f"LLM 返回不是 JSON 数组,类型为 {type(result).__name__}"
+        )
+
+    if len(result) != expected_len:
+        raise ValueError(
+            f"LLM 返回 {len(result)} 条记录,期望 {expected_len} 条。"
+            f"该批次将回退为 unchanged 并重新请求。"
+        )
+
+    return result
+
+
+# --------------------------------------------------------------------------
+# 5. 主流程
+# --------------------------------------------------------------------------
+
+def run_fusion(
+    episode_id: str,
+    output_dir: str,
+    no_ai: bool = False,
+    batch_size: int = 35,
+) -> dict:
+    """
+    主流程:
+    1. 读 output_dir/B稿_v2.txt → b_lines（断言行数>0）
+    2. 读 output_dir/asr_v2_timed.txt → asr_lines
+    3. align_asr_context 生成等长 context
+    4. 按 batch_size 分块；每块结果落缓存,已存在则复用（断点续跑）
+    5. 逐块 review_batch,汇总所有记录
+    6. 硬校验（任一不过就 raise,不写出文件）
+    7. 写 output_dir/融合B稿.txt
+    8. 写 output_dir/fusion_review.csv
+    9. 返回统计 dict
+    """
+    out_dir = Path(output_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    b_path = out_dir / "B稿_v2.txt"
+    asr_path = out_dir / "asr_v2_timed.txt"
+
+    if not b_path.exists():
+        raise FileNotFoundError(f"B稿_v2.txt 不存在: {b_path}")
+    if not asr_path.exists():
+        raise FileNotFoundError(f"asr_v2_timed.txt 不存在: {asr_path}")
+
+    # Step 1: 解析 B 稿
+    b_lines = parse_timed_lines(b_path)
+    assert len(b_lines) > 0, f"B稿_v2.txt 解析后为空: {b_path}"
+
+    # Step 2: 解析 ASR
+    asr_lines = parse_timed_lines(asr_path)
+
+    # Step 3: 对齐 ASR 上下文
+    contexts = align_asr_context(b_lines, asr_lines)
+    assert len(contexts) == len(b_lines), (
+        f"context 长度 {len(contexts)} != B 稿行数 {len(b_lines)}"
+    )
+
+    # Step 4: 分块 + 缓存
+    cache_dir = out_dir / ".c3_cache"
+    cache_dir.mkdir(parents=True, exist_ok=True)
+
+    all_records = []
+    total_batches = (len(b_lines) + batch_size - 1) // batch_size
+
+    for batch_idx in range(total_batches):
+        start = batch_idx * batch_size
+        end = min(start + batch_size, len(b_lines))
+        batch_b = b_lines[start:end]
+        batch_ctx = contexts[start:end]
+
+        cache_path = cache_dir / f"batch_{batch_idx}.json"
+
+        if cache_path.exists():
+            # 断点续跑:复用缓存
+            try:
+                cached = json.loads(cache_path.read_text(encoding="utf-8"))
+                print(f"[fusion_review] 复用缓存 batch_{batch_idx} ({len(cached)} 条)")
+                all_records.extend(cached)
+                continue
+            except Exception as e:
+                print(
+                    f"[fusion_review] 缓存 batch_{batch_idx} 损坏,重新计算: {e}",
+                    file=sys.stderr,
+                )
+
+        print(
+            f"[fusion_review] 复审 batch {batch_idx + 1}/{total_batches} "
+            f"(行 {start + 1}-{end})..."
+        )
+        try:
+            batch_records = review_batch(batch_b, batch_ctx, no_ai=no_ai)
+        except Exception as e:
+            print(
+                f"[fusion_review] batch {batch_idx + 1} 失败,跳过缓存写入: {e}",
+                file=sys.stderr,
+            )
+            # 不写缓存,下次重跑时重新请求该批
+            continue
+
+        # 写入缓存
+        cache_path.write_text(
+            json.dumps(batch_records, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+        all_records.extend(batch_records)
+
+    # Step 6: 硬校验
+    _hard_validate(all_records, b_lines)
+
+    # Step 6.5: 修正语义——final_text 等于 B 原文的行强制归为 unchanged
+    _normalize_unchanged_when_no_edit(all_records, b_lines)
+
+    # Step 7: 写 融合B稿.txt
+    fused_path = out_dir / "融合B稿.txt"
+    fused_lines = []
+    for rec, bl in zip(all_records, b_lines):
+        fused_lines.append(f"[{bl['ts_raw']}] {rec['final_text']}")
+    fused_path.write_text("\n".join(fused_lines) + "\n", encoding="utf-8")
+
+    # Step 8: 写 fusion_review.csv
+    csv_path = out_dir / "fusion_review.csv"
+    csv_rows = [
+        "line_no,timestamp,b_original,asr_context,final_text,change_type,confidence,reason"
+    ]
+    for rec, bl, ctx in zip(all_records, b_lines, contexts):
+        if rec["change_type"] == "unchanged" and rec["confidence"] >= 0.8:
+            continue  # 只写需要 review 的行
+        # CSV 转义: 字段含逗号或引号时用双引号包裹
+        row_fields = [
+            str(rec["line_no"]),
+            bl["ts_raw"],
+            bl["text"],
+            ctx,
+            rec["final_text"],
+            rec["change_type"],
+            str(rec["confidence"]),
+            rec["reason"],
+        ]
+        csv_rows.append(_csv_row(row_fields))
+    csv_path.write_text("\n".join(csv_rows) + "\n", encoding="utf-8")
+
+    # Step 9: 统计
+    stats = {
+        "total_lines": len(b_lines),
+        "change_counts": {},
+        "review_lines": 0,
+    }
+    for rec in all_records:
+        ct = rec["change_type"]
+        stats["change_counts"][ct] = stats["change_counts"].get(ct, 0) + 1
+        if ct != "unchanged" or rec["confidence"] < 0.8:
+            stats["review_lines"] += 1
+
+    print(f"[fusion_review] 融合完成: 总行数={stats['total_lines']}")
+    print(f"[fusion_review] 各 change_type 计数: {stats['change_counts']}")
+    print(f"[fusion_review] 进 review 行数: {stats['review_lines']}")
+    print(f"[fusion_review] 融合B稿: {fused_path}")
+    print(f"[fusion_review] review CSV: {csv_path}")
+
+    stats["fused_path"] = str(fused_path)
+    stats["csv_path"] = str(csv_path)
+
+    return stats
+
+
+def _hard_validate(records: List[dict], b_lines: List[dict]) -> None:
+    """硬校验,任一不过就 raise ValueError,不写出文件。"""
+    # 校验 1: 行数必须相等
+    if len(records) != len(b_lines):
+        raise ValueError(
+            f"行数不一致: records={len(records)}, B稿={len(b_lines)}"
+        )
+
+    # 校验 2: 逐行时间戳不能被动
+    for i, (rec, bl) in enumerate(zip(records, b_lines)):
+        rec_line_no = rec.get("line_no")
+        expected_line_no = bl["idx"]
+        if rec_line_no != expected_line_no:
+            raise ValueError(
+                f"第 {i} 条记录 line_no={rec_line_no}, 期望 {expected_line_no}"
+            )
+
+    # 校验 3: change_type 枚举
+    for rec in records:
+        ct = rec.get("change_type", "")
+        if ct not in CHANGE_TYPE_ENUM:
+            raise ValueError(
+                f"行 {rec.get('line_no')}: 非法 change_type='{ct}'"
+            )
+
+
+def _normalize_unchanged_when_no_edit(
+    records: List[dict], b_lines: List[dict]
+) -> None:
+    """修正语义:final_text 等于 B 原文的行,强制归为 unchanged。
+
+    LLM 有时会把"考虑后决定保留 B 稿"标成 minor_edit,
+    但实际 final_text == b_original, 这不在留痕范围内。
+    """
+    b_text_by_idx = {bl["idx"]: bl["text"] for bl in b_lines}
+    fixed = 0
+    for rec in records:
+        line_no = rec.get("line_no")
+        b_orig = b_text_by_idx.get(line_no)
+        if b_orig is not None and rec.get("final_text") == b_orig:
+            if rec.get("change_type") != "unchanged":
+                rec["change_type"] = "unchanged"
+                rec["confidence"] = 1.0
+                rec["reason"] = ""
+                fixed += 1
+    if fixed:
+        print(
+            f"[fusion_review] 修正 {fixed} 行: final_text==B原文但change_type≠unchanged, 已强制归为 unchanged"
+        )
+
+
+def _csv_row(fields: List[str]) -> str:
+    """将字段列表格式化为 CSV 行,处理逗号和引号转义。"""
+    escaped = []
+    for f in fields:
+        s = str(f)
+        if "," in s or '"' in s or "\n" in s:
+            s = s.replace('"', '""')
+            s = f'"{s}"'
+        escaped.append(s)
+    return ",".join(escaped)