test: 多线程字幕顺序验证（全量真实数据夹具）+ 文档更新

- 新增 TestSubtitleOcrOrderThreading：模拟真实 vlm-ocr API 返回（不走真实
  OCR），全量 14236 帧真实数据 + 种子化可变延迟（约 8% 慢帧，模拟真实 OCR
  延迟波动），验证 4/16 线程产物与单线程确认结果逐字节一致、并发真实发生、
  完成顺序乱序但输出保序、时间轴严格递增、每条字幕与起始时刻帧文本对齐
- 全量夹具：frames_manifest_full.json + ocr_frames_full.json（真实任务
  run_ac7f480a3ccb 全部 14236 帧），常驻测试不跳过
- AGENTS.md：更新 llm-filter 参数说明与测试资产清单

AGENTS.md

@@ -54,9 +54,9 @@ vrsub/
 | `faster-whisper` | `audio_uri`（16kHz 单声道） | `srt_uri` | 参数：`language`、`task`、`model_path`、`device`、`compute_type`、`beam_size`、`vad_filter`（默认开）、`condition_on_previous_text`、`chunk_seconds` |
 | `llm-translate` | `srt_uri` | `cn_srt_uri` | 参数：`target_language`、`model` |
 | `vlm-ocr` | `image_uri` | `text`、`text_uri` | 直接调本地 Ollama 多模态模型（glm-ocr）的 `/api/chat` 做视频帧 OCR（流式 + 5s 上限），参数：`model`、`ollama_host`、`prompt`、`timeout_seconds`、`keep_alive`、`num_predict`、`temperature`、`repeat_penalty` |
