feat: 转写支持静音段幻觉抑制参数（HST）

无 VAD 的 decode_full 会在无语音段"编"出字幕（`こんにちは`/`おはようございます`/
`東京都交通局8800形電車`）。实测这类幻觉与"呻吟间隙里的真实短台词"在
`no_speech_prob`、`avg_logprob`、silero VAD 与音频能量四个维度上都不可分，
只能用 faster-whisper 自带的 `hallucination_silence_threshold`（HST）抑制：
怀疑幻觉时跳过超过阈值的静音部分（需 `word_timestamps=True`）。

- `nodes/whisper.py` 透传 `word_timestamps` / `hallucination_silence_threshold` /
  `no_speech_threshold` / `log_prob_threshold` / `compression_ratio_threshold`，
  未配置的键不传，保持 faster-whisper 默认值与改造前行为。
- `learn-translate` 默认 HST=2.0 + word_timestamps：同一片头 120 秒无对话段由 15 条
  字幕降到 4 条且无套话幻觉残留，呻吟段基本保留；代价转写约慢 1.8×。
- 实测数据与取舍记录见 docs/workflows.md#decode_full-与幻觉呻吟清洗。

docs/configuration.md

@@ -33,7 +33,12 @@ http://127.0.0.1:8000/docs          API 文档
 | `WOV_CLEANUP_GRACE_SECONDS` | `3600` | 孤儿清理宽限期（秒） |
 | `WOV_BATCH_ENABLED` | `1` | 开启文件夹批量处理引擎（处理 source=batch 任务） |
 | `WOV_BATCH_INTERVAL_SECONDS` | `1.0` | 批量引擎轮询间隔 |
