vrsub/tests/app/test_batch/test_batch.py

"""src/wov_app/batch.py 的模块级测试（数据 → 测试过程 → 验证结果）。

被测模块：`src/wov_app/batch.py`（文件夹批量处理：扫描定位、旁挂字幕跳过、
引擎执行与产物放置），可独立调用。用例在临时目录构造真实视频/字幕文件与
真实 SQLite 记录，使用真实 echo 节点跑通执行链路。
"""

from __future__ import annotations

import json
from contextlib import contextmanager
from pathlib import Path

import pytest

from wov_app import registry
from wov_app.batch import (
    KEEP_MODEL_FLAG,
    MARKER_NAME,
    SUBTITLE_EXTENSIONS,
    VIDEO_EXTENSIONS,
    BatchWorker,
    _sidecar_product_name,
    create_job,
    list_sidecar_subtitles,
    load_marker,
    remove_job_workspace,
    scan_videos,
)
from wov_app.db import Database
from wov_sdk.models import InvokeRequest, InvokeResponse, WorkflowDefinition


@pytest.fixture(autouse=True)
def _isolate_registry():
    """用例前清空注册表、用例后恢复快照：保证用例看到的是干净基线，
    不受其他模块（如 main 生命周期 register_all）的注册结果影响。"""
    snapshot = dict(registry._registry)
    registry._registry.clear()
    yield
    registry._registry.clear()
    registry._registry.update(snapshot)


def _make_video(path: Path) -> Path:
    """创建真实可读的视频文件（内容不重要，但必须是真实文件）。"""
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(b"\x00\x00\x00\x18ftypmp42" + b"\x00" * 64)
    return path


def _echo_definition(final_output: str = "step.file_uri") -> WorkflowDefinition:
    """单 echo 节点的真实工作流定义。"""
    return WorkflowDefinition.from_dict({
        "name": "批量流程",
        "version": 1,
        "nodes": [{"id": "step", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}}],
        "edges": [],
        "entry_inputs": {"video_uri": "file"},
        "final_outputs": {"result": final_output},
    })


def _published_db(tmp_path: Path, workflow_id: str = "wf") -> Database:
    """建好已发布工作流（含版本）的临时库。"""
    db = Database(tmp_path / "wov.db")
    db.upsert_workflow({
        "id": workflow_id, "name": "批量流程", "description": "", "published": 1,
        "latest_version": 1,
    })
    db.create_workflow_version(workflow_id, 1, _echo_definition().to_dict())
    return db


def _staged_definition() -> WorkflowDefinition:
    """三节点分阶段链路：prep(echo) → translate(llm) → post(echo)，产物为 srt。"""
    return WorkflowDefinition.from_dict({
        "name": "分阶段流程",
        "version": 1,
        "nodes": [
            {"id": "prep", "node_type": "echo", "params": {"node_tag": "prep"},
             "inputs": {"file_uri": "input.video_uri"}},
            {"id": "translate", "node_type": "llm-translate", "params": {"node_tag": "translate"},
             "inputs": {"file_uri": "prep.file_uri"}},
            {"id": "post", "node_type": "echo", "params": {"node_tag": "post"},
             "inputs": {"file_uri": "translate.file_uri"}},
        ],
        "edges": [{"from": "prep", "to": "translate"}, {"from": "translate", "to": "post"}],
        "entry_inputs": {"video_uri": "file"},
        "final_outputs": {"cn_srt": "post.file_uri"},
    })


def _staged_db(tmp_path: Path, workflow_id: str = "wf") -> Database:
    """建好已发布的三节点分阶段工作流库。"""
    db = Database(tmp_path / "wov.db")
    db.upsert_workflow({
        "id": workflow_id, "name": "分阶段流程", "description": "", "published": 1,
        "latest_version": 1,
    })
    db.create_workflow_version(workflow_id, 1, _staged_definition().to_dict())
    return db


