fix: 保持产物收尾幂等并保护批量成品完整性

AGENTS.md

@@ -135,8 +135,13 @@ whisper 节点按以下顺序解析模型路径，默认避免从远端下载：
 等假名，真实短对话（そこ/やばい/ねえ/やだ/えへへ）天然不命中。仅 decode_full
 生效，demo 等 VAD 链路不受影响。
 
-最终产物按 `上传文件名.标识.时间戳` 重命名（如 `test01.zh-CN.20260815123000.srt`），
-标识优先取节点的 `target_language` 参数，否则用产物别名。
+最终产物按 `上传文件名.标识.时间戳` 命名（如 `test01.zh-CN.20260815123000.srt`），
+标识优先取节点的 `target_language` 参数，否则用产物别名。**审查 R03 修复**：
+保留节点原始文件及 URI，把成品副本存入 `runs/<run_id>/finals/output-<编码别名>/`；
+时间戳固定取 run 创建时间，重复收尾覆盖相同路径，多个别名分目录避免冲突。
+复制先写同目录临时文件，再原子替换目标，失败不登记残缺文件；`final_outputs`
+声明的引用或文件缺失时任务失败，不能标完成。旧版本原文件已改名但最终别名记录
+仍指向有效文件时允许复用；原文件与成品都丢失时明确报错。
 
 ### 任务参数覆盖（前端框选）
 
@@ -356,6 +361,10 @@ http://127.0.0.1:8000/docs      API 文档
   存为 `<视频名>.CN.srt`、双目字幕存为 `<视频名>.CN_dual_eye.ass`（稳定无时间戳，
   `_sidecar_product_name` 映射，其余扩展名产物保留原文件名；同名目标直接覆盖）。
   文件名含视频主名，下次批量扫描会命中"已有字幕"规则直接跳过该视频。
+- **成品放置校验（审查 R03）**：先预检全部 `final_outputs` 对应的记录与文件，
+  缺任一项即失败，不开始覆盖视频旁成品；全部齐备后逐文件原子替换。复制失败时
+  视频记 FAILED，保留 run、工作空间和已放置的完整成品，供修复后幂等重试。
+  原子替换保证单文件完整，不代表多个成品的跨文件事务或断电持久性。
 - **过程文件清理（2026-09 起）**：视频收尾完成后删除该视频的整个工作空间与
   run 记录（音频/分块/帧图/节点产物不留残），防止媒体库把切片数据当视频入库。
   工作空间位于**应用私有目录** `data/storage/batch/<job_id>/<bv_id>/`

docs/代码审查问题跟踪.md

@@ -10,7 +10,7 @@
 | --- | --- | --- | --- | --- |
 | R01 | P0 | 普通删除接口可能删除批量源视频目录；清理逻辑从输入路径推导删除范围 | 已修复 | 普通接口拒绝单独删除批量 run；手动与自动清理只删除该上传任务的私有目录；视频、旁挂字幕及其他任务文件不受影响 |
 | R02 | P1 | 自适应线程池限流后可能扩容；退出标记位于积压队列尾部，缩容不及时 | 已修复 | 限流不增加并发；降低目标后不再超额提交；真实积压队列验证 |
-| R03 | P1 | 最后节点执行时暂停再继续会覆盖有效下载 URI；批量缺产物仍清理并完成 | 待处理 | 收尾幂等，恢复后全部必需产物可下载；缺产物不清理工作空间 |
+| R03 | P1 | 最后节点执行时暂停再继续会覆盖有效下载 URI；批量缺产物仍清理并完成 | 已修复 | 收尾幂等，恢复后全部必需产物可下载；缺产物不清理工作空间 |
 | R04 | P1 | 环形 DAG 可通过保存校验，执行失败后仍在 QUEUED 堵塞队列 | 待处理 | 保存/发布拒绝无效 DAG；历史无效任务进入 FAILED，不阻塞后续任务 |
 | R05 | P1 | 翻译按固定四行解析 SRT；补齐行数不能保证文本与时间轴对应 | 待处理 | 合法多行 cue 正确解析；按稳定 ID 回填译文并校验缺失、重复项 |
 | R06 | P1 | OCR 跨空白帧合并相同字幕；临时请求失败被永久存为无文字 | 待处理 | 空白帧结束当前字幕段；无文字和可重试失败分开存档 |
