feat: 批量流水线按 GPU 资源调度（非 GPU 阶段并行、GPU 阶段互斥）

此前组内阶段是串行的（先全部提音、再全部转写、再全部翻译），LLM 走线上端点时
翻译阶段不占显存、GPU 全程空转——实测占整轮挂钟约 40%（19.6W / 272MiB）。

- `_run_job` 改为按组启动在途流水线：每个视频独立推进自己的阶段，最多
  `WOV_BATCH_PIPELINE_WORKERS`（默认 4）个阶段在途。
- 派发只看资源：`stage_gpu_need_mb` 为 0 的阶段（提音、线上翻译、ASS）立刻派发，
  可与其它视频的转写并行；需要 GPU 的阶段由 `GpuGate` 互斥准入，并按"阶段索引
  最小者优先"派发，组内仍是先跑完全部转写再进翻译——本机 Ollama 模型每组只
  加载一次，不需要按"是否云端"写分支。
- 同一阶段只在途一份（派发即标记 running），单视频异常不带走整组；暂停沿用
  run 级 paused.flag，暂停后不再派发新阶段。
- 测试：远端翻译与其它视频转写重叠、本机端点下全部转写先于翻译且翻译互斥、
  提音与转写重叠，以及既有分组/暂停/失败隔离用例。

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

@@ -85,6 +85,23 @@
   “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

@@ -104,15 +104,20 @@
   run 的 `current_node_id` 在 DAG 拓扑序中的位置推导、节点类型映射中文标签。
   粒度限制：`progress` 只有**节点边界**粒度，句级进度（转写分块、翻译批次、OCR 帧）
   不落库、只在控制台日志里。
-- **分块流水线执行（本地模型只加载一次）**：批量引擎把待处理视频按
-  `WOV_BATCH_STAGE_GROUP_SIZE`（默认 8）分组，**组内按节点顺序跑完全部视频**
-  （先全部 extract、再全部 ASR、再全部 LLM 翻译、最后 ASS）再进入下一组。每个
-  视频的 run 在阶段边界保持 RUNNING（`execute_run(stop_after=节点)`），下一阶段
-  从产物表跳过已完成节点继续，因此本地模型每组只加载一次、卸载一次，而不是每个
-  视频来回加载卸载；产物仍按组增量落地。LLM 阶段执行时引擎在 run 根目录写
-  `keep_model.flag`，节点据此不在每次调用后卸载模型（`nodes/llm.py`），阶段
-  结束由引擎调 `release_local_model()` 统一释放显存，让下一组的 ASR 拿到 GPU
-  （否则本地模型常驻显存会让 whisper 直接 CUDA OOM）。设为 1 即回到「每个视频
+- **按资源调度的在途流水线（2026-09 起）**：批量引擎把待处理视频按
+  `WOV_BATCH_STAGE_GROUP_SIZE`（默认 8）分组，组内**每个视频独立推进自己的
+  阶段**，最多 `WOV_BATCH_PIPELINE_WORKERS`（默认 4）个阶段在途。判定只看资源：
+  阶段是否需要 GPU 由 `wov_app.resources.stage_gpu_need_mb` 给出（提音、**线上
+  端点**的翻译、ASS 都不需要），不需要 GPU 的阶段立刻派发，于是转写能与线上翻译
+  并行、提音能与转写并行（此前组内阶段是串行的，翻译时 GPU 全程空转）。需要 GPU
+  的阶段由进程内 `GpuGate` 串行准入并按"阶段索引最小者优先"派发，因此组内仍是
+  先跑完全部转写再进翻译——**本机 Ollama 模型每组只加载一次**，不需要按"是否
+  云端"写分支。每个视频的 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 已

src/wov_app/batch.py

@@ -11,15 +11,19 @@
   字幕文件（`.srt/.ass/.ssa/.vtt`），说明该视频已有字幕，直接记为 SKIPPED，
   不为它触发任何流水线。运行时（BatchWorker）只消费已定位好的明细列表，
   **不再重新扫描文件夹**（运行期间新增/删除的视频不会改变本次任务的范围）。