@contextmanager
def _recording_nodes(
    trace: list[tuple[str, str]],
    llm_flag_state: list[bool] | None = None,
    fail_stage: tuple[str, str] | None = None,
    flag_trace: list[tuple[str, bool]] | None = None,
):
    """把 echo / llm-translate 节点换成记录调用顺序的假节点（覆盖真实注册表条目）。

    假节点把上游传来的视频名写成 payload.srt 透传给下一节点，因此每个阶段都
    知道自己在处理哪个视频；记录 (节点标签, 视频名)，llm 节点额外记录
    keep_model.flag 是否存在；fail_stage 指定的 (标签, 视频名) 组合返回失败。
    """
    registry.register_all()

    def handler(request: InvokeRequest) -> InvokeResponse:
        tag = str(request.params.get("node_tag"))
        source = str(request.inputs.get("file_uri") or "")
        if source and Path(source).suffix.lower() in VIDEO_EXTENSIONS:
            # 首阶段的输入就是视频文件，后续阶段拿到的是上一阶段的 payload。
            video_name = Path(source).name
        else:
            video_name = Path(source).read_text(encoding="utf-8").strip() if source else ""
        trace.append((tag, video_name))
        run_root = Path(request.output_dir).parent.parent
        if llm_flag_state is not None and tag == "translate":
            llm_flag_state.append((run_root / KEEP_MODEL_FLAG).exists())
        if flag_trace is not None:
            flag_trace.append((tag, (run_root / KEEP_MODEL_FLAG).exists()))
        if fail_stage == (tag, video_name):
            return InvokeResponse(status="failed", error="模拟阶段失败")
        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")
        return InvokeResponse(status="completed", outputs={"file_uri": str(output)})

    for node_type in ("echo", "llm-translate"):
        registry.register(registry.get_node(node_type), handler)
    yield


# ---------------------------------------------------------------------------
# 扫描与旁挂字幕判定
# ---------------------------------------------------------------------------


def test_scan_videos_recursive_and_flat(tmp_path: Path) -> None:
    """递归扫描包含子目录视频；非递归只扫顶层；结果按路径排序。"""
    # 数据：顶层 2 个视频 + 子目录 1 个视频 + 1 个非视频文件。
    _make_video(tmp_path / "b.mp4")
    _make_video(tmp_path / "a.mkv")
    _make_video(tmp_path / "sub" / "c.mp4")
    (tmp_path / "note.txt").write_text("x", encoding="utf-8")

    # 测试过程
    recursive = [p.name for p in scan_videos(tmp_path, recursive=True)]
    flat = [p.name for p in scan_videos(tmp_path, recursive=False)]

    # 验证结果
    assert recursive == ["a.mkv", "b.mp4", "c.mp4"]
    assert flat == ["a.mkv", "b.mp4"]


def test_video_extensions_are_lowercase_dotted() -> None:
    """视频扩展名集合为小写带点形式（与 suffix.lower() 比较一致）。"""
    # 数据：模块常量。
    # 测试过程与验证结果
    assert all(ext.startswith(".") and ext.islower() for ext in VIDEO_EXTENSIONS)
    assert ".mp4" in VIDEO_EXTENSIONS and ".mkv" in VIDEO_EXTENSIONS


def test_list_sidecar_subtitles_matches_by_stem(tmp_path: Path) -> None:
    """视频旁含视频主名的字幕文件被识别（含 CN/dual_eye 等约定命名）。"""
    # 数据：视频 + 三种约定命名的字幕 + 一个无关文件。
    video = _make_video(tmp_path / "movie.mp4")
    expected = [
        tmp_path / "movie.CN.srt",
        tmp_path / "movie.CN_dual_eye.ass",
        tmp_path / "movie.srt",
    ]
    for path in expected:
        path.write_text("1\n", encoding="utf-8")
    (tmp_path / "other.srt").write_text("1\n", encoding="utf-8")

    # 测试过程
    found = list_sidecar_subtitles(video)

    # 验证结果：三个匹配、无关文件不在结果里。
    assert set(found) == set(expected)
    assert (tmp_path / "other.srt") not in found