-| `frame-extract` | `video_uri` | `frames_manifest`、`frame_count` | 按**帧间隔**抽帧（解析 fps → step=round(间隔秒×fps)，ffmpeg select 按帧号精确取帧，帧时间=帧号/fps 无累计偏差）并 crop 裁切字幕区域，参数：`interval_seconds`（默认 0.5）、`crop`（[x,y,w,h] 0~1） |
+| `frame-extract` | `video_uri` | `frames_manifest`、`frame_count` | 按**帧间隔**抽帧（解析 fps → step=round(间隔秒×fps)，ffmpeg select 按帧号精确取帧，帧时间=帧号/fps 无累计偏差）并 crop 裁切字幕区域，参数：`interval_seconds`（默认 0.5）、`crop`（[x,y,w,h] 0~1）。**帧文件必须按帧号数值排序读取**（`_sorted_frame_files`）：ffmpeg `%04d` 编号超过 9999 帧后扩为 5 位，字典序 `sorted()` 会把 5 位编号排在 4 位之前导致时间与图像错位（真实发生于 run_339ec7ee437f 的 14236 帧任务，回归测试见 `test_frame_files_read_order_matches_frame_number`） |
 | `subtitle-ocr` | `frames_manifest` | `srt_uri`、`count` | 自适应线程池并发逐帧调 vlm-ocr → 垃圾过滤（无文字帧）→ 相同字幕合并（记录最后可见帧）→ 组装 SRT，消失时间=最后可见帧+采样间隔（间隔从帧清单推导），参数：`min_chars`、`min_alnum_ratio`、`garbage_tokens`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold` |
-| `llm-filter` | `srt_uri` | `srt_uri`、`kept`、`removed` | LLM 过滤无意义字幕（自适应线程池并发判断）：每条连同前后各 `context_size`（默认 10）条纯文本（不含时间戳）分批给 LLM，仅判断目标字幕是否多余/无意义，判定删除则该条连同时间戳移除并重新编号，参数：`context_size`、`model`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold` |
+| `llm-filter` | `srt_uri` | `srt_uri`、`kept`、`removed` | 两级过滤：①**规则层**（不调 LLM）直接删横线装饰/HTML 水印 token/URL/邮箱/单双 ASCII 字符；②**LLM 五类分类**（garbage/overlay/noise 删，repeat/dialogue 留，未识别回退保留）每条连同前后各 `context_size`（默认 10）条纯文本分批判断，**按文本去重**（忽略空白/大小写，相同文本只调一次 LLM，上下文取首次出现）保证判定一致并省调用，**长文本保护**（≥`min_keep_len` 默认 12 时 noise 不构成删除依据）。参数：`context_size`、`min_keep_len`、`overlay_tokens`（JSON 数组）、`dedupe`（默认开）、`model`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold`。回归数据：testdata/ocr_srt_run_ac7f480a3ccb.srt（真实任务 1666 条 OCR 输出） |
 | `srt-to-dual-eye-ass` | `cn_srt_uri` | `ass_uri` | 参数：`resolution`，如 `3840x1920` |
 
 ### 模型权重解析（本地优先）
@@ -269,7 +269,12 @@ http://127.0.0.1:8000/docs      API 文档
   缺失时测试跳过而非现场生成。大体积视频素材放 `data/testdata/`（gitignored）。OCR 相关资产：
 `ocr_text.png`（有文字）、`ocr_notext.png`（无文字帧）、`subtitle_10s.mp4`
 （烧录 SUB 001@1-4s / SUB 002@6-9s 的 10s 测试视频）、`test_real_hav_sub.png`
-（真实视频字幕截图，VLM 集成测试期望识别出"还有没有什么困扰 或者奇怪的地方吗"）。
+（真实视频字幕截图，VLM 集成测试期望识别出"还有没有什么困扰 或者奇怪的地方吗"）、
+`ocr_srt_run_ac7f480a3ccb.srt`（真实任务 1666 条 OCR 输出，llm-filter 回归）、
+`frames_manifest_full.json` + `ocr_frames_full.json`（真实任务 run_ac7f480a3ccb
+**全部 14236 帧**的帧清单与逐帧 OCR 文本，多线程顺序测试常驻夹具；配合
+`ocr_srt_run_ac7f480a3ccb.srt` 作为单线程确认基线，见
+`tests/test_subtitle_ocr_order_threading.py`）。
 - **开发流程强制 TDD（红-绿-重构）**：任何新功能/修复必须先写失败测试（红），
   再实现最小代码让其通过（绿），最后重构保持整洁；不允许先写实现后补测试。
 - 测试运行：`uv run pytest`；全部测试位于 `tests/`。

nodes/frame_extract.py

@@ -111,6 +111,24 @@ def _parse_progress_line(line: str) -> int | None:
     except ValueError:
         return None
 
+
+def _sorted_frame_files(frames_dir: Path) -> list[Path]:
+    """按文件名中的帧号数值排序返回帧文件列表（自然排序，非字典序）。
+
+    关键点：ffmpeg 的 %04d 编号在超过 9999 帧后会自动扩为 5 位
+    （frame_10000.png 等），此时 sorted() 默认的字典序会把 5 位编号排在
+    4 位编号之前（如 "frame_10009" < "frame_1009"），导致帧号回退、
+    manifest 时间与图像错位（曾真实发生于 run_339ec7ee437f 的 14236 帧
+    任务，全片后半段时间轴全部错乱）。必须解析出帧号按数值排序，
+    才能保证"第 k 个文件 = 第 k 个选中帧 = 时间 index*step/fps"成立。
+    """
+    def frame_number(path: Path) -> int:
+        # 文件名形如 frame_0001.png，取下划线后的数字部分。
+        return int(path.stem.split("_", 1)[1])
+
+    return sorted(frames_dir.glob("frame_*.png"), key=frame_number)
+
+
 def invoke(request: InvokeRequest) -> InvokeResponse:
     """按帧间隔抽取并裁切视频帧，输出 frames.json 清单。"""
     video_uri = request.inputs.get("video_uri")
@@ -203,7 +221,7 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
         return InvokeResponse(status="failed", error=stderr[-500:] or "ffmpeg failed")
 
     # 第 k 个输出文件对应原始帧号 k×step，时间 = 帧号 / fps（帧精确，无累计偏差）。
-    files = sorted(frames_dir.glob("frame_*.png"))
+    files = _sorted_frame_files(frames_dir)
     manifest = [
         {"time": round((index * step) / fps, 3), "image_uri": str(path)}
         for index, path in enumerate(files)

nodes/llm_filter.py

@@ -1,9 +1,30 @@
 """LLM 字幕过滤节点。
 
-对 OCR 识别出的 SRT 字幕做二次过滤：为避免字幕上下文过长，把每条字幕连同其
-前后各 context_size 条字幕（纯文本，**不含时间戳**）分批提供给 LLM，模型仅判断
-目标字幕是否属于多余、无意义的字符（如重复、残缺、无实际语义的杂项）；
-判定无意义则删除该条（连同其时间戳），其余字幕保持原样并重新编号输出。
+对 OCR 识别出的 SRT 字幕做二次过滤，两级判断：
+
+1. **确定性规则层**（不调用 LLM）：横线装饰、HTML/水印 token、URL/邮箱、
+   单双 ASCII 字符等 OCR 噪声直接删除——这些模式是稳定可判的，走规则
+   既省 token 又保证结果确定（真实数据中占删除量的 65%+）。
+
+2. **LLM 五类分类层**：把目标字幕连同前后各 context_size 条纯文本分批
+   提供给 LLM，模型输出五个类别之一：
+   - garbage：垃圾字符（乱码、残缺、装饰性符号）→ 删除
+   - overlay：水印/网页/播放器等覆盖层文本 → 删除
+   - noise：与上下文无关、无实际语义的杂项 → 删除
+   - repeat：内容性重复（语气词、呻吟、重复感叹，属于内容本身）→ 保留
+   - dialogue：正常对话 → 保留
+   未识别输出一律回退 dialogue（宁滥勿缺，避免误删真实对话）。
+
+3. **文本去重**：相同文本（忽略全部空白与大小写差异）只调一次 LLM，
+   上下文取首次出现位置，结果缓存复用——修复"同一句字幕 5 留 6 删"的
+   判定不一致，同时把长视频的 LLM 调用量降到唯一文本数。
+
+4. **长文本保护**：长度 ≥ min_keep_len 的文本，仅 garbage/overlay 两个
+   明确垃圾类别可删，noise 不构成删除依据，防止长对话被误删。
+
+参考真实任务 run_ac7f480a3ccb（2026-08，OCR 1666 条）：旧实现把 394 条
+真实对话当噪声删掉（占删除 35%）、同文本判定不一致；新实现按上述机制
+回归测试已固化在 tests/test_llm_filter.py。
 """
 
 from __future__ import annotations
@@ -13,6 +34,7 @@ import os
 import re
 import urllib.request
 from pathlib import Path
+
 from nodes.adaptive_pool import AdaptiveThreadPool
 from wov_app.logging import get_logger
 from wov_sdk.models import InvokeRequest, InvokeResponse
@@ -31,6 +53,37 @@ TARGET_MARK = "【目标】"
 # 默认上下文窗口：目标字幕前后各取 10 条。
 DEFAULT_CONTEXT_SIZE = 10
 
+# LLM 输出类别：garbage/overlay/noise 删除；repeat/dialogue 保留。
+CATEGORY_GARBAGE = "garbage"
+CATEGORY_OVERLAY = "overlay"
+CATEGORY_NOISE = "noise"
+CATEGORY_REPEAT = "repeat"
+CATEGORY_DIALOGUE = "dialogue"
+_ALL_CATEGORIES = (
+    CATEGORY_GARBAGE,
+    CATEGORY_OVERLAY,
+    CATEGORY_NOISE,
+    CATEGORY_REPEAT,
+    CATEGORY_DIALOGUE,
+)
+DELETE_CATEGORIES = {CATEGORY_GARBAGE, CATEGORY_OVERLAY, CATEGORY_NOISE}
+
+# 规则层正则：横线装饰（含全角/半角横线、下划线、中点、句点等符号组合）。
+_DASH_RE = re.compile(r"^[\s\-—_~=•・。．.、]+$")
+# 规则层正则：URL / 邮箱。
+_URL_OR_MAIL_RE = re.compile(r"^(https?://|www\.)\S+$|^[\w.+-]+@[\w.-]+\.\w+$")
+# 默认水印/覆盖层 token（casefold 后比较，可经 overlay_tokens 参数覆盖）。
+DEFAULT_OVERLAY_TOKENS = frozenset(
+    {"html", "background", "___", "cleaning", "buffering", "loading"}
+)
+# 长文本保护阈值：≥ 该长度的文本，noise 类别不构成删除依据。
+DEFAULT_MIN_KEEP_LEN = 12
+
+
+def _has_cjk(text: str) -> bool:
+    """是否含 CJK 汉字：单字"嗯/好"等可能是内容，规则层不直接删。"""
+    return any("\u4e00" <= ch <= "\u9fff" for ch in text)
+
 
 def parse_srt(text: str) -> list[dict]:
     """解析 SRT 文本为条目列表：[{"start", "end", "text"}]。"""
@@ -55,13 +108,58 @@ def serialize_srt(entries: list[dict]) -> str:
     return "\n\n".join(blocks) + "\n"
 
 
-def _judge_target(
+def _rule_verdict(text: str, overlay_tokens: set[str]) -> bool | None:
+    """确定性规则层：返回 True（删除）/ None（交给 LLM 多维判断）。
+
+    规则覆盖 OCR 噪声的稳定模式：空文本、纯横线装饰、URL/邮箱、水印 token
+    （HTML/background 等）、单双 ASCII 字符。含 CJK 的短文本不算垃圾，
+    因为"嗯/好"等可能是内容；其余情况返回 None 交由 LLM 分类。
+    """
+    t = text.strip()
+    if not t:
+        return True
+    if _DASH_RE.match(t):
+        return True
+    if _URL_OR_MAIL_RE.match(t):
+        return True
+    if t.casefold() in overlay_tokens:
+        return True
+    if len(t) <= 2 and not _has_cjk(t):
+        return True
+    return None
+
+
+def _dedup_key(text: str) -> str:
+    """文本去重键：去掉全部空白并统一大小写。
+
+    OCR 同一句字幕常带/不带空格（"可没法胜任" vs "可 没法胜任"），
+    视为同一文本以保证判定一致；也用于把 LLM 调用量降到唯一文本数。
+    """
+    return "".join(text.split()).casefold()
+
+
+def _should_delete(category: str, text: str, min_keep_len: int) -> bool:
+    """按 LLM 类别与长文本保护决定是否删除。
+
+    repeat/dialogue 一律保留；garbage/overlay 一律删除（含长文本——
+    这两个是明确的垃圾信号）；noise 对短文本删除，但 ≥min_keep_len 的
+    长文本不删（noise 太模糊，不足以推翻一句完整台词）。
+    """
+    if category not in DELETE_CATEGORIES:
+        return False
+    if category == CATEGORY_NOISE and len(text) >= min_keep_len:
+        return False
+    return True
+
+
+def _judge_category(
     entries: list[dict], index: int, context_size: int, params: dict
-) -> bool:
-    """调用 LLM 判断目标字幕是否多余/无意义；返回 True 表示应删除。
+) -> str:
+    """调用 LLM 把目标字幕归入五类之一，返回类别词（未识别回退 dialogue）。
 
     请求体只含目标字幕及其前后各 context_size 条字幕的纯文本（无时间戳），
-    目标字幕用 TARGET_MARK 标记；模型只需回答"保留"或"删除"。
+    目标字幕用 TARGET_MARK 标记；模型只输出一个类别英文单词。
+    输出无法识别（空/乱码/旧式"保留"）时回退 dialogue，宁滥勿缺。
     """
     start = max(0, index - context_size)
     end = min(len(entries), index + context_size + 1)
@@ -81,9 +179,14 @@ def _judge_target(
     model = str(params.get("model") or os.getenv("LLM_MODEL", "Qwen/Qwen3.6-35B-A3B"))
     system_prompt = (
         "你是字幕质量过滤器。用户会提供一段字幕序列（纯文本，不含时间戳），"
-        f"其中用{TARGET_MARK}标记的字幕是需要判断的目标。请判断该字幕是否属于"
-        "多余、无意义的字符（如重复、残缺、无实际语义的杂项）。"
-        "只回答两个字：保留 或 删除，不要输出其他内容。"
+        f"其中用{TARGET_MARK}标记的字幕是需要判断的目标。"
+        "请把目标字幕归入以下五个类别之一：\n"
+        "garbage：垃圾字符（乱码、残缺、装饰性符号、横线）\n"
+        "overlay：水印、网页/播放器/字幕组等覆盖层文本，不是视频对白\n"
+        "noise：与上下文无关、无实际语义的杂项\n"
+        "repeat：内容性重复（如语气词、呻吟、重复的感叹或对话），属于内容本身\n"
+        "dialogue：正常对话\n"
+        "只输出一个类别英文单词，不要输出其他内容。"
     )
     body = {
         "model": model,
@@ -94,7 +197,7 @@ def _judge_target(
         # 关闭推理模式：Qwen3 等模型默认会把思考过程写入 reasoning_content，
         # 导致 content 为空或包含多余内容。
         "enable_thinking": False,
-        # 只输出"保留/删除"，输出上限给得很小即可。
+        # 类别词很短，输出上限给得很小即可。
         "max_tokens": 16,
     }
     headers = {"Content-Type": "application/json"}
@@ -109,13 +212,23 @@ def _judge_target(
     )
     with urllib.request.urlopen(request, timeout=request_timeout) as response:
         payload = json.loads(response.read().decode("utf-8"))
-    content = str(payload["choices"][0]["message"]["content"])
-    # 模型回答含"删除"即视为该条无意义；其余情况（保留/异常）一律保留，宁多勿删。
-    return "删除" in content
+    content = str(payload["choices"][0]["message"]["content"]).strip().lower()
+    # 精确匹配五个类别词（兼容"garbagexxx"这类多余输出）。
+    for category in _ALL_CATEGORIES:
+        if content == category or content.startswith(category):
+            return category
+    # 旧式"删除/保留"回答兼容：含"删除"视为垃圾，其余一律保留。
+    if "删除" in content:
+        return CATEGORY_GARBAGE
+    return CATEGORY_DIALOGUE
 
 
 def invoke(request: InvokeRequest) -> InvokeResponse:
-    """过滤 SRT 中多余/无意义的字幕，产物为 filtered.srt。"""
+    """过滤 SRT 中多余/无意义的字幕，产物为 filtered.srt。
+
+    流程：规则层（确定性删除）→ LLM 层（去重后按唯一文本多维分类）→
+    长文本保护 → 保留条重新编号输出。
+    """
     srt_uri = request.inputs.get("srt_uri")
     if not srt_uri:
         return InvokeResponse(status="failed", error="srt_uri is required")
@@ -125,38 +238,78 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
 
     entries = parse_srt(srt_path.read_text(encoding="utf-8"))
     context_size = int(request.params.get("context_size", DEFAULT_CONTEXT_SIZE))
-    # 单条判断的工作函数：返回 True 表示该条应删除。
-    def judge_one(index) -> bool:
-        return _judge_target(entries, index, context_size, request.params)
+    min_keep_len = int(request.params.get("min_keep_len", DEFAULT_MIN_KEEP_LEN))
+    # 水印 token 可经参数覆盖（JSON 数组字符串或列表），默认内置常见覆盖层词。
+    raw_tokens = request.params.get("overlay_tokens")
+    if isinstance(raw_tokens, str) and raw_tokens.strip():
+        raw_tokens = json.loads(raw_tokens)
+    overlay_tokens = {
+        str(t).casefold() for t in (raw_tokens or DEFAULT_OVERLAY_TOKENS)
+    }
+    # 去重开关：默认开；关掉时每个条目独立调用 LLM（不省调用，判定各自独立）。
+    dedupe = str(request.params.get("dedupe", "1")) not in ("0", "false", "False")
 
-    # 进度日志：打印已判定条数、总数与平均处理速度（条/s）。
-    def log_progress(done: int, total: int, rate: float) -> None:
-        logger.info("字幕判定进度 %d/%d 条 (%.1f 条/s)", done, total, rate)
+    # 阶段 1：确定性规则层（不调 LLM）。
+    rule_verdicts = [_rule_verdict(entry["text"], overlay_tokens) for entry in entries]
+    llm_needed = [i for i, verdict in enumerate(rule_verdicts) if verdict is None]
 
-    # 自适应并发调用 LLM：10s 窗口内平均响应 < 0.3s 则加 1 线程（上限
-    # pool_max_workers），> pool_slow_threshold 则减 1 线程（下限 1），
-    # 按实测负载弹性伸缩，避免盲目并发压垮 LLM 接口。
-    pool = AdaptiveThreadPool(
-        worker=judge_one,
-        on_progress=log_progress,
-        min_workers=int(request.params.get("pool_min_workers", 1)),
-        max_workers=int(request.params.get("pool_max_workers", 16)),
-        window_seconds=float(request.params.get("pool_window_seconds", 10.0)),
-        fast_threshold=float(request.params.get("pool_fast_threshold", 0.3)),
-        slow_threshold=float(request.params.get("pool_slow_threshold", 1.0)),
-    )
-    verdicts = pool.map(range(len(entries)))
+    # 阶段 2：LLM 分类层（去重：相同文本只判一次，上下文取首次出现）。
+    cat_by_index: dict[int, str] = {}
+    if llm_needed:
+        if dedupe:
+            first_of_key: dict[str, int] = {}
+            pool_indices: list[int] = []
+            for i in llm_needed:
+                key = _dedup_key(entries[i]["text"])
+                if key not in first_of_key:
+                    first_of_key[key] = i
+                    pool_indices.append(i)
+        else:
+            pool_indices = llm_needed
 
+        # 单条判断的工作函数：返回类别词。
+        def judge_one(index: int) -> str:
+            return _judge_category(entries, index, context_size, request.params)
+
+        # 进度日志：打印已判定条数、总数与平均处理速度（条/s）。
+        def log_progress(done: int, total: int, rate: float) -> None:
+            logger.info("字幕判定进度 %d/%d 条 (%.1f 条/s)", done, total, rate)
+
+        # 自适应并发调用 LLM：按实测负载弹性伸缩，避免压垮 LLM 接口。
+        pool = AdaptiveThreadPool(
+            worker=judge_one,
+            on_progress=log_progress,
+            min_workers=int(request.params.get("pool_min_workers", 1)),
+            max_workers=int(request.params.get("pool_max_workers", 16)),
+            window_seconds=float(request.params.get("pool_window_seconds", 10.0)),
+            fast_threshold=float(request.params.get("pool_fast_threshold", 0.3)),
+            slow_threshold=float(request.params.get("pool_slow_threshold", 1.0)),
+        )
+        categories = pool.map(pool_indices)
+
+        for index, category in zip(pool_indices, categories):
+            # 并行下 LLM 异常被线程池隔离为异常结果：任一条失败即整体失败，
+            # 避免静默输出未过滤结果。
+            if isinstance(category, Exception):
+                return InvokeResponse(status="failed", error=str(category))
+            cat_by_index[index] = category
+        if dedupe:
+            # 去重填充：与首次出现同键的条目复用同一类别，保证判定一致。
+            for i in llm_needed:
+                if i not in cat_by_index:
+                    cat_by_index[i] = cat_by_index[first_of_key[_dedup_key(entries[i]["text"])]]
+
+    # 阶段 3：合并规则与 LLM 判定，应用长文本保护并输出。
     kept: list[dict] = []
     removed = 0
-    for index, (entry, verdict) in enumerate(zip(entries, verdicts)):
-        # 并行下 LLM 异常被线程池隔离为异常结果：任一条失败即整体失败，
-        # 避免静默输出未过滤结果。
-        if isinstance(verdict, Exception):
-            return InvokeResponse(status="failed", error=str(verdict))
-        if verdict:
+    for i, (entry, rule_verdict) in enumerate(zip(entries, rule_verdicts)):
+        # 规则层命中即删（True）；未命中则按 LLM 类别与长文本保护判定。
+        if rule_verdict is True:
             removed += 1
-            logger.info("删除无意义字幕 %d: %r", index + 1, entry["text"][:40])
+            continue
+        if _should_delete(cat_by_index[i], entry["text"], min_keep_len):
+            removed += 1
+            logger.info("删除无意义字幕 %d: %r", i + 1, entry["text"][:40])
         else:
             kept.append(entry)
 

nodes/subtitle_ocr.py

@@ -63,6 +63,28 @@ def _assemble_srt(
     return lines
 
 
+def _merge_kept(
+    manifest: list[dict], texts: list[str]
+) -> list[tuple[float, float, str]]:
+    """按 manifest 时间轴把逐帧 OCR 文本合并为字幕条目。
+
+    kept 元素为 (起始帧时间, 最后可见帧时间, 文本)：空文本（无文字帧）跳过；
+    连续帧相同字幕合并为一条（字幕停留多帧属正常现象），仅更新最后可见帧
+    时间，起始时间保持首次出现。要求 texts 与 manifest 按帧顺序一一对应
+    （调用方保证），重组装旧数据时也复用此逻辑保证行为一致。
+    """
+    kept: list[tuple[float, float, str]] = []
+    for index, text in enumerate(texts):
+        if not text:
+            continue
+        time = float(manifest[index]["time"])
+        if kept and kept[-1][2] == text:
+            kept[-1] = (kept[-1][0], time, text)
+            continue
+        kept.append((time, time, text))
+    return kept
+
+
 def invoke(request: InvokeRequest) -> InvokeResponse:
     """逐帧 OCR 并汇总字幕，产物为 subtitle.srt。"""
     manifest_uri = request.inputs.get("frames_manifest")
@@ -138,19 +160,8 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
     )
     texts = pool.map(list(enumerate(manifest)))
 
-    # kept 元素为 (起始帧时间, 最后可见帧时间, 文本)；按帧顺序合并连续相同字幕。
-    kept: list[tuple[float, float, str]] = []
-    for index, text in enumerate(texts):
-        if not text:
-            continue
-        time = float(manifest[index]["time"])
-        # 连续帧相同字幕合并为一条（字幕停留多帧属正常现象）：
-        # 仅更新最后可见帧时间，起始时间保持首次出现。
-        if kept and kept[-1][2] == text:
-            kept[-1] = (kept[-1][0], time, text)
-            continue
-        kept.append((time, time, text))
-
+    # 按帧顺序合并连续相同字幕（与重组装旧数据共用 _merge_kept）。
+    kept = _merge_kept(manifest, texts)
     output_dir = Path(request.output_dir)
     output_dir.mkdir(parents=True, exist_ok=True)
     output_path = output_dir / "subtitle.srt"

scripts/regenerate_filter_ac7f480a3ccb.py

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+"""用修复后的 llm_filter 对 run_ac7f480a3ccb 的真实 OCR 输出重新过滤。
+
+- 加载 .env（LLM Key/模型配置）
+- 复用节点 invoke 真实代码路径（与调度器执行完全一致）
+- 池参数按该 run 的 workflow v4 配置（单线程 pool 1/1）
+- 产物写入 run 的 filter 步骤目录（filtered.srt）
+"""
+import sys
+import time
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv(".env")
+
+from wov_sdk.models import InvokeRequest
+from nodes.llm_filter import invoke
+
+RUN = Path("data/storage/runs/run_ac7f480a3ccb")
+SRT_URI = RUN / "steps/ocr/subtitle.srt"
+OUT_DIR = RUN / "steps/filter"
+
+print(f"输入: {SRT_URI}", flush=True)
+print(f"输出目录: {OUT_DIR}", flush=True)
+
+started = time.monotonic()
+response = invoke(
+    InvokeRequest(
+        run_id="run_ac7f480a3ccb",
+        node_instance_id="",
+        inputs={"srt_uri": str(SRT_URI)},
+        # 与 workflow v4（单线程）保持一致。
+        params={"pool_min_workers": 1, "pool_max_workers": 1},
+        output_dir=str(OUT_DIR),
+    )
+)
+elapsed = time.monotonic() - started
+print(f"耗时 {elapsed:.0f}s", flush=True)
+if response.status != "completed":
+    print(f"FAILED: {response.error}", flush=True)
+    sys.exit(1)
+
+print(f"kept={response.outputs['kept']} removed={response.outputs['removed']}", flush=True)
+print(f"产物: {response.outputs['srt_uri']}", flush=True)

tests/test_integration_reassemble_ocr.py

@@ -0,0 +1,97 @@
+"""真实任务 OCR 数据重组装集成测试。
+
+run_339ec7ee437f 是 2026-08 真实跑过的 ocr-subtitle 任务（14236 帧 / 2 小时
+视频）。该任务产出时期存在 frame-extract 帧文件排序 bug：ffmpeg 的 %04d 编号
+超过 9999 帧后扩为 5 位，sorted() 字典序把 5 位编号排在 4 位之前，导致
+frames.json 中 time 与图像错位（13236/14236 条位置↔帧号错位），最终 SRT
+后半段时间轴全部错乱。
+
+本测试读取该任务**已落盘的逐帧 OCR 文本**（ocr_frames/<位置>/ocr.txt，由
+vlm-ocr 节点写入的清洗后文本），用 image_uri 解析真实帧号重建正确时间轴，
+复用生产合并/组装逻辑生成 SRT，并断言"每条字幕的起始时刻 = 其文本来源帧的
+真实时间"这一核心正确性。
+
+数据位于 gitignored 的 data/storage，缺失时跳过（与
+test_integration_subtitle_ocr 的真实资产约定一致）。
+"""
+
+import json
+import re
+from pathlib import Path
+
+import pytest
+
+from nodes.subtitle_ocr import _assemble_srt
+from nodes.subtitle_ocr import _merge_kept
+from nodes.subtitle_ocr import _sampling_interval
+
+# 单体根目录：tests/ 的上一级。
+WORKSPACE = Path(__file__).resolve().parent.parent
+RUN_DIR = WORKSPACE / "data/storage/runs/run_339ec7ee437f"
+MANIFEST = RUN_DIR / "steps/extract/frames.json"
+OCR_DIR = RUN_DIR / "steps/ocr/ocr_frames"
+
+# 源视频时长（ffprobe: 02:00:24.80）。
+VIDEO_DURATION = 7224.8
+# 该视频烧录过的字幕行，被 test_real_hav_sub.png 集成测试确认过识别结果。
+KNOWN_LINE = "还有没有什么困扰"
+
+
+def _frame_number(uri: str) -> int:
+    """从 image_uri 文件名解析真实帧号（frame_0001.png -> 1）。"""
+    return int(re.match(r".*frame_(\d+)\.png", Path(uri).name).group(1))
+
+
+@pytest.mark.integration
+def test_reassemble_real_run_ocr_data(tmp_path) -> None:
+    """真实任务逐帧 OCR 数据按真实帧时间轴重组装为正确 SRT。"""
+    if not (MANIFEST.is_file() and OCR_DIR.is_dir()):
+        pytest.skip("缺少真实任务 run_339ec7ee437f 数据，跳过集成测试")
+
+    manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
+    total = len(manifest)
+    # 等间隔采样：step/fps 从旧 manifest 相邻 time 差恢复（本任务 0.507s/帧）。
+    dt = manifest[1]["time"] - manifest[0]["time"]
+    assert dt > 0
+
+    # ① 读取逐帧 OCR 文本：位置 i -> ocr_frames/{i:04d}/ocr.txt（真实数据）。
+    texts_by_pos: list[str] = []
+    for i in range(total):
+        p = OCR_DIR / f"{i:04d}" / "ocr.txt"
+        if not p.is_file():
+            pytest.fail(f"缺少逐帧 OCR 数据: {p}")
+        texts_by_pos.append(p.read_text(encoding="utf-8").strip())
+
+    # ② 旧 manifest 位置 -> 真实帧号：image_uri 才是实际被 OCR 的图像，
+    # 旧 time 字段按位置推导已错位，不可用。
+    frame_by_pos = [_frame_number(item["image_uri"]) for item in manifest]
+    assert sorted(frame_by_pos) == list(range(1, total + 1))  # 一一对应，无缺无重
+
+    # ③ 按真实帧号重建正确时间轴：第 k 帧（1-based）时间 = (k-1)*dt。
+    ocr_by_frame = {n: texts_by_pos[i] for i, n in enumerate(frame_by_pos)}
+    manifest_corrected = [{"time": round((k - 1) * dt, 3)} for k in range(1, total + 1)]
+    texts_corrected = [ocr_by_frame[k] for k in range(1, total + 1)]
+
+    # ④ 复用生产合并 + 组装逻辑（与 subtitle_ocr.invoke 完全同一路径）。
+    kept = _merge_kept(manifest_corrected, texts_corrected)
+    interval = _sampling_interval(manifest_corrected, dt)
+    srt = "\n".join(_assemble_srt(kept, interval)) + "\n"
+    out = tmp_path / "subtitle.srt"
+    out.write_text(srt, encoding="utf-8")
+
+    # ⑤ 核心正确性：每条字幕起始时刻对应的帧，其 OCR 文本必须就是本条字幕
+    # 文本（修复前旧 SRT 此处大面积不一致：文本来自其他时刻的帧）。
+    for start, _end, text in kept:
+        frame_no = int(round(start / dt)) + 1
+        assert ocr_by_frame[frame_no] == text, f"帧 {frame_no} 时间错位: {text!r}"
+
+    # ⑥ 时间轴严格递增且不超出片长（含尾部一个采样间隔的消失余量）。
+    assert all(
+        s1 < s2 for (s1, _1, _), (s2, _2, _) in zip(kept, kept[1:])
+    )
+    assert kept[0][0] >= 0.0
+    assert kept[-1][1] + interval <= VIDEO_DURATION + interval + 1e-6
+
+    # ⑦ 已知烧录字幕行必须出现，且字幕条数合理（空帧被跳过、连续相同字幕合并）。
+    assert KNOWN_LINE in srt
+    assert len(kept) > 100

tests/test_llm_filter.py

@@ -1,14 +1,30 @@
 """LLM 字幕过滤节点测试。
 
-覆盖 SRT 解析/序列化、±N 上下文窗口组装（纯文本无时间戳、目标标记）、
-LLM 调用（按 I/O 边界 mock urlopen）与删除判定、invoke 全链路与异常路径。
+覆盖：
+- SRT 解析/序列化（多行、末条无空行、重新编号）；
+- 确定性规则层（横线装饰、HTML/水印 token、URL、单双 ASCII 字符直接删，不调 LLM）；
+- LLM 五类分类判断（garbage/overlay/noise 删，repeat/dialogue 留，未识别回退保留）；
+- 文本去重（相同文本只调一次 LLM，上下文取首次出现，判定结果一致）；
+- 长文本保护（≥min_keep_len 时 noise 类不删，需明确垃圾类别）；
+- 真实任务 run_ac7f480a3ccb 留存数据回归（OCR 1666 条：垃圾删除、对话保留、时间轴单调、去重后 LLM 调用数）；
+- invoke 全链路与异常路径。
 """
 
 import json
 import urllib.error
 from pathlib import Path
 
-from nodes.llm_filter import invoke, parse_srt, serialize_srt
+import pytest
+
+from nodes.llm_filter import (
+    _dedup_key,
+    _judge_category,
+    _rule_verdict,
+    _should_delete,
+    invoke,
+    parse_srt,
+    serialize_srt,
+)
 from wov_sdk.models import InvokeRequest
 
 # 4 条字幕的 SRT：第 3 条为"答："开头的无意义杂项，模拟 OCR 噪声。
@@ -19,6 +35,10 @@ _SRT = (
     "4\n00:00:13,000 --> 00:00:16,000\n第二句正常字幕\n"
 )
 
+# 真实任务 run_ac7f480a3ccb 的 OCR 输出（1666 条，2026-08 单线程 v4 跑完整 2 小时视频）。
+WORKSPACE = Path(__file__).resolve().parent.parent
+REAL_SRT = WORKSPACE / "testdata" / "ocr_srt_run_ac7f480a3ccb.srt"
+
 
 class FakeResponse:
     """模拟 urllib 响应：read() 返回 LLM 兼容接口的 JSON 载荷。"""
@@ -37,9 +57,9 @@ class FakeResponse:
 
 
 class FakeLLM:
-    """模拟 LLM 兼容接口：记录请求体，按策略返回"保留/删除"。
+    """模拟 LLM 兼容接口：记录请求体，按策略返回类别词。
 
-    支持两种策略：contents（按队列顺序，用于单次直调 _judge_target 的
+    支持两种策略：contents（按队列顺序，用于单次直调 _judge_category 的
     确定性测试）或 decision_fn（按请求体内容决策，用于并发 invoke 测试，
     保证任何线程执行顺序下判定结果都确定）。
     """
@@ -70,12 +90,17 @@ def _patch_llm(monkeypatch, contents: list[str] | None = None, decision_fn=None)
 
 
 def _decision_by_target(body) -> str:
-    """按目标字幕内容决策：含"答："判为删除，其余保留（与 _SRT 的噪声对应）。"""
+    """按目标字幕内容决策：含"答："判为 garbage，其余为 dialogue。"""
     target = next(
         line for line in body["messages"][1]["content"].splitlines()
         if line.startswith("【目标】")
     )
-    return "删除" if "答：" in target else "保留"
+    return "garbage" if "答：" in target else "dialogue"
+
+
+# ---------------------------------------------------------------------------
+# SRT 解析 / 序列化
+# ---------------------------------------------------------------------------
 
 
 def test_parse_srt_multiline_and_last_block() -> None:
@@ -105,14 +130,49 @@ def test_serialize_srt_renumbers() -> None:
     )
 
 
-def test_judge_target_window_and_keep(monkeypatch) -> None:
-    """窗口只含纯文本（无时间戳）、目标带标记；模型答"保留"则返回 False。"""
-    from nodes.llm_filter import _judge_target
+# ---------------------------------------------------------------------------
+# 确定性规则层
+# ---------------------------------------------------------------------------
 
+
+def test_rule_verdict_deletes_garbage_patterns() -> None:
+    """规则层直接删除：横线装饰、HTML/水印 token、URL/邮箱、单双 ASCII 字符。"""
+    tokens = {"html", "background", "___"}
+    for text in (
+        "---", "------", "------------------", "--- ---", "___", "= = =",
+        "…", "・", "HTML", "html", "Background", "background", "___",
+        "https://example.com/x", "www.example.com", "a@b.com",
+        "A", "V", "1", "DQ", "P4",
+    ):
+        assert _rule_verdict(text, tokens) is True, text
+    # 空文本/纯空白也删除。
+    assert _rule_verdict("   ", tokens) is True
+
+
+def test_rule_verdict_passes_cjk_and_dialogue() -> None:
+    """规则层不误伤：CJK 单字、正常对话、内容性短句交给 LLM 判断（None）。"""
+    tokens = {"html"}
+    for text in ("嗯", "好", "谢谢你 松井小姐", "不这么做的话 可没法胜任患者的对象", "好舒服"):
+        assert _rule_verdict(text, tokens) is None, text
+
+
+def test_rule_verdict_custom_overlay_tokens() -> None:
+    """overlay_tokens 参数化：自定义 token 同样直接删除。"""
+    assert _rule_verdict("Cleaning", {"cleaning"}) is True
+    assert _rule_verdict("Cleaning", set()) is None
+
+
+# ---------------------------------------------------------------------------
+# LLM 分类判断
+# ---------------------------------------------------------------------------
+
+
+def test_judge_category_window_and_dialogue(monkeypatch) -> None:
+    """窗口只含纯文本（无时间戳）、目标带标记；模型答 dialogue 则保留。"""
     entries = parse_srt(_SRT)
-    fake = _patch_llm(monkeypatch, ["保留"])
+    fake = _patch_llm(monkeypatch, ["dialogue"])
     # context_size=1，目标为第 2 条（index=1）：窗口 0..2 共 3 行，目标在中间。
-    assert _judge_target(entries, 1, context_size=1, params={}) is False
+    assert _judge_category(entries, 1, context_size=1, params={}) == "dialogue"
     body = fake.bodies[0]
     lines = body["messages"][1]["content"].splitlines()
     assert len(lines) == 3
@@ -125,32 +185,71 @@ def test_judge_target_window_and_keep(monkeypatch) -> None:
     assert body["max_tokens"] == 16
 
 
-def test_judge_target_delete(monkeypatch) -> None:
-    """模型答"删除"时返回 True（判定该条无意义）。"""
-    from nodes.llm_filter import _judge_target
-
+def test_judge_category_delete_classes(monkeypatch) -> None:
+    """模型答 garbage/overlay/noise 时返回对应类别（删除类）。"""
     entries = parse_srt(_SRT)
+    for cat in ("garbage", "overlay", "noise"):
+        _patch_llm(monkeypatch, [cat])
+        assert _judge_category(entries, 2, context_size=10, params={}) == cat
+
+
+def test_judge_category_repeat_and_unknown_kept(monkeypatch) -> None:
+    """repeat/dialogue 返回保留类；未识别输出（空/乱码/旧式"保留"）回退 dialogue。"""
+    entries = parse_srt(_SRT)
+    for answer in ("repeat", "dialogue", "", "???", "保留"):
+        _patch_llm(monkeypatch, [answer])
+        got = _judge_category(entries, 2, context_size=10, params={})
+        assert got in ("repeat", "dialogue"), (answer, got)
+    # 旧式"删除"回答兼容为垃圾类。
     _patch_llm(monkeypatch, ["删除"])
-    assert _judge_target(entries, 2, context_size=10, params={}) is True
+    assert _judge_category(entries, 2, context_size=10, params={}) == "garbage"
 
 
-def test_judge_target_model_and_auth(monkeypatch) -> None:
+def test_judge_category_model_and_auth(monkeypatch) -> None:
     """模型名从参数取；配置 API Key 时附带 Bearer 鉴权头。"""
-    from nodes.llm_filter import _judge_target
-
     entries = parse_srt(_SRT)
     monkeypatch.setenv("LLM_API_KEY", "sk-test")
-    fake = _patch_llm(monkeypatch, ["保留"])
-    assert _judge_target(entries, 0, context_size=10, params={"model": "m/1"}) is False
+    fake = _patch_llm(monkeypatch, ["dialogue"])
+    assert _judge_category(entries, 0, context_size=10, params={"model": "m/1"}) == "dialogue"
     assert fake.bodies[0]["model"] == "m/1"
     assert fake.headers[0]["Authorization"] == "Bearer sk-test"
 
 
+# ---------------------------------------------------------------------------
+# 去重键与删除判定
+# ---------------------------------------------------------------------------
+
+
+def test_dedup_key_normalizes_whitespace_and_case() -> None:
+    """去重键：去掉全部空白并统一大小写，空格差异视为同一文本。"""
+    assert _dedup_key("不这么做的话 可没法胜任患者的对象") == _dedup_key(
+        "不这么做的话可没法胜任患者的对象"
+    )
+    assert _dedup_key("HTML") == _dedup_key("html")
+    assert _dedup_key("你好 世界") != _dedup_key("你好世界2")
+
+
+def test_should_delete_long_text_noise_protected() -> None:
+    """长文本保护：≥min_keep_len 时 noise 不删，garbage/overlay 仍删；短文本三类都删。"""
+    long_text = "不这么做的话 可没法胜任患者的对象"
+    assert _should_delete("noise", long_text, min_keep_len=12) is False
+    assert _should_delete("garbage", long_text, min_keep_len=12) is True
+    assert _should_delete("overlay", long_text, min_keep_len=12) is True
+    assert _should_delete("noise", "答：无意义杂项", min_keep_len=12) is True
+    assert _should_delete("repeat", long_text, min_keep_len=12) is False
+    assert _should_delete("dialogue", long_text, min_keep_len=12) is False
+
+
+# ---------------------------------------------------------------------------
+# invoke 全链路
+# ---------------------------------------------------------------------------
+
+
 def test_invoke_filters_and_renumbers(monkeypatch, tmp_path) -> None:
-    """全链路（并发）：按 LLM 判定删除无意义条，保留条重新编号输出。"""
+    """全链路（并发）：按 LLM 类别判定删除无意义条，保留条重新编号输出。"""
     srt = tmp_path / "in.srt"
     srt.write_text(_SRT, encoding="utf-8")
-    # 内容决策：目标字幕含"答："判删除，其余保留（任何线程顺序下结果确定）。
+    # 内容决策：目标字幕含"答："判 garbage，其余 dialogue（任何线程顺序下结果确定）。
     _patch_llm(monkeypatch, decision_fn=_decision_by_target)
     response = invoke(
         InvokeRequest(
@@ -172,6 +271,79 @@ def test_invoke_filters_and_renumbers(monkeypatch, tmp_path) -> None:
     assert "00:00:13,000 --> 00:00:16,000\n第二句正常字幕\n" in out
 
 
+def test_invoke_rules_skip_llm(monkeypatch, tmp_path) -> None:
+    """规则层命中时直接删除且不调用 LLM：横线/HTML/单字符全部清理。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\n---\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\nHTML\n\n"
+        "3\n00:00:09,000 --> 00:00:12,000\nV\n\n"
+        "4\n00:00:13,000 --> 00:00:16,000\n正常对话\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, ["dialogue"])
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    assert response.outputs["kept"] == 1
+    assert response.outputs["removed"] == 3
+    assert len(fake.bodies) == 1  # 仅"正常对话"需 LLM，其余全走规则。
+
+
+def test_invoke_dedup_single_llm_call(monkeypatch, tmp_path) -> None:
+    """去重：相同文本（含空格变体）只调一次 LLM，判定结果一致。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\n好舒服\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\n好 舒 服\n\n"
+        "3\n00:00:09,000 --> 00:00:12,000\n好舒服\n\n"
+        "4\n00:00:13,000 --> 00:00:16,000\n别的台词\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, decision_fn=lambda body: "repeat")
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    # 2 个唯一文本（好舒服/别的台词）→ 恰好 2 次 LLM 调用。
+    assert len(fake.bodies) == 2
+    # repeat 类保留 → 4 条全部保留。
+    assert response.outputs["kept"] == 4
+    assert response.outputs["removed"] == 0
+
+
+def test_invoke_dedupe_disabled(monkeypatch, tmp_path) -> None:
+    """dedupe=0 关闭去重：每个条目都调一次 LLM。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\n同一句\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\n同一句\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, decision_fn=lambda body: "dialogue")
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={"dedupe": "0"},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    assert len(fake.bodies) == 2
+
+
 def test_invoke_context_size_param(monkeypatch, tmp_path) -> None:
     """context_size 参数生效：窗口大小=2×context_size+1（两端截断除外）。"""
     srt = tmp_path / "in.srt"
@@ -195,6 +367,52 @@ def test_invoke_context_size_param(monkeypatch, tmp_path) -> None:
     assert len(window) == 3
 
 
+def test_invoke_overlay_tokens_param(monkeypatch, tmp_path) -> None:
+    """overlay_tokens 参数：自定义 token 走规则层删除，不调用 LLM。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\nCleaning\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\n正常对话\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, decision_fn=lambda body: "dialogue")
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={"overlay_tokens": ["cleaning"]},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    assert response.outputs["removed"] == 1
+    assert response.outputs["kept"] == 1
+    assert len(fake.bodies) == 1  # 只判正常对话。
+
+
+def test_invoke_overlay_tokens_json_string(monkeypatch, tmp_path) -> None:
+    """overlay_tokens 以 JSON 字符串形式传入（工作流参数常见形态）同样生效。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\nXXLogo\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\n正常对话\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, ["dialogue"])
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={"overlay_tokens": '["xxlogo", "xx"]'},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    assert response.outputs["removed"] == 1
+    assert response.outputs["kept"] == 1
+    assert len(fake.bodies) == 1
+
+
 def test_invoke_missing_input(tmp_path) -> None:
     """缺少 srt_uri 时返回失败。"""
     response = invoke(
@@ -255,3 +473,79 @@ def test_invoke_empty_srt(monkeypatch, tmp_path) -> None:
     assert response.outputs["kept"] == 0
     assert response.outputs["removed"] == 0
     assert fake.bodies == []
+
+
+# ---------------------------------------------------------------------------
+# 真实任务留存数据回归（testdata/ocr_srt_run_ac7f480a3ccb.srt）
+# ---------------------------------------------------------------------------
+
+
+def test_real_run_rules_and_dialogue_regression(monkeypatch, tmp_path) -> None:
+    """真实数据回归：规则层清理垃圾、对话保留、时间轴单调、去重生效。
+
+    夹具为 run_ac7f480a3ccb（单线程 v4 跑完整 2 小时视频）的 OCR 输出 1666 条。
+    假 LLM 对非规则条目一律答 dialogue：验证规则层删掉全部垃圾（横线/水印/
+    单字符），真实对话全部保留，且相同文本只调一次 LLM（去重）。
+    """
+    srt = REAL_SRT
+    if not srt.is_file():
+        pytest.skip("缺少 testdata/ocr_srt_run_ac7f480a3ccb.srt，跳过回归测试")
+    fake = _patch_llm(monkeypatch, decision_fn=lambda body: "dialogue")
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+
+    entries = parse_srt(srt.read_text(encoding="utf-8"))
+    out = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
+
+    # 规则层删除量 = 夹具中直接命中规则文本的条数（真实数据动态计算）。
+    # 规则层删除量 = 夹具中直接命中规则文本的条数（与 invoke 默认 token 集一致）。
+    from nodes.llm_filter import DEFAULT_OVERLAY_TOKENS
+
+    tokens = set(DEFAULT_OVERLAY_TOKENS)
+    rule_removed = sum(
+        1 for e in entries if _rule_verdict(e["text"], tokens) is True
+    )
+    assert response.outputs["removed"] == rule_removed
+    assert response.outputs["kept"] == len(entries) - rule_removed
+
+    # 垃圾全部清出输出（横线、HTML、单字符都不再出现；只检查文本字段，
+    # 排除 SRT 的序号行与时间轴行）。
+    for entry in parse_srt(out):
+        assert _rule_verdict(entry["text"], tokens) is not True, entry["text"]
+    # 真实对话保留（修复前被误删的自我介绍、请求对话、呻吟内容行）。
+    # 注意：片头标题卡（淫魔病院…）真实 LLM 判为 overlay（烧录标题），
+    # 不在必保留列表内；本测试假 LLM 一律答 dialogue，仅验证机制正确。
+    for kept_line in (
+        "谢谢你 松井小姐",
+        "我姓泷本 请多多指教",
+        "不这么做的话 可没法胜任患者的对象",
+        "你要看着北冈小姐的脸 你们俩相互看看嘛",
+        "好舒服",
+    ):
+        assert kept_line in out, kept_line
+
+    # 时间轴单调递增（输出顺序即时间顺序）。
+    times = []
+    for m in __import__("re").finditer(
+        r"(\d{2}:\d{2}:\d{2},\d{3})\s*-->", out
+    ):
+        t = m.group(1).replace(",", ".")
+        h, mi, s = t.split(":")
+        times.append(int(h) * 3600 + int(mi) * 60 + float(s))
+    assert all(a < b for a, b in zip(times, times[1:]))
+
+    # 去重：LLM 调用数 == 非规则条目中的唯一文本数（远小于条目数）。
+    unique_llm = len({
+        _dedup_key(e["text"])
+        for e in entries
+        if _rule_verdict(e["text"], tokens) is None
+    })
+    assert len(fake.bodies) == unique_llm
+    assert len(fake.bodies) < len(entries)  # 去重确实省调用。

tests/test_ocr_flow.py

@@ -305,6 +305,26 @@ def test_frame_extract_one_second_exact_frames(tmp_path) -> None:
     assert len(manifest) == 10
 
 
+def test_frame_files_read_order_matches_frame_number(tmp_path) -> None:
+    """真实任务留存数据：帧文件按帧号数值排序读取，而非字典序。
+
+    回归用例：ffmpeg 的 %04d 编号在超过 9999 帧后自动扩为 5 位
+    （frame_10000.png 等），此时 sorted() 默认字典序会把 5 位编号排在
+    4 位编号之前（如 frame_10009 < frame_1009），导致帧号回退、manifest
+    时间与图像错位。testdata/frames_boundary/ 是 2026-08 真实任务
+    run_339ec7ee437f（14236 帧 / 2 小时视频）中跨越该边界的真实帧文件。
+    """
+    from nodes.frame_extract import _sorted_frame_files
+
+    boundary = TESTDATA / "frames_boundary"
+    files = _sorted_frame_files(boundary)
+    # 从文件名解析帧号：读取顺序必须等于帧号数值递增序（无回退）。
+    nums = [int(p.stem.split("_", 1)[1]) for p in files]
+    assert nums == sorted(nums)
+    # 边界关键对：5 位编号必须排在 4 位编号之后，禁止字典序错位。
+    assert nums.index(10000) > nums.index(9999)
+    assert nums.index(10009) > nums.index(1009)
+
 # ---------------------------------------------------------------------------
 # subtitle-ocr：OCR 循环 + 长度上限 + 合并 + SRT 组装
 # ---------------------------------------------------------------------------

tests/test_subtitle_ocr_order_threading.py

@@ -0,0 +1,215 @@
+"""多线程下字幕顺序正确性测试（模拟真实 API 返回，全量真实数据）。
+
+目标：验证 subtitle-ocr 在**多线程**执行时能否正确处理字幕顺序。
+
+- 不走真实 vlm-ocr（Ollama）网络调用：registry.invoke 被替换为
+  FakeVlmOcrApi，按 image_uri 文件名中的帧号，直接从测试数据（真实任务
+  run_ac7f480a3ccb 的**全量**逐帧 OCR 结果）取该帧文本返回，模拟真实
+  API 返回结构；输入可以是文件名（frames_manifest / image_uri）；
+- **真实 OCR 延迟模拟**：真实 vlm-ocr 每次调用延迟不可预判——大部分帧快、
+  少量帧明显慢（复杂画面/模型排队，真实约 0.1s~5s，测试按比例缩放）。
+  假 API 用种子化随机生成同样的快/慢分布（约 8% 慢帧），制造真实波动下
+  的乱序完成，对线程池的保序能力施加最贴近真实情况的压力；
+- 核心断言：多线程（4/16 线程）产出的 SRT 与**用户确认过的精确结果**逐字节
+  一致——该结果正是真实任务 run_ac7f480a3ccb 以**单线程**（workflow v4
+  pool 1/1）运行产出并经用户确认的，因此逐字节一致即证明"多线程 == 单线程"；
+  同时断言并发真实发生、完成顺序确实乱序、全量时间轴严格递增、每条字幕
+  与其起始时刻帧的文本对齐。
+
+全量真实数据（testdata/，常驻夹具，真实任务 run_ac7f480a3ccb 全部 14236 帧）：
+- frames_manifest_full.json：完整帧清单（14236 条，image_uri 改为文件名）；
+- ocr_frames_full.json：{帧号: 该帧 OCR 文本}（位置 p ↔ 帧 p+1）；
+- ocr_srt_run_ac7f480a3ccb.srt：该任务单线程运行产出、用户确认过的精确结果（1666 条）。
+"""
+
+import json
+import random
+import re
+import threading
+import time
+from pathlib import Path
+
+from wov_sdk.models import InvokeRequest, InvokeResponse
+
+WORKSPACE = Path(__file__).resolve().parent.parent
+TESTDATA = WORKSPACE / "testdata"
+FULL_MANIFEST = TESTDATA / "frames_manifest_full.json"
+FULL_OCR_TEXTS = TESTDATA / "ocr_frames_full.json"
+CONFIRMED_SRT = TESTDATA / "ocr_srt_run_ac7f480a3ccb.srt"
+TOTAL_FRAMES = 14236
+
+# 帧号解析：image_uri 文件名形如 frame_0411.png。
+_FRAME_RE = re.compile(r".*frame_(\d+)\.png")
+
+
+def _frame_no(uri: str) -> int:
+    """从 image_uri 文件名解析帧号。"""
+    return int(_FRAME_RE.match(Path(uri).name).group(1))
+
+
+def _ts_to_seconds(ts: str) -> float:
+    """SRT 时间戳（HH:MM:SS,mmm）转秒。"""
+    h, m, s = ts.replace(",", ".").split(":")
+    return int(h) * 3600 + int(m) * 60 + float(s)
+
+
+class FakeVlmOcrApi:
+    """模拟真实 vlm-ocr API：不发起真实网络调用，从测试数据返回该帧 OCR 文本。
+
+    - 输入是 image_uri 文件名，解析帧号后从全量真实 OCR 结果（{帧号: 文本}）
+      取该帧文本返回，模拟真实 API 返回结构（status=completed / outputs.text）；
+    - **可变延迟模拟真实 OCR**：每次调用独立随机，大部分帧快
+      （0.2~2ms，对应真实约 0.1~1s），约 slow_ratio 比例的帧明显慢
+      （6~18ms，对应真实约 3~9s，如复杂画面/模型排队）。延迟不可预判，
+      对线程池的乱序恢复能力施加与真实情况一致的随机压力；
+    - 线程安全地记录并发峰值、完成顺序与每次延迟，供断言"多线程确实发生、
+      乱序完成、且存在明显慢帧"。
+    """
+
+    def __init__(self, texts_by_frame: dict[int, str], seed: int = 20260817,
+                 fast_ms: float = 1.0, slow_ms: float = 12.0,
+                 slow_ratio: float = 0.08) -> None:
+        self._texts_by_frame = texts_by_frame
+        self._fast_ms = fast_ms
+        self._slow_ms = slow_ms
+        self._slow_ratio = slow_ratio
+        # 种子化随机源：延迟波动可复现（固定种子 → 测试确定性，不 flaky）。
+        self._rng = random.Random(seed)
+        self._lock = threading.Lock()
+        self.active = 0
+        self.max_active = 0
+        # 完成顺序（帧号）：用于断言乱序完成确实发生。
+        self.completed_frames: list[int] = []
+        # 每次调用的实际延迟（毫秒）：用于断言快/慢分布确实发生。
+        self.delays_ms: list[float] = []
+
+    def __call__(self, node_id: str, request: InvokeRequest) -> InvokeResponse:
+        frame_no = _frame_no(request.inputs["image_uri"])
+        with self._lock:
+            self.active += 1
+            self.max_active = max(self.max_active, self.active)
+        try:
+            # 真实 OCR 延迟：慢帧比例固定，具体哪帧慢由随机决定（不可预判）。
+            if self._rng.random() < self._slow_ratio:
+                delay = self._slow_ms * (0.5 + self._rng.random())
+            else:
+                delay = self._fast_ms * (0.2 + self._rng.random() * 1.8)
+            self.delays_ms.append(delay)
+            time.sleep(delay / 1000.0)
+            text = self._texts_by_frame[frame_no]
+        finally:
+            with self._lock:
+                self.active -= 1
+                self.completed_frames.append(frame_no)
+        # 与真实 vlm-ocr 节点一致的返回结构。
+        return InvokeResponse(status="completed", outputs={"text": text})
+
+
+def _run_ocr(monkeypatch, manifest_path: Path, texts_by_frame: dict[int, str],
+             pool_min: int, pool_max: int, output_dir: Path, seed: int,
+             ) -> tuple[Path, FakeVlmOcrApi]:
+    """用给定线程配置运行 subtitle-ocr，返回 (产物路径, 假 API 实例)。"""
+    from nodes.subtitle_ocr import invoke as ocr_invoke
+
+    fake = FakeVlmOcrApi(texts_by_frame, seed=seed)
+    monkeypatch.setattr("wov_app.registry.invoke", fake)
+    response = ocr_invoke(
+        InvokeRequest(
+            run_id="order_test",
+            node_instance_id="",
+            inputs={"frames_manifest": str(manifest_path)},
+            params={"pool_min_workers": pool_min, "pool_max_workers": pool_max},
+            output_dir=str(output_dir),
+        )
+    )
+    assert response.status == "completed", response.error
+    return Path(response.outputs["srt_uri"]), fake
+
+
+def _load_full_data() -> tuple[list[dict], dict[int, str]]:
+    """加载全量夹具：manifest 与 {帧号: 文本}。"""
+    manifest = json.loads(FULL_MANIFEST.read_text(encoding="utf-8"))
+    texts_by_frame = {
+        int(key): value for key, value in json.loads(FULL_OCR_TEXTS.read_text(encoding="utf-8")).items()
+    }
+    return manifest, texts_by_frame
+
+
+def _assert_alignment(srt_text: str, manifest: list[dict], texts_by_frame: dict[int, str]) -> None:
+    """核心对齐断言：每条字幕的起始时刻对应的帧，其 OCR 文本必须就是本条字幕文本。
+
+    这正是"多线程下顺序正确"的最终验证：无论线程如何并发/乱序完成，
+    每条字幕贴的时刻必须是它真实来源帧的时刻。
+    """
+    time_text = {
+        round(float(entry["time"]), 3): texts_by_frame[_frame_no(entry["image_uri"])]
+        for entry in manifest
+    }
+    blocks = re.findall(
+        r"(\d{2}:\d{2}:\d{2},\d{3})\s*-->\s*(\d{2}:\d{2}:\d{2},\d{3})\s*\n(.*?)(?=\n\s*\d+\s*\n|\Z)",
+        srt_text, re.DOTALL,
+    )
+    times = []
+    for start, _end, text in blocks:
+        start_s = _ts_to_seconds(start)
+        # 帧时间与 SRT 时间戳间允许 ±2ms 容差：format_timestamp 用 int 截断，
+        # 浮点 256.258 会以 256.25799.. 截断为 256,257（真实运行同样行为）。
+        best = min(time_text, key=lambda t: abs(t - start_s))
+        assert abs(best - start_s) <= 0.002, f"字幕起始时刻 {start_s}s 无对应帧"
+        assert time_text[best] == text.strip(), f"时刻 {start_s}s 的文本与帧不一致"
+        times.append(start_s)
+    assert all(a < b for a, b in zip(times, times[1:])), "时间轴必须严格递增"
+
+
+class TestSubtitleOcrOrderUnderThreading:
+    """多线程下字幕顺序正确性测试类（全量 14236 帧真实数据 + 真实可变延迟）。
+
+    确认结果（testdata/ocr_srt_run_ac7f480a3ccb.srt）是真实任务 run_ac7f480a3ccb
+    以**单线程**（workflow v4 pool 1/1）运行产出、经用户确认的精确结果：
+    - 多线程（4/16 线程）+ 真实可变延迟下重放全量数据，产物必须与它逐字节一致；
+    - 同时必须真实发生"多线程 + 乱序完成 + 明显慢帧"，而顺序仍正确。
+    数据全部为常驻夹具，测试始终执行（不依赖 gitignored 数据）。
+    """
+
+    def test_full_real_data_variable_latency_keeps_order(self, monkeypatch, tmp_path) -> None:
+        """全量真实数据 + 真实可变延迟：多线程产物与单线程确认结果逐字节一致。"""
+        manifest, texts_by_frame = _load_full_data()
+        confirmed = CONFIRMED_SRT.read_text(encoding="utf-8")
+
+        outputs: dict[tuple, str] = {}
+        fakes: dict[tuple, FakeVlmOcrApi] = {}
+        for pool_min, pool_max in ((4, 4), (16, 16)):
+            out_dir = tmp_path / f"p{pool_min}-{pool_max}"
+            srt_path, fake = _run_ocr(
+                monkeypatch, FULL_MANIFEST, texts_by_frame,
+                pool_min=pool_min, pool_max=pool_max,
+                output_dir=out_dir, seed=20260817,
+            )
+            outputs[(pool_min, pool_max)] = srt_path.read_text(encoding="utf-8")
+            fakes[(pool_min, pool_max)] = fake
+
+        # ① 多线程产物与用户确认过的精确结果（真实单线程运行）逐字节一致。
+        assert outputs[(4, 4)] == confirmed, "4 线程产物与确认结果不一致"
+        assert outputs[(16, 16)] == confirmed, "16 线程产物与确认结果不一致"
+        assert outputs[(4, 4)] == outputs[(16, 16)]
+
+        # ② 多线程确实发生（并发峰值>1），且可变延迟下完成顺序乱序（靠后帧
+        # 先完成/慢帧滞后），但输出仍与确认结果一致——线程池正确恢复了顺序。
+        for (pool_min, pool_max), fake in fakes.items():
+            assert fake.max_active > 1, f"{pool_max} 线程配置下应真实并发"
+            assert fake.completed_frames != sorted(fake.completed_frames), \
+                f"{pool_max} 线程下可变延迟应产生乱序完成"
+
+        # ③ 延迟模拟符合真实 OCR：存在明显慢帧（>3 倍快帧均值），也含极快帧。
+        for (_, pool_max), fake in fakes.items():
+            delays = fake.delays_ms
+            assert len(delays) == TOTAL_FRAMES, "每帧都应产生一次调用延迟"
+            assert max(delays) > 3.0, f"{pool_max} 线程下应存在明显慢帧（真实 OCR 延迟波动）"
+            assert min(delays) < 1.0, "应存在快帧（大部分帧响应快）"
+
+        # ④ 全量时间轴严格递增 + 每条字幕与其起始时刻帧的文本对齐。
+        _assert_alignment(confirmed, manifest, texts_by_frame)
+
+        # ⑤ 已知真实内容存在于结果中（确认结果的代表性条目）。
+        for line in ("北冈小姐", "这是特别病房患者的病历表", "应该已经察觉到 至今为止的一切了吧"):
+            assert line in confirmed, line

workflows/ocr-subtitle.json

@@ -2,10 +2,10 @@
   "id": "ocr-subtitle",
   "name": "字幕OCR提取",
   "description": "抽帧并 OCR 提取视频烧录字幕，经 LLM 过滤无意义内容后生成带时间轴的 SRT 基准数据。",
-  "version": 3,
+  "version": 4,
   "definition": {
     "name": "字幕OCR提取",
-    "version": 3,
+    "version": 4,
     "nodes": [
       {
         "id": "extract",
@@ -27,7 +27,9 @@
         "id": "ocr",
         "node_type": "subtitle-ocr",
         "params": {
-          "prompt": "提取图像中的文字，不要描述图片中的内容"
+          "prompt": "提取图像中的文字，不要描述图片中的内容",
+          "pool_min_workers": 1,
+          "pool_max_workers": 1
         },
         "inputs": {
           "frames_manifest": "extract.frames_manifest"
@@ -36,7 +38,10 @@
       {
         "id": "filter",
         "node_type": "llm-filter",
-        "params": {},
+        "params": {
+          "pool_min_workers": 1,
+          "pool_max_workers": 1
+        },
         "inputs": {
           "srt_uri": "ocr.srt_uri"
         }