-| `WOV_BATCH_STAGE_GROUP_SIZE` | `8` | 批量「分块流水线」分组大小：每组视频按节点顺序跑完全部阶段（全部 extract → 全部 ASR → 全部翻译 → 全部 ASS）再进入下一组，本地模型每组只加载一次；设为 1 等价于每个视频各跑完整链路（产物逐视频落地最及时） |
+| `WOV_BATCH_STAGE_GROUP_SIZE` | `8` | 批量流水线分组大小：组内每个视频独立推进阶段（详见 [operations.md](./operations.md#文件夹批量处理)），GPU 阶段组内串行、非 GPU 阶段并行；本机模型每组只加载一次 |
+| `WOV_BATCH_PIPELINE_WORKERS` | `4` | 批量组内在途阶段上限：提音/线上翻译/ASS 等不吃 GPU 的阶段可并行，GPU 阶段仍互斥 |
+| `WOV_DB_BUSY_TIMEOUT_SECONDS` | `30` | SQLite 写锁等待时长（并发写产物/进度时排队而不是立刻报错）；库自动开启 WAL |
+| `WOV_LOCAL_MODEL_HOSTS` | 空 | 视为「本机模型」的额外主机名/IP（逗号分隔）：Ollama 跑在本机局域网地址上时必须声明，否则会被当成不占显存的远端端点 |
+| `WOV_LOCAL_LLM_RESERVE_MB` | `22528` | 本机 LLM 端点占用的显存预留（MB）：常驻期间转写阶段会等待显存 |
+| `WOV_LOCAL_VLM_RESERVE_MB` | `8192` | 本机 VLM/OCR 端点占用的显存预留（MB） |
 | `WOV_AUTO_VAD` | `1` | 开启每视频自适应 VAD 调参（详见 [adaptive_vad.md](./adaptive_vad.md)） |
 | `WHISPER_MODEL_PATH` | 见 [模型权重解析](./node-protocol.md#模型权重解析本地优先) | 显式指定 whisper 模型路径 |
 | `WHISPER_DEVICE` | `auto` | 转写设备 |

docs/decisions.md

@@ -71,6 +71,37 @@
   并用半透明填充 + 半透明描边降低遮挡感。
 - **调研全文**：[VR双目字幕景深与遮挡调查报告.md](./VR双目字幕景深与遮挡调查报告.md)。
 
+## 批量任务先写明细再排队（CREATING 状态）
+
+- **现象**：新建批量任务后偶发任务被标 COMPLETED、`done=0/N`，剩余视频永远不再
+  被处理（例：`batch_ac585c5458de` 唯一明细还 PENDING 而任务已 COMPLETED）。
+- **根因**：`create_job` 先把任务行以 QUEUED 入库（此时引擎就看得见），再逐条登记
+  明细（扫描媒体库时 500+ 条要数秒）；引擎轮询到的快照可能没包含剩余明细，收尾时
+  “无未结束明细”检查也看不到它们，于是把任务标 COMPLETED。
+- **结论**：任务行改以 `CREATING` 入库，明细全部登记完才置 QUEUED（引擎只取
+  QUEUED）；登记中途异常置 FAILED 并抛给路由；启动时把上一进程遗留的 CREATING
+  统一置 FAILED（`fail_creating_batch_jobs`），避免静默残留。
+- **旧数据修复**：`scripts/fix_zombie_batch_jobs.py` 仍用于处理历史
+  “COMPLETED 但仍有未结束明细”的脏数据。回归测试见 `test_batch.py` 的
+  `test_job_hidden_from_engine_until_details_written`。
+
+## 批量流水线按 GPU 资源调度（而不是按"是否云端"分支）
+
+- **现象**：LLM 走线上端点时翻译阶段不占显存，但组内阶段仍是"先全部提音、再全部
+  转写、再全部翻译"的串行推进，翻译期间 GPU 全程空闲（实测 19.6W / 272MiB，
+  占整轮挂钟约 40%）。
+- **结论**：阶段能否启动只看资源。`wov_app.resources.stage_gpu_need_mb` 给出每个
+  阶段的显存需求（whisper 按权重 ×1.45、本机 LLM 端点按预留、线上端点为 0），
+  进程内 `GpuGate` 负责准入：不需要 GPU 的阶段立刻放行（转写与线上翻译并行）；
+  需要 GPU 的阶段互斥，并按"阶段索引最小者优先"派发，组内因此仍是先跑完全部转写
+  再进翻译——本机 Ollama 模型每组只加载一次，行为与旧实现一致。
+- **为什么不是"配置分支"**：代码不判断端点是不是云端，只读"这个阶段要不要占
+  GPU、现在够不够"。同一份代码在本机模型下自动退化为串行、在线上模型下自动并行。
+- **降级**：探测不到 `nvidia-smi` 时退化为 GPU 阶段互斥（与串行结果一致）；
+  局域网地址上的本机 Ollama 需用 `WOV_LOCAL_MODEL_HOSTS` 声明，否则会被当成远端。
+- **并发写**：流水线多线程写产物/进度，SQLite 开 WAL + busy timeout
+  （`WOV_DB_BUSY_TIMEOUT_SECONDS`）。
+
 ## 相关文档
 
 - 当前生效的参数与协议：[node-protocol.md](./node-protocol.md)

docs/operations.md

@@ -64,7 +64,10 @@
   空列表不报 500）。只暴露目录名，不返回文件内容。
 - **创建任务时一次性定位（2026-09 起）**：`POST /api/batch/jobs {folder,
   workflow_id, recursive}` 只扫描一次文件夹并把每个视频登记为 `batch_videos`
-  明细（PENDING/RUNNING/PAUSED/COMPLETED/FAILED/SKIPPED）。**视频所在目录
+  明细（PENDING/RUNNING/PAUSED/COMPLETED/FAILED/SKIPPED）。任务行先以
+  `CREATING` 入库、**明细全部登记完才置 QUEUED**：否则引擎会在明细写一半时拾起
+  任务、收尾把任务误标 COMPLETED（理由见
+  [decisions.md](./decisions.md#批量任务先写明细再排队creating-状态)）。**视频所在目录
   （视频旁）若已存在文件名含视频名的字幕文件**（`.srt/.ass/.ssa/.vtt`，
   `list_sidecar_subtitles` 判定，如 `movie.CN.srt`、`movie.CN_dual_eye.ass`），
   说明该视频已有字幕，创建即记 **SKIPPED**——不为它触发任何流水线。运行时
@@ -73,8 +76,8 @@
   无版本/一个视频都没有）返回 422。
 - **产物放在视频旁**：每个视频处理完成后，把工作流 `final_outputs` 对应的最终
   产物文件（字幕流水线即中文 `.srt` 与双目 `.ass`）**复制一份到视频所在目录**，
-  与 .mp4 放在一起（`_place_products`）。文件名**对齐媒体库既有约定**：中文字幕
+  与 .mp4 放在一起（`_place_products`）。文件名**对齐媒体库既有约定**：日语转写
-  存为 `<视频名>.CN.srt`、双目字幕存为 `<视频名>.CN_dual_eye.ass`（稳定无时间戳，
+  存为 `<视频名>.JA.srt`、中文字幕存为 `<视频名>.CN.srt`、双目字幕存为 `<视频名>.CN_dual_eye.ass`（稳定无时间戳，
   `_sidecar_product_name` 映射，其余扩展名产物保留原文件名；同名目标直接覆盖）。
   文件名含视频主名，下次批量扫描会命中"已有字幕"规则直接跳过该视频。
 - **成品放置校验（审查 R03）**：先预检全部 `final_outputs` 对应的记录与文件，
@@ -101,15 +104,20 @@
   run 的 `current_node_id` 在 DAG 拓扑序中的位置推导、节点类型映射中文标签。
   粒度限制：`progress` 只有**节点边界**粒度，句级进度（转写分块、翻译批次、OCR 帧）
   不落库、只在控制台日志里。
-- **分块流水线执行（本地模型只加载一次）**：批量引擎把待处理视频按
+- **按资源调度的在途流水线（2026-09 起）**：批量引擎把待处理视频按
-  `WOV_BATCH_STAGE_GROUP_SIZE`（默认 8）分组，**组内按节点顺序跑完全部视频**
+  `WOV_BATCH_STAGE_GROUP_SIZE`（默认 8）分组，组内**每个视频独立推进自己的
-  （先全部 extract、再全部 ASR、再全部 LLM 翻译、最后 ASS）再进入下一组。每个
+  阶段**，最多 `WOV_BATCH_PIPELINE_WORKERS`（默认 4）个阶段在途。判定只看资源：
-  视频的 run 在阶段边界保持 RUNNING（`execute_run(stop_after=节点)`），下一阶段
+  阶段是否需要 GPU 由 `wov_app.resources.stage_gpu_need_mb` 给出（提音、**线上
-  从产物表跳过已完成节点继续，因此本地模型每组只加载一次、卸载一次，而不是每个
+  端点**的翻译、ASS 都不需要），不需要 GPU 的阶段立刻派发，于是转写能与线上翻译
-  视频来回加载卸载；产物仍按组增量落地。LLM 阶段执行时引擎在 run 根目录写
+  并行、提音能与转写并行（此前组内阶段是串行的，翻译时 GPU 全程空转）。需要 GPU
-  `keep_model.flag`，节点据此不在每次调用后卸载模型（`nodes/llm.py`），阶段
+  的阶段由进程内 `GpuGate` 串行准入并按"阶段索引最小者优先"派发，因此组内仍是
-  结束由引擎调 `release_local_model()` 统一释放显存，让下一组的 ASR 拿到 GPU
+  先跑完全部转写再进翻译——**本机 Ollama 模型每组只加载一次**，不需要按"是否
-  （否则本地模型常驻显存会让 whisper 直接 CUDA OOM）。设为 1 即回到「每个视频
+  云端"写分支。每个视频的 run 在阶段边界保持 RUNNING（`execute_run(stop_after=节点)`），
+  下一阶段从产物表跳过已完成节点继续。LLM 阶段执行时引擎在 run 根目录写
+  `keep_model.flag`，节点据此不在每次调用后卸载模型（`nodes/llm.py`），组末由
+  引擎调 `release_local_model()` 统一释放显存，让下一组的 ASR 拿到 GPU（否则
+  本机模型常驻显存会让 whisper 直接 CUDA OOM）。显存探测不到（无 `nvidia-smi`）
+  时退化为"GPU 阶段互斥"，行为与串行一致。设为 1 即回到「每个视频
   跑完整链路」的旧行为。
 - **失败视频不跨阶段推进**：某阶段失败的视频只在**它失败节点的那个阶段**重试
   （下一次引擎循环从断点续跑），不会在后续阶段里重跑前序节点——避免本地 LLM 已
@@ -160,6 +168,40 @@
 - **环境变量**：`WOV_BATCH_ENABLED`（默认 1）、`WOV_BATCH_INTERVAL_SECONDS`
   （默认 1.0），完整列表见 [configuration.md](./configuration.md)。
 
+## 媒体库历史字幕重建（统计 → 备份 → 批量重跑）
+
+模型或管线切换后，媒体库里由旧管线生成的字幕需要整批重跑。批量引擎按
+“视频旁已有字幕即 SKIPPED”判定，所以重跑前必须先统计范围、再把旧字幕改名
+备份，否则建出来的任务会把所有视频全部跳过。工具是 `scripts/plan_regenerate_subtitles.py`
+（**不依赖仓库**，可拷到媒体库主机直接跑：CIFS 挂载下逐文件 ffprobe 与
+目录扫描都慢得多，在 NAS 本地跑 524 个视频只要几秒）。
+
+生成时间的判定口径：只看本流水线产出的 CN 产物（`<视频名>.CN.srt`、
+`<视频名>.CN_dual_eye.ass`），取它们**最早**的 mtime——双目 `.ass` 可能被样式
+统一脚本原地改写而“变新”，中文字幕的 mtime 才是真实生成时间。
+
+```bash
+# 1) 统计：数量 / 时长 / 体积 / 按月分布，并写出逐条明细 JSON
+python3 scripts/plan_regenerate_subtitles.py /vol1/1000/123 \
+    --before 2026-09-01 --out data/experiments/regen_plan/plan_2026-09-01.json
+
+# 2) 备份旧字幕（默认预览，--apply 才落盘）：改名 <原名>.old-<当天日期>
+python3 scripts/plan_regenerate_subtitles.py /vol1/1000/123 --backup-old
+python3 scripts/plan_regenerate_subtitles.py /vol1/1000/123 --backup-old --apply \
+    --select sample10.txt          # 只处理清单里的视频，支持 # 注释
+
+# 3) 建批量任务（批量页或 POST /api/batch/jobs）
+```
+
+- 备份文件名带 `.old-<日期>` 后缀，扩展名不再是字幕后缀，批量引擎不会再把它当
+  旁挂字幕；确认新字幕无误后删除备份，需要回退则去掉后缀改回原名。
+- 未备份的视频仍有旁挂字幕 → 创建任务时记 SKIPPED，因此“整库建一个任务”即可，
+  实际只会处理被备份的那批；已重跑的视频产出新 CN 字幕（mtime 变新），全量重建
+  时自动归入“已是最新”，不会重复处理。
+- 实测数据（380 个待重生成 / 207.88 小时；10 个样本 6.54 小时素材跑 77.7 分钟、
+  GPU 平均 289.6 W、约 0.375 kWh，全量推算约 41 小时 / 整机 15 kWh）见
+  `data/experiments/regen_plan/REPORT.md`。
+
 ## 相关文档
 
 - 环境变量全表：[configuration.md](./configuration.md)

docs/testing.md

@@ -51,6 +51,7 @@ tests/
 ├── web/test_crop/               # 对应 web/assets/（框选几何换算）
 ├── web/test_batch/              # 对应 web/assets/（批量页渲染）
 ├── scripts/test_fix_zombie_batch_jobs/  # 对应 scripts/（僵尸批量任务修复）
+├── scripts/test_plan_regenerate_subtitles/  # 对应 scripts/（字幕重生成计划统计）
 └── shared/                      # 跨模块公共设施
     ├── realdata_contract.py     # 真实数据契约与对齐量化
     ├── srt_entries.py           # 按秒解析 SRT
@@ -161,6 +162,7 @@ tests/
 | `web/assets/crop.js` | 框选几何换算 | `tests/web/test_crop/`（真实 node 执行） | 已覆盖 |
 | `web/assets/batch.js` | 批量页明细/进度渲染 | `tests/web/test_batch/`（真实 node 执行） | 已覆盖 |
 | `scripts/fix_zombie_batch_jobs.py` | 僵尸批量任务诊断与修复 | `tests/scripts/test_fix_zombie_batch_jobs/` | 已覆盖 |
+| `scripts/plan_regenerate_subtitles.py` | 媒体库字幕重生成计划统计（数量/时长/分类） | `tests/scripts/test_plan_regenerate_subtitles/` | 已覆盖 |
 | `tests/shared/srt_entries.py` | 按秒解析 SRT（测试公共设施） | `tests/shared/test_srt_entries/` | 已覆盖 |
 | `tests/shared/realdata_contract.py` | 真实数据契约与对齐量化 | `tests/shared/test_alignment/` | 已覆盖 |
 | `tests/shared/env_isolation.py` | 环境/临时目录隔离 | 被 `tests/app/test_config` 等间接覆盖 | 已覆盖（间接） |

docs/workflows.md

@@ -37,6 +37,26 @@ JSON 不支持注释，因此"参数理由"以节点 `params` 内 `_note_<参数
 等假名，真实短对话（そこ/やばい/ねえ/やだ/えへへ）天然不命中。仅 decode_full
 生效，learn-translate 等 VAD 链路不受影响。
 
+**重复伪影过滤**（2026-09）：whisper 在单个窗口内卡住重复时，会把一个单元写满整条
+cue（实测 30 秒、重复 74–446 次，如 `チン`×111）；这种正文会让下游 LLM 跟着重复、
+把输出预算耗在思考上而报结构错误（表现为 `translation alignment failed …
+Expecting value: line 1 column 1 (char 0)`），也会直接渲染成超长字幕行。whisper
+转录后按**模式判据**（无字符词表）**整条删除**（`remove_repetition_entries`）：
+展示时长 ≥15s 且同一 1–6 字单元连续重复 ≥6 次且覆盖正文 ≥70%；真实短促呻吟
+（`ぇ`×15、`ああああああ`）靠时长区分，零误删。VAD 开关都会生效。
+
+**静音段幻觉抑制**（2026-09）：decode_full 的固有副作用是"无语音段照样写字幕"——
+whisper 会在静音/音乐段输出 `こんにちは`、`おはようございます`、`東京都交通局8800形電車`
+这类短幻觉。实测这类幻觉与"呻吟间隙里的真实短台词"在 `no_speech_prob`、`avg_logprob`、
+silero VAD 与音频能量四个维度上**都不可分**（数据见 `data/experiments/regen_plan/REPORT.md`），
+所以走 faster-whisper 自带的 `hallucination_silence_threshold`（HST，需
+`word_timestamps=True`）：怀疑该段是幻觉时，跳过超过阈值的静音部分。learn-translate
+默认 `hallucination_silence_threshold=2.0` + `word_timestamps=true`：同一片头 120 秒
+无对话段由 15 条字幕降到 4 条（无套话幻觉残留），呻吟段基本保留；代价是转写约慢 1.8×。
+参数由工作流直接传给节点（`_guard_transcribe_params` 透传 `word_timestamps` /
+`hallucination_silence_threshold` / `no_speech_threshold` / `log_prob_threshold` /
+`compression_ratio_threshold`，未配置的键保持 faster-whisper 默认值）。
+
 调研过程与结论见 [调研-whisper漏句与decode_full验证.md](./调研-whisper漏句与decode_full验证.md)。
 
 ## 切换模型不改代码
@@ -57,7 +77,9 @@ JSON 不支持注释，因此"参数理由"以节点 `params` 内 `_note_<参数
 保留节点原始文件及 URI，把成品副本存入 `runs/<run_id>/finals/output-<编码别名>/`；
 时间戳固定取 run 创建时间，重复收尾覆盖相同路径，多个别名分目录避免冲突。
 复制先写同目录临时文件，再原子替换目标，失败不登记残缺文件；`final_outputs`
-声明的引用或文件缺失时任务失败，不能标完成。旧版本原文件已改名但最终别名记录
+声明的引用或文件缺失时任务失败，不能标完成。learn-translate 声明三份最终产物：
+`ja_srt`（asr 节点清洗后的日语转写，翻译前的归档）、`cn_srt`（中文译文）、`ass`（双目）。
+旧版本原文件已改名但最终别名记录
 仍指向有效文件时允许复用；原文件与成品都丢失时明确报错。批量场景下成品还会
 复制到视频旁，命名约定见 [operations.md](./operations.md#文件夹批量处理)。
 

nodes/subtitle_cleanup.py

@@ -10,7 +10,8 @@ ASR（whisper 无 VAD 解码）与 LLM 翻译在无语音段会输出固定套
 - JAPANESE_HALLUCINATION_TOKENS：日文（whisper 无 VAD 解码直接输出的套话）。
 
 删除只针对展示时长 ≥ 阈值的长条目：短时相同词可能是剧情真实台词（如真实
-互道晚安），必须保留。另外删除纯呻吟/喘息碎片（判据见 _is_pure_moan）。
+互道晚安），必须保留。另外删除纯呻吟/喘息碎片（判据见 _is_pure_moan）与
+ whisper 窗口内重复循环造成的重复伪影（判据见 _is_repetition_artifact）。
 
 纯函数，不修改输入。
 """
@@ -63,6 +64,56 @@ MOAN_CHARS = frozenset(
 # /んふふ）有效假名 ≤3；>3（如ああああ）或含非呻吟字符的一律保留。
 DEFAULT_MOAN_MAX_CHARS = 3
 
+# 重复伪影判据（whisper 循环解码会把一个单元写满整条 cue）：
+# 单元最长 6 字（更长的重复单元在真实数据里未见），重复段占正文 ≥70%。
+# 长度维度两档：长条（≥15s）重复 ≥6 次即删；短条则要求极端重复（≥20 次）——
+# 真实呻吟重复次数实测 ≤15（ぇ×15 / しゅ×7），而短时伪影实测 3.7 秒填了 111 次。
+REPETITION_UNIT_MAX_CHARS = 6
+REPETITION_MIN_REPEATS = 6
+REPETITION_HARD_REPEATS = 20
+REPETITION_MIN_COVERAGE = 0.7
+
+
+def _longest_repeated_run(text: str, unit_max: int) -> tuple[str, int] | None:
+    """返回正文中最长的'同一单元连续重复'片段（原文, 重复次数）。
+
+    单元长度 1..unit_max 由短到长尝试，取片段最长的一次；无重复返回 None。
+    注意用 finditer 而不是 re.search(..., pos)：模块级 re.search 的第三个位置
+    参数是 flags，拿它当偏移会导致原地重搜、死循环。
+    """
+    best: tuple[str, int] | None = None
+    for size in range(1, unit_max + 1):
+        for match in re.finditer(rf"(.{{1,{size}}})\1+", text):
+            run = match.group(0)
+            if best is None or len(run) > len(best[0]):
+                best = (run, len(run) // len(match.group(1)))
+    return best
+
+
+def _is_repetition_artifact(text: str, duration: float, threshold_seconds: float,
+                            min_repeats: int, min_coverage: float) -> bool:
+    """判断一条 cue 是否为 whisper 重复循环伪影（整条删除判据）。
+
+    同时满足：同一 1–6 字单元连续重复 ≥ min_repeats 次、重复段占正文
+    ≥ min_coverage，且**时长 ≥ 阈值或重复次数 ≥ REPETITION_HARD_REPEATS**。
+    长度维度分两档是必要的：只按"长条才删"会漏掉短时循环（实测 3.7 秒的 cue
+    被填了 111 个假名，翻译后变成 56 个"哈"进成品）；而真实呻吟的重复次数实测
+    ≤15（ぇ×15 占 3.2 秒），所以短条用"极端重复"判据即可区分，不会误删。
+    threshold_seconds ≤ 0 关闭过滤。
+    """
+    if threshold_seconds <= 0:
+        return False
+    stripped = text.replace("\n", "")
+    found = _longest_repeated_run(stripped, REPETITION_UNIT_MAX_CHARS)
+    if found is None:
+        return False
+    run, repeats = found
+    if not stripped or repeats < min_repeats:
+        return False
+    if len(run) / len(stripped) < min_coverage:
+        return False
+    return duration >= threshold_seconds or repeats >= REPETITION_HARD_REPEATS
+
 def _moan_chars(text: str) -> int:
     """返回 text 中'有效假名字符'数量（呻吟判据的一部分）。
 
@@ -141,6 +192,31 @@ def remove_short_moan_entries(
     return _remove_cues_by_predicate(srt_text, _keep)
 
 
+def remove_repetition_entries(
+    srt_text: str,
+    threshold_seconds: float = DEFAULT_THRESHOLD_SECONDS,
+    min_repeats: int = REPETITION_MIN_REPEATS,
+    min_coverage: float = REPETITION_MIN_COVERAGE,
+) -> str:
+    """删除 SRT 中 whisper 重复循环造成的**整条重复伪影**（ASR 产物去噪）。
+
+    模型在窗口内卡住重复时会把同一单元写满整条 cue（实测 30 秒、重复 74–446
+    次）：这种正文会让下游 LLM 跟着重复、把预算耗在思考上而报结构错误，也会
+    直接渲染成超长字幕行，因此在产生处整条删除。判据见 _is_repetition_artifact：
+    长条（时长 ≥ 阈值）或极端重复（≥ REPETITION_HARD_REPEATS 次）+ 同一短单元
+    占满正文；真实短促呻吟（重复 ≤15 次）不会命中。
+    threshold_seconds ≤ 0 时关闭过滤。纯函数，不修改输入。
+    """
+
+    def _keep(start_sec: float, end_sec: float, text: str) -> bool:
+        """保留判据：非重复伪影才保留。"""
+        return not _is_repetition_artifact(
+            text, end_sec - start_sec, threshold_seconds, min_repeats, min_coverage,
+        )
+
+    return _remove_cues_by_predicate(srt_text, _keep)
+
+
 def _remove_cues_by_predicate(
     srt_text: str,
     keep: callable,

nodes/whisper.py

@@ -170,6 +170,24 @@ def _wav_duration_seconds(path: Path, fallback: float) -> float:
         return fallback
 
 
+# 静音段幻觉抑制参数：无 VAD 的整段解码会在无语音处"编"出字幕，这些参数直接
+# 交给 faster-whisper（不传时保持其内置默认值，行为与改造前一致）。其中
+# `hallucination_silence_threshold` 需要 `word_timestamps=True` 才生效：它按词级
+# 时间戳跳过幻觉段里的静音部分，是"无语音段别写字幕"的主要开关。
+_GUARD_PARAM_KEYS = (
+    "word_timestamps",
+    "hallucination_silence_threshold",
+    "no_speech_threshold",
+    "log_prob_threshold",
+    "compression_ratio_threshold",
+)
+
+
+def _guard_transcribe_params(params: dict) -> dict:
+    """挑出工作流显式传入的幻觉抑制参数，未传的键交给 faster-whisper 默认值。"""
+    return {key: params[key] for key in _GUARD_PARAM_KEYS if params.get(key) is not None}
+
+
 def _append_srt_lines(lines: list[str], segments, offset: float, start_index: int) -> int:
     """把一段转写结果按 SRT 格式追加到 lines，时间加上 offset 偏移。
 
@@ -287,6 +305,8 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
                 condition_on_previous_text=bool(
                     request.params.get("condition_on_previous_text", False)
                 ),
+                # 静音段幻觉抑制（未配置时不传，保持默认行为）。
+                **_guard_transcribe_params(request.params),
             )
             # 进度日志：块序号/总数、单块耗时、实时倍率（块音频时长/墙钟耗时）
             # 与转写累计耗时，直观反映数据处理速度。
@@ -300,24 +320,24 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
                 chunk_seconds / chunk_elapsed if chunk_elapsed > 0 else 0.0,
                 time.monotonic() - transcribe_started,
             )
-        # decode_full（无 VAD）的副作用清理：无语音段的长时寒暄幻觉与纯语气词
+        # 噪声清理：重复伪影（窗口内卡住重复，VAD 开关都会有）整条删除；
-        # 碎片都是噪声，整条删除（序列号重排，不留 '-' 占位污染下游）；判据与
+        # decode_full 另加无语音段的长时寒暄幻觉与纯语气词碎片。都是整条删除、
-        # 细节见 nodes/subtitle_cleanup.py。仅 decode_full 时启用。
+        # 序号重排，不留 '-' 占位污染下游；判据见 nodes/subtitle_cleanup.py。
-        if decode_full:
+        from nodes.subtitle_cleanup import (
-            from nodes.subtitle_cleanup import (
+            clean_japanese_hallucinations,
-                clean_japanese_hallucinations,
+            remove_repetition_entries,
-                remove_short_moan_entries,
+            remove_short_moan_entries,
-            )
+        )
 
-            body = clean_japanese_hallucinations("\n".join(lines))
+        body = remove_repetition_entries("\n".join(lines))
+        if decode_full:
+            body = clean_japanese_hallucinations(body)
             # short_moan_max_chars：有效假名 ≤ 该值的纯呻吟碎片整条删除，
             # 设 0 关闭（真实短对话不会命中，判据见 subtitle_cleanup）。
             body = remove_short_moan_entries(
                 body,
                 max_chars=int(request.params.get("short_moan_max_chars", 3)),
             )
-        else:
-            body = "\n".join(lines)
         output_path = output_dir / "transcript.srt"
         output_path.write_text(body, encoding="utf-8")
         return InvokeResponse(status="completed", outputs={"srt_uri": str(output_path)})

scripts/plan_regenerate_subtitles.py

@@ -0,0 +1,320 @@
+"""统计媒体库中需要重新生成字幕的视频数量与时长，产出可复用的重生成计划。
+
+判定口径（`--before` 可调）：
+
+- 只看本流水线产出的 CN 字幕：`<视频名>.CN.srt` 与 `<视频名>.CN_dual_eye.ass`；
+- 生成时间取这些产物中**最早**的 mtime。双目 `.ass` 可能被样式统一脚本原地
+  改写而"变新"，中文字幕的 mtime 才反映真实生成时间，取最早值两者兼容；
+- 生成时间早于 `--before`（默认 2026-09-01，本地时区）即列为待重新生成；
+- 完全没有 CN 字幕的视频单独统计，不算"重生成"（那属于首次生成）。
+
+脚本不依赖仓库，可拷到媒体库所在主机直接跑（共享目录挂载方式下逐文件
+ffprobe 太慢）：目录内容只读，不改动任何文件；结果写 JSON 计划供批量重跑。
+
+用法：
+    python3 scripts/plan_regenerate_subtitles.py /mnt/fnOS/123
+    python3 scripts/plan_regenerate_subtitles.py /vol1/1000/123 --before 2026-09-01 \
+        --out data/experiments/regen_plan.json
+
+批量引擎看到视频旁已有字幕就会跳过该视频，重生成前需先改名备份旧字幕：
+
+    python3 scripts/plan_regenerate_subtitles.py /vol1/1000/123 --backup-old
+    python3 scripts/plan_regenerate_subtitles.py /vol1/1000/123 --backup-old --apply \
+        --select sample.txt
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import sys
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime
+from pathlib import Path
+
+# 与 src/wov_app/batch.py 的 VIDEO_EXTENSIONS / SUBTITLE_EXTENSIONS 保持一致，
+# 本脚本要在媒体库主机上独立运行，故不复用仓库常量。
+VIDEO_EXTENSIONS = {
+    ".mp4", ".mkv", ".avi", ".mov", ".webm", ".flv", ".ts",
+    ".m4v", ".wmv", ".mpg", ".mpeg", ".3gp",
+}
+SUBTITLE_EXTENSIONS = {".srt", ".ass", ".ssa", ".vtt"}
+
+# CN 产物识别：这两个后缀由本流水线生成（见 batch.py 的产物命名约定）；
+# 媒体库里其它 .srt 多为片源自带字幕，不参与"字幕生成时间"判定。
+CN_PRODUCT_MARKERS = (".cn.srt", ".cn_dual_eye.ass")
+
+PENDING = "pending"
+CURRENT = "current"
+MISSING = "missing"
+
+
+def iter_videos(root: Path) -> list[Path]:
+    """递归列出媒体库下全部视频文件，按路径排序保证结果可复现。"""
+    paths = [
+        p for p in root.rglob("*")
+        if p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS
+    ]
+    return sorted(paths)
+
+
+def cn_products(video: Path) -> list[Path]:
+    """列出视频旁由本流水线产出的 CN 字幕（同一目录、文件名含视频主名）。"""
+    stem = video.stem.lower()
+    found: list[Path] = []
+    for item in video.parent.iterdir():
+        if not item.is_file():
+            continue
+        name = item.name.lower()
+        if item.suffix.lower() not in SUBTITLE_EXTENSIONS:
+            continue
+        if stem in name and name.endswith(CN_PRODUCT_MARKERS):
+            found.append(item)
+    return sorted(found)
+
+
+def sidecar_subtitles(video: Path) -> list[Path]:
+    """列出批量引擎认定为“已有字幕”的旁挂字幕文件（与 batch.py 判定一致）。
+
+    同目录、字幕扩展名、文件名包含视频主名；主名只有单个字符时只接受
+    “主名.”前缀，避免 a.mp4 误配 apple.srt。
+    """
+    stem = video.stem.lower()
+    try:
+        siblings = list(video.parent.iterdir())
+    except OSError:
+        return []
+    found: list[Path] = []
+    for item in siblings:
+        try:
+            if not item.is_file() or item.suffix.lower() not in SUBTITLE_EXTENSIONS:
+                continue
+        except OSError:
+            continue
+        name = item.name.lower()
+        if len(stem) <= 1:
+            matched = name.startswith(stem + ".")
+        else:
+            matched = stem in name
+        if matched:
+            found.append(item)
+    return sorted(found)
+
+
+def read_select_list(path: Path) -> set[str]:
+    """读取 `--select` 文件里的视频标识（每行一个，支持行首或行尾 `#` 注释）。"""
+    keys: set[str] = set()
+    for line in Path(path).read_text(encoding="utf-8").splitlines():
+        text = line.split(" #", 1)[0].strip()
+        if text and not text.startswith("#"):
+            keys.add(text)
+    return keys
+
+
+def backup_old_subtitles(items: list[dict], suffix: str, apply: bool = False,
+                        select: set[str] | None = None) -> list[tuple[Path, Path]]:
+    """把待重生成视频的旁挂字幕原地改名，让批量引擎不再把该视频判为已有字幕。
+
+    改名目标为 `<原名><suffix>`；目标已存在（同一批重复执行）时跳过，不覆盖
+    旧备份。apply=False 只返回改名为计划，不动磁盘。select 按绝对路径、
+    相对路径或文件名筛选要处理的视频。
+    """
+    changed: list[tuple[Path, Path]] = []
+    for item in items:
+        video = Path(item["video"])
+        if select is not None:
+            keys = (str(video), item.get("rel"), video.name)
+            if not any(key in select for key in keys if key):
+                continue
+        for sub in sidecar_subtitles(video):
+            target = sub.with_name(sub.name + suffix)
+            if target.exists():
+                continue
+            if apply:
+                sub.rename(target)
+            changed.append((sub, target))
+    return changed
+
+
+def probe_duration(video: Path, timeout: float = 120.0) -> float | None:
+    """用 ffprobe 读取容器时长（秒）；失败返回 None，由调用方单独计数。"""
+    try:
+        proc = subprocess.run(
+            [
+                "ffprobe", "-v", "error",
+                "-show_entries", "format=duration",
+                "-of", "default=nw=1:nk=1", str(video),
+            ],
+            capture_output=True, text=True, timeout=timeout,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+    if proc.returncode != 0:
+        return None
+    try:
+        return float(proc.stdout.strip())
+    except ValueError:
+        return None
+
+
+def build_items(root: Path, workers: int, probe: bool = True) -> list[dict]:
+    """扫描全部视频，逐条产出路径、大小、时长与 CN 字幕生成时间。"""
+    videos = iter_videos(root)
+    durations: list[float | None] = [None] * len(videos)
+    if probe and videos:
+        with ThreadPoolExecutor(max_workers=workers) as pool:
+            durations = list(pool.map(probe_duration, videos))
+
+    items: list[dict] = []
+    for video, duration in zip(videos, durations):
+        products = cn_products(video)
+        mtimes = [p.stat().st_mtime for p in products]
+        items.append({
+            "video": str(video),
+            "rel": str(video.relative_to(root)),
+            "size": video.stat().st_size,
+            "duration": duration,
+            "generated_at": datetime.fromtimestamp(min(mtimes)).isoformat() if mtimes else None,
+            "cn_products": [
+                {"path": str(p), "mtime": datetime.fromtimestamp(p.stat().st_mtime).isoformat()}
+                for p in products
+            ],
+        })
+    return items
+
+
+def classify(items: list[dict], before: datetime) -> dict[str, list[dict]]:
+    """按 CN 字幕生成时间把视频分为待重生成/已最新/无 CN 字幕三组。"""
+    groups: dict[str, list[dict]] = {PENDING: [], CURRENT: [], MISSING: []}
+    for item in items:
+        if item["generated_at"] is None:
+            groups[MISSING].append(item)
+        elif datetime.fromisoformat(item["generated_at"]) < before:
+            groups[PENDING].append(item)
+        else:
+            groups[CURRENT].append(item)
+    return groups
+
+
+def _agg(items: list[dict]) -> dict:
+    """汇总一组的数量、总时长、总大小，时长缺失的视频单独计数。"""
+    return {
+        "count": len(items),
+        "seconds": sum(i["duration"] for i in items if i["duration"] is not None),
+        "bytes": sum(i["size"] for i in items),
+        "no_duration": sum(1 for i in items if i["duration"] is None),
+    }
+
+
+def _fmt_hours(seconds: float) -> str:
+    return f"{seconds / 3600:.2f} 小时"
+
+
+def _fmt_duration(probed: bool, seconds: float) -> str:
+    """未探测时长时（--backup-old 等只统计数量的场景）不显示 0 小时。"""
+    return _fmt_hours(seconds) if probed else "未探测时长"
+
+
+def _fmt_size(num_bytes: int) -> str:
+    return f"{num_bytes / (1024 ** 3):.1f} GiB"
+
+
+def format_summary(root: Path, items: list[dict], groups: dict[str, list[dict]],
+                   before: datetime) -> str:
+    """生成人类可读的统计报告（数量 / 时长 / 体积 / 按月分布）。"""
+    probed = any(i["duration"] is not None for i in items)
+    lines = [
+        f"媒体库: {root}",
+        f"视频总数: {len(items)} 个 / {_fmt_duration(probed, _agg(items)['seconds'])} / "
+        f"{_fmt_size(_agg(items)['bytes'])}",
+        "",
+        f"待重新生成（CN 字幕早于 {before:%Y-%m-%d}）: ",
+    ]
+    pending = _agg(groups[PENDING])
+    duration_text = _fmt_duration(probed, pending["seconds"])
+    lines.append(f"  {pending['count']} 个 / {duration_text} / {_fmt_size(pending['bytes'])}")
+
+    by_month: dict[str, dict] = {}
+    for item in groups[PENDING]:
+        month = item["generated_at"][:7]
+        bucket = by_month.setdefault(month, {"count": 0, "seconds": 0.0})
+        bucket["count"] += 1
+        bucket["seconds"] += item["duration"] or 0.0
+    for month in sorted(by_month):
+        bucket = by_month[month]
+        lines.append(f"    {month}: {bucket['count']} 个 / "
+                     f"{_fmt_duration(probed, bucket['seconds'])}")
+
+    current = _agg(groups[CURRENT])
+    lines.append("")
+    lines.append(f"已是最新（生成时间 >= {before:%Y-%m-%d}）: {current['count']} 个 / "
+                 f"{_fmt_duration(probed, current['seconds'])} / {_fmt_size(current['bytes'])}")
+    missing = _agg(groups[MISSING])
+    lines.append(f"无 CN 字幕（首次生成）: {missing['count']} 个 / "
+                 f"{_fmt_duration(probed, missing['seconds'])} / {_fmt_size(missing['bytes'])}")
+
+    no_duration = sum(1 for i in items if i["duration"] is None)
+    if no_duration and probed:
+        lines.append(f"警告: {no_duration} 个视频未能读出时长，未计入时长合计")
+    return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="统计需要重新生成字幕的视频数量与时长")
+    parser.add_argument("root", help="媒体库根目录")
+    parser.add_argument("--before", default="2026-09-01",
+                        help="重生成分界线，早于该日期生成的 CN 字幕待重跑（默认 2026-09-01）")
+    parser.add_argument("--out", help="计划 JSON 输出路径（默认只打印统计）")
+    parser.add_argument("--workers", type=int, default=8, help="ffprobe 并发数（默认 8）")
+    parser.add_argument("--no-probe", action="store_true", help="跳过时长探测（只统计数量）")
+    parser.add_argument("--list-pending", action="store_true", help="额外打印待重生成的视频路径")
+    parser.add_argument("--backup-old", action="store_true",
+                        help="把待重生成视频的旁挂字幕改名备份（默认预览，需 --apply 落盘）")
+    parser.add_argument("--apply", action="store_true", help="与 --backup-old 一起用时真正改名")
+    parser.add_argument("--select", help="只处理该文件列出的视频（每行一个路径或文件名）")
+    parser.add_argument("--suffix", help="备份后缀，默认 .old-<当天日期>")
+    args = parser.parse_args(argv)
+
+    root = Path(args.root).expanduser().resolve()
+    if not root.is_dir():
+        print(f"目录不存在: {root}", file=sys.stderr)
+        return 2
+    before = datetime.fromisoformat(args.before)
+
+    need_duration = args.out is not None or args.list_pending or not args.backup_old
+    items = build_items(root, max(1, args.workers), probe=need_duration and not args.no_probe)
+    groups = classify(items, before)
+    print(format_summary(root, items, groups, before))
+    if args.list_pending:
+        print("\n待重新生成的视频:")
+        for item in groups[PENDING]:
+            print(f"  {item['video']}")
+
+    if args.backup_old:
+        suffix = args.suffix or f".old-{datetime.now():%Y%m%d}"
+        select = read_select_list(Path(args.select)) if args.select else None
+        changed = backup_old_subtitles(groups[PENDING], suffix, apply=args.apply, select=select)
+        verb = "已改名" if args.apply else "待改名（预览，未落盘）"
+        print(f"\n旧字幕备份（后缀 {suffix}）: {verb} {len(changed)} 个文件")
+        for source, target in changed:
+            print(f"  {source} -> {target.name}")
+
+    if args.out:
+        out = Path(args.out).expanduser()
+        out.parent.mkdir(parents=True, exist_ok=True)
+        status_of = {id(i): name for name, group in groups.items() for i in group}
+        payload = {
+            "root": str(root),
+            "before": before.isoformat(),
+            "scanned_at": datetime.now().isoformat(),
+            "totals": {name: _agg(group) for name, group in groups.items()},
+            "items": [{**i, "status": status_of[id(i)]} for i in items],
+        }
+        out.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8")
+        print(f"\n计划 JSON 已写入: {out}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/wov_app/batch.py

@@ -11,15 +11,19 @@
   字幕文件（`.srt/.ass/.ssa/.vtt`），说明该视频已有字幕，直接记为 SKIPPED，
   不为它触发任何流水线。运行时（BatchWorker）只消费已定位好的明细列表，
   **不再重新扫描文件夹**（运行期间新增/删除的视频不会改变本次任务的范围）。
-- **分块流水线执行**：视频按 `WOV_BATCH_STAGE_GROUP_SIZE` 分组，组内按节点
+- **按资源调度的在途流水线**：视频按 `WOV_BATCH_STAGE_GROUP_SIZE` 分组，组内
-  顺序跑完全部视频（先全部 extract、再全部 ASR、再全部 LLM 翻译、最后 ASS）
+  每个视频独立推进自己的阶段（最多 `WOV_BATCH_PIPELINE_WORKERS` 个阶段在途）。
-  再进入下一组——本地模型每组只加载一次、卸载一次，产物按组增量落地。
+  阶段是否需要 GPU 由 `wov_app.resources` 判定：提音、线上翻译与 ASS 不需要，
+  与其它视频的转写并行（GPU 不再空转）；需要 GPU 的阶段由进程内门控
+  `GpuGate` 串行准入，并按"阶段索引最小者优先"派发，于是组内先跑完全部转写
+  再进翻译——本机 Ollama 模型每组只加载一次（线上端点则完全不受该顺序约束）。
   阶段边界用 `execute_run(stop_after=节点)` 停在节点（任务保持 RUNNING），
-  LLM 阶段靠 `keep_model.flag` 让节点保持模型常驻，阶段结束由引擎统一释放
+  LLM 阶段靠 `keep_model.flag` 让节点保持模型常驻，组末由引擎统一释放显存
-  显存（详见 docs/operations.md#文件夹批量处理）。
+  （详见 docs/operations.md#文件夹批量处理）。
 - **产物放在视频旁**：每个视频处理完成后，把工作流 `final_outputs` 对应的
-  最终产物文件（字幕流水线即中文 `.srt` 与双目 `.ass`）**复制一份到视频的
+  最终产物文件（字幕流水线即日语转写 `.srt`、中文 `.srt` 与双目 `.ass`）
-  所在目录**，与 .mp4 放在一起；文件名**对齐媒体库既有约定**：中文字幕存为
+  **复制一份到视频的所在目录**，与 .mp4 放在一起；文件名按 `final_outputs`
+  别名**对齐媒体库既有约定**：日语转写存为 `<视频名>.JA.srt`、中文字幕存为
   `<视频名>.CN.srt`、双目字幕存为 `<视频名>.CN_dual_eye.ass`（文件名稳定且
   含视频主名，媒体库可自动匹配，下次批量扫描也会命中"已有字幕"规则跳过）。
 - **过程文件清理**：视频收尾完成后删除该视频的整个工作空间与 run 记录，
@@ -46,13 +50,20 @@ import shutil
 import threading
 import time
 import uuid
+from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
 from datetime import datetime, timezone
 from pathlib import Path
 
 from wov_app import registry
-from wov_app.config import BATCH_INTERVAL_SECONDS, BATCH_STAGE_GROUP_SIZE, STORAGE_DIR
+from wov_app.config import (
+    BATCH_INTERVAL_SECONDS,
+    BATCH_PIPELINE_WORKERS,
+    BATCH_STAGE_GROUP_SIZE,
+    STORAGE_DIR,
+)
 from wov_app.db import Database
 from wov_app.logging import get_logger
+from wov_app.resources import GpuGate, stage_gpu_need_mb
 from wov_app.scheduler import WorkflowScheduler, topological_sort
 from wov_app.storage import atomic_copy
 from wov_sdk.models import WorkflowDefinition, WorkflowNode
@@ -72,6 +83,14 @@ VIDEO_EXTENSIONS = {
 # 最终产物（.srt/.ass）也在该集合内，保证下次扫描能命中同一规则直接跳过。
 SUBTITLE_EXTENSIONS = {".srt", ".ass", ".ssa", ".vtt"}
 
+# 组内流水线的轮询间隔（秒）：等第一个阶段结束时顺带检查暂停与资源放行。
+PIPELINE_POLL_SECONDS = 0.2
+
+
+def _make_gate() -> GpuGate:
+    """创建任务级 GPU 门控（测试通过替换本函数注入假探测结果）。"""
+    return GpuGate()
+
 # 暂停信号文件名：与节点约定一致，位于 run 根目录（<work_dir>/runs/<run_id>/）。
 PAUSE_FLAG = "paused.flag"
 
@@ -149,16 +168,28 @@ def list_sidecar_subtitles(video: Path) -> list[Path]:
     return sorted(found)
 
 
-def _sidecar_product_name(video: Path, source: Path) -> str:
+# final_outputs 别名 → 视频旁文件名后缀：日语转写与中文译文同为 `.srt`，
+# 只按扩展名映射会让两份产物撞名互相覆盖，必须按别名区分语言。
+_PRODUCT_SUFFIXES = {
+    "ja_srt": ".JA.srt",
+    "cn_srt": ".CN.srt",
+    "ass": ".CN_dual_eye.ass",
+}
+
+
+def _sidecar_product_name(video: Path, source: Path, alias: str | None = None) -> str:
     """把最终产物映射为放在视频旁时的标准字幕文件名。
 
     对齐媒体库既有约定（文件名稳定、无时间戳，媒体库可按视频主名自动匹配）：
+    - 日语转写（`ja_srt`）→ `<视频名>.JA.srt`；
     - 中文 `.srt` 产物 → `<视频名>.CN.srt`；
     - 双目 `.ass` 产物 → `<视频名>.CN_dual_eye.ass`；
     - 其余扩展名的最终产物保留原文件名（含时间戳），避免误改语义。
-    注：当前内置字幕工作流（learn-translate / zh-direct）在视频旁放置的 `.srt/.ass` 即
+
-    中文字幕与双目字幕；若未来出现"非中文 .srt"类产物需在此按产物来源区分。
+    别名来自 `final_outputs` 的键，未登记别名时按扩展名兜底（历史工作流兼容）。
     """
+    if alias in _PRODUCT_SUFFIXES:
+        return f"{video.stem}{_PRODUCT_SUFFIXES[alias]}"
     suffix = source.suffix.lower()
     if suffix == ".srt":
         return f"{video.stem}.CN.srt"
@@ -221,7 +252,10 @@ def create_job(
         "folder_path": str(folder),
         "workflow_id": workflow_id,
         "recursive": int(recursive),
-        "status": "QUEUED",
+        # CREATING：明细未写完前引擎看不见本任务。逐条登记 500+ 条明细要数秒，
+        # 若此刻已是 QUEUED，引擎会读到半个快照并在收尾时把任务误标 COMPLETED，
+        # 剩下的视频就再也不会被处理（自愈分支只碰运气）。写完明细立刻置 QUEUED。
+        "status": "CREATING",
         "progress": 0,
         "total": 0,
         "done": 0,
@@ -233,27 +267,35 @@ def create_job(
     })
     pending = 0
     skipped = 0
-    for video in videos:
+    try:
-        # 视频所在目录已存在对应字幕文件 → 已处理过，直接跳过不触发流水线。
+        for video in videos:
-        if list_sidecar_subtitles(video):
+            # 视频所在目录已存在对应字幕文件 → 已处理过，直接跳过不触发流水线。
-            status = "SKIPPED"
+            if list_sidecar_subtitles(video):
-            skipped += 1
+                status = "SKIPPED"
-        else:
+                skipped += 1
-            status = "PENDING"
+            else:
-            pending += 1
+                status = "PENDING"
-        video_id = f"bv_{uuid.uuid4().hex[:12]}"
+                pending += 1
-        db.create_batch_video({
+            video_id = f"bv_{uuid.uuid4().hex[:12]}"
-            "id": video_id,
+            db.create_batch_video({
-            "job_id": job_id,
+                "id": video_id,
-            "video_path": str(video),
+                "job_id": job_id,
-            # 私有工作空间：storage/batch/<job_id>/<bv_id>/，与媒体库隔离。
+                "video_path": str(video),
-            "work_dir": str(BATCH_WORK_ROOT / job_id / video_id),
+                # 私有工作空间：storage/batch/<job_id>/<bv_id>/，与媒体库隔离。
-            "run_id": None,
+                "work_dir": str(BATCH_WORK_ROOT / job_id / video_id),
-            "status": status,
+                "run_id": None,
-            "error": None,
+                "status": status,
-            "created_at": now,
+                "error": None,
-            "updated_at": now,
+                "created_at": now,
-        })
+                "updated_at": now,
+            })
+    except Exception as exc:  # noqa: BLE001
+        # 明细写到一半失败：记 FAILED 留可见记录（CREATING 状态没人会拾起，
+        # 沉默的残留任务会让用户以为什么都没发生），然后把异常交给路由层。
+        db.update_batch_job(
+            job_id, status="FAILED", error=str(exc), updated_at=_now_iso(),
+        )
+        raise
     # total = 本批真正需要处理（无字幕）的视频数；已有字幕被 SKIPPED 的
     # 不计入总数也不计入完成数——进度条只反映"实际待处理"的这批。
     if pending == 0:
@@ -263,7 +305,10 @@ def create_job(
             updated_at=_now_iso(),
         )
     else:
-        db.update_batch_job(job_id, total=pending, updated_at=_now_iso())
+        # 明细全部就位后才排队，引擎从此拿到的快照一定是完整的。
+        db.update_batch_job(
+            job_id, status="QUEUED", total=pending, updated_at=_now_iso(),
+        )
     logger.info(
         "创建批量任务 %s: 文件夹 %s, 工作流 %s, 共 %d 个视频（%d 待处理, %d 已有字幕跳过）",
         job_id, folder, workflow_id, len(videos), pending, skipped,
@@ -291,6 +336,11 @@ class BatchWorker:
             return
         # 独立运行时确保节点已注册；重复注册幂等。
         registry.register_all()
+        # 上一进程中断留下的 CREATING 任务（明细登记中途被杀/热重载）没人会推进，
+        # 启动时统一记为 FAILED，避免用户以为任务还在创建中。
+        stale = self.db.fail_creating_batch_jobs("创建明细中断（进程中断），请重新创建任务")
+        if stale:
+            logger.warning("启动清理 %d 个未登记完的批量任务（CREATING → FAILED）", stale)
         self._stopping = False
         self._thread = threading.Thread(
             target=self._loop,
@@ -389,44 +439,23 @@ class BatchWorker:
         total = int(job["total"] or 0)
 
         group_size = max(1, int(BATCH_STAGE_GROUP_SIZE))
+        gate = _make_gate()
         for start in range(0, len(items), group_size):
             group = items[start:start + group_size]
-            for stage_index, node_id in enumerate(order):
+            outcome = self._run_group(
-                node_spec = node_by_id[node_id]
+                job=job,
-                # 末阶段不传 stop_after：让调度器收尾（final_outputs + COMPLETED）。
+                group=group,
-                is_last_stage = stage_index == len(order) - 1
+                order=order,
-                executed = False
+                node_by_id=node_by_id,
-                for item in group:
+                definition=definition,
-                    # 暂停检查：批量任务被暂停后停止处理后续视频，等待用户继续。
+                version=version,
-                    current = self.db.get_batch_job(job_id)
+                gate=gate,
-                    if current is None or current["status"] == "PAUSED":
+            )
-                        # 停下前先把已完成/失败项入账，让暂停中的前端看到真实进度。
+            if outcome == "PAUSED":
-                        self.db.sync_batch_job_progress(job_id)
+                # 停下前先把已完成/失败项入账，让暂停中的前端看到真实进度。
-                        logger.info("批量任务 %s 已暂停，停止在视频 %s", job_id, item["video_path"])
+                self.db.sync_batch_job_progress(job_id)
-                        return
+                logger.info("批量任务 %s 已暂停，等待用户继续", job_id)
-                    self.db.update_batch_job(job_id, current_video=str(item["video_path"]), updated_at=_now_iso())
+                return
-                    try:
-                        outcome = self._run_stage(
-                            job, item, version, definition, node_spec, order, is_last_stage,
-                        )
-                    except Exception as exc:  # noqa: BLE001
-                        # 单视频兜底：不中断整个批量任务，记录错误后继续下一个视频。
-                        logger.exception(
-                            "批量任务 %s 视频 %s 阶段 %s 处理异常", job_id, item["video_path"], node_id,
-                        )
-                        self.db.update_batch_video(item["id"], status="FAILED", error=str(exc), updated_at=_now_iso())
-                        outcome = "FAILED"
-                    executed = executed or outcome is not None
-                    # 每个视频每个阶段后实时同步一次汇总，让进度尽快入账。
-                    self.db.sync_batch_job_progress(job_id)
-                    # 阶段内被暂停（节点内的 paused.flag）：任务保持 PAUSED 等续跑。
-                    if outcome == "PAUSED":
-                        self.db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso())
-                        return
-                # 阶段收尾：LLM 阶段结束时统一释放本地模型显存，让下一组的
-                # whisper（ASR）拿到 GPU，否则下一个视频转写会 CUDA OOM。
-                if executed and node_spec.node_type.startswith(LLM_NODE_PREFIX):
-                    self._release_llm_model(node_spec.params)
 
         # 先按明细实时对齐汇总（done 不计 SKIPPED），再判断能否收尾。
         # 仍有未结束视频时不能标 COMPLETED，否则会出现“还有待处理视频却已完成”
@@ -465,6 +494,203 @@ class BatchWorker:
             job_id, total, done, failed,
         )
 
+    def _is_job_paused(self, job_id: str) -> bool:
+        """批量任务是否已被暂停（或记录已消失）：暂停后不再派发新阶段。"""
+        job = self.db.get_batch_job(job_id)
+        return job is None or job["status"] == "PAUSED"
+
+    def _next_stage_node(
+        self,
+        pipeline: dict,
+        order: list[str],
+        node_by_id: dict[str, WorkflowNode],
+    ) -> WorkflowNode | None:
+        """返回该视频下一个待执行阶段的节点；已跑完或已失败时返回 None。"""
+        if pipeline["state"] != "ready" or pipeline["stage"] >= len(order):
+            return None
+        return node_by_id[order[pipeline["stage"]]]
+
+    def _submit_group_stage(
+        self,
+        job: dict,
+        pipeline: dict,
+        node_spec: WorkflowNode,
+        order: list[str],
+        definition: WorkflowDefinition,
+        version: dict,
+        running: dict,
+        pool: ThreadPoolExecutor,
+        lease_key: str | None = None,
+    ) -> None:
+        """把一个阶段交给线程池执行（异常在池内兜底，不让线程池任务抛出去）。"""
+        item = pipeline["item"]
+        is_last_stage = node_spec.id == order[-1]
+        # 标记在途：同一视频同一阶段只允许有一个执行体，否则会被重复派发。
+        pipeline["state"] = "running"
+        if lease_key:
+            pipeline["lease_key"] = lease_key
+        if node_spec.node_type.startswith(LLM_NODE_PREFIX):
+            pipeline["executed_llm"] = True
+            pipeline["llm_params"] = node_spec.params
+        self.db.update_batch_job(job["id"], current_video=str(item["video_path"]), updated_at=_now_iso())
+        logger.info(
+            "批量任务 %s 视频 %s 进入阶段 %s（在途 %d）",
+            job["id"], Path(item["video_path"]).name, node_spec.id, len(running) + 1,
+        )
+
+        def _worker() -> str:
+            """线程内的阶段执行：单视频异常不中断整组。"""
+            try:
+                outcome = self._run_stage(
+                    job, item, version, definition, node_spec, order, is_last_stage,
+                )
+            except Exception as exc:  # noqa: BLE001
+                logger.exception(
+                    "批量任务 %s 视频 %s 阶段 %s 处理异常",
+                    job["id"], item["video_path"], node_spec.id,
+                )
+                self.db.update_batch_video(
+                    item["id"], status="FAILED", error=str(exc), updated_at=_now_iso(),
+                )
+                return "FAILED"
+            return outcome or "RUNNING"
+
+        running[pool.submit(_worker)] = pipeline
+
+    def _dispatch_group_stages(
+        self,
+        job: dict,
+        pipelines: list[dict],
+        order: list[str],
+        node_by_id: dict[str, WorkflowNode],
+        definition: WorkflowDefinition,
+        version: dict,
+        gate: GpuGate,
+        running: dict,
+        pool: ThreadPoolExecutor,
+        workers: int,
+    ) -> None:
+        """派发就绪阶段：非 GPU 阶段可并行，GPU 阶段互斥且按上游优先。"""
+        # 1) 不需要 GPU 的阶段（提音 / 线上翻译 / ASS）：立刻派发，与其它视频的
+        # 转写并行，GPU 不再空转。
+        for pipeline in pipelines:
+            if len(running) >= workers:
+                break
+            node_spec = self._next_stage_node(pipeline, order, node_by_id)
+            if node_spec is None:
+                continue
+            if stage_gpu_need_mb(node_spec.node_type, node_spec.params) > 0:
+                continue
+            self._submit_group_stage(job, pipeline, node_spec, order, definition, version, running, pool)
+        # 2) 需要 GPU 的阶段：一次只跑一个，且选"阶段索引最小"的视频——组内因此
+        # 先把转写跑完再进翻译，本机 Ollama 模型仍每组只加载一次。
+        if len(running) >= workers or gate.holder is not None:
+            return
+        candidates = [
+            (pipeline["stage"], pipeline["index"], pipeline)
+            for pipeline in pipelines
+            if self._next_stage_node(pipeline, order, node_by_id) is not None
+        ]
+        gpu_candidates = [
+            entry for entry in candidates
+            if stage_gpu_need_mb(
+                node_by_id[order[entry[2]["stage"]]].node_type,
+                node_by_id[order[entry[2]["stage"]]].params,
+            ) > 0
+        ]
+        if not gpu_candidates:
+            return
+        _, _, pipeline = min(gpu_candidates, key=lambda entry: (entry[0], entry[1]))
+        node_spec = node_by_id[order[pipeline["stage"]]]
+        need = stage_gpu_need_mb(node_spec.node_type, node_spec.params)
+        lease_key = f"{pipeline['item']['id']}:{node_spec.id}"
+        # 显存/在途不满足时本轮跳过，等其它阶段释放后再试（不阻塞派发线程）。
+        if not gate.try_acquire(lease_key, need):
+            return
+        self._submit_group_stage(
+            job, pipeline, node_spec, order, definition, version, running, pool, lease_key,
+        )
+
+    def _finish_group_stage(
+        self,
+        job: dict,
+        pipeline: dict,
+        future: Future,
+        order: list[str],
+    ) -> None:
+        """收集一个阶段的执行结果并推进该视频的流水线。"""
+        try:
+            outcome = future.result()
+        except Exception:  # noqa: BLE001 - 池内已兜底，这里只保证组不被带崩
+            logger.exception("批量任务 %s 阶段执行线程异常", job["id"])
+            pipeline["state"] = "failed"
+            return
+        self.db.sync_batch_job_progress(job["id"])
+        if outcome == "PAUSED":
+            # 节点在边界（分块/批次/帧）停下：任务保持 PAUSED 等用户继续。
+            pipeline["state"] = "paused"
+            self.db.update_batch_job(job["id"], status="PAUSED", updated_at=_now_iso())
+            return
+        if outcome == "FAILED":
+            pipeline["state"] = "failed"
+            return
+        pipeline["stage"] += 1
+        pipeline["state"] = "done" if pipeline["stage"] >= len(order) else "ready"
+
+    def _run_group(
+        self,
+        job: dict,
+        group: list[dict],
+        order: list[str],
+        node_by_id: dict[str, WorkflowNode],
+        definition: WorkflowDefinition,
+        version: dict,
+        gate: GpuGate,
+    ) -> str:
+        """组内在途流水线：每个视频独立推进阶段，GPU 阶段按上游优先串行。
+
+        阶段是否需要 GPU 由 resources.stage_gpu_need_mb 判定（提音/线上翻译/ASS
+        不需要），于是它们与其它视频的转写并行；需要 GPU 的阶段由 GpuGate 准入，
+        并按"阶段索引最小者优先"派发，组内因此先把转写跑完再进翻译，本机 Ollama
+        模型仍每组只加载一次。
+
+        返回 "PAUSED" 表示组内被暂停（调用方停止任务），其余情况返回 "DONE"。
+        """
+        job_id = job["id"]
+        workers = max(1, int(BATCH_PIPELINE_WORKERS))
+        pipelines = [
+            {"item": item, "stage": 0, "index": index, "state": "ready", "executed_llm": False}
+            for index, item in enumerate(group)
+        ]
+        with ThreadPoolExecutor(max_workers=workers) as pool:
+            running: dict[Future, dict] = {}
+            while True:
+                paused = self._is_job_paused(job_id)
+                if not paused:
+                    self._dispatch_group_stages(
+                        job, pipelines, order, node_by_id, definition, version,
+                        gate, running, pool, workers,
+                    )
+                if not running:
+                    break
+                done, _ = wait(
+                    list(running), timeout=PIPELINE_POLL_SECONDS,
+                    return_when=FIRST_COMPLETED,
+                )
+                for future in done:
+                    pipeline = running.pop(future)
+                    lease_key = pipeline.pop("lease_key", None)
+                    if lease_key:
+                        gate.release(lease_key)
+                    self._finish_group_stage(job, pipeline, future, order)
+                if paused and not running:
+                    return "PAUSED"
+        # 组末统一释放本地模型：否则下一组的 whisper 转写会 CUDA OOM。
+        llm_params = next((p["llm_params"] for p in pipelines if p.get("llm_params")), None)
+        if llm_params is not None:
+            self._release_llm_model(llm_params)
+        return "PAUSED" if self._is_job_paused(job_id) else "DONE"
+
     def _run_stage(
         self,
         job: dict,
@@ -581,8 +807,8 @@ class BatchWorker:
     def _place_products(self, run_id: str, video: Path, definition: WorkflowDefinition) -> list[str]:
         """把最终产物文件复制到视频所在目录（视频旁），返回放置的文件名。
 
-        只为 `final_outputs` 声明的最终产物放置副本：字幕流水线的产物即中文
+        只为 `final_outputs` 声明的最终产物放置副本：字幕流水线的产物即日语
-        `.srt` 与双目 `.ass`，按库内约定命名（见 _sidecar_product_name），
+        转写 `.srt`、中文 `.srt` 与双目 `.ass`，按库内约定命名（见 _sidecar_product_name），
         文件名稳定且含视频主名——媒体库按主名匹配字幕，下次批量扫描也会命中
         "已有字幕"规则跳过该视频。同名目标直接覆盖：可能是上一次运行/旧工作流
         留下的旧内容，应以本次产物为准。
@@ -596,7 +822,7 @@ class BatchWorker:
             source = Path(artifact["uri"])
             if not source.is_file():
                 raise ValueError(f"missing final artifact file: {alias} ({source})")
-            target = video.parent / _sidecar_product_name(video, source)
+            target = video.parent / _sidecar_product_name(video, source, alias)
             products.append((source, target))
         placed: list[str] = []
         for source, target in products:

src/wov_app/config.py

@@ -17,6 +17,10 @@ DATA_DIR = Path(os.getenv("WOV_DATA_DIR", str(WORKSPACE_ROOT / "data")))
 DB_PATH = Path(os.getenv("WOV_DB_PATH", str(DATA_DIR / "wov.db")))
 STORAGE_DIR = Path(os.getenv("WOV_STORAGE_DIR", str(DATA_DIR / "storage")))
 
+# SQLite 写锁等待时长（秒）：批量流水线多线程并发写产物/进度，短时竞争要排队
+# 而不是立刻抛 database is locked。
+DB_BUSY_TIMEOUT_SECONDS = float(os.getenv("WOV_DB_BUSY_TIMEOUT_SECONDS", "30"))
+
 # 调度器轮询排队任务的间隔（秒）。
 SCHEDULER_INTERVAL_SECONDS = float(os.getenv("WOV_SCHEDULER_INTERVAL_SECONDS", "1.0"))
 
@@ -25,6 +29,10 @@ SCHEDULER_INTERVAL_SECONDS = float(os.getenv("WOV_SCHEDULER_INTERVAL_SECONDS", "
 BATCH_ENABLED = os.getenv("WOV_BATCH_ENABLED", "1") == "1"
 BATCH_INTERVAL_SECONDS = float(os.getenv("WOV_BATCH_INTERVAL_SECONDS", "1.0"))
 
+# 批量流水线的在途阶段数：组内每个视频独立推进，非 GPU 阶段（提音、线上翻译、
+# ASS）并行执行；GPU 阶段仍由资源门控串行化（见 docs/operations.md）。
+BATCH_PIPELINE_WORKERS = int(os.getenv("WOV_BATCH_PIPELINE_WORKERS", "4"))
+
 # 批量"分块流水线"分组大小：每组视频按节点顺序跑完全部阶段（全部 extract → 全部
 # ASR → 全部翻译 → 全部 ASS）再处理下一组，使本地模型每组只加载一次；产物仍按
 # 组增量落地（详见 docs/operations.md#文件夹批量处理）。

src/wov_app/db.py

@@ -13,6 +13,8 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Iterator
 
+from wov_app.config import DB_BUSY_TIMEOUT_SECONDS
+
 
 def _now_iso() -> str:
     """返回当前 UTC 时间的 ISO 格式字符串（与批量引擎的时间戳一致）。"""
@@ -30,12 +32,19 @@ class Database:
 
     @contextmanager
     def _connect(self) -> Iterator[sqlite3.Connection]:
-        """提供带事务提交的数据库连接上下文。"""
+        """提供带事务提交的数据库连接上下文。
-        conn = sqlite3.connect(self.path)
+
+        批量流水线会在多个线程里并发写产物与进度：开 WAL 让读写并行，busy
+        timeout 让短时写锁竞争排队等待，而不是直接抛 database is locked。
+        """
+        conn = sqlite3.connect(self.path, timeout=DB_BUSY_TIMEOUT_SECONDS)
         # 按列名读取结果，返回 dict 更直观。
         conn.row_factory = sqlite3.Row
         # 开启外键约束，保证子表记录引用有效。
         conn.execute("PRAGMA foreign_keys = ON")
+        # WAL 是持久设置，重复执行无副作用；不支持 WAL 的文件系统会保持原模式。
+        conn.execute("PRAGMA journal_mode = WAL")
+        conn.execute("PRAGMA synchronous = NORMAL")
         try:
             yield conn
             conn.commit()
@@ -590,6 +599,20 @@ class Database:
             conn.execute(f"UPDATE batch_jobs SET {assignments} WHERE id = ?", values)
 
 
+    def fail_creating_batch_jobs(self, reason: str) -> int:
+        """把停留在 CREATING 的批量任务置 FAILED，返回处理条数。
+
+        CREATING 只存在于 create_job 逐条登记明细期间；进程被杀或热重载后没有任何
+        线程会推进它（引擎只取 QUEUED），启动时统一收尾成可见的失败记录。
+        """
+        with self._connect() as conn:
+            cursor = conn.execute(
+                "UPDATE batch_jobs SET status = 'FAILED', error = ?, updated_at = ? "
+                "WHERE status = 'CREATING'",
+                (reason, _now_iso()),
+            )
+            return int(cursor.rowcount or 0)
+
     def sync_batch_job_progress(self, job_id: str) -> None:
         """按视频明细实时对齐任务的 total/done/failed 汇总并落库。
 

src/wov_app/resources.py

@@ -0,0 +1,217 @@
+"""GPU 资源门控：按"当前可用显存 + 在途申请"决定阶段能否启动。
+
+批量流水线里各阶段对 GPU 的需求不同：`faster-whisper` 必须独占显存；走本机
+Ollama 端点的 `llm-*`/OCR 阶段（模型常驻显存）同样必须独占；而**线上端点**的
+`llm-translate`、提音与 ASS 合成都不占显存。用资源现状判定，就不需要在代码里
+按"是否走云端"分叉：
+
+- 显存不够或已有 GPU 阶段在跑 → 后续 GPU 阶段排队等待（与串行等价，不抢显存）；
+- 不需要 GPU → 立刻放行，于是转写能与线上翻译并行，GPU 不再空转。
+
+探测不到 `nvidia-smi`（纯 CPU 机器、无 GPU 容器）时无法核对显存，退化为
+"GPU 阶段互斥"这一保守规则：仍不并发，但与串行结果一致。
+
+本模块只做准入判定与需求估算，不加载/卸载任何模型：显存的实际释放由节点与
+引擎负责（本地模型常驻的让位见 docs/operations.md#文件夹批量处理）。
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import socket
+import subprocess
+import threading
+import urllib.parse
+from pathlib import Path
+
+# 安全余量：显存刚好"够"时仍可能因分配碎片化抛 CUDA OOM，留 800MB 缓冲。
+DEFAULT_SAFETY_MB = 800
+
+# 权重之外的推理开销系数：CTranslate2 除权重外还需推理工作区、激活与 CUDA 上下文。
+# 实测（large-v2 权重 2.87GB）峰值增量约 4097MB ≈ 权重 × 1.39，取 1.45 覆盖余量。
+WEIGHTS_MEMORY_FACTOR = 1.45
+
+# 本机 LLM/VLM 端点（Ollama）的显存预留：模型常驻期间 whisper 必须让位。
+# 默认按整卡预留——本机大模型加载后实测占用 20–21GB，宁可保守也不要 OOM。
+DEFAULT_LOCAL_LLM_RESERVE_MB = 22528
+DEFAULT_LOCAL_VLM_RESERVE_MB = 8192
+
+# 判定"端点在本机"的主机名集合：默认只看回环；局域网地址的本机 Ollama 需要
+# 通过 WOV_LOCAL_MODEL_HOSTS 显式声明（见 docs/configuration.md）。
+_LOOPBACK_HOSTS = {"localhost", "127.0.0.1", "::1", "0.0.0.0"}
+
+
+def _env_int(name: str, default: int) -> int:
+    """读取整数环境变量；空值或非法值回退默认值。"""
+    raw = os.getenv(name)
+    if raw is None or not str(raw).strip():
+        return default
+    try:
+        return int(float(raw))
+    except ValueError:
+        return default
+
+
+def local_model_hosts() -> set[str]:
+    """返回被视为"本机模型"的主机名集合（回环 + WOV_LOCAL_MODEL_HOSTS）。"""
+    hosts = set(_LOOPBACK_HOSTS)
+    for item in (os.getenv("WOV_LOCAL_MODEL_HOSTS") or "").split(","):
+        item = item.strip().lower()
+        if item:
+            hosts.add(item)
+    return hosts
+
+
+def _endpoint_host(value: str) -> str:
+    """从 URL 或 host:port 字符串里取出主机名（小写）。"""
+    text = (value or "").strip()
+    if not text:
+        return ""
+    if "://" not in text:
+        text = "http://" + text
+    host = urllib.parse.urlsplit(text).hostname or ""
+    return host.lower()
+
+
+def is_local_endpoint(value: str) -> bool:
+    """判断 LLM/VLM 端点是否指向本机（模型占本机显存）。
+
+    回环地址一律算本机；局域网地址（如另一张网卡上的 Ollama）需要显式列入
+    WOV_LOCAL_MODEL_HOSTS——本机多网卡时无法可靠推断，宁可要求配置。
+    """
+    host = _endpoint_host(value)
+    if not host:
+        return False
+    if host in local_model_hosts():
+        return True
+    # 主机名解析成本低，顺带覆盖 localhost 之外的别名（如容器内的 host.docker.internal）。
+    try:
+        if socket.gethostbyname(host) in {ip for name in _LOOPBACK_HOSTS
+                                          for ip in [socket.gethostbyname(name)] if ip}:
+            return True
+    except OSError:
+        return False
+    return False
+
+
+def probe_free_memory_mb() -> int | None:
+    """返回当前最大可用显存（MB）；无 NVIDIA GPU 或探测失败时返回 None。
+
+    多卡时取可用显存最大的那张卡（推理默认只用一张卡）。
+    """
+    if shutil.which("nvidia-smi") is None:
+        return None
+    try:
+        completed = subprocess.run(
+            [
+                "nvidia-smi",
+                "--query-gpu=memory.free",
+                "--format=csv,noheader,nounits",
+            ],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return None
+    if completed.returncode != 0:
+        return None
+    values = [int(line.strip()) for line in completed.stdout.splitlines() if line.strip().isdigit()]
+    return max(values) if values else None
+
+
+def whisper_need_mb(model_dir: Path) -> int:
+    """按 CTranslate2 权重体量估算转写所需显存（MB）。"""
+    weights = Path(model_dir) / "model.bin"
+    size_mb = weights.stat().st_size / (1024 * 1024) if weights.is_file() else 0.0
+    return int(size_mb * WEIGHTS_MEMORY_FACTOR)
+
+
+def _llm_need_mb() -> int:
+    """LLM 阶段的显存需求：本机端点按预留整卡，线上端点为 0。"""
+    base = os.getenv("LLM_API_BASE", "")
+    if base and is_local_endpoint(base):
+        return _env_int("WOV_LOCAL_LLM_RESERVE_MB", DEFAULT_LOCAL_LLM_RESERVE_MB)
+    return 0
+
+
+def _vlm_need_mb() -> int:
+    """帧 OCR 阶段的显存需求：本机 Ollama 按预留，远端端点为 0。"""
+    base = os.getenv("OLLAMA_HOST") or os.getenv("VLM_API_BASE") or ""
+    if base and is_local_endpoint(base):
+        return _env_int("WOV_LOCAL_VLM_RESERVE_MB", DEFAULT_LOCAL_VLM_RESERVE_MB)
+    return 0
+
+
+def stage_gpu_need_mb(node_type: str, params: dict) -> int:
+    """估算某个节点在当前配置下的显存需求（MB）；0 表示不需要 GPU。
+
+    只认节点类型本身的能力，不看"这是批量还是单任务"——主调度器与批量引擎
+    可以共用同一套判定。
+    """
+    if node_type == "faster-whisper":
+        # 延迟导入：本地模型解析会读环境变量与盘上权重，不在模块导入期执行。
+        from nodes.whisper import resolve_model_path
+
+        return whisper_need_mb(Path(resolve_model_path(params)))
+    if node_type.startswith("llm"):
+        return _llm_need_mb()
+    if node_type in ("vlm-ocr", "subtitle-ocr"):
+        return _vlm_need_mb()
+    # 提音/抽帧/ASS/echo 等走 CPU 或本地磁盘，不占显存。
+    return 0
+
+
+class GpuGate:
+    """进程内 GPU 准入：GPU 阶段互斥 + 显存余量检查，非 GPU 阶段不受限。
+
+    互斥（而不是按显存叠加并发）是刻意的：GPU 算力才是瓶颈，两个转写同时跑
+    只会互相抢算力而整体更慢；需要并发时再按实测调整（见 operations.md）。
+    """
+
+    def __init__(self, probe=None, safety_mb: int = DEFAULT_SAFETY_MB) -> None:
+        self._probe = probe if probe is not None else probe_free_memory_mb
+        self._safety_mb = safety_mb
+        self._lock = threading.Lock()
+        self._holder: str | None = None
+        self._holder_need_mb = 0
+
+    def try_acquire(self, key: str, memory_mb: int) -> bool:
+        """尝试为 key 取得准入；False 表示当前资源不满足，调用方稍后重试。
+
+        同一 key 重复申请视为成功（重入），避免调用方因重复申请把自己挡住。
+        """
+        if memory_mb <= 0:
+            # 不需要 GPU：不占车道，直接放行（提音/线上翻译/ASS 可并行）。
+            return True
+        with self._lock:
+            if self._holder == key:
+                return True
+            if self._holder is not None:
+                return False
+            free = self._probe()
+            if free is not None and free < memory_mb + self._safety_mb:
+                return False
+            self._holder = key
+            self._holder_need_mb = memory_mb
+            return True
+
+    def release(self, key: str) -> None:
+        """释放 key 持有的 GPU 车道（非持有者调用无副作用）。"""
+        with self._lock:
+            if self._holder == key:
+                self._holder = None
+                self._holder_need_mb = 0
+
+    @property
+    def holder(self) -> str | None:
+        """当前持有 GPU 车道的阶段标识（无人持有时为 None）。"""
+        with self._lock:
+            return self._holder
+
+    @property
+    def holder_need_mb(self) -> int:
+        """当前持有者声明的显存需求（MB）。"""
+        with self._lock:
+            return self._holder_need_mb

tests/app/test_batch/test_batch.py

@@ -14,6 +14,7 @@ from pathlib import Path
 import pytest
 
 from wov_app import registry
+from wov_app.resources import GpuGate
 from wov_app.batch import (
     KEEP_MODEL_FLAG,
     MARKER_NAME,
@@ -247,6 +248,17 @@ def test_sidecar_product_name_maps_srt_and_ass() -> None:
     assert _sidecar_product_name(video, Path("/tmp/x.ass")) == "movie.CN_dual_eye.ass"
 
 
+def test_sidecar_product_name_maps_language_variants_by_alias() -> None:
+    """日语转写与中文译文都是 .srt：按 final_outputs 别名区分，互不覆盖。"""
+    # 数据：同一个视频的两份 .srt 产物（日语转写、中文译文）。
+    video = Path("/videos/movie.mp4")
+
+    # 测试过程与验证结果：按别名映射成不同语言后缀。
+    assert _sidecar_product_name(video, Path("/tmp/transcript.srt"), "ja_srt") == "movie.JA.srt"
+    assert _sidecar_product_name(video, Path("/tmp/cn.srt"), "cn_srt") == "movie.CN.srt"
+    assert _sidecar_product_name(video, Path("/tmp/x.ass"), "ass") == "movie.CN_dual_eye.ass"
+
+
 def test_sidecar_product_name_keeps_other_extensions() -> None:
     """其他扩展名产物保留原文件名（不误改语义）。"""
     # 数据：vtt 产物。
@@ -432,6 +444,34 @@ def test_worker_processes_pending_video_end_to_end(tmp_path: Path, monkeypatch)
     assert list(folder.glob("movie.*")), "应在视频旁放置最终产物"
 
 
+def test_final_outputs_place_japanese_transcript_by_language(tmp_path: Path, monkeypatch) -> None:
+    """日语转写（过滤后待翻译）与中文译文都放到视频旁，文件名按语言区分。"""
+    # 数据：prep 阶段产出日语转写、translate 阶段产出中文译文，两者都是 .srt。
+    folder = tmp_path / "videos"
+    _make_video(folder / "movie.mp4")
+    db = Database(tmp_path / "wov.db")
+    db.upsert_workflow({
+        "id": "wf", "name": "两产物流程", "description": "", "published": 1,
+        "latest_version": 1,
+    })
+    definition = _staged_definition().to_dict()
+    definition["final_outputs"] = {"ja_srt": "prep.file_uri", "cn_srt": "translate.file_uri"}
+    db.create_workflow_version("wf", 1, WorkflowDefinition.from_dict(definition).to_dict())
+    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
+
+    # 测试过程：驱动一轮批量处理。
+    with _recording_nodes([]):
+        job_id = create_job(db, str(folder), "wf")
+        BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))
+
+    # 验证结果：两份产物都在视频旁且不互相覆盖。
+    placed = sorted(
+        p.name for p in folder.iterdir()
+        if p.name.startswith("movie.") and p.suffix.lower() in (".srt", ".ass")
+    )
+    assert placed == ["movie.CN.srt", "movie.JA.srt"]
+
+
 def test_worker_skips_video_with_sidecar_subtitle(tmp_path: Path, monkeypatch) -> None:
     """已有旁挂字幕的视频不触发流水线（SKIPPED 不产生 run）。"""
     # 数据：一个已带字幕的视频。
@@ -522,7 +562,7 @@ def test_worker_processes_recovered_zombie_job(tmp_path: Path, monkeypatch) -> N
 
 
 def test_worker_runs_grouped_stage_pipeline(tmp_path: Path, monkeypatch) -> None:
-    """分块流水线：组内按节点顺序跑完全部视频，而不是每个视频跑完整链路。"""
+    """分组流水线：组内每个视频独立推进阶段，阶段顺序仍按 DAG，组间分批。"""
     # 数据：3 个视频 + 三节点链路，分组大小 2（前两个一组、第三个一组）。
     folder = tmp_path / "videos"
     for name in ("a.mp4", "b.mp4", "c.mp4"):
@@ -537,13 +577,12 @@ def test_worker_runs_grouped_stage_pipeline(tmp_path: Path, monkeypatch) -> None
         job_id = create_job(db, str(folder), "wf")
         BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))
 
-    # 验证结果：组1 三个阶段各跑 a/b，再轮到组2 的 c。
+    # 验证结果：每个视频的三阶段按 DAG 顺序各跑一次；组 1（a/b）全部跑完才轮到组 2（c）。
-    assert trace == [
+    for name in ("a.mp4", "b.mp4", "c.mp4"):
-        ("prep", "a.mp4"), ("prep", "b.mp4"),
+        assert [tag for tag, video in trace if video == name] == ["prep", "translate", "post"]
-        ("translate", "a.mp4"), ("translate", "b.mp4"),
+    first_group = [i for i, entry in enumerate(trace) if entry[1] in ("a.mp4", "b.mp4")]
-        ("post", "a.mp4"), ("post", "b.mp4"),
+    second_group = [i for i, entry in enumerate(trace) if entry[1] == "c.mp4"]
-        ("prep", "c.mp4"), ("translate", "c.mp4"), ("post", "c.mp4"),
+    assert max(first_group) < min(second_group)
-    ]
     # 三个视频都完成且产物按约定名落到视频旁。
     assert db.get_batch_job(job_id)["status"] == "COMPLETED"
     assert sorted(p.name for p in folder.glob("*.srt")) == ["a.CN.srt", "b.CN.srt", "c.CN.srt"]
@@ -577,6 +616,155 @@ def test_worker_releases_local_llm_once_per_group(tmp_path: Path, monkeypatch) -
     assert not list((tmp_path / "storage").rglob(KEEP_MODEL_FLAG))
 
 
+@contextmanager
+def _timed_nodes(durations: dict[str, float]):
+    """把三类节点换成"记录起止时间"的假节点，用于断言阶段之间的并行/互斥。
+
+    节点类型决定显存需求（`faster-whisper` 需 GPU、`llm-translate` 看端点、
+    `echo` 不需要），因此可以直接观察资源门控的调度结果。
+    """
+    import time as _time
+
+    registry.register_all()
+    events: list[tuple[str, str, str, float]] = []
+
+    def handler(request: InvokeRequest) -> InvokeResponse:
+        tag = str(request.params.get("node_tag"))
+        # 首阶段可能从 audio_uri 拿到视频路径，后续阶段从 file_uri 拿到上游产物。
+        source = str(request.inputs.get("file_uri") or request.inputs.get("audio_uri") or "")
+        if source and Path(source).suffix.lower() in VIDEO_EXTENSIONS:
+            video_name = Path(source).name
+        else:
+            video_name = Path(source).read_text(encoding="utf-8").strip() if source else ""
+        events.append((tag, video_name, "start", _time.monotonic()))
+        # 时长可按 "阶段:视频" 细化（用于制造"A 已转写、B 还在提音"的时序）。
+        _time.sleep(durations.get(f"{tag}:{video_name}", durations.get(tag, 0.05)))
+        output_dir = Path(request.output_dir)
+        output_dir.mkdir(parents=True, exist_ok=True)
+        output = output_dir / "payload.srt"
+        output.write_text(video_name, encoding="utf-8")
+        events.append((tag, video_name, "end", _time.monotonic()))
+        return InvokeResponse(status="completed", outputs={"file_uri": str(output)})
+
+    for node_type in ("echo", "llm-translate", "faster-whisper"):
+        registry.register(registry.get_node(node_type), handler)
+    yield events
+
+
+def _overlaps(events, first: tuple[str, str], second: tuple[str, str]) -> bool:
+    """两个 (阶段, 视频) 执行区间是否重叠（用于断言并行/串行）。"""
+    def window(key):
+        start = next(e[3] for e in events if (e[0], e[1], e[2]) == (*key, "start"))
+        end = next(e[3] for e in events if (e[0], e[1], e[2]) == (*key, "end"))
+        return start, end
+
+    a_start, a_end = window(first)
+    b_start, b_end = window(second)
+    return a_start < b_end and b_start < a_end
+
+
+def _asr_translate_db(tmp_path: Path) -> Database:
+    """两阶段链路 asr(faster-whisper) → translate(llm) 的临时库。"""
+    db = Database(tmp_path / "wov.db")
+    db.upsert_workflow({
+        "id": "wf", "name": "转写翻译", "description": "", "published": 1, "latest_version": 1,
+    })
+    definition = {
+        "name": "转写翻译", "version": 1,
+        "nodes": [
+            {"id": "asr", "node_type": "faster-whisper", "params": {"node_tag": "asr"},
+             "inputs": {"audio_uri": "input.video_uri"}},
+            {"id": "translate", "node_type": "llm-translate", "params": {"node_tag": "translate"},
+             "inputs": {"file_uri": "asr.file_uri"}},
+        ],
+        "edges": [{"from": "asr", "to": "translate"}],
+        "entry_inputs": {"video_uri": "file"},
+        "final_outputs": {"cn_srt": "translate.file_uri"},
+    }
+    db.create_workflow_version("wf", 1, WorkflowDefinition.from_dict(definition).to_dict())
+    return db
+
+
+def test_remote_translate_overlaps_other_video_transcription(tmp_path: Path, monkeypatch) -> None:
+    """线上翻译不占 GPU：A 在翻译时，B 的转写可以同时跑（GPU 不空转）。"""
+    # 数据：2 个视频 + asr→translate 链路；翻译端点在远端，显存充裕。
+    folder = tmp_path / "videos"
+    for name in ("a.mp4", "b.mp4"):
+        _make_video(folder / name)
+    db = _asr_translate_db(tmp_path)
+    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
+    monkeypatch.setenv("LLM_API_BASE", "https://api.siliconflow.cn/v1/chat/completions")
+    monkeypatch.setattr("wov_app.batch._make_gate", lambda: GpuGate(probe=lambda: 24000))
+
+    # 测试过程
+    with _timed_nodes({"asr": 0.4, "translate": 0.6}) as events:
+        job_id = create_job(db, str(folder), "wf")
+        BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))
+
+    # 验证结果：A 的翻译与 B 的转写时间区间重叠。
+    assert _overlaps(events, ("translate", "a.mp4"), ("asr", "b.mp4"))
+
+
+def test_local_llm_endpoint_serializes_and_runs_all_asr_first(tmp_path: Path, monkeypatch) -> None:
+    """本机 LLM 端点占显存：GPU 阶段互斥，且组内先跑完全部转写再翻译（模型只载一次）。"""
+    # 数据：2 个视频 + asr→translate 链路；翻译端点在本机（占整卡显存）。
+    folder = tmp_path / "videos"
+    for name in ("a.mp4", "b.mp4"):
+        _make_video(folder / name)
+    db = _asr_translate_db(tmp_path)
+    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
+    monkeypatch.setenv("LLM_API_BASE", "http://localhost:11434/v1/chat/completions")
+    monkeypatch.setattr("wov_app.batch._make_gate", lambda: GpuGate(probe=lambda: 24000))
+    # 本用例只验调度顺序，不必真去卸载本机模型。
+    monkeypatch.setattr("wov_app.batch.release_local_model", lambda model=None: None)
+
+    # 测试过程
+    with _timed_nodes({"asr": 0.3, "translate": 0.3}) as events:
+        job_id = create_job(db, str(folder), "wf")
+        BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))
+
+    # 验证结果：全部转写先于任何翻译；翻译之间互斥不重叠。
+    last_asr = max(e[3] for e in events if e[0] == "asr" and e[2] == "end")
+    first_translate = min(e[3] for e in events if e[0] == "translate" and e[2] == "start")
+    assert last_asr < first_translate
+    assert _overlaps(events, ("translate", "a.mp4"), ("translate", "b.mp4")) is False
+
+
+def test_extract_overlaps_transcription_of_other_video(tmp_path: Path, monkeypatch) -> None:
+    """提音（不需要 GPU）与其它视频的转写并行——组内不再"先全部提音再转写"。"""
+    # 数据：2 个视频 + extract(echo) → asr(faster-whisper) 链路。
+    folder = tmp_path / "videos"
+    for name in ("a.mp4", "b.mp4"):
+        _make_video(folder / name)
+    db = Database(tmp_path / "wov.db")
+    db.upsert_workflow({
+        "id": "wf", "name": "提音转写", "description": "", "published": 1, "latest_version": 1,
+    })
+    definition = {
+        "name": "提音转写", "version": 1,
+        "nodes": [
+            {"id": "extract", "node_type": "echo", "params": {"node_tag": "extract"},
+             "inputs": {"file_uri": "input.video_uri"}},
+            {"id": "asr", "node_type": "faster-whisper", "params": {"node_tag": "asr"},
+             "inputs": {"audio_uri": "extract.file_uri"}},
+        ],
+        "edges": [{"from": "extract", "to": "asr"}],
+        "entry_inputs": {"video_uri": "file"},
+        "final_outputs": {"cn_srt": "asr.file_uri"},
+    }
+    db.create_workflow_version("wf", 1, WorkflowDefinition.from_dict(definition).to_dict())
+    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
+    monkeypatch.setattr("wov_app.batch._make_gate", lambda: GpuGate(probe=lambda: 24000))
+
+    # 测试过程
+    with _timed_nodes({"extract:a.mp4": 0.05, "extract:b.mp4": 0.8, "asr": 0.4}) as events:
+        job_id = create_job(db, str(folder), "wf")
+        BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))
+
+    # 验证结果：A 的转写与 B 的提音重叠。
+    assert _overlaps(events, ("asr", "a.mp4"), ("extract", "b.mp4"))
+
+
 def test_worker_defers_video_failed_in_earlier_stage(tmp_path: Path, monkeypatch) -> None:
     """上一阶段失败的视频不在后续阶段重跑（避免 LLM 已常驻时重跑 ASR 抢显存）。"""
     # 数据：2 个视频（同一组）+ 三节点链路，prep 阶段让 a 失败。
@@ -748,3 +936,88 @@ def test_worker_clears_stale_keep_model_flag_before_stage(tmp_path: Path, monkey
     # 验证结果：prep（非 LLM）看不到残留标志；translate（LLM 阶段）才写入；
     # post（非 LLM）不再看到它。
     assert flag_trace == [("prep", False), ("translate", True), ("post", False)]
+
+
+def test_job_hidden_from_engine_until_details_written(tmp_path: Path, monkeypatch) -> None:
+    """建任务期间任务对引擎不可见：明细写完前不被拾起，避免半成品被标完成。
+
+    曾因任务行先入库、524 条明细后写，引擎在明细写一半时拾起任务、收尾时快照
+    里没有剩余明细，把任务误标 COMPLETED（视频永远不再被处理）。
+    """
+    # 数据：3 个待处理视频的媒体库。
+    folder = tmp_path / "videos"
+    for name in ("a.mp4", "b.mp4", "c.mp4"):
+        _make_video(folder / name)
+    db = _published_db(tmp_path)
+    probes: list[str | None] = []
+    original = db.create_batch_video
+
+    def _probe_after_insert(item: dict) -> None:
+        """每写完一条明细，立刻问一次引擎队列（模拟轮询线程的拾取时机）。"""
+        original(item)
+        picked = db.next_queued_batch_job()
+        probes.append(picked["id"] if picked else None)
+
+    monkeypatch.setattr(db, "create_batch_video", _probe_after_insert)
+
+    # 测试过程
+    job_id = create_job(db, str(folder), "wf", recursive=False)
+
+    # 验证结果：写入过程中引擎始终取不到任务；写完后才是 QUEUED 且明细完整。
+    assert probes == [None, None, None]
+    job = db.get_batch_job(job_id)
+    assert job["status"] == "QUEUED"
+    assert job["total"] == 3
+    assert len(db.list_batch_videos(job_id)) == 3
+    assert db.next_queued_batch_job()["id"] == job_id
+
+
+def test_create_job_marks_failed_when_detail_insert_breaks(tmp_path: Path, monkeypatch) -> None:
+    """明细写入中途失败时任务记 FAILED 并留下错误，不产生看不见的残留任务。"""
+    # 数据：2 个视频，第二条明细写入时抛异常。
+    folder = tmp_path / "videos"
+    for name in ("a.mp4", "b.mp4"):
+        _make_video(folder / name)
+    db = _published_db(tmp_path)
+    original = db.create_batch_video
+    calls = {"n": 0}
+
+    def _fail_second(item: dict) -> None:
+        calls["n"] += 1
+        if calls["n"] == 2:
+            raise RuntimeError("磁盘写满")
+        original(item)
+
+    monkeypatch.setattr(db, "create_batch_video", _fail_second)
+
+    # 测试过程 + 验证结果：异常继续抛出，任务可被观察到且为 FAILED。
+    with pytest.raises(RuntimeError):
+        create_job(db, str(folder), "wf", recursive=False)
+    job = db.list_batch_jobs(limit=10)[0]
+    assert job["status"] == "FAILED"
+    assert "磁盘写满" in (job["error"] or "")
+    assert db.next_queued_batch_job() is None
+
+
+def test_worker_start_fails_leftover_creating_job(tmp_path: Path) -> None:
+    """进程中断留下的 CREATING 任务在引擎启动时记为 FAILED，不静默残留。"""
+    # 数据：一条只登记到一半的任务（模拟明细写入中途进程被杀/热重载）。
+    db = _published_db(tmp_path)
+    db.create_batch_job({
+        "id": "batch_halfway", "folder_path": "/videos", "workflow_id": "wf",
+        "recursive": 0, "status": "CREATING", "progress": 0, "total": 0, "done": 0,
+        "failed": 0, "current_video": None, "error": None,
+        "created_at": "2026-09-18T00:00:00+00:00", "updated_at": "2026-09-18T00:00:00+00:00",
+    })
+    worker = BatchWorker(db, interval_seconds=999)
+
+    # 测试过程
+    worker.start()
+    try:
+        # 验证结果：任务变为可见的 FAILED，且不会被引擎当排队任务拾起。
+        job = db.get_batch_job("batch_halfway")
+        assert job["status"] == "FAILED"
+        assert "中断" in (job["error"] or "")
+        assert db.next_queued_batch_job() is None
+    finally:
+        worker.stop()

tests/app/test_db/test_database.py

@@ -381,3 +381,62 @@ def test_recover_interrupted_batch_jobs_requeues_zombie_completed(db_with_workfl
     assert count == 1
     assert db_with_workflow.get_batch_job("zombie")["status"] == "QUEUED"
     assert db_with_workflow.get_batch_job("done")["status"] == "COMPLETED"
+
+
+def test_wal_mode_enabled_for_concurrent_pipeline_writes(tmp_path: Path) -> None:
+    """数据：临时库文件。
+
+    过程：打开一次连接并读取日志模式。
+
+    验证：处于 WAL 模式——批量流水线多线程并发写产物/进度时，读写可并行，
+    不会互相阻塞成 database is locked。
+    """
+    db = Database(tmp_path / "wov.db")
+
+    with db._connect() as conn:
+        mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
+
+    assert mode.lower() == "wal"
+
+
+def test_concurrent_artifact_writes_survive(tmp_path: Path) -> None:
+    """数据：四个线程各自往同一任务写入 25 条产物记录。
+
+    过程：并发调用 create_artifact（批量流水线里多个视频同时落产物的场景）。
+
+    验证：全部写入成功且记录数正确，没有因写锁竞争丢数据或抛异常。
+    """
+    import threading
+
+    db = Database(tmp_path / "wov.db")
+    db.upsert_workflow({"id": "wf", "name": "流程", "description": "", "published": 1, "latest_version": 1})
+    db.create_workflow_version("wf", 1, {"nodes": []})
+    db.create_run({
+        "id": "run-concurrent", "workflow_id": "wf", "workflow_version": 1,
+        "status": "RUNNING", "source": "batch", "created_at": "t1", "updated_at": "t1",
+    })
+    errors: list[Exception] = []
+
+    def writer(worker: int) -> None:
+        try:
+            for index in range(25):
+                db.create_artifact({
+                    "run_id": "run-concurrent", "node_id": "asr",
+                    "name": f"w{worker}-{index}", "uri": f"/tmp/w{worker}-{index}.srt",
+                    "mime_type": "text/plain", "size": index,
+                })
+        except Exception as exc:  # noqa: BLE001 - 用例要报告任意写失败
+            errors.append(exc)
+
+    threads = [threading.Thread(target=writer, args=(n,)) for n in range(4)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert errors == []
+    with db._connect() as conn:
+        count = conn.execute(
+            "SELECT COUNT(*) FROM artifacts WHERE run_id = ?", ("run-concurrent",),
+        ).fetchone()[0]
+    assert count == 100

tests/app/test_resources/test_gate.py

@@ -0,0 +1,150 @@
+"""src/wov_app/resources.py 的模块级测试（数据 → 测试过程 → 验证结果）。
+
+被测模块：GPU 资源门控（显存探测 + 需求估算 + 准入租约）。批量流水线用它决定
+"某个阶段现在能不能起"——用资源现状而不是"是否走云端"来分支。
+
+`nvidia-smi` 探测属 I/O 边界：用例注入确定的探测结果；需求估算用**真实权重文件**
+验证（写入指定大小的文件，不是伪造结构）。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from wov_app.resources import (
+    GpuGate,
+    is_local_endpoint,
+    stage_gpu_need_mb,
+    whisper_need_mb,
+)
+
+
+def _probe(free_mb: int | None):
+    """构造固定返回值的显存探测函数（None 表示探测不到 GPU）。"""
+    return lambda: free_mb
+
+
+def test_stage_without_gpu_never_waits() -> None:
+    """数据：一个已持有 GPU 租约的阶段 + 一个不需要 GPU 的阶段。
+
+    过程：不需要 GPU 的阶段申请准入。
+
+    验证：直接放行——线上模型阶段与 ffmpeg/ASS 阶段不该被 GPU 占用挡住。
+    """
+    gate = GpuGate(probe=_probe(24000))
+    assert gate.try_acquire("video-a:asr", 4200) is True
+
+    assert gate.try_acquire("video-b:translate", 0) is True
+
+
+def test_gpu_stages_are_exclusive() -> None:
+    """数据：显存充足（探测 24000MB），但已有 GPU 阶段在跑。
+
+    过程：另一个 GPU 阶段申请准入。
+
+    验证：拒绝——显存够也不并发跑两个 GPU 阶段（GPU 算力才是瓶颈，两个 whisper
+    同时跑只会互相争抢而整体更慢）。
+    """
+    gate = GpuGate(probe=_probe(24000))
+    assert gate.try_acquire("video-a:asr", 4200) is True
+
+    assert gate.try_acquire("video-b:asr", 4200) is False
+
+
+def test_gpu_stage_needs_free_memory_above_need_plus_margin() -> None:
+    """数据：显存探测结果分别低于/高于"需求 + 安全余量"。
+
+    过程：用同一需求申请准入。
+
+    验证：不足时拒绝，充足时放行——避免临界卡上因碎片化直接 CUDA OOM。
+    """
+    need, margin = 4200, 800
+    tight = GpuGate(probe=_probe(need + margin - 1), safety_mb=margin)
+    roomy = GpuGate(probe=_probe(need + margin), safety_mb=margin)
+
+    assert tight.try_acquire("video-a:asr", need) is False
+    assert roomy.try_acquire("video-a:asr", need) is True
+
+
+def test_probe_missing_falls_back_to_exclusivity() -> None:
+    """数据：探测不到 GPU（无 nvidia-smi，例如纯 CPU 机器或容器）。
+
+    过程：连续申请两个 GPU 阶段。
+
+    验证：无法核对显存时退化为"GPU 阶段互斥"——仍然不会并发，行为与串行等价。
+    """
+    gate = GpuGate(probe=_probe(None))
+
+    assert gate.try_acquire("video-a:asr", 4200) is True
+    assert gate.try_acquire("video-b:asr", 4200) is False
+    gate.release("video-a:asr")
+    assert gate.try_acquire("video-b:asr", 4200) is True
+
+
+def test_release_frees_the_gpu_lane() -> None:
+    """数据：一个已释放租约的阶段。
+
+    过程：释放后再次申请。
+
+    验证：GPU 车道重新可用（视频间不会互相永久阻塞）。
+    """
+    gate = GpuGate(probe=_probe(24000))
+    gate.try_acquire("video-a:asr", 4200)
+    gate.release("video-a:asr")
+
+    assert gate.try_acquire("video-b:asr", 4200) is True
+
+
+def test_whisper_need_scales_with_real_weights(tmp_path: Path) -> None:
+    """数据：真实写入的 100MB 权重文件（CTranslate2 的 model.bin）。
+
+    过程：按权重体量估算推理显存需求。
+
+    验证：需求 = 权重 × 系数（推理工作区与 CUDA 上下文另占约 45%）。
+    """
+    model_dir = tmp_path / "faster-whisper-large-v2"
+    model_dir.mkdir()
+    (model_dir / "model.bin").write_bytes(b"\x00" * (100 * 1024 * 1024))
+
+    assert whisper_need_mb(model_dir) == int(100 * 1.45)
+
+
+def test_stage_need_zero_for_non_gpu_nodes(tmp_path: Path) -> None:
+    """数据：ffmpeg 提音、ASS 合成、echo 三类不需要 GPU 的节点。
+
+    过程：估算其显存需求。
+
+    验证：都是 0，因此可以与 GPU 阶段并行（阶段内提音/合成不再阻塞转写）。
+    """
+    for node_type in ("ffmpeg-extract", "srt-to-dual-eye-ass", "echo"):
+        assert stage_gpu_need_mb(node_type, {}) == 0
+
+
+def test_stage_need_for_local_and_remote_llm(monkeypatch) -> None:
+    """数据：LLM 端点分别是本机（占显存常驻）与远端（不占显存）。
+
+    过程：估算 llm-translate 阶段的显存需求。
+
+    验证：本机端点按预留显存计入（挡住 whisper），远端端点为 0（放行转写）。
+    """
+    monkeypatch.setenv("WOV_LOCAL_MODEL_HOSTS", "")
+    monkeypatch.setenv("LLM_API_BASE", "http://localhost:11434/v1/chat/completions")
+    local_need = stage_gpu_need_mb("llm-translate", {})
+    monkeypatch.setenv("LLM_API_BASE", "https://api.siliconflow.cn/v1/chat/completions")
+    remote_need = stage_gpu_need_mb("llm-translate", {})
+
+    assert local_need > 0
+    assert remote_need == 0
+
+
+def test_local_endpoint_accepts_configured_lan_host(monkeypatch) -> None:
+    """数据：Ollama 跑在本机另一张网卡的局域网地址上（如 192.168.123.70）。
+
+    过程：判断端点是否本机模型。
+
+    验证：通过 WOV_LOCAL_MODEL_HOSTS 显式声明后视为本机，按其显存需求预留。
+    """
+    monkeypatch.setenv("WOV_LOCAL_MODEL_HOSTS", "192.168.123.70")
+
+    assert is_local_endpoint("http://192.168.123.70:11434/v1/chat/completions") is True
+    assert is_local_endpoint("https://api.siliconflow.cn/v1/chat/completions") is False

tests/nodes/test_subtitle_cleanup/data/repetition_cues.json

@@ -0,0 +1,62 @@
+{
+ "artifacts": [
+  {
+   "start": "00:10:00,064",
+   "end": "00:10:30,064",
+   "text": "チンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチンチ",
+   "repeats": 111,
+   "unit": "チン"
+  },
+  {
+   "start": "00:36:29,999",
+   "end": "00:36:59,999",
+   "text": "ああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああああ",
+   "repeats": 446,
+   "unit": "あ"
+  },
+  {
+   "start": "00:40:20,199",
+   "end": "00:40:50,199",
+   "text": "ハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハッハ",
+   "repeats": 111,
+   "unit": "ハッ"
+  }
+ ],
+ "moans": [
+  {
+   "start": "00:01:28,832",
+   "end": "00:01:31,252",
+   "text": "しゅしゅしゅしゅしゅしゅしゅ",
+   "repeats": 7,
+   "unit": "しゅ"
+  },
+  {
+   "start": "00:01:31,991",
+   "end": "00:01:33,991",
+   "text": "しゅしゅしゅしゅしゅしゅ",
+   "repeats": 6,
+   "unit": "しゅ"
+  },
+  {
+   "start": "00:03:00,096",
+   "end": "00:03:02,276",
+   "text": "ウウウウウウウウウウ",
+   "repeats": 10,
+   "unit": "ウ"
+  }
+ ],
+ "normal": {
+  "start": "00:00:12,000",
+  "end": "00:00:15,000",
+  "text": "そんなにご褒美欲しかったの?"
+ },
+ "short_artifacts": [
+  {
+   "start": "00:22:56,363",
+   "end": "00:23:00,093",
+   "text": "ははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははははは�",
+   "repeats": 111,
+   "unit": "は"
+  }
+ ]
+}

tests/nodes/test_subtitle_cleanup/test_cleanup.py

@@ -8,6 +8,9 @@ whisper（日语链路）与 llm-translate（中文链路）复用，纯函数
 
 from __future__ import annotations
 
+import json
+from pathlib import Path
+
 from nodes.subtitle_cleanup import (
     DEFAULT_MOAN_MAX_CHARS,
     HALLUCINATION_TOKENS,
@@ -15,6 +18,7 @@ from nodes.subtitle_cleanup import (
     clean_japanese_hallucinations,
     clean_srt_text,
     remove_hallucination_entries,
+    remove_repetition_entries,
     remove_short_moan_entries,
 )
 from tests.shared.srt_entries import parse_srt_entries
@@ -246,3 +250,100 @@ def test_multiline_moan_entry_removed_as_one_cue() -> None:
     # 验证结果：只剩第二条并重编号。
     assert [e["text"] for e in parse_srt_entries(cleaned)] == ["そこ"]
     assert cleaned.startswith("1\n")
+
+
+# 真实转写抽样（data/repetition_cues.json）：30 秒窗口被同一单元填满的 whisper
+# 重复伪影 + 真实短呻吟 + 真实台词，用于"重复伪影"判据的正反例。
+_REPETITION_CUES = json.loads(
+    (Path(__file__).parent / "data" / "repetition_cues.json").read_text(encoding="utf-8")
+)
+
+
+def test_remove_repetition_entries_deletes_whisper_loops_only() -> None:
+    """数据：真实 ASR 产物——3 条 30 秒重复伪影（重复 74/111/446 次）、
+    3 条 2–3 秒真实呻吟（重复 6–10 次）、1 条正常台词。
+
+    过程：调用 remove_repetition_entries 清理整份 SRT。
+
+    验证：只删 30 秒伪影，真实呻吟与台词原样保留，剩余 cue 序号连续。
+    """
+    artifacts = _REPETITION_CUES["artifacts"]
+    moans = _REPETITION_CUES["moans"]
+    normal = _REPETITION_CUES["normal"]
+    srt = _srt(
+        *[(c["start"], c["end"], c["text"]) for c in artifacts],
+        *[(c["start"], c["end"], c["text"]) for c in moans],
+        (normal["start"], normal["end"], normal["text"]),
+    )
+
+    cleaned = remove_repetition_entries(srt)
+
+    for cue in artifacts:
+        assert cue["text"] not in cleaned
+    for cue in moans:
+        assert cue["text"] in cleaned
+    assert normal["text"] in cleaned
+    entries = parse_srt_entries(cleaned)
+    assert [e["text"] for e in entries] == [c["text"] for c in moans] + [normal["text"]]
+    # 序号/时间轴重建后从 1 连续编号，不留空号（合法 SRT）。
+    numbers = [line for line in cleaned.splitlines() if line.strip().isdigit()]
+    assert numbers == [str(i) for i in range(1, len(entries) + 1)]
+
+
+def test_remove_repetition_entries_keeps_short_repeated_moan() -> None:
+    """数据：3 秒内重复 15 次的真实呻吟（时长不足阈值）。
+
+    过程：调用 remove_repetition_entries。
+
+    验证：保留——时长阈值是"窗口被填满"的判据，短促重复属真实发声。
+    """
+    srt = _srt(("00:00:01,000", "00:00:04,200", "ぇ" * 15))
+
+    cleaned = remove_repetition_entries(srt)
+
+    assert "ぇ" * 15 in cleaned
+
+
+def test_remove_repetition_entries_keeps_mixed_long_line() -> None:
+    """数据：30 秒长条但正文以正常台词为主，只有少量重复。
+
+    过程：调用 remove_repetition_entries。
+
+    验证：保留——重复片段未占正文 70% 以上，不构成重复伪影。
+    """
+    text = "そうですね、それでいいと思いますよ" * 3 + "ああ"
+    srt = _srt(("00:00:01,000", "00:00:31,000", text))
+
+    cleaned = remove_repetition_entries(srt)
+
+    assert text in cleaned
+
+
+def test_remove_repetition_entries_can_be_disabled() -> None:
+    """数据：一条 30 秒重复伪影，阈值设为 0（关闭）。
+
+    过程：调用 remove_repetition_entries(threshold_seconds=0)。
+
+    验证：原样返回，便于按需走旧行为。
+    """
+    artifact = _REPETITION_CUES["artifacts"][0]
+    srt = _srt((artifact["start"], artifact["end"], artifact["text"]))
+
+    assert remove_repetition_entries(srt, threshold_seconds=0) == srt
+
+
+def test_remove_repetition_entries_deletes_short_but_extreme_repetition() -> None:
+    """数据：真实产物里的短时伪影——3.7 秒的 cue 被同一个假名填了 111 次。
+
+    过程：调用 remove_repetition_entries。
+
+    验证：整条删除。判据不能只看时长：这种短条会一路翻成 56 个"哈"进成品字幕，
+    但真实呻吟的重复次数实测 ≤15，用重复次数上限即可区分。
+    """
+    artifact = _REPETITION_CUES["short_artifacts"][0]
+    srt = _srt((artifact["start"], artifact["end"], artifact["text"]))
+
+    cleaned = remove_repetition_entries(srt)
+
+    assert artifact["text"] not in cleaned
+    assert parse_srt_entries(cleaned) == []

tests/nodes/test_whisper/test_transcribe.py

@@ -543,3 +543,75 @@ def test_real_whisper_transcribes_real_speech(tmp_path: Path) -> None:
     starts = [e["start"] for e in entries]
     assert starts == sorted(starts)
     assert max(starts) <= 62.0
+
+
+def test_invoke_drops_repetition_artifact_in_decode_full(tmp_path: Path, monkeypatch) -> None:
+    """decode_full 下删除 30 秒重复伪影：它会带着翻译层一起进重复循环。
+
+    数据：假模型返回一段 30 秒窗口被同一单元填满的伪影 + 一条真实台词。
+    过程：调用 invoke（decode_full=True）。
+    验证：伪影整条删除、真实台词保留，产物里不再出现超长重复正文。
+    """
+    artifact = "チン" * 111
+    model = FakeModel([
+        FakeSegment(0.0, 30.0, artifact),
+        FakeSegment(30.0, 33.0, "そこ、だめ"),
+    ])
+    _inject_model(monkeypatch, model)
+
+    response = invoke(_request(tmp_path, SPEECH_WAV, chunk_seconds=0, decode_full=True))
+
+    assert response.status == "completed", response.error
+    content = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
+    assert artifact not in content
+    assert "そこ、だめ" in content
+
+
+def test_invoke_passes_hallucination_guard_params(tmp_path: Path, monkeypatch) -> None:
+    """数据：工作流传入幻觉抑制参数（HST + word_timestamps + 两个阈值）。
+
+    过程：调用 invoke。
+
+    验证：参数原样传给 faster-whisper——静音段幻觉只能靠这些参数抑制：
+    `hallucination_silence_threshold` 需要 word_timestamps 才生效。
+    """
+    # 数据：无语音段也能"编"出字幕的假模型 + 显式传入的抑制参数。
+    model = FakeModel([FakeSegment(0.0, 2.0, "こんにちは")])
+    _inject_model(monkeypatch, model)
+
+    # 测试过程
+    invoke(_request(
+        tmp_path, SPEECH_WAV, chunk_seconds=0, language="ja",
+        word_timestamps=True, hallucination_silence_threshold=2.0,
+        no_speech_threshold=0.3, log_prob_threshold=-1.2,
+        compression_ratio_threshold=2.4,
+    ))
+
+    # 验证结果
+    call = model.calls[0]
+    assert call["word_timestamps"] is True
+    assert call["hallucination_silence_threshold"] == 2.0
+    assert call["no_speech_threshold"] == 0.3
+    assert call["log_prob_threshold"] == -1.2
+    assert call["compression_ratio_threshold"] == 2.4
+
+
+def test_invoke_omits_guard_params_by_default(tmp_path: Path, monkeypatch) -> None:
+    """数据：不传抑制参数（默认工作流）。
+
+    过程：调用 invoke。
+
+    验证：不透传这些键，保持 faster-whisper 自身默认值，行为与改造前一致。
+    """
+    # 数据：普通假模型。
+    model = FakeModel([FakeSegment(0.0, 1.0, "x")])
+    _inject_model(monkeypatch, model)
+
+    # 测试过程
+    invoke(_request(tmp_path, SPEECH_WAV, chunk_seconds=0))
+
+    # 验证结果
+    call = model.calls[0]
+    assert "hallucination_silence_threshold" not in call
+    assert "no_speech_threshold" not in call
+    assert "word_timestamps" not in call

tests/scripts/test_plan_regenerate_subtitles/test_plan_regenerate_subtitles.py

@@ -0,0 +1,317 @@
+"""plan_regenerate_subtitles.py 的模块级测试（数据 → 测试过程 → 验证结果）。
+
+被测模块：字幕重生成计划统计脚本。用例在临时目录里放**真实视频文件**与
+真实字幕文件，调用脚本的真实函数；只有 ffprobe 子进程这一 I/O 边界在需要
+确定性时长时注入固定值，其余用例跑真实 ffprobe（缺可执行文件时跳过）。
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+from scripts.plan_regenerate_subtitles import (
+    CURRENT,
+    MISSING,
+    PENDING,
+    _agg,
+    backup_old_subtitles,
+    build_items,
+    classify,
+    cn_products,
+    format_summary,
+    main,
+    probe_duration,
+    read_select_list,
+    sidecar_subtitles,
+)
+
+DATA = Path(__file__).parent / "data"
+CLIP = DATA / "clip_10s.mp4"  # 真实 10 秒 mp4，供扫描与时长探测使用
+BEFORE = datetime(2026, 9, 1)
+
+
+def _set_mtime(path: Path, when: str) -> None:
+    """把文件 mtime 设为指定日期（本地时区），模拟不同时期生成的字幕。"""
+    ts = datetime.fromisoformat(when).timestamp()
+    os.utime(path, (ts, ts))
+
+
+def _make_video(tmp_path: Path, name: str) -> Path:
+    """在临时目录放一份真实视频文件，返回其路径。"""
+    video = tmp_path / f"{name}.mp4"
+    shutil.copyfile(CLIP, video)
+    return video
+
+
+def _make_subtitle(video: Path, suffix: str, when: str) -> Path:
+    """在视频旁写一个真实字幕文件（内容为合法 SRT/ASS 片段）并设置 mtime。"""
+    sub = video.with_name(video.name[: -len(video.suffix)] + suffix)
+    if suffix.endswith(".srt"):
+        sub.write_text("1\n00:00:01,000 --> 00:00:02,000\n你好\n", encoding="utf-8")
+    else:
+        sub.write_text("[Script Info]\nTitle: t\n", encoding="utf-8")
+    _set_mtime(sub, when)
+    return sub
+
+
+def test_products_pick_earliest_mtime_so_style_rewrite_stays_old(tmp_path: Path) -> None:
+    """数据：中文字幕生成于 2025-12，双目 .ass 被样式脚本在 2026-09 原地改写。
+
+    过程：扫描该视频并读取 CN 产物生成时间。
+
+    验证：只认本流水线产物，生成时间取最早值（2025-12），视频判为待重生成。
+    """
+    video = _make_video(tmp_path, "movie")
+    _make_subtitle(video, ".srt", "2025-11-01")  # 片源自带字幕，不参与判定
+    _make_subtitle(video, ".CN.srt", "2025-12-01")
+    _make_subtitle(video, ".CN_dual_eye.ass", "2026-09-06")
+
+    items = build_items(tmp_path, workers=2, probe=False)
+
+    assert [p.name for p in cn_products(video)] == ["movie.CN.srt", "movie.CN_dual_eye.ass"]
+    assert items[0]["generated_at"].startswith("2025-12-01")
+    assert [i["video"] for i in classify(items, BEFORE)[PENDING]] == [str(video)]
+
+
+def test_classify_groups_by_generation_time(tmp_path: Path) -> None:
+    """数据：三个视频——旧字幕、新字幕、完全没有 CN 字幕。
+
+    过程：扫描后按 2026-09-01 分界线分类。
+
+    验证：分别落入待重生成 / 已最新 / 无 CN 字幕三组，边界时刻本身算已最新。
+    """
+    old = _make_video(tmp_path, "old")
+    _make_subtitle(old, ".CN.srt", "2026-04-05")
+    fresh = _make_video(tmp_path, "fresh")
+    _make_subtitle(fresh, ".CN.srt", "2026-09-01")
+    _make_subtitle(fresh, ".CN_dual_eye.ass", "2026-09-06")
+    never = _make_video(tmp_path, "never")
+
+    groups = classify(build_items(tmp_path, workers=2, probe=False), BEFORE)
+
+    assert [i["video"] for i in groups[PENDING]] == [str(old)]
+    assert [i["video"] for i in groups[CURRENT]] == [str(fresh)]
+    assert [i["video"] for i in groups[MISSING]] == [str(never)]
+
+
+def test_name_with_extra_suffix_still_matches_product(tmp_path: Path) -> None:
+    """数据：手工改过名的产物 `<视频名>666.CN.srt`（媒体库里真实存在这种命名）。
+
+    过程：扫描该视频的 CN 产物。
+
+    验证：按「文件名含视频主名 + CN 产物后缀」命中，与批量引擎的旁挂判定一致。
+    """
+    video = _make_video(tmp_path, "clip")
+    _make_subtitle(video, "666.CN.srt", "2025-12-14")
+
+    items = build_items(tmp_path, workers=2, probe=False)
+
+    assert items[0]["generated_at"].startswith("2025-12-14")
+    assert classify(items, BEFORE)[PENDING]
+
+
+def test_aggregate_sums_duration_and_size(tmp_path: Path) -> None:
+    """数据：两个待重生成视频，时长各有值、其中一个探测失败。
+
+    过程：汇总该组的数量、时长、体积。
+
+    验证：时长合计跳过探测失败项并单独计数，体积按全部文件累计。
+    """
+    video = _make_video(tmp_path, "a")
+    _make_subtitle(video, ".CN.srt", "2025-12-01")
+    items = [
+        {"size": 100, "duration": 12.5},
+        {"size": 50, "duration": None},
+    ]
+
+    agg = _agg(items)
+
+    assert agg == {"count": 2, "seconds": 12.5, "bytes": 150, "no_duration": 1}
+
+
+def test_main_writes_plan_json(tmp_path: Path) -> None:
+    """数据：真实 10 秒视频 + 旧 CN 字幕 + 新 CN 字幕各一个。
+
+    过程：调用 main() 全流程并写出计划 JSON。
+
+    验证：统计报告含待重生成/已最新行数，JSON 里两条明细状态与时长正确。
+    """
+    if shutil.which("ffprobe") is None:
+        pytest.skip("环境缺少 ffprobe，无法探测真实视频时长")
+    old = _make_video(tmp_path, "old")
+    _make_subtitle(old, ".CN.srt", "2025-12-01")
+    fresh = _make_video(tmp_path, "fresh")
+    _make_subtitle(fresh, ".CN.srt", "2026-09-10")
+    out = tmp_path / "plan.json"
+
+    code = main([str(tmp_path), "--before", "2026-09-01", "--out", str(out), "--workers", "2"])
+
+    assert code == 0
+    payload = json.loads(out.read_text(encoding="utf-8"))
+    by_name = {Path(i["video"]).name: i for i in payload["items"]}
+    assert by_name["old.mp4"]["status"] == PENDING
+    assert by_name["old.mp4"]["duration"] == pytest.approx(10.0, abs=0.5)
+    assert by_name["fresh.mp4"]["status"] == CURRENT
+    assert payload["totals"][PENDING]["count"] == 1
+    assert payload["totals"][PENDING]["seconds"] == pytest.approx(10.0, abs=0.5)
+
+
+def test_probe_duration_reads_real_video() -> None:
+    """数据：真实 10 秒 mp4（模块目录内素材）。
+
+    过程：调用真实 ffprobe 读取时长。
+
+    验证：返回约 10 秒，说明统计脚本的时长口径来自容器真实时长。
+    """
+    if shutil.which("ffprobe") is None:
+        pytest.skip("环境缺少 ffprobe，无法探测真实视频时长")
+
+    assert probe_duration(CLIP) == pytest.approx(10.0, abs=0.1)
+
+
+def test_format_summary_reports_all_groups() -> None:
+    """数据：三组各一条明细（旧/新/无字幕），时长与体积已知。
+
+    过程：格式化统计报告。
+
+    验证：报告包含总量、待重生成、已最新、无 CN 字幕四段关键信息。
+    """
+    root = Path("/media")
+    items = [
+        {"duration": 3600.0, "size": 0, "generated_at": "2025-12-01T00:00:00"},
+        {"duration": 1800.0, "size": 0, "generated_at": "2026-09-10T00:00:00"},
+        {"duration": None, "size": 0, "generated_at": None},
+    ]
+
+    text = format_summary(root, items, classify(items, BEFORE), BEFORE)
+
+    assert "视频总数: 3 个" in text
+    assert "待重新生成（CN 字幕早于 2026-09-01）" in text
+    assert "已是最新" in text
+    assert "无 CN 字幕（首次生成）" in text
+    assert "警告: 1 个视频未能读出时长" in text
+
+
+def test_sidecar_subtitles_matches_batch_engine_rule(tmp_path: Path) -> None:
+    """数据：视频旁有片源 .srt、CN 产物与无关文件各一份。
+
+    过程：列出批量引擎会认定为"已有字幕"的旁挂文件。
+
+    验证：只收文件名含视频主名的字幕，不含无关图片与别人的字幕。
+    """
+    video = _make_video(tmp_path, "movie")
+    source = _make_subtitle(video, ".srt", "2025-11-01")
+    cn_srt = _make_subtitle(video, ".CN.srt", "2025-12-01")
+    cn_ass = _make_subtitle(video, ".CN_dual_eye.ass", "2026-09-06")
+    (tmp_path / "movie-poster.jpg").write_bytes(b"poster")
+    (tmp_path / "other.srt").write_text("1\n", encoding="utf-8")
+
+    found = sidecar_subtitles(video)
+
+    assert found == [cn_srt, cn_ass, source]
+
+
+def test_backup_renames_every_sidecar_and_unsets_skip(tmp_path: Path) -> None:
+    """数据：待重生成视频带片源 .srt + CN 产物 + 既有备份目标冲突。
+
+    过程：执行改名备份。
+
+    验证：全部旁挂字幕改名成 .old 后缀、原文件消失；改名后批量引擎再也
+    不会把该视频判为"已有字幕"。
+    """
+    video = _make_video(tmp_path, "movie")
+    source = _make_subtitle(video, ".srt", "2025-11-01")
+    cn_srt = _make_subtitle(video, ".CN.srt", "2025-12-01")
+    cn_ass = _make_subtitle(video, ".CN_dual_eye.ass", "2026-09-06")
+
+    changed = backup_old_subtitles([{"video": str(video)}], suffix=".old", apply=True)
+
+    assert len(changed) == 3
+    assert sidecar_subtitles(video) == []
+    for path in (source, cn_srt, cn_ass):
+        assert not path.exists()
+        assert (path.parent / (path.name + ".old")).read_text(encoding="utf-8")
+
+
+def test_backup_dry_run_only_reports(tmp_path: Path) -> None:
+    """数据：一个带旧字幕的视频。
+
+    过程：不传 apply 走预览。
+
+    验证：返回改名计划，磁盘文件保持不变。
+    """
+    video = _make_video(tmp_path, "movie")
+    sub = _make_subtitle(video, ".CN.srt", "2025-12-01")
+
+    changed = backup_old_subtitles([{"video": str(video)}], suffix=".old", apply=False)
+
+    assert changed == [(sub, sub.with_name(sub.name + ".old"))]
+    assert sub.exists()
+
+
+def test_backup_does_not_overwrite_existing_backup(tmp_path: Path) -> None:
+    """数据：视频旁已有同名备份文件（同一批重复执行）。
+
+    过程：再次执行改名备份。
+
+    验证：已存在的备份不被覆盖，原文件也不被删除，返回空变更。
+    """
+    video = _make_video(tmp_path, "movie")
+    sub = _make_subtitle(video, ".CN.srt", "2025-12-01")
+    backup = sub.with_name(sub.name + ".old")
+    backup.write_text("first\n", encoding="utf-8")
+
+    changed = backup_old_subtitles([{"video": str(video)}], suffix=".old", apply=True)
+
+    assert changed == []
+    assert backup.read_text(encoding="utf-8") == "first\n"
+    assert sub.exists()
+
+
+def test_backup_respects_select_list(tmp_path: Path) -> None:
+    """数据：两个待重生成视频，只选中其中一个（选择文件带行尾注释）。
+
+    过程：执行改名备份。
+
+    验证：只改选中视频的旁挂字幕，另一个保持原样。
+    """
+    chosen = _make_video(tmp_path, "chosen")
+    chosen_sub = _make_subtitle(chosen, ".CN.srt", "2025-12-01")
+    kept = _make_video(tmp_path, "kept")
+    kept_sub = _make_subtitle(kept, ".CN.srt", "2025-12-02")
+    select = tmp_path / "select.txt"
+    select.write_text(f"# 只处理这一个\n{chosen}  # 40.3min\n\n", encoding="utf-8")
+
+    changed = backup_old_subtitles(
+        [{"video": str(chosen)}, {"video": str(kept)}],
+        suffix=".old", apply=True, select=read_select_list(select),
+    )
+
+    assert changed == [(chosen_sub, chosen_sub.with_name(chosen_sub.name + ".old"))]
+    assert kept_sub.exists()
+
+
+def test_format_summary_marks_missing_durations(tmp_path: Path) -> None:
+    """数据：只做分类统计、未探测时长的明细（--backup-old 场景）。
+
+    过程：格式化统计报告。
+
+    验证：报告不显示 0 小时，而是标注未探测时长，也不出现时长缺失警告。
+    """
+    root = Path("/media")
+    items = [
+        {"duration": None, "size": 10, "generated_at": "2025-12-01T00:00:00"},
+        {"duration": None, "size": 20, "generated_at": "2026-09-10T00:00:00"},
+    ]
+
+    text = format_summary(root, items, classify(items, BEFORE), BEFORE)
+
+    assert "未探测时长" in text
+    assert "0.00 小时" not in text
+    assert "警告" not in text

tests/shared/test_alignment/test_alignment.py

@@ -45,6 +45,12 @@ def _sample_dir() -> Path | None:
     return directory if directory.is_dir() else None
 
 
+MEDIA_SUFFIXES = {
+    ".wav", ".mp3", ".flac", ".m4a", ".aac", ".ogg",
+    ".mp4", ".mkv", ".mov", ".webm", ".ts",
+}
+
+
 @pytest.mark.integration
 def test_real_alignment_candidates_discovered() -> None:
     """对齐素材探查：能找到成对的媒体与参考字幕（数据契约可用）。"""
@@ -52,6 +58,12 @@ def test_real_alignment_candidates_discovered() -> None:
     directory = _sample_dir()
     if directory is None:
         pytest.skip("缺少 tests/shared/data/alignment 目录")
+    # 真实媒体素材不入库（.gitignore 掉的 mp4/wav），没拷过素材属环境状态：跳过
+    # 而不是失败；只要有一个媒体文件就说明素材已就位，此时配对不全才是真回归。
+    has_media = any(p.is_file() and p.suffix.lower() in MEDIA_SUFFIXES
+                    for p in directory.iterdir())
+    if not has_media:
+        pytest.skip("缺少真实媒体素材（按需从媒体库复制，不入库）")
 
     # 测试过程
     candidates = alignment_candidates(directory)

web/assets/app.js

@@ -382,11 +382,13 @@ function progressBar(percent, failed) {
   return `<div class="progress"><div class="progress-bar ${failed ? "error" : ""}" style="width:${Math.min(100, percent)}%"></div></div>`;
 }
 
-// 为已完成任务生成最终产物下载链接（cn_srt / ass）。
+// 为已完成任务生成最终产物下载链接（ja_srt / cn_srt / ass）。
+// 日语转写是翻译前的中间产物，归档下来便于回看与对照译文。
 function artifactLinks(runId, status) {
   if (status !== "COMPLETED") return "-";
   return `
     <div class="downloads-inline">
+      <a class="download-link" href="/api/runs/${encodeURIComponent(runId)}/artifacts/ja_srt">日语 SRT</a>
       <a class="download-link" href="/api/runs/${encodeURIComponent(runId)}/artifacts/cn_srt">中文 SRT</a>
       <a class="download-link" href="/api/runs/${encodeURIComponent(runId)}/artifacts/ass">VR ASS</a>
     </div>`;

workflows/learn-translate.json

@@ -31,6 +31,10 @@
           "condition_on_previous_text": false,
           "chunk_seconds": 60,
           "vad_filter": false,
+          "word_timestamps": true,
+          "hallucination_silence_threshold": 2.0,
+          "_note_word_timestamps": "开启词级时间戳：hallucination_silence_threshold 只在该模式下生效（它按词级时间戳跳过幻觉段里的静音部分）。代价是解码更慢（实测同段音频约 1.8×），换来的是无语音段不再产生\"こんにちは/おはようございます\"这类套话幻觉。",
+          "_note_hallucination_silence_threshold": "静音段幻觉抑制：怀疑该段是幻觉时，跳过超过 2 秒的静音。实测片头无对话段的字幕条数由 15 降到 4、且不再出现套话幻觉，呻吟段基本保留。设 0 或删除该键即关闭（回到改造前行为）。",
           "beam_size": 1,
           "_note_decode_full": "【本次修复核心】decode_full=true 强制无 VAD 整段解码，绕过 silero VAD 对呻吟/轻语/快速讲解/BGM 混叠人声的切段误杀。实测 savr-1054 全片：生产 VAD 仅召回 115 条，decode_full 召回 340 条（弱语音全找回）。正例：教师快速带过一句'つまりね'（弱语音），decode_full 能捕捉；反例（decode_full=false 默认）：该句被 silero 概率<阈值当静音剔除，字幕整句消失。代价是无语音段会产生长时'おやすみなさい/ご視聴ありがとうございました'幻觉——处理方式：whisper 转录后立即**连带时间戳把整条 cue 删除**（不留下 '-' 占位污染下游，占位会渲染进 ASS 成减号），短时（≤15s）相同词可能是剧情真实道晚安则保留。实测 speech_60s：前 30s 无语音幻觉被整条剔除，字幕直接从 30s 真实内容开始、序号连续。",
           "_note_vad_filter": "本工作流 decode_full=true 时 vad_filter 被强制置 false（两者互斥，decode_full 优先）。保留 vad_filter=false 仅为显式声明'不启用 VAD 切段'。正例：学习视频讲解者偶有停顿、翻页声，无 VAD 不误删。反例：vad_filter=true + 讲解者语速快/带气声，弱音节被整段吞掉（见 decode_full 反例）。",
@@ -89,6 +93,7 @@
       "video_uri": "file"
     },
     "final_outputs": {
+      "ja_srt": "asr.srt_uri",
       "cn_srt": "translate.cn_srt_uri",
       "ass": "ass.ass_uri"
     }