def test_list_sidecar_subtitles_ignores_non_subtitle_files(tmp_path: Path) -> None:
    """同名但非字幕扩展名的文件不算旁挂字幕。"""
    # 数据：视频 + 同名字幕 + 同名文本。
    video = _make_video(tmp_path / "movie.mp4")
    srt = tmp_path / "movie.srt"
    srt.write_text("1\n", encoding="utf-8")
    (tmp_path / "movie.txt").write_text("x", encoding="utf-8")

    # 测试过程
    found = list_sidecar_subtitles(video)

    # 验证结果
    assert found == [srt]


def test_list_sidecar_subtitles_short_stem_requires_dot_prefix(tmp_path: Path) -> None:
    """视频主名只有一个字符时只接受"主名."前缀，避免 a.mp4 误配 apple.srt。"""
    # 数据：a.mp4 + apple.srt（不应命中）+ a.srt（应命中）。
    video = _make_video(tmp_path / "a.mp4")
    (tmp_path / "apple.srt").write_text("1\n", encoding="utf-8")
    good = tmp_path / "a.srt"
    good.write_text("1\n", encoding="utf-8")

    # 测试过程
    found = list_sidecar_subtitles(video)

    # 验证结果
    assert found == [good]


def test_subtitle_extensions_cover_common_formats() -> None:
    """字幕扩展名覆盖 srt/ass/ssa/vtt。"""
    # 数据：模块常量。
    # 测试过程与验证结果
    assert SUBTITLE_EXTENSIONS == {".srt", ".ass", ".ssa", ".vtt"}


# ---------------------------------------------------------------------------
# 产物命名映射
# ---------------------------------------------------------------------------


def test_sidecar_product_name_maps_srt_and_ass() -> None:
    """最终产物映射为媒体库约定名（.srt → CN.srt，.ass → CN_dual_eye.ass）。"""
    # 数据：视频与两类最终产物。
    video = Path("/videos/movie.mp4")

    # 测试过程与验证结果
    assert _sidecar_product_name(video, Path("/tmp/x.zh-CN.20260101.srt")) == "movie.CN.srt"
    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 产物。
    # 测试过程与验证结果
    assert _sidecar_product_name(Path("/v/movie.mp4"), Path("/tmp/movie.vtt")) == "movie.vtt"


# ---------------------------------------------------------------------------
# 完成标记与工作空间清理
# ---------------------------------------------------------------------------


def test_load_marker_reads_valid_json(tmp_path: Path) -> None:
    """旧版完成标记（batch.done.json）可读回字典。"""
    # 数据：真实标记文件。
    work = tmp_path / "work"
    work.mkdir()
    payload = {"finals": {"cn_srt_uri": "/videos/movie.CN.srt"}}
    (work / MARKER_NAME).write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")

    # 测试过程
    marker = load_marker(work)

    # 验证结果
    assert marker == payload


def test_load_marker_returns_none_for_corrupt_or_missing(tmp_path: Path) -> None:
    """标记缺失或内容损坏时返回 None（走旁挂字幕判定，不报错）。"""
    # 数据：不存在标记 + 损坏标记。
    work = tmp_path / "work"
    work.mkdir()
    assert load_marker(work) is None
    (work / MARKER_NAME).write_text("{broken", encoding="utf-8")

    # 测试过程与验证结果
    assert load_marker(work) is None


def test_remove_job_workspace_only_touches_private_dir(tmp_path: Path, monkeypatch) -> None:
    """删除任务只清理应用私有工作空间，不触碰用户视频目录。"""
    # 数据：私有工作空间 + 用户媒体目录。
    private = tmp_path / "storage" / "batch" / "job-1"
    private.mkdir(parents=True)
    (private / "temp.wav").write_bytes(b"x")
    media = tmp_path / "media"
    _make_video(media / "movie.mp4")
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")

    # 测试过程
    remove_job_workspace("job-1")

    # 验证结果：私有空间被删，用户目录完整。
    assert not private.exists()
    assert (media / "movie.mp4").is_file()