@@ -59,6 +59,18 @@
 - 补充重试额度用例后：`uv run pytest tests/test_adaptive_pool.py -q`，22 passed，0.27 秒。连续限流后的第二轮 map 保持 1 并发，进度序号重新从 1 开始。`git diff --check` 通过。
 - 状态：修复及验证完成，纳入 `fix/review-improvements` 分支；未部署。尚未对真实服务配额下的吞吐做性能结论；已发出请求的终止依赖节点自身超时。
 
+## R03 修复记录
+
+- 根因：最终产物原地改名后，节点输出 URI 未同步，恢复时旧路径覆盖有效最终记录；批量放置对缺失产物直接跳过，随后仍清理并标记完成。
+- 实现：[scheduler.py](../src/wov_app/scheduler.py) 保留节点原文件，把命名成品复制至 run 的 finals 别名目录；时间戳固定取 run 创建时间，重复收尾路径稳定。缺少最终引用或文件时进入 FAILED。
+- 兼容：旧版本已经改名但最终别名记录仍有效时复用该路径，不再覆盖为失效节点 URI；原始与最终文件均丢失时明确报错。
+- 文件操作：[storage.py](../src/wov_app/storage.py) 提供同目录临时文件复制和原子替换，失败清理临时文件、保留源及原目标。[batch.py](../src/wov_app/batch.py) 先预检全部必需成品，复制成功后才清理 run 和工作空间；中途失败保留断点供重试。
+- TDD 红：新增/强化的 8 个场景失败，覆盖暂停恢复链接、旧改名记录、多别名同源、缺少输出引用/文件、批量缺产物清理、节点原始 URI 保留及复制故障。
+- 相关回归：`uv run pytest tests/test_finalization.py tests/test_scheduler.py tests/test_batch.py tests/test_apps_api.py tests/test_db.py -q --tb=short`，97 passed，3.22 秒。
+- 补充批量复制故障及安全回归：`uv run pytest tests/test_finalization.py tests/test_maintenance.py tests/test_run_deletion_safety.py -q --tb=short`，22 passed，5.83 秒。真实视频/字幕验证写满故障后旧成品和工作空间完整，故障解除可成功放置并清理；API 下载验证原始文件与两个最终别名均可读取。仅有既存 Starlette/httpx 弃用警告。
+- 状态：修复及验证完成，纳入 `fix/review-improvements` 分支；未部署。原子性为单文件级，多文件中途失败允许部分完整成品已经更新，重试会重新放置；不承诺断电持久性。
+- 剩余边界：批量固定工作流版本仍由 R08 跟踪；此修复不会自动重建已经丢失的历史成品。
+
 ## 审查基线
 
 - 修复前全套测试：`uv run pytest`，369 passed、6 skipped，76.75 秒。

src/wov_app/batch.py

@@ -48,6 +48,7 @@ from wov_app.config import BATCH_INTERVAL_SECONDS, STORAGE_DIR
 from wov_app.db import Database
 from wov_app.logging import get_logger
 from wov_app.scheduler import WorkflowScheduler
+from wov_app.storage import atomic_copy
 from wov_sdk.models import WorkflowDefinition
 
 # 批量引擎运行日志：任务进度、视频逐个处理与暂停/续跑等状态变化。