-- **分块流水线执行**：视频按 `WOV_BATCH_STAGE_GROUP_SIZE` 分组，组内按节点
-  顺序跑完全部视频（先全部 extract、再全部 ASR、再全部 LLM 翻译、最后 ASS）
-  再进入下一组——本地模型每组只加载一次、卸载一次，产物按组增量落地。
+- **按资源调度的在途流水线**：视频按 `WOV_BATCH_STAGE_GROUP_SIZE` 分组，组内
+  每个视频独立推进自己的阶段（最多 `WOV_BATCH_PIPELINE_WORKERS` 个阶段在途）。
+  阶段是否需要 GPU 由 `wov_app.resources` 判定：提音、线上翻译与 ASS 不需要，
+  与其它视频的转写并行（GPU 不再空转）；需要 GPU 的阶段由进程内门控
+  `GpuGate` 串行准入，并按"阶段索引最小者优先"派发，于是组内先跑完全部转写
+  再进翻译——本机 Ollama 模型每组只加载一次（线上端点则完全不受该顺序约束）。
   阶段边界用 `execute_run(stop_after=节点)` 停在节点（任务保持 RUNNING），
-  LLM 阶段靠 `keep_model.flag` 让节点保持模型常驻，阶段结束由引擎统一释放
-  显存（详见 docs/operations.md#文件夹批量处理）。
+  LLM 阶段靠 `keep_model.flag` 让节点保持模型常驻，组末由引擎统一释放显存
+  （详见 docs/operations.md#文件夹批量处理）。
 - **产物放在视频旁**：每个视频处理完成后，把工作流 `final_outputs` 对应的
-  最终产物文件（字幕流水线即中文 `.srt` 与双目 `.ass`）**复制一份到视频的
-  所在目录**，与 .mp4 放在一起；文件名**对齐媒体库既有约定**：中文字幕存为
+  最终产物文件（字幕流水线即日语转写 `.srt`、中文 `.srt` 与双目 `.ass`）
+  **复制一份到视频的所在目录**，与 .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"
 
@@ -420,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):
-                node_spec = node_by_id[node_id]
-                # 末阶段不传 stop_after：让调度器收尾（final_outputs + COMPLETED）。
-                is_last_stage = stage_index == len(order) - 1
-                executed = False
-                for item in group:
-                    # 暂停检查：批量任务被暂停后停止处理后续视频，等待用户继续。
-                    current = self.db.get_batch_job(job_id)
-                    if current is None or current["status"] == "PAUSED":
+            outcome = self._run_group(
+                job=job,
+                group=group,
+                order=order,
+                node_by_id=node_by_id,
+                definition=definition,
+                version=version,
+                gate=gate,
+            )
+            if outcome == "PAUSED":
                 # 停下前先把已完成/失败项入账，让暂停中的前端看到真实进度。
                 self.db.sync_batch_job_progress(job_id)
-                        logger.info("批量任务 %s 已暂停，停止在视频 %s", job_id, item["video_path"])
+                logger.info("批量任务 %s 已暂停，等待用户继续", job_id)
                 return
-                    self.db.update_batch_job(job_id, current_video=str(item["video_path"]), updated_at=_now_iso())
-                    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，否则会出现“还有待处理视频却已完成”
@@ -496,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,

src/wov_app/config.py

@@ -29,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#文件夹批量处理）。

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,
@@ -561,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"):
@@ -576,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。
-    assert trace == [
-        ("prep", "a.mp4"), ("prep", "b.mp4"),
-        ("translate", "a.mp4"), ("translate", "b.mp4"),
-        ("post", "a.mp4"), ("post", "b.mp4"),
-        ("prep", "c.mp4"), ("translate", "c.mp4"), ("post", "c.mp4"),
-    ]
+    # 验证结果：每个视频的三阶段按 DAG 顺序各跑一次；组 1（a/b）全部跑完才轮到组 2（c）。
+    for name in ("a.mp4", "b.mp4", "c.mp4"):
+        assert [tag for tag, video in trace if video == name] == ["prep", "translate", "post"]
+    first_group = [i for i, entry in enumerate(trace) if entry[1] in ("a.mp4", "b.mp4")]
+    second_group = [i for i, entry in enumerate(trace) if entry[1] == "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"]
@@ -616,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 失败。