# ---------------------------------------------------------------------------
# 创建批量任务：一次性定位
# ---------------------------------------------------------------------------


def test_create_job_registers_pending_and_skipped(tmp_path: Path) -> None:
    """创建任务一次性定位视频：无字幕记 PENDING，已有字幕记 SKIPPED。"""
    # 数据：3 个视频，其中一个已有旁挂字幕。
    folder = tmp_path / "videos"
    _make_video(folder / "a.mp4")
    _make_video(folder / "b.mp4")
    _make_video(folder / "c.mp4")
    (folder / "b.CN.srt").write_text("1\n", encoding="utf-8")
    db = _published_db(tmp_path)

    # 测试过程
    job_id = create_job(db, str(folder), "wf", recursive=False)
    videos = db.list_batch_videos(job_id)

    # 验证结果：状态分布正确，总数只算待处理。
    statuses = {Path(v["video_path"]).name: v["status"] for v in videos}
    assert statuses == {"a.mp4": "PENDING", "b.mp4": "SKIPPED", "c.mp4": "PENDING"}
    job = db.get_batch_job(job_id)
    assert job["total"] == 2
    assert job["status"] == "QUEUED"


def test_create_job_completes_immediately_when_all_skipped(tmp_path: Path) -> None:
    """全部视频都已有字幕时任务直接完成，不排队不触发流水线。"""
    # 数据：两个视频都带字幕。
    folder = tmp_path / "videos"
    _make_video(folder / "a.mp4")
    _make_video(folder / "b.mp4")
    (folder / "a.srt").write_text("1\n", encoding="utf-8")
    (folder / "b.srt").write_text("1\n", encoding="utf-8")
    db = _published_db(tmp_path)

    # 测试过程
    job_id = create_job(db, str(folder), "wf")

    # 验证结果
    job = db.get_batch_job(job_id)
    assert job["status"] == "COMPLETED"
    assert job["total"] == 0
    assert db.next_queued_batch_job() is None


def test_create_job_uses_private_work_dir(tmp_path: Path, monkeypatch) -> None:
    """明细的工作空间位于应用私有目录（与用户媒体库隔离）。"""
    # 数据：一个视频。
    folder = tmp_path / "videos"
    _make_video(folder / "a.mp4")
    db = _published_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")

    # 测试过程
    job_id = create_job(db, str(folder), "wf")
    video = db.list_batch_videos(job_id)[0]

    # 验证结果：work_dir 在私有 storage/batch 下，且不在视频目录内。
    work_dir = Path(video["work_dir"])
    assert (tmp_path / "storage" / "batch") in work_dir.parents
    assert folder not in work_dir.parents


@pytest.mark.parametrize(
    ("folder_setup", "workflow_id", "message"),
    [
        ("missing", "wf", "folder not found"),
        ("empty", "wf", "no videos found"),
        ("ok", "unknown", "published workflow not found"),
    ],
)
def test_create_job_rejects_invalid_input(tmp_path: Path, folder_setup: str, workflow_id: str, message: str) -> None:
    """校验失败时抛 ValueError（路由层转 422）：目录缺失/无视频/工作流未发布。"""
    # 数据：按参数准备目录与工作流。
    folder = tmp_path / "videos"
    if folder_setup == "empty":
        folder.mkdir()
    elif folder_setup == "ok":
        _make_video(folder / "a.mp4")
    db = _published_db(tmp_path)

    # 测试过程与验证结果
    with pytest.raises(ValueError, match=message):
        create_job(db, str(folder), workflow_id)