@@ -531,18 +532,22 @@ class BatchWorker:
         "已有字幕"规则跳过该视频。同名目标直接覆盖：可能是上一次运行/旧工作流
         留下的旧内容，应以本次产物为准。
         """
-        placed: list[str] = []
-        video.parent.mkdir(parents=True, exist_ok=True)
+        # 先校验全部必需输出，缺文件时不覆盖任何视频旁成品，更不能继续清理。
+        products: list[tuple[Path, Path]] = []
         for alias in definition.final_outputs:
             artifact = self.db.get_artifact(run_id, alias)
             if artifact is None:
-                continue
+                raise ValueError(f"missing final artifact: {alias}")
             source = Path(artifact["uri"])
             if not source.is_file():
-                continue
+                raise ValueError(f"missing final artifact file: {alias} ({source})")
             target = video.parent / _sidecar_product_name(video, source)
-            # 目标名稳定 → 直接覆盖写入，避免旧同名产物被"大小一致复用"误保留。
-            shutil.copy2(source, target)
+            products.append((source, target))
+        placed: list[str] = []
+        for source, target in products:
+            # 稳定目标名 + 原子替换：复制失败保留旧成品，异常交调用方记录 FAILED。
+            # 多文件中途失败仍保留完整工作空间，下次可以幂等地重新放置。
+            atomic_copy(source, target)
             placed.append(target.name)
         return placed
 

src/wov_app/scheduler.py

@@ -18,6 +18,7 @@ from wov_app import registry
 from wov_app.config import SCHEDULER_INTERVAL_SECONDS
 from wov_app.db import Database
 from wov_app.logging import get_logger
+from wov_app.storage import atomic_copy
 
 # 调度器运行日志：节点进度、暂停/续跑等状态变化。
 logger = get_logger("scheduler")
@@ -237,8 +238,10 @@ class WorkflowScheduler:
             # 处理 final_outputs，为用户端提供简洁的下载别名。
             for alias, ref in definition.final_outputs.items():
                 resolved = self._resolve_ref(ref, run.get("input_uri"), outputs_by_node)
-                if resolved is not None:
-                    # 最终产物按 上传文件名.标识.时间戳 重命名，区分语言与版本。
+                if resolved is None:
+                    raise ValueError(f"missing final output: {alias} ({ref})")
+                # 成品按 上传文件名.标识.时间戳 命名，保留节点原始文件与 URI。
+                # 同一个 run 重复收尾使用稳定路径，不再因暂停/重启反复改名。
                 resolved = self._final_artifact_uri(resolved, run, definition, alias, ref)
                 self.db.create_artifact(
                     {
@@ -285,25 +288,35 @@ class WorkflowScheduler:
         alias: str,
         ref: str,
     ) -> str:
-        """把最终产物重命名为 上传文件名.标识.时间戳 并返回新 URI。
+        """生成命名成品副本并返回稳定 URI，节点原始产物始终保留。
 
-        标识优先取产出节点的 target_language 参数（如 zh-CN），否则回退为
-        产物别名；时间戳取当前时刻，用于区分同一上传文件的多次运行版本。
-        重命名在原地进行（同目录），不复制文件。
+        标识优先取产出节点的 target_language，否则回退别名；时间戳固定为
+        run 创建时间。每个别名单独目录，防止相同语言/扩展名的输出互相覆盖。
+        复制使用原子替换；收尾重复执行覆盖相同目标，不生成新的时间戳副本。
         """
         source = Path(resolved)
-        # 续跑等场景下源文件可能已被上次收尾重命名过：不再重命名，原样返回。
         if not source.is_file():
-            return resolved
+            # 兼容旧版本：原文件已改名，但最终别名仍记录有效路径时直接复用。
+            existing = self.db.get_artifact(run["id"], alias)
+            if existing is not None and Path(existing["uri"]).is_file():
+                return existing["uri"]
+            raise ValueError(f"missing final output file: {alias} ({resolved})")
         # 基础名来自上传文件名；无上传文件时退回通用名称 subtitle。
         base = Path(run["input_uri"]).stem if run.get("input_uri") else "subtitle"
         # 通过最终输出引用定位产出节点，取其语言参数作为标识。
         node_id = ref.partition(".")[0]
         node = next((item for item in definition.nodes if item.id == node_id), None)
         tag = (node.params.get("target_language") if node else None) or alias
