plan(tree/repair): 三项改造实现计划(遥测加固+断点续跑+并发)

6 个 Task: telemetry 防御加固 → call_id 根因修复 → detector L2/L1 扩展 → progress 管理 → 并发编排+CLI → lint+全量测试
2026-07-09 00:08:23 -04:00
parent 8a49ef18e6
commit dbc9d38cd7
9 changed files with 1915 additions and 3 deletions
@@ -0,0 +1,786 @@
+# 建树修复管线三项改造 实现计划
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 根治遥测主键冲突导致的熔断误触发，为修复管线增加视频级断点续跑和 16 路并发。
+
+**Architecture:** 三层改动互相独立——(1) adapters 层遥测防御加固 + call_id 根因修复，对所有 GovernedLLMClient 使用方生效；(2) detector 扩展 L2/L1 空字段检测，零 LLM 成本；(3) tools/repair_trees.py 编排层并发 + 断点续跑。
+
+**Tech Stack:** Python 3.11, asyncio, sqlite3, loguru
+
+**本计划不涉及核心算法迁移，保真校验不适用。**
+
+---
+
+### Task 1: 遥测防御性加固
+
+**Files:**
+- Modify: `adapters/telemetry.py:47-55`（`_INSERT_SQL`）、`adapters/telemetry.py:69-114`（`_write`）
+- Test: `tests/unit/test_telemetry.py`
+
+- [ ] **Step 1: 写失败测试 — 重复 call_id 写入不抛异常**
+
+在 `tests/unit/test_telemetry.py` 末尾追加：
+
+```python
+@pytest.mark.asyncio
+async def test_duplicate_call_id_does_not_raise(recorder, db_path):
+    """重复 call_id 写入应静默忽略（INSERT OR IGNORE），不抛异常。"""
+    kwargs = _make_call_kwargs()
+    await recorder.record_llm_call(**kwargs)
+    # 第二次用相同 call_id 写入不应抛异常
+    await recorder.record_llm_call(**kwargs)
+
+    conn = sqlite3.connect(str(db_path))
+    rows = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone()
+    conn.close()
+    assert rows[0] == 1  # 只有一条记录
+```
+
+- [ ] **Step 2: 运行测试验证失败**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_telemetry.py::test_duplicate_call_id_does_not_raise -v`
+预期: FAIL — `sqlite3.IntegrityError: UNIQUE constraint failed`
+
+- [ ] **Step 3: 写失败测试 — DB 错误不冒泡**
+
+```python
+@pytest.mark.asyncio
+async def test_db_error_does_not_propagate(tmp_path):
+    """SQLite 写入失败时 record_llm_call 应静默降级（logger.warning），不抛异常。"""
+    # 用目录路径当 db_path —— sqlite3.connect 对目录名会在真正操作时报错
+    bad_recorder = SQLiteTelemetryRecorder(db_path=tmp_path / "nonexistent_dir" / "bad.db")
+    kwargs = _make_call_kwargs()
+    # 不应抛异常
+    await bad_recorder.record_llm_call(**kwargs)
+```
+
+- [ ] **Step 4: 运行测试验证失败**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_telemetry.py::test_db_error_does_not_propagate -v`
+预期: FAIL — `sqlite3.OperationalError: unable to open database file`
+
+- [ ] **Step 5: 写失败测试 — 并发写不报锁错**
+
+```python
+@pytest.mark.asyncio
+async def test_concurrent_writes_no_lock_error(recorder, db_path):
+    """16 路并发 record_llm_call 应全部成功，无 database is locked 错误。"""
+    import asyncio
+
+    tasks = []
+    for _ in range(16):
+        kwargs = _make_call_kwargs()  # 每次生成不同 call_id
+        tasks.append(recorder.record_llm_call(**kwargs))
+    await asyncio.gather(*tasks)
+
+    conn = sqlite3.connect(str(db_path))
+    count = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone()[0]
+    conn.close()
+    assert count == 16
+```
+
+- [ ] **Step 6: 运行测试验证失败**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_telemetry.py::test_concurrent_writes_no_lock_error -v`
+预期: 可能 PASS（取决于并发时序）或 FAIL — `database is locked`
+
+- [ ] **Step 7: 实现遥测三层加固**
+
+修改 `adapters/telemetry.py`：
+
+1. `_INSERT_SQL`: `INSERT INTO` → `INSERT OR IGNORE INTO`
+2. `_write` 方法: `sqlite3.connect()` 后追加 WAL + busy_timeout pragma，整个方法体包 `try/except sqlite3.Error`
+
+```python
+_INSERT_SQL = """
+    INSERT OR IGNORE INTO llm_calls (
+        call_id, parent_call_id, session_id,
+        model_name, provider, messages, response, thinking,
+        prompt_tokens, completion_tokens, latency_ms,
+        ttft_ms, max_inter_token_ms,
+        cache_hit, error
+    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
+    """
+
+def _write(self, *, call_id, parent_call_id, session_id, model_name,
+           provider, messages, response, thinking, prompt_tokens,
+           completion_tokens, latency_ms, ttft_ms, max_inter_token_ms,
+           cache_hit, error):
+    """同步写入一条 LLM 调用记录到 SQLite。"""
+    try:
+        conn = sqlite3.connect(str(self._db_path), timeout=10.0)
+        try:
+            conn.execute("PRAGMA journal_mode=WAL")
+            conn.execute("PRAGMA busy_timeout=5000")
+            self._ensure_table(conn)
+            conn.execute(
+                self._INSERT_SQL,
+                (call_id, parent_call_id, session_id, model_name,
+                 provider, messages, response, thinking, prompt_tokens,
+                 completion_tokens, latency_ms, ttft_ms,
+                 max_inter_token_ms, int(cache_hit), error),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+    except sqlite3.Error as exc:
+        logger.warning("遥测写入失败（已降级），call_id={}: {}", call_id, exc)
+```
+
+在文件顶部添加 `from loguru import logger`。
+
+- [ ] **Step 8: 运行全部遥测测试验证通过**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_telemetry.py -v`
+预期: 全部 PASS（含三个新测试）
+
+- [ ] **Step 9: 提交**
+
+```bash
+git add adapters/telemetry.py tests/unit/test_telemetry.py
+git commit -m "fix(telemetry): INSERT OR IGNORE + WAL + try/except 三层防御加固
+
+根治遥测写入主键冲突(UNIQUE constraint)和并发写锁(database is locked)
+导致的异常冒泡，遥测侧信道错误不再污染 LLM 重试链。"
+```
+
+---
+
+### Task 2: GovernedLLMClient call_id 移入重试循环
+
+**Files:**
+- Modify: `adapters/llm.py:298-299`（`call_id` 生成位置）、`adapters/llm.py:358-369`（成功路径 `response.call_id`）
+- Test: `tests/unit/test_governed_llm.py`
+
+- [ ] **Step 1: 更新现有测试断言 — call_id 应每次重试独立**
+
+`test_governed_llm.py:200-202` 当前断言 `assert len(call_ids) == 1`（所有重试记录共用一个 call_id），需要反转为 `assert len(call_ids) == 3`（每次 attempt 独立）。
+
+找到 `test_transient_error_retries_and_records_telemetry` 中：
+
+```python
+    # 所有遥测记录应使用同一个 call_id（Important 2 修复验证）
+    call_ids = {c["call_id"] for c in telemetry.calls}
+    assert len(call_ids) == 1
+```
+
+替换为：
+
+```python
+    # 每次 attempt 应使用独立的 call_id（根因修复：防遥测主键冲突）
+    call_ids = {c["call_id"] for c in telemetry.calls}
+    assert len(call_ids) == 3  # 2 次失败 + 1 次成功 = 3 个独立 call_id
+```
+
+- [ ] **Step 2: 运行测试验证失败**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_governed_llm.py::test_transient_error_retries_and_records_telemetry -v`
+预期: FAIL — `assert 1 == 3`（当前还是共用一个 call_id）
+
+- [ ] **Step 3: 将 call_id 生成移入重试循环**
+
+修改 `adapters/llm.py`。将第 298-299 行的 `call_id = str(uuid4())` **从重试循环外移入循环内**：
+
+找到（约 line 298-338）：
+
+```python
+        # ② call_id 生成
+        call_id = str(uuid4())
+
+        # ③ 缓存查询（cache 为 None 时跳过）
+        cached = await self._cache.get(self._model, messages) if self._cache is not None else None
+        if cached is not None:
+            ...
+            return response
+
+        # ④ 重试循环 + 流式消费
+        last_exc: Exception | None = None
+        for attempt in range(self._max_retries):
+```
+
+改为：
+
+```python
+        # ② 缓存查询（cache 为 None 时跳过）— call_id 在缓存路径独立生成
+        cached = await self._cache.get(self._model, messages) if self._cache is not None else None
+        if cached is not None:
+            cache_call_id = str(uuid4())
+            response = LLMResponse(
+                ...
+                call_id=cache_call_id,
+            )
+            await self._telemetry.record_llm_call(
+                call_id=cache_call_id,
+                ...
+            )
+            return response
+
+        # ③ 重试循环 + 流式消费（每次 attempt 独立 call_id）
+        last_exc: Exception | None = None
+        for attempt in range(self._max_retries):
+            call_id = str(uuid4())
+            attempt_start = time.monotonic()
+```
+
+注意：缓存命中路径的 `call_id` 改为独立的 `cache_call_id`（保持在循环外生成，因为不走重试）。
+
+- [ ] **Step 4: 运行全部 GovernedLLM 测试验证通过**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_governed_llm.py -v`
+预期: 全部 PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add adapters/llm.py tests/unit/test_governed_llm.py
+git commit -m "fix(llm): call_id 移入重试循环，每 attempt 独立
+
+消除重试时遥测主键冲突的根因。每次 attempt 独立记录，
+parent_call_id 不受影响（循环外固定），更利于事后诊断重试轨迹。"
+```
+
+---
+
+### Task 3: 检测器扩展 L2/L1 空字段
+
+**Files:**
+- Modify: `app/tree/repair/detector.py:57-68`（L1 循环）、`app/tree/repair/detector.py:73-84`（L2 循环）
+- Test: `tests/unit/test_repair_detector.py`
+
+- [ ] **Step 1: 写失败测试 — L2 event_description 为空触发 empty_field**
+
+在 `tests/unit/test_repair_detector.py` 追加：
+
+```python
+def test_detects_l2_empty_event_description():
+    """L2 event_description 为空应报 empty_field。"""
+    l3 = L3Node(
+        id="l1_0_l2_0_l3_0",
+        card=L3Card("正常帧", ["实体"], ["动作"], [], "居中", {}),
+        timestamp=1.0,
+    )
+    l2 = L2Node(
+        id="l1_0_l2_0",
+        card=L2Card("", [], [], [], [], "", None),  # event_description 为空
+        time_range=(0.0, 10.0),
+        children=[l3],
+    )
+    l1 = L1Node(
+        id="l1_0",
+        card=L1Card("正常场景", "", [], [], [], [], ""),
+        time_range=(0.0, 10.0),
+        children=[l2],
+    )
+    index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1])
+    issues = detect_issues(index)
+    l2_empties = [i for i in issues if i.level == 2 and i.issue_type == "empty_field"]
+    assert len(l2_empties) == 1
+    assert "event_description" in l2_empties[0].details
+```
+
+- [ ] **Step 2: 写失败测试 — L1 scene_summary 为空触发 empty_field**
+
+```python
+def test_detects_l1_empty_scene_summary():
+    """L1 scene_summary 为空应报 empty_field。"""
+    l3 = L3Node(
+        id="l1_0_l2_0_l3_0",
+        card=L3Card("正常帧", ["实体"], ["动作"], [], "居中", {}),
+        timestamp=1.0,
+    )
+    l2 = L2Node(
+        id="l1_0_l2_0",
+        card=L2Card("正常事件", [], [], [], [], "", None),
+        time_range=(0.0, 10.0),
+        children=[l3],
+    )
+    l1 = L1Node(
+        id="l1_0",
+        card=L1Card("", "", [], [], [], [], ""),  # scene_summary 为空
+        time_range=(0.0, 10.0),
+        children=[l2],
+    )
+    index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1])
+    issues = detect_issues(index)
+    l1_empties = [i for i in issues if i.level == 1 and i.issue_type == "empty_field"]
+    assert len(l1_empties) == 1
+    assert "scene_summary" in l1_empties[0].details
+```
+
+- [ ] **Step 3: 运行测试验证失败**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_repair_detector.py::test_detects_l2_empty_event_description tests/unit/test_repair_detector.py::test_detects_l1_empty_scene_summary -v`
+预期: 两个都 FAIL
+
+- [ ] **Step 4: 实现 L2/L1 空字段检测**
+
+修改 `app/tree/repair/detector.py` 的 `detect_issues` 函数：
+
+在 L2 的 `if not l2.children:` 检查之后（`continue` 之前），插入 L2 空字段检测：
+
+```python
+        for l2 in l1.children:
+            # L2: event_description 不为空
+            if not l2.card.event_description:
+                issues.append(
+                    NodeIssue(
+                        node_id=l2.id,
+                        level=2,
+                        issue_type="empty_field",
+                        details="L2 节点字段为空: event_description",
+                    )
+                )
+
+            # L2: children 不为空
+            if not l2.children:
+                ...
+```
+
+在 L1 的 `if not l1.children:` 检查之前，插入 L1 空字段检测：
+
+```python
+    for l1 in index.roots:
+        # L1: scene_summary 不为空
+        if not l1.card.scene_summary:
+            issues.append(
+                NodeIssue(
+                    node_id=l1.id,
+                    level=1,
+                    issue_type="empty_field",
+                    details="L1 节点字段为空: scene_summary",
+                )
+            )
+
+        # L1: children 不为空
+        if not l1.children:
+            ...
+```
+
+同步更新 docstring 的"检查项"列表。
+
+- [ ] **Step 5: 运行全部 detector 测试验证通过**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_repair_detector.py -v`
+预期: 全部 PASS
+
+- [ ] **Step 6: 提交**
+
+```bash
+git add app/tree/repair/detector.py tests/unit/test_repair_detector.py
+git commit -m "feat(detector): 扩展空字段检测到 L2 event_description / L1 scene_summary
+
+断点续跑判据需要 L2/L1 层的 empty_field 检测。零 LLM 成本。"
+```
+
+---
+
+### Task 4: 断点续跑 — progress 文件管理
+
+**Files:**
+- Modify: `tools/repair_trees.py`（新增 progress 读写函数 + 跳过逻辑）
+- Test: `tests/unit/test_repair_progress.py`（新建）
+
+- [ ] **Step 1: 写失败测试 — progress 读写与幂等性**
+
+新建 `tests/unit/test_repair_progress.py`：
+
+```python
+"""修复管线断点续跑 progress 管理测试。"""
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+
+import pytest
+
+
+def test_load_progress_missing_file(tmp_path):
+    """progress 文件不存在时返回空集合。"""
+    from tools.repair_trees import load_progress
+
+    result = load_progress(tmp_path / "nonexistent.json")
+    assert result == set()
+
+
+def test_load_progress_valid_file(tmp_path):
+    """正常读取已有 progress 文件。"""
+    from tools.repair_trees import load_progress
+
+    path = tmp_path / "progress.json"
+    path.write_text(json.dumps({"finished_video_ids": ["vid_a", "vid_b"]}))
+    result = load_progress(path)
+    assert result == {"vid_a", "vid_b"}
+
+
+def test_load_progress_corrupted_file(tmp_path):
+    """损坏的 JSON 文件返回空集合（不抛异常）。"""
+    from tools.repair_trees import load_progress
+
+    path = tmp_path / "progress.json"
+    path.write_text("{invalid json")
+    result = load_progress(path)
+    assert result == set()
+
+
+@pytest.mark.asyncio
+async def test_save_progress_atomic(tmp_path):
+    """save_progress 原子写入，并发调用不丢失更新。"""
+    from tools.repair_trees import save_progress
+
+    path = tmp_path / "progress.json"
+    lock = asyncio.Lock()
+
+    await save_progress(path, lock, "vid_a")
+    await save_progress(path, lock, "vid_b")
+
+    data = json.loads(path.read_text())
+    assert set(data["finished_video_ids"]) == {"vid_a", "vid_b"}
+
+
+@pytest.mark.asyncio
+async def test_save_progress_concurrent(tmp_path):
+    """16 路并发 save_progress 不丢失更新。"""
+    from tools.repair_trees import save_progress
+
+    path = tmp_path / "progress.json"
+    lock = asyncio.Lock()
+
+    tasks = [save_progress(path, lock, f"vid_{i}") for i in range(16)]
+    await asyncio.gather(*tasks)
+
+    data = json.loads(path.read_text())
+    assert len(data["finished_video_ids"]) == 16
+
+
+def test_should_skip_finished(tmp_path):
+    """已在 finished 集合中的视频应跳过。"""
+    from tools.repair_trees import should_skip_video
+
+    finished = {"vid_a", "vid_b"}
+    assert should_skip_video("vid_a", finished, reaggregate_all=False) is True
+    assert should_skip_video("vid_c", finished, reaggregate_all=False) is False
+
+
+def test_should_skip_reaggregate_all_forces_rerun(tmp_path):
+    """--reaggregate-all 标志强制不跳过。"""
+    from tools.repair_trees import should_skip_video
+
+    finished = {"vid_a"}
+    assert should_skip_video("vid_a", finished, reaggregate_all=True) is False
+```
+
+- [ ] **Step 2: 运行测试验证失败**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_repair_progress.py -v`
+预期: FAIL — `ImportError: cannot import name 'load_progress'`
+
+- [ ] **Step 3: 在 repair_trees.py 实现 progress 管理函数**
+
+在 `tools/repair_trees.py` 的 `_build_clients()` 函数之前插入：
+
+```python
+# ---------------------------------------------------------------------------
+# 断点续跑 — progress 文件管理
+# ---------------------------------------------------------------------------
+
+PROGRESS_FILE = "repair_progress.json"
+
+
+def load_progress(path: Path) -> set[str]:
+    """读取 progress 文件，返回已完成视频 ID 集合。
+
+    参数:
+        path: progress JSON 文件路径。
+
+    返回:
+        已完成视频 ID 集合。文件不存在或损坏时返回空集。
+    """
+    if not path.exists():
+        return set()
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+        return set(data.get("finished_video_ids", []))
+    except (json.JSONDecodeError, KeyError, TypeError):
+        logger.warning("progress 文件损坏，忽略: {}", path)
+        return set()
+
+
+async def save_progress(path: Path, lock: asyncio.Lock, vid: str) -> None:
+    """原子追加一个视频 ID 到 progress 文件。
+
+    参数:
+        path: progress JSON 文件路径。
+        lock: asyncio.Lock，防并发读改写丢更新。
+        vid: 要追加的视频 ID。
+    """
+    async with lock:
+        finished = load_progress(path)
+        finished.add(vid)
+        tmp = path.with_suffix(".tmp")
+        tmp.write_text(
+            json.dumps({"finished_video_ids": sorted(finished)}, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+        os.replace(str(tmp), str(path))
+
+
+def should_skip_video(vid: str, finished: set[str], *, reaggregate_all: bool) -> bool:
+    """判断是否跳过该视频。
+
+    参数:
+        vid: 视频 ID。
+        finished: progress 中已完成的视频 ID 集合。
+        reaggregate_all: --reaggregate-all 标志。
+
+    返回:
+        True 表示跳过。
+    """
+    if reaggregate_all:
+        return False
+    return vid in finished
+```
+
+- [ ] **Step 4: 运行测试验证通过**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_repair_progress.py -v`
+预期: 全部 PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add tools/repair_trees.py tests/unit/test_repair_progress.py
+git commit -m "feat(repair): 断点续跑 progress 文件管理
+
+load_progress / save_progress(asyncio.Lock + os.replace 原子写入)
+/ should_skip_video。支持并发安全的读改写和 --reaggregate-all 兜底。"
+```
+
+---
+
+### Task 5: 并发编排 + CLI 参数 + 熔断阈值适配
+
+**Files:**
+- Modify: `tools/repair_trees.py`（`main_async` 并发化 + `parse_args` 新参数 + `_build_clients` 阈值适配）
+- Modify: `.env.example:44`（阈值注释更新）
+- Test: 手动集成测试（`--dry-run` 模式验证并发 + 跳过逻辑）
+
+- [ ] **Step 1: 更新 parse_args 增加 --concurrency 和 --reaggregate-all**
+
+修改 `tools/repair_trees.py` 的 `parse_args` 函数，在 `--dry-run` 之后追加：
+
+```python
+    parser.add_argument(
+        "--concurrency",
+        type=int,
+        default=16,
+        help="并发修复视频数（默认: 16）",
+    )
+    parser.add_argument(
+        "--reaggregate-all",
+        action="store_true",
+        help="强制全量重聚合，忽略 progress 文件",
+    )
+```
+
+- [ ] **Step 2: 更新 _build_clients 接受 concurrency 参数适配熔断阈值**
+
+修改 `_build_clients` 签名为 `_build_clients(concurrency: int = 16)`。
+
+在读取 `breaker_threshold` 后追加：
+
+```python
+    breaker_threshold = int(os.getenv("LLM_CIRCUIT_BREAKER_THRESHOLD", "5"))
+    breaker_threshold = max(breaker_threshold, concurrency * 2)
+```
+
+传入两个 `CircuitBreaker` 实例时使用适配后的 `breaker_threshold`。
+
+- [ ] **Step 3: 重写 main_async 实现并发 + 断点续跑**
+
+将 `main_async` 的串行 for 循环替换为 Semaphore 并发编排：
+
+```python
+async def main_async(args: argparse.Namespace) -> None:
+    """异步主流程：并发修复视频。"""
+    videos_dir = Path(args.videos_dir)
+    srt_dir = Path(args.srt_dir)
+    questions_dir = Path(args.questions_dir)
+    concurrency = args.concurrency
+    reaggregate_all = args.reaggregate_all
+
+    # 扫描所有视频
+    vid_dirs = sorted(
+        d for d in videos_dir.iterdir()
+        if d.is_dir() and (d / "tree.json").exists()
+    )
+    logger.info("发现 {} 个视频", len(vid_dirs))
+
+    # 加载 progress
+    progress_path = PROJECT_ROOT / "logs" / PROGRESS_FILE
+    finished = load_progress(progress_path)
+    if finished:
+        logger.info("已完成 {} 个视频（从 progress 文件加载）", len(finished))
+
+    if args.dry_run:
+        logger.info("=== DRY RUN 模式：仅检测不修复 ===")
+
+    # 构建客户端（dry_run 模式不需要）
+    llm, vlm = (None, None) if args.dry_run else _build_clients(concurrency)
+
+    # 过滤跳过的视频
+    pending = []
+    skipped_count = 0
+    for vid_dir in vid_dirs:
+        vid = vid_dir.name
+        if should_skip_video(vid, finished, reaggregate_all=reaggregate_all):
+            skipped_count += 1
+            continue
+        pending.append(vid_dir)
+
+    if skipped_count:
+        logger.info("跳过 {} 个已完成视频，待处理 {} 个", skipped_count, len(pending))
+
+    # 并发编排
+    sem = asyncio.Semaphore(concurrency)
+    progress_lock = asyncio.Lock()
+    all_stats: list[dict] = []
+    stats_lock = asyncio.Lock()
+    start_time = time.time()
+    completed = 0
+
+    async def _process(vid_dir: Path) -> None:
+        nonlocal completed
+        async with sem:
+            vid = vid_dir.name
+            tree_path = vid_dir / "tree.json"
+            frames_dir = vid_dir
+
+            logger.info("开始修复 {}", vid)
+
+            stats = await _repair_one_video(
+                vid, tree_path, frames_dir, srt_dir, questions_dir,
+                llm, vlm, dry_run=args.dry_run,
+            )
+
+            async with stats_lock:
+                all_stats.append(stats)
+                completed += 1
+
+            # 无 error 且非 dry_run 才记 finished
+            if stats["error"] is None and not args.dry_run:
+                await save_progress(progress_path, progress_lock, vid)
+
+            # 进度日志
+            if completed % 10 == 0:
+                elapsed = time.time() - start_time
+                rate = completed / elapsed * 60
+                logger.info(
+                    "进度: {}/{}, 已用 {:.0f}s, 速率 {:.1f} 视频/分钟",
+                    completed, len(pending), elapsed, rate,
+                )
+
+    tasks = [asyncio.create_task(_process(vd)) for vd in pending]
+    await asyncio.gather(*tasks)
+
+    # 最终汇总
+    elapsed = time.time() - start_time
+    total_issues = sum(s["issues_found"] for s in all_stats)
+    total_repaired = sum(s["l3_repaired"] for s in all_stats)
+    total_injected = sum(s["facts_injected"] for s in all_stats)
+    total_errors = sum(1 for s in all_stats if s["error"])
+
+    logger.info("=" * 60)
+    logger.info("修复完成")
+    logger.info("  视频总数: {}", len(all_stats))
+    logger.info("  跳过数: {}", skipped_count)
+    logger.info("  问题总数: {}", total_issues)
+    logger.info("  L3 修复数: {}", total_repaired)
+    logger.info("  事实注入数: {}", total_injected)
+    logger.info("  失败数: {}", total_errors)
+    logger.info("  总耗时: {:.0f}s", elapsed)
+    logger.info("  并发数: {}", concurrency)
+    logger.info("=" * 60)
+
+    if total_errors > 0:
+        logger.warning("以下视频修复失败:")
+        for s in all_stats:
+            if s["error"]:
+                logger.warning("  {}: {}", s["vid"], s["error"])
+```
+
+- [ ] **Step 4: 更新 .env.example 熔断阈值注释**
+
+将 `.env.example` 中：
+
+```
+LLM_CIRCUIT_BREAKER_THRESHOLD=5
+```
+
+改为：
+
+```
+LLM_CIRCUIT_BREAKER_THRESHOLD=5  # 实际阈值 = max(此值, concurrency*2)
+```
+
+- [ ] **Step 5: dry-run 模式冒烟测试**
+
+运行: `conda activate Video-Tree-TRM && python tools/repair_trees.py --dry-run --concurrency 4 2>&1 | head -30`
+预期: 看到"发现 N 个视频"、"跳过 M 个已完成视频"、并发日志，无报错
+
+- [ ] **Step 6: 提交**
+
+```bash
+git add tools/repair_trees.py .env.example
+git commit -m "feat(repair): asyncio.Semaphore 并发 + 断点续跑 + CLI 参数
+
+--concurrency 默认 16，--reaggregate-all 强制全量重聚合。
+Semaphore 限视频并发数，视频内四步串行。progress 文件
+asyncio.Lock + os.replace 原子写入。熔断阈值 max(.env, concurrency*2)。"
+```
+
+---
+
+### Task 6: Lint + 全量测试 + 最终验证
+
+**Files:**
+- 无新文件
+
+- [ ] **Step 1: ruff 格式化与检查**
+
+运行:
+```bash
+conda activate Video-Tree-TRM && ruff format adapters/telemetry.py adapters/llm.py app/tree/repair/detector.py tools/repair_trees.py
+conda activate Video-Tree-TRM && ruff check adapters/telemetry.py adapters/llm.py app/tree/repair/detector.py tools/repair_trees.py --fix
+```
+预期: 无错误
+
+- [ ] **Step 2: 全量测试**
+
+运行: `conda activate Video-Tree-TRM && pytest tests/unit/ -v --tb=short`
+预期: 全部 PASS
+
+- [ ] **Step 3: 提交 lint 修正（如有）**
+
+```bash
+git add -u
+git commit -m "style: ruff format adapters + detector + repair_trees"
+```
+
+- [ ] **Step 4: 确认改动文件范围**
+
+运行: `git diff --stat main`
+预期改动文件：
+
+| 文件 | 性质 |
+|------|------|
+| `adapters/telemetry.py` | 修改 |
+| `adapters/llm.py` | 修改 |
+| `app/tree/repair/detector.py` | 修改 |
+| `tools/repair_trees.py` | 修改 |
+| `.env.example` | 修改 |
+| `tests/unit/test_telemetry.py` | 修改 |
+| `tests/unit/test_governed_llm.py` | 修改 |
+| `tests/unit/test_repair_detector.py` | 修改 |
+| `tests/unit/test_repair_progress.py` | 新建 |