def test_create_job_rejects_workflow_without_version(tmp_path: Path) -> None:
    """已发布但无版本记录的工作流被拒绝（无法执行）。"""
    # 数据：有工作流记录但无版本。
    folder = tmp_path / "videos"
    _make_video(folder / "a.mp4")
    db = Database(tmp_path / "wov.db")
    db.upsert_workflow({"id": "wf", "name": "无版本", "description": "", "published": 1})

    # 测试过程与验证结果
    with pytest.raises(ValueError, match="no version"):
        create_job(db, str(folder), "wf")


# ---------------------------------------------------------------------------
# 引擎：执行、产物放置、暂停
# ---------------------------------------------------------------------------


def test_worker_processes_pending_video_end_to_end(tmp_path: Path, monkeypatch) -> None:
    """引擎处理待处理视频：跑通流水线、产物放到视频旁、明细与任务标记完成。"""
    # 数据：一个视频 + echo 单节点工作流（产物为复制后的输入文件）。
    folder = tmp_path / "videos"
    _make_video(folder / "movie.mp4")
    db = _published_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    registry.register_all()
    job_id = create_job(db, str(folder), "wf")

    # 测试过程：直接驱动一轮处理（避免后台线程时序不确定）。
    worker = BatchWorker(db, interval_seconds=999)
    worker._process_job(db.get_batch_job(job_id))

    # 验证结果：明细与任务完成，产物按约定名放到视频旁，工作空间被清理。
    item = db.list_batch_videos(job_id)[0]
    assert item["status"] == "COMPLETED"
    assert db.get_batch_job(job_id)["status"] == "COMPLETED"
    # echo 节点产物是 .txt，按约定保留原文件名放置在视频旁。
    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）。"""
    # 数据：一个已带字幕的视频。
    folder = tmp_path / "videos"
    _make_video(folder / "movie.mp4")
    (folder / "movie.CN.srt").write_text("1\n", encoding="utf-8")
    _make_video(folder / "other.mp4")
    db = _published_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    registry.register_all()

    # 测试过程
    job_id = create_job(db, str(folder), "wf")

    # 验证结果：SKIPPED 视频没有 run_id。
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job_id)}
    assert videos["movie.mp4"]["status"] == "SKIPPED"
    assert videos["movie.mp4"]["run_id"] is None
    assert videos["other.mp4"]["status"] == "PENDING"


def test_worker_pause_sets_job_paused(tmp_path: Path, monkeypatch) -> None:
    """暂停批量任务：任务状态置 PAUSED，待处理视频不被推进。"""
    # 数据：一个待处理视频的任务。
    folder = tmp_path / "videos"
    _make_video(folder / "a.mp4")
    db = _published_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    job_id = create_job(db, str(folder), "wf")

    # 测试过程
    db.update_batch_job(job_id, status="PAUSED", updated_at="2026-09-01T01:00:00+00:00")

    # 验证结果：不会被 next_queued_batch_job 拾起（等待显式 resume）。
    assert db.next_queued_batch_job() is None
    assert db.get_batch_job(job_id)["status"] == "PAUSED"


def test_worker_start_stop_idempotent(tmp_path: Path) -> None:
    """引擎 start 重复调用不产生多余线程；stop 正常结束。"""
    # 数据：空库。
    db = Database(tmp_path / "wov.db")
    worker = BatchWorker(db, interval_seconds=999)

    # 测试过程
    worker.start()
    first = worker._thread
    worker.start()
    second = worker._thread
    worker.stop()

    # 验证结果
    assert first is second
    assert worker._thread is None


def test_worker_processes_recovered_zombie_job(tmp_path: Path, monkeypatch) -> None:
    """僵尸任务（COMPLETED 但明细仍 PENDING）被恢复后能真正处理完剩余视频。

    曾出现「任务已完成、视频仍未处理」的僵尸状态（完成标记先于视频收尾写出），
    而引擎只拾取 QUEUED：不恢复就永远不会再处理那个视频。
    """
    # 数据：已发布工作流 + 一个视频；创建任务后伪造成僵尸状态。
    folder = tmp_path / "videos"
    _make_video(folder / "movie.mp4")
    db = _published_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    registry.register_all()
    job_id = create_job(db, str(folder), "wf")
    db.update_batch_job(job_id, status="COMPLETED", progress=1.0,
                        updated_at="2026-09-01T00:00:00+00:00")

    # 测试过程：重启恢复把僵尸任务放回队列，引擎拾起后处理剩余视频。
    db.recover_interrupted_batch_jobs("2026-09-01T01:00:00+00:00")
    worker = BatchWorker(db, interval_seconds=999)
    worker._process_job(db.get_batch_job(job_id))

    # 验证结果：视频真的处理完、产物放到视频旁、任务保持完成。
    item = db.list_batch_videos(job_id)[0]
    assert item["status"] == "COMPLETED"
    assert db.get_batch_job(job_id)["status"] == "COMPLETED"
    assert list(folder.glob("movie.*")), "应在视频旁放置最终产物"


# ---------------------------------------------------------------------------
# 引擎：分块流水线（阶段化执行）
# ---------------------------------------------------------------------------


def test_worker_runs_grouped_stage_pipeline(tmp_path: Path, monkeypatch) -> None:
    """分块流水线：组内按节点顺序跑完全部视频，而不是每个视频跑完整链路。"""
    # 数据：3 个视频 + 三节点链路，分组大小 2（前两个一组、第三个一组）。
    folder = tmp_path / "videos"
    for name in ("a.mp4", "b.mp4", "c.mp4"):
        _make_video(folder / name)
    db = _staged_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    monkeypatch.setattr("wov_app.batch.BATCH_STAGE_GROUP_SIZE", 2)
    trace: list[tuple[str, str]] = []

    # 测试过程：用记录调用顺序的假节点驱动引擎跑一轮。
    with _recording_nodes(trace):
        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"),
    ]
    # 三个视频都完成且产物按约定名落到视频旁。
    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"]


def test_worker_releases_local_llm_once_per_group(tmp_path: Path, monkeypatch) -> None:
    """LLM 阶段结束由引擎统一释放显存：每组一次，而不是每个视频一次。"""
    # 数据：3 个视频 + 分组 2，记录释放调用与 LLM 调用时的常驻信号状态。
    folder = tmp_path / "videos"
    for name in ("a.mp4", "b.mp4", "c.mp4"):
        _make_video(folder / name)
    db = _staged_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    monkeypatch.setattr("wov_app.batch.BATCH_STAGE_GROUP_SIZE", 2)
    releases: list[str | None] = []
    monkeypatch.setattr(
        "wov_app.batch.release_local_model",
        lambda model=None: releases.append(model),
    )
    trace: list[tuple[str, str]] = []
    flag_state: list[bool] = []

    # 测试过程
    with _recording_nodes(trace, llm_flag_state=flag_state):
        job_id = create_job(db, str(folder), "wf")
        BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))

    # 验证结果：两组各释放一次；每次 LLM 调用都在“保持常驻”信号下执行；信号已清理。
    assert releases == [None, None]
    assert flag_state == [True, True, True]
    assert not list((tmp_path / "storage").rglob(KEEP_MODEL_FLAG))


def test_worker_defers_video_failed_in_earlier_stage(tmp_path: Path, monkeypatch) -> None:
    """上一阶段失败的视频不在后续阶段重跑（避免 LLM 已常驻时重跑 ASR 抢显存）。"""
    # 数据：2 个视频（同一组）+ 三节点链路，prep 阶段让 a 失败。
    folder = tmp_path / "videos"
    for name in ("a.mp4", "b.mp4"):
        _make_video(folder / name)
    db = _staged_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    monkeypatch.setattr("wov_app.batch.BATCH_STAGE_GROUP_SIZE", 2)
    monkeypatch.setattr("wov_app.batch.release_local_model", lambda model=None: None)
    trace: list[tuple[str, str]] = []

    # 测试过程
    with _recording_nodes(trace, fail_stage=("prep", "a.mp4")):
        job_id = create_job(db, str(folder), "wf")
        BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))

    # 验证结果：a 只在 prep 出现一次并记为 FAILED；b 三阶段跑完并落地产物。
    assert [entry for entry in trace if entry[1] == "a.mp4"] == [("prep", "a.mp4")]
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job_id)}
    assert videos["a.mp4"]["status"] == "FAILED"
    assert videos["b.mp4"]["status"] == "COMPLETED"
    assert (folder / "b.CN.srt").is_file()
    assert not (folder / "a.CN.srt").exists()
    assert db.get_batch_job(job_id)["failed"] == 1


# ---------------------------------------------------------------------------
# 引擎：创建期间拾起任务（明细未登记完）的自愈
# ---------------------------------------------------------------------------


def test_worker_requeues_job_without_details(tmp_path: Path, monkeypatch) -> None:
    """任务行先于明细写入：拾起到无明细的任务时保持 QUEUED，不按空任务收尾。"""
    # 数据：只有任务行、还没写任何视频明细的批量任务。
    db = _published_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    db.create_batch_job({
        "id": "batch-registering", "folder_path": str(tmp_path), "workflow_id": "wf",
        "recursive": 1, "status": "QUEUED", "progress": 0, "total": 0, "done": 0,
        "failed": 0, "current_video": None, "error": None,
        "created_at": "2026-09-01T00:00:00+00:00", "updated_at": "2026-09-01T00:00:00+00:00",
    })

    # 测试过程
    worker = BatchWorker(db, interval_seconds=999)
    worker._process_job(db.get_batch_job("batch-registering"))

    # 验证结果：任务仍在排队等待登记完成，而不是被标成 COMPLETED。
    assert db.get_batch_job("batch-registering")["status"] == "QUEUED"
    assert db.next_queued_batch_job() is not None


def test_worker_requeues_job_when_video_registered_mid_pass(tmp_path: Path, monkeypatch) -> None:
    """明细在引擎处理中途才登记进来：本轮结束后置回 QUEUED，下一轮续跑完成。"""
    # 数据：1 个视频 + 单节点工作流；处理首个阶段时登记第二个视频（模拟创建中拾起）。
    folder = tmp_path / "videos"
    _make_video(folder / "first.mp4")
    db = _published_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    registry.register_all()
    job_id = create_job(db, str(folder), "wf")
    worker = BatchWorker(db, interval_seconds=999)
    original_stage = worker._run_stage
    injected = {"done": False}

    def stage_with_late_video(*args, **kwargs):
        # 模拟 create_job 仍在写明细：引擎快照之后新视频才出现在数据库里。
        if not injected["done"]:
            injected["done"] = True
            late_video = _make_video(folder / "second.mp4")
            db.create_batch_video({
                "id": "bv_late", "job_id": job_id, "video_path": str(late_video),
                "work_dir": str(tmp_path / "storage" / "batch" / job_id / "bv_late"),
                "run_id": None, "status": "PENDING", "error": None,
                "created_at": "2026-09-01T00:00:01+00:00", "updated_at": "2026-09-01T00:00:01+00:00",
            })
        return original_stage(*args, **kwargs)

    monkeypatch.setattr(worker, "_run_stage", stage_with_late_video)

    # 测试过程：第一轮只看到 first.mp4。
    worker._process_job(db.get_batch_job(job_id))

    # 验证结果：job 被置回 QUEUED 等待下一轮，新视频还没被处理。
    assert db.get_batch_job(job_id)["status"] == "QUEUED"
    statuses = {Path(v["video_path"]).name: v["status"] for v in db.list_batch_videos(job_id)}
    assert statuses == {"first.mp4": "COMPLETED", "second.mp4": "PENDING"}

    # 测试过程：下一轮引擎拾起后处理剩余视频并收尾。
    second = BatchWorker(db, interval_seconds=999)
    second._process_job(db.get_batch_job(job_id))

    # 验证结果：两个视频都完成、任务完成、产物都在视频旁（真实 echo 节点产物为 echo.txt）。
    statuses = {Path(v["video_path"]).name: v["status"] for v in db.list_batch_videos(job_id)}
    assert statuses == {"first.mp4": "COMPLETED", "second.mp4": "COMPLETED"}
    assert db.get_batch_job(job_id)["status"] == "COMPLETED"
    # 真实 echo 节点的最终产物保留原扩展名（非 .srt/.ass），按视频主名放置。
    assert list(folder.glob("first.*")) and list(folder.glob("second.*"))


def test_worker_removes_empty_job_workspace_after_completion(tmp_path: Path, monkeypatch) -> None:
    """任务全部完成后删掉任务级工作空间目录（每视频工作空间已各自清理）。"""
    # 数据：一个视频的批量任务。
    folder = tmp_path / "videos"
    _make_video(folder / "movie.mp4")
    db = _published_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    registry.register_all()
    job_id = create_job(db, str(folder), "wf")

    # 测试过程
    BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))

    # 验证结果：任务完成，任务级目录（收尾后只剩空壳）被删除。
    assert db.get_batch_job(job_id)["status"] == "COMPLETED"
    assert not (tmp_path / "storage" / "batch" / job_id).exists()


def test_worker_keeps_job_workspace_when_video_failed(tmp_path: Path, monkeypatch) -> None:
    """有失败视频时保留任务工作空间（失败视频的中间产物供断点重试）。"""
    # 数据：三节点链路，prep 阶段让视频失败。
    folder = tmp_path / "videos"
    _make_video(folder / "a.mp4")
    db = _staged_db(tmp_path)
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch")
    monkeypatch.setattr("wov_app.batch.BATCH_STAGE_GROUP_SIZE", 2)
    monkeypatch.setattr("wov_app.batch.release_local_model", lambda model=None: None)

    # 测试过程
    with _recording_nodes([], fail_stage=("prep", "a.mp4")):
        job_id = create_job(db, str(folder), "wf")
        BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))

    # 验证结果：视频失败、任务工作空间仍在（可重试）。
    assert db.get_batch_job(job_id)["failed"] == 1
    assert (tmp_path / "storage" / "batch" / job_id).exists()


def test_worker_clears_stale_keep_model_flag_before_stage(tmp_path: Path, monkeypatch) -> None:
    """强杀残留的 keep_model.flag 不会带到后续阶段：非 LLM 节点不应看到它。"""
    # 数据：两节点链路 + 已存在的 run（工作空间里残留强杀时的 keep_model.flag）。
    folder = tmp_path / "videos"
    _make_video(folder / "a.mp4")
    db = _staged_db(tmp_path)
    work_root = tmp_path / "storage" / "batch"
    monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", work_root)
    monkeypatch.setattr("wov_app.batch.BATCH_STAGE_GROUP_SIZE", 8)
    monkeypatch.setattr("wov_app.batch.release_local_model", lambda model=None: None)
    job_id = create_job(db, str(folder), "wf")
    video = db.list_batch_videos(job_id)[0]
    run_id = "run_stale_flag"
    db.update_batch_video(video["id"], run_id=run_id, updated_at="2026-09-01T00:00:00+00:00")
    db.create_run({
        "id": run_id, "workflow_id": "wf", "workflow_version": 1, "status": "QUEUED",
        "current_node_id": None, "progress": 0.0, "error": None,
        "input_uri": str(folder / "a.mp4"), "param_overrides": None, "source": "batch",
        "created_at": "2026-09-01T00:00:00+00:00", "updated_at": "2026-09-01T00:00:00+00:00",
    })
    run_dir = Path(video["work_dir"]) / "runs" / run_id
    run_dir.mkdir(parents=True, exist_ok=True)
    (run_dir / KEEP_MODEL_FLAG).write_text("", encoding="utf-8")
    flag_trace: list[tuple[str, bool]] = []

    # 测试过程
    with _recording_nodes([], flag_trace=flag_trace):
        BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id))

    # 验证结果：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()