-        timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
-        new_path = source.with_name(f"{base}.{tag}.{timestamp}{source.suffix}")
-        source.rename(new_path)
+        # 编码目录别名避免路径分隔符；保留常规 cn_srt/ass 名称便于排查。
+        from urllib.parse import quote
+
+        timestamp = datetime.fromisoformat(run["created_at"]).strftime("%Y%m%d%H%M%S")
+        final_dir = self.storage_dir / "runs" / run["id"] / "finals" / ("output-" + quote(alias, safe=""))
+        filename = f"{base}.{tag}.{timestamp}{source.suffix}"
+        if Path(filename).name != filename or "\\" in filename:
+            raise ValueError(f"invalid final output filename: {alias}")
+        new_path = final_dir / filename
+        atomic_copy(source, new_path)
         return str(new_path)
 
     @staticmethod

src/wov_app/storage.py

@@ -0,0 +1,26 @@
+"""产物文件操作：以同目录临时文件复制并原子替换，避免暴露写到一半的成品。"""
+
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+
+def atomic_copy(source: Path, target: Path) -> None:
+    """保留源文件，完整复制后替换目标；失败清理临时文件并保留原目标。
+
+    临时文件与目标位于同一目录，确保 replace 不跨文件系统；关闭句柄后再
+    copy2/replace，兼容 Windows。原子性针对单个文件，不承诺多文件事务或
+    断电持久性；调用方仅在全部必需文件复制成功后登记完成并清理工作空间。
+    """
+    target.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(
+        dir=target.parent, prefix=f".{target.name}.", suffix=".tmp", delete=False
+    ) as handle:
+        temporary = Path(handle.name)
+    try:
+        shutil.copy2(source, temporary)
+        temporary.replace(target)
+    finally:
+        temporary.unlink(missing_ok=True)

tests/test_batch.py

@@ -682,13 +682,16 @@ def test_batch_worker_pending_leftover_never_marks_completed(tmp_path, monkeypat
     monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _CrashScheduler)
     # 放开假调度器的崩溃：第二次调用时不再删 run，让真实调度器跑通。
     class _RecoveringScheduler:
-        """第二轮：不再崩溃，把 run 置 COMPLETED（由引擎收尾放产物）。"""
+        """第二轮：真实执行工作流并产生成品，由引擎收尾放置。"""
 
         def __init__(self, db: Database, work_dir: Path) -> None:
             self.db = db
 
         def execute_run(self, run_id: str) -> None:
-            self.db.update_run(run_id, status="COMPLETED", updated_at=_now_iso())
+            from wov_app.scheduler import WorkflowScheduler
+
+            video = self.db.list_batch_videos(job["id"])[0]
+            WorkflowScheduler(self.db, Path(video["work_dir"])).execute_run(run_id)
 
     monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _RecoveringScheduler)
     worker._process_job(job)
@@ -701,7 +704,7 @@ def test_batch_worker_already_completed_runs_finalized(tmp_path) -> None:
     """run 已完成但视频未标记（收尾前中断）：直接放置产物、清理并标记完成。
 
     覆盖 _place_products 各分支：产物齐全（复制）、产物记录存在但文件丢失
-    （跳过）、无产物记录（跳过）——完成后均清理工作空间与 run 记录。
+    （报错）、无产物记录（报错）——只有成品齐全才清理工作空间与 run 记录。
     """
     db = _db(tmp_path)
     _seed_echo_workflow(db)
@@ -714,6 +717,7 @@ def test_batch_worker_already_completed_runs_finalized(tmp_path) -> None:
         """创建 COMPLETED run；可选产物文件与产物记录。"""
         _create_run(db, run_id, folder, status="COMPLETED", name=name)
         work_dir = Path(videos[name]["work_dir"])
+        work_dir.mkdir(parents=True)
         if with_artifact:
             path = work_dir / "runs" / run_id / "steps" / "step" / f"{name}.result.txt"
             db.create_artifact(
@@ -739,20 +743,24 @@ def test_batch_worker_already_completed_runs_finalized(tmp_path) -> None:
 
     BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job["id"]))
     videos = {Path(v["video_path"]).stem: v for v in db.list_batch_videos(job["id"])}
-    assert all(videos[name]["status"] == "COMPLETED" for name in ("a", "b", "c"))
+    assert videos["a"]["status"] == "COMPLETED"
+    assert all(videos[name]["status"] == "FAILED" for name in ("b", "c"))
     # a 的产物复制到视频旁；b/c 无产物可复制。
     product = _find_product(folder, "a", "result")
     assert product.read_text(encoding="utf-8") == "产物 a"
     assert not list(folder.glob("b.result.*")) and not list(folder.glob("c.result.*"))
-    # 三个视频的工作空间与 run 记录都被清理。
-    for name in ("a", "b", "c"):
-        assert not Path(videos[name]["work_dir"]).exists()
-        assert db.get_run(f"run_done_{name}") is None
-    assert db.list_runs() == []
+    # a 清理完成；b/c 必须保留工作空间、关联 run 与错误供修复后重试。
+    assert not Path(videos["a"]["work_dir"]).exists()
+    assert db.get_run("run_done_a") is None
+    for name in ("b", "c"):
+        assert Path(videos[name]["work_dir"]).exists()
+        assert db.get_run(f"run_done_{name}") is not None
+        assert videos[name]["run_id"] == f"run_done_{name}"
+        assert "result" in videos[name]["error"]
 
 
 def test_batch_worker_place_products_branches(tmp_path) -> None:
-    """_place_products：无产物记录/文件丢失跳过；.srt/.ass 按库内约定命名覆盖；
+    """_place_products：无产物记录/文件丢失报错；.srt/.ass 按库内约定命名覆盖；
     其他扩展名保留原文件名复制。"""
     db = _db(tmp_path)
     _seed_echo_workflow(db)
@@ -805,6 +813,14 @@ def test_batch_worker_place_products_branches(tmp_path) -> None:
         {"run_id": run_id, "node_id": "step", "name": "alias_txt", "uri": str(src_txt), "mime_type": "text/plain", "size": src_txt.stat().st_size}
     )
 
+    # 所有必需成品预检通过前不开始覆盖，缺失原因中包含具体别名。
+    with pytest.raises(ValueError, match="alias_none"):
+        worker._place_products(run_id, video, WorkflowDefinition.from_dict(definition))
+    del definition["final_outputs"]["alias_none"]
+    with pytest.raises(ValueError, match="alias_lost"):
+        worker._place_products(run_id, video, WorkflowDefinition.from_dict(definition))
+    assert target_srt.read_text(encoding="utf-8") == "旧字幕内容"
+    del definition["final_outputs"]["alias_lost"]
     placed = worker._place_products(run_id, video, WorkflowDefinition.from_dict(definition))
     assert placed == ["a.CN.srt", "a.CN_dual_eye.ass", "a.other.20260902120000.txt"]
     # srt/ass 被改名为标准名且覆盖旧文件；txt 保留原名。

tests/test_finalization.py

@@ -0,0 +1,169 @@
+"""产物收尾回归：真实 SQLite、字幕文件和 API 下载验证暂停及故障恢复。"""
+
+import shutil
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from wov_app.db import Database
+from wov_app.main import app
+from wov_app.scheduler import WorkflowScheduler
+from wov_sdk.models import WorkflowDefinition
+
+
+@pytest.fixture
+def subtitle(tmp_path):
+    """复用真实字幕资产，不在测试中生成模型输出或占位媒体。"""
+    source = Path(__file__).resolve().parent.parent / "testdata/ocr_srt_run_ac7f480a3ccb.srt"
+    if not source.is_file():
+        pytest.skip("缺少真实字幕资产")
+    target = tmp_path / "original.srt"
+    shutil.copy2(source, target)
+    return target
+
+
+def _seed(db, source, finals):
+    """登记已完成节点的真实产物，模拟节点执行结束后的断点。"""
+    definition = {"name": "finalization", "version": 1,
+                  "nodes": [{"id": "step", "node_type": "echo"}],
+                  "final_outputs": finals}
+    db.upsert_workflow({"id": "finalization", "name": "收尾回归"})
+    db.create_workflow_version("finalization", 1, definition)
+    db.create_run({"id": "run_finalization", "workflow_id": "finalization",
+                   "workflow_version": 1, "status": "QUEUED", "input_uri": str(source),
+                   "created_at": "2026-01-01T00:00:00+00:00", "updated_at": "2026-01-01T00:00:00+00:00"})
+    db.create_artifact({"run_id": "run_finalization", "node_id": "step",
+                        "name": "step.file_uri", "uri": str(source)})
+    return WorkflowDefinition.from_dict(definition)
+
+
+def test_pause_during_final_registration_keeps_downloads(subtitle, tmp_path, monkeypatch):
+    """在第一个最终别名登记后暂停，再继续，两别名均可下载且 URI 不漂移。"""
+    db = Database(tmp_path / "test.db")
+    _seed(db, subtitle, {"result": "step.file_uri", "copy": "step.file_uri"})
+    original = subtitle.read_bytes()
+    create = db.create_artifact
+
+    def pause_once(artifact):
+        create(artifact)
+        if artifact["name"] == "result":
+            db.pause_run("run_finalization", "2026-01-01T00:00:00+00:00")
+
+    scheduler = WorkflowScheduler(db, tmp_path / "storage")
+    with monkeypatch.context() as patch:
+        patch.setattr(db, "create_artifact", pause_once)
+        scheduler.execute_run("run_finalization")
+    first_uri = db.get_artifact("run_finalization", "result")["uri"]
+    assert db.get_run("run_finalization")["status"] == "PAUSED"
+    db.resume_run("run_finalization", "2026-01-01T00:00:00+00:00")
+    scheduler.execute_run("run_finalization")
+    assert db.get_run("run_finalization")["status"] == "COMPLETED"
+    assert db.get_artifact("run_finalization", "result")["uri"] == first_uri
+    with TestClient(app) as client:
+        monkeypatch.setattr(app.state, "db", db)
+        for alias in ("result", "copy", "step.file_uri"):
+            response = client.get(f"/api/runs/run_finalization/artifacts/{alias}")
+            assert response.status_code == 200
+            assert response.content == original
+
+
+def test_legacy_renamed_final_survives_resume(subtitle, tmp_path):
+    """兼容旧版已改名但最终记录有效的断点，恢复不能覆盖成不存在的旧路径。"""
+    db = Database(tmp_path / "test.db")
+    _seed(db, subtitle, {"result": "step.file_uri"})
+    renamed = subtitle.with_name("legacy.srt")
+    subtitle.rename(renamed)
+    db.create_artifact({"run_id": "run_finalization", "node_id": "step",
+                        "name": "result", "uri": str(renamed)})
+    WorkflowScheduler(db, tmp_path / "storage").execute_run("run_finalization")
+    assert db.get_run("run_finalization")["status"] == "COMPLETED"
+    assert Path(db.get_artifact("run_finalization", "result")["uri"]).read_bytes() == renamed.read_bytes()
+
+
+def test_failed_final_copy_preserves_source_and_can_retry(subtitle, tmp_path, monkeypatch):
+    """复制中途写入失败不登记残缺成品，源字幕可用，恢复后可成功收尾。"""
+    db = Database(tmp_path / "test.db")
+    _seed(db, subtitle, {"result": "step.file_uri"})
+    expected = subtitle.read_bytes()
+    scheduler = WorkflowScheduler(db, tmp_path / "storage")
+
+    def fail_copy(source, destination, **kwargs):
+        # 文件 I/O 边界模拟磁盘写满：真实写出截断内容后报错。
+        Path(destination).write_bytes(Path(source).read_bytes()[:64])
+        raise OSError("disk full")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(shutil, "copy2", fail_copy)
+        scheduler.execute_run("run_finalization")
+    assert db.get_run("run_finalization")["status"] == "FAILED"
+    assert db.get_artifact("run_finalization", "result") is None
+    assert subtitle.read_bytes() == expected
+    assert not list((tmp_path / "storage").rglob("*.tmp"))
+    db.update_run("run_finalization", status="QUEUED", updated_at="2026-01-01T00:00:00+00:00")
+    scheduler.execute_run("run_finalization")
+    assert db.get_run("run_finalization")["status"] == "COMPLETED"
+    assert Path(db.get_artifact("run_finalization", "result")["uri"]).read_bytes() == expected
+
+
+def test_batch_copy_failure_preserves_old_product_and_workspace(subtitle, tmp_path, monkeypatch):
+    """批量成品写到一半失败：旧字幕不变、run/工作空间保留，重试可收尾。"""
+    from wov_app.batch import BatchWorker, create_job
+
+    video_asset = Path(__file__).resolve().parent.parent / "testdata/subtitle_10s.mp4"
+    if not video_asset.is_file():
+        pytest.skip("缺少真实视频资产")
+    library = tmp_path / "library"
+    library.mkdir()
+    video = library / "movie.mp4"
+    shutil.copy2(video_asset, video)
+    db = Database(tmp_path / "test.db")
+    _seed(db, subtitle, {"result": "step.file_uri"})
+    db.upsert_workflow({"id": "finalization", "name": "收尾回归", "published": 1, "latest_version": 1})
+    job_id = create_job(db, str(library), "finalization")
+    item = db.list_batch_videos(job_id)[0]
+    work_dir = Path(item["work_dir"])
+    work_dir.mkdir(parents=True)
+    source = work_dir / "result.srt"
+    shutil.copy2(subtitle, source)
+    now = "2026-01-01T00:00:00+00:00"
+    db.update_batch_video(item["id"], run_id="run_finalization", updated_at=now)
+    db.update_run("run_finalization", status="COMPLETED", updated_at=now)
+    db.create_artifact({"run_id": "run_finalization", "node_id": "step",
+                        "name": "result", "uri": str(source)})
+    target = library / "movie.CN.srt"
+    shutil.copy2(subtitle, target)
+    original = target.read_bytes()
+    worker = BatchWorker(db)
+
+    def fail_copy(source, destination, **kwargs):
+        Path(destination).write_bytes(Path(source).read_bytes()[:64])
+        raise OSError("disk full")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(shutil, "copy2", fail_copy)
+        worker._process_job(db.get_batch_job(job_id))
+    assert db.get_batch_video(item["id"])["status"] == "FAILED"
+    assert target.read_bytes() == original
+    assert db.get_run("run_finalization") is not None
+    assert source.read_bytes() == original
+    assert not list(library.glob("*.tmp"))
+    worker._process_job(db.get_batch_job(job_id))
+    assert db.get_batch_video(item["id"])["status"] == "COMPLETED"
+    assert db.get_run("run_finalization") is None
+    assert not work_dir.exists()
+    assert target.read_bytes() == original
+    assert video.read_bytes() == video_asset.read_bytes()
+
+
+@pytest.mark.parametrize("missing_ref", [False, True])
+def test_missing_final_fails_run(subtitle, tmp_path, missing_ref):
+    """必需最终输出无引用或文件丢失都应 FAILED，不能成功登记失效链接。"""
+    db = Database(tmp_path / "test.db")
+    _seed(db, subtitle, {"result": "step.missing" if missing_ref else "step.file_uri"})
+    if not missing_ref:
+        subtitle.unlink()
+    WorkflowScheduler(db, tmp_path / "storage").execute_run("run_finalization")
+    run = db.get_run("run_finalization")
+    assert run["status"] == "FAILED"
+    assert "result" in run["error"]

tests/test_scheduler.py

@@ -429,9 +429,9 @@ def test_final_artifact_renamed_with_language_tag(tmp_path) -> None:
     assert filename.startswith("movie01.zh-CN.")
     assert filename.endswith(".txt")
     assert Path(final["uri"]).is_file()
-    # 原始未重命名文件不应残留。
+    # 节点原始 URI 必须保持有效；成品副本与原始文本一致，保证断点可恢复。
     step_artifacts = [item for item in artifacts if item["name"] == "step.file_uri"]
-    assert not Path(step_artifacts[0]["uri"]).exists()
+    assert Path(step_artifacts[0]["uri"]).read_bytes() == Path(final["uri"]).read_bytes()
 
 
 def test_final_artifact_renamed_fallback_base_and_tag(tmp_path